package gateway import ( "context" "errors" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" ) // 本文件拥有编排器的聊天历史持久化钩子(design/16 // §3.3):用户消息流在提交时写入,助手回复在 // 回合结算时写入。未接线 history 存储时二者都是静默空操作, // 因此编排器无存储也能原样运行(测试 / 未接线的网关)。 // modeChat 是回合未携带行为模式时所持久化的模式(design/01 §3, // design/16 §5.1)。它也是 IM 路径唯一会写入的模式,因为 IM // 只携带聊天文本。 const modeChat = "chat" // modeOrDefault 返回 mode,调用方未提供时返回 modeChat,使 // 持久化消息绝不携带空的行为模式(design/16 §5.1)。 func modeOrDefault(mode string) string { if mode == "" { return modeChat } return mode } // messageModeContextKey 是携带回合行为模式的 context 键 // (design/01 §3、design/16 §5.1),从 HTTP/WS 边界引入编排器的 // 持久化,使每条持久化消息被它发送时的模式标注。 type messageModeContextKey struct{} // withMode 返回携带 mode 的 context,使编排器能标注 // 持久化的用户消息与助手回复。空 mode 是空操作 // (持久化层回退到 modeChat)。 func withMode(ctx context.Context, mode string) context.Context { return context.WithValue(ctx, messageModeContextKey{}, mode) } // modeFrom 返回 ctx 携带的行为模式,未设置时返回 ""。 func modeFrom(ctx context.Context) string { v, _ := ctx.Value(messageModeContextKey{}).(string) return v } // persistUserMessages 把一次已提交回合中每条 RoleUser 消息写入 // 历史存储,使原始用户消息流在回合结算前就持久,随后 // 通知传输层,使同一账号的每个浏览器都能追加它 // 而无需手动刷新(design/16 §3.3、§4.2)。deviceID 用发送客户端标注 // 消息(design/16 §4.5),mode 是该回合发送时的行为模式 // (design/16 §5.1),为空时回退到 modeChat,而 personaID 是 // 该回合说话时的角色(design/22 §7-5)—— 即当时的角色,因此 // 之后切换角色的会话仍保有准确的「谁说了这个」流。未接线 // history 存储时它是空操作。被拒绝的回合(空、待处理已满) // 绝不触达它,因此不会持久化任何空内容。 // // ctx 上若带着客户端的幂等键(G4,clientMsgHintFrom),第一条用户消息会带上它 // 与编排器预分配的消息 id:写入因此可被 (session_id, client_msg_id) 的唯一索引 // 判重。省略幂等键时 ctx 上没有该值,写入与加入这一列之前逐字相同。 func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string, messages []model.Message, deviceID, mode, personaID string) { if o.history == nil { return } hint := clientMsgHintFrom(ctx) for _, m := range messages { if m.Role != model.RoleUser { continue } text, image := messageTextImage(m) if text == "" && image == "" { continue } msg := history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image, DeviceID: deviceID, Mode: modeOrDefault(mode), PersonaID: personaID} // 幂等键只标注**第一条**用户消息:一个回合至多只有一条用户消息,而把 // 同一个键盖到多条上会让它们互相冲突(唯一索引把它们判成同一条)。 if hint.ClientMsgID != "" { msg.ClientMsgID, msg.ID = hint.ClientMsgID, hint.MessageID hint = clientMsgHint{} } full, err := o.history.AppendFull(ctx, msg) if err != nil { o.log.Error("orchestrator persist user", "err", err, "session", sessionID, "device_id", deviceID) // 落库失败就撤销认领:否则客户端拿同一个 clientMsgId 重试时会被判成 // 「重复到达」而既不落库也不起回合 —— 这条消息从此再也进不来。 if msg.ClientMsgID != "" { o.releaseClientMsg(sessionID, msg.ClientMsgID, msg.ID) } continue } if msg.ClientMsgID != "" && full.ID != msg.ID { // 存储侧唯一索引裁决出既有行(例如本进程重启后另一次写入已落库): // 把权威 id 覆盖进认领表,别让后续重复请求拿到一个悬空的 messageId。 o.rememberClientMsg(sessionID, msg.ClientMsgID, full.ID) } o.notifyPersisted(full) } } // persistAssistantReply 把已结算回合的最终助手文本写入 // 历史存储,然后通知传输层(design/16 §3.3、§4.2)。deviceID 用 // 发送该回合的客户端标注回复(design/16 §4.5),mode 是 // 该回合的行为模式(design/16 §5.1),为空时回退到 modeChat, // 而 personaID 是产出该回复的角色(design/22 §7-5)。Mood // 是该回合中人格的 mood(design/01 §2 情绪状态机),取自 loop // 结果,因此已存/DTO/`message_persisted` 的 mood 是真实数据,而不是 // 永远为空的列;不跟踪 mood 的回合存一个空 mood。它 // 在未接线 history 存储,或该回合没有产出文本 // 时是空操作(被中断/出错的回合保持 res 为 nil)。该写入在一个全新的 // 后台 context 下运行,因此被取消的回合 context 绝不可能中断它。 func (o *Orchestrator) persistAssistantReply(sessionID string, res *loop.Result, deviceID, mode, personaID string) { if o.history == nil || res == nil || res.Text == "" { return } full, err := o.history.AppendFull(context.Background(), history.Message{SessionID: sessionID, Role: history.RoleAssistant, Content: res.Text, Mood: res.Mood, DeviceID: deviceID, Mode: modeOrDefault(mode), PersonaID: personaID}) if err != nil { o.log.Error("orchestrator persist assistant", "err", err, "session", sessionID, "device_id", deviceID) return } o.notifyPersisted(full) } // messageTextImage 返回消息内容部分拼接后的文本与首个图片 // URL,供持久化进 chat_messages 各列。 func messageTextImage(m model.Message) (text, image string) { for _, c := range m.Content { if c.Type == model.ContentTypeText { text += c.Text } else if c.Type == model.ContentTypeImage && image == "" { image = c.ImageURL } } return text, image } // --- G4 客户端幂等键(clientMsgId)------------------------------------------- // // 全链路此前没有幂等:客户端超时重发同一段文本就是两条用户消息、两个回合。 // 这一节给编排器加判重,与存储侧的 (session_id, client_msg_id) 部分唯一索引 // 配合:编排器决定「这一轮还要不要跑」,存储决定「这一行还要不要写」。 // // 两层职责刻意分开:内存认领表(sessionState.clientMsgs)让**同进程并发**只有一个 // 赢家(这是唯一索引覆盖不到的窗口 —— 两次重发都还没插入),唯一索引则让判重 // 跨重启依然成立(单二进制单进程,不存在跨进程并发)。 // maxClientMsgsPerSession 限制每会话保留在内存里的认领条数。淘汰最旧条目只会 // 让一次很晚到达的重复多问一次存储,绝不会让重复被放行 —— 权威在存储。 const maxClientMsgsPerSession = 64 // clientMsgHint 是一条幂等提示(G4)。ClientMsgID 是客户端自带的去重键 // (空 = 老客户端,行为逐字不变);MessageID 是编排器为这条消息**预分配**的 // id,使并发的重复请求在第一条真正落库之前就能拿到一个确定(且终将存在)的 // messageId 去对齐本地乐观气泡。 type clientMsgHint struct { ClientMsgID string MessageID string } // clientMsgHintKey 是携带 clientMsgHint 的 context 键。它从 HTTP/WS 边界 // (REST 请求体 / ws chat 帧的 clientMsgId)引入编排器,与 deviceID、mode // 走同一种机制。 type clientMsgHintKey struct{} // withClientMsgHint 返回携带该提示的 context。空 ClientMsgID 是空操作:省略 // clientMsgId 的请求在 ctx 上不留任何痕迹。 func withClientMsgHint(ctx context.Context, h clientMsgHint) context.Context { if h.ClientMsgID == "" { return ctx } return context.WithValue(ctx, clientMsgHintKey{}, h) } // clientMsgHintFrom 返回 ctx 携带的幂等提示,未设置时返回零值。 func clientMsgHintFrom(ctx context.Context) clientMsgHint { h, _ := ctx.Value(clientMsgHintKey{}).(clientMsgHint) return h } // clientMsgIDFrom 返回 ctx 携带的客户端幂等键,未设置时为 ""。 func clientMsgIDFrom(ctx context.Context) string { return clientMsgHintFrom(ctx).ClientMsgID } // clientMsgIDReader 是 HistoryStore 的一个**可选**更窄视图(与 persona_env.go // 的 sessionPersonaReader 同一做法):按 (会话, clientMsgId) 定点读一条既有 // 消息,供进程重启后的判重(内存认领表已随进程消失)。未实现该视图的存储替身 // 或后端退回「内存认领表」判定,不因此报错,也无需为此实现新方法。 type clientMsgIDReader interface { FindByClientMsgID(ctx context.Context, sessionID, clientMsgID string) (history.Message, error) } // admit 是一次提交能否进入会话状态机的判定结果(design/15 §4 的队列容量 + // G4 的幂等判重)。两种为真的情形都意味着「不落库、不起回合」,但它们对客户端 // 的含义相反:full 是「稍后重试」(队列满),dup 是「不必重试」(消息早已在库)。 type admit struct { // full 报告该会话的待处理队列已满。 full bool // dup 报告这次提交的 clientMsgId 已经到达过。 dup bool // mode 是判定时的会话忙碌通道,供 PendingFull 事件如实报告(在锁内取值, // 解锁后再广播)。 mode Mode // hint 是本次的幂等提示:dup 为真时指向既有/已认领的消息,否则携带本次 // 落库要用的预分配 id(省略 clientMsgId 时是零值)。 hint clientMsgHint } // admitLocked 在状态锁下判定一次提交能否进入状态机,并在可以进入时完成 G4 的 // 认领。它必须在 o.mu 下调用,且**不**在锁内广播任何状态事件 —— 调用方解锁后 // 再发(与既有 PendingFull 路径同一纪律)。 // // 顺序是刻意的:**队列满的拒绝排在认领之前**。被拒的回合没有落库,若在拒绝前 // 认领了 clientMsgId,客户端用同一个键重试时会被判成「重复到达」,这条消息就 // 再也进不来了 —— 那正是 G2 那类静默丢失的翻版。 func (o *Orchestrator) admitLocked(ctx context.Context, st *sessionState, sessionID string) admit { if st.mode != ModeIdle && len(st.pending) >= o.maxPending { return admit{full: true, mode: st.mode} } hint, dup := o.claimClientMsgLocked(ctx, st, sessionID, clientMsgIDFrom(ctx)) return admit{dup: dup, mode: st.mode, hint: hint} } // claimClientMsgLocked 在状态锁下判重并认领一个 clientMsgId(G4)。duplicate 为 // true 表示这条消息此前已经到达(hint.MessageID 指向既有/已认领的消息),调用方 // 必须原地返回:既不落库、也不启动回合。duplicate 为 false 时 hint.MessageID 是 // 本次落库要用的预分配 id(省略 clientMsgId 时 hint 与 duplicate 都是零值)。 // // 必须在 o.mu 下调用:认领与状态读取必须是同一个临界区,否则两个并发的重复请求 // 会双双认领成功、各起一个回合(这正是「先查再插」的经典竞态)。存储查询也是有界 // 的定点读(部分唯一索引至多命中一行),与 startTurnMessages 的转录读取同一量级, // 因此不违反「状态锁不持有慢操作」。 func (o *Orchestrator) claimClientMsgLocked(ctx context.Context, st *sessionState, sessionID, clientMsgID string) (clientMsgHint, bool) { if clientMsgID == "" { return clientMsgHint{}, false } if id, ok := st.clientMsgs[clientMsgID]; ok { return clientMsgHint{ClientMsgID: clientMsgID, MessageID: id}, true } if id, ok := o.findPersistedClientMsg(ctx, sessionID, clientMsgID); ok { o.rememberClientMsgLocked(st, clientMsgID, id) return clientMsgHint{ClientMsgID: clientMsgID, MessageID: id}, true } id, err := history.NewMessageID() if err != nil { // 生成失败就退回「不判重」:宁可多跑一轮重复,也不把这条消息挡在门外 // —— 挡在门外就是静默丢失,正是本次要消灭的东西。 o.log.Error("orchestrator client msg id", "err", err, "session", sessionID) return clientMsgHint{}, false } o.rememberClientMsgLocked(st, clientMsgID, id) return clientMsgHint{ClientMsgID: clientMsgID, MessageID: id}, false } // findPersistedClientMsg 报告该会话是否已有携带该 clientMsgId 的消息,是时返回 // 其消息 id。存储未接、未实现该可选视图、或查询失败时一律报「没有」并记日志: // 判重失败的方向必须是**放行**(多一条重复),绝不是拒绝(丢一条消息)。 func (o *Orchestrator) findPersistedClientMsg(ctx context.Context, sessionID, clientMsgID string) (string, bool) { reader, ok := o.history.(clientMsgIDReader) if !ok { return "", false } msg, err := reader.FindByClientMsgID(ctx, sessionID, clientMsgID) if err != nil { if !errors.Is(err, history.ErrMessageNotFound) { o.log.Error("orchestrator client msg lookup", "err", err, "session", sessionID) } return "", false } return msg.ID, msg.ID != "" } // rememberClientMsgLocked 认领一个 clientMsgId(必须在 o.mu 下调用),并按 // maxClientMsgsPerSession 有界淘汰最旧的条目。 func (o *Orchestrator) rememberClientMsgLocked(st *sessionState, clientMsgID, msgID string) { if st.clientMsgs == nil { st.clientMsgs = make(map[string]string, maxClientMsgsPerSession) } if _, ok := st.clientMsgs[clientMsgID]; !ok { st.clientMsgOrder = append(st.clientMsgOrder, clientMsgID) if len(st.clientMsgOrder) > maxClientMsgsPerSession { oldest := st.clientMsgOrder[0] st.clientMsgOrder = st.clientMsgOrder[1:] delete(st.clientMsgs, oldest) } } st.clientMsgs[clientMsgID] = msgID } // rememberClientMsg 是 rememberClientMsgLocked 的加锁包装,供锁外的持久化路径 // 在拿到存储裁决出的权威 id 后修正认领(见 persistUserMessages)。会话状态不 // 存在时它是空操作(那意味着本次提交没有走过状态机,没有认领需要修正)。 func (o *Orchestrator) rememberClientMsg(sessionID, clientMsgID, msgID string) { if clientMsgID == "" || msgID == "" { return } o.mu.Lock() defer o.mu.Unlock() if st := o.states[sessionID]; st != nil { o.rememberClientMsgLocked(st, clientMsgID, msgID) } } // releaseClientMsg 撤销一次认领(落库失败时调用),使同一个 clientMsgId 的 // 下一次重试仍然能落库,而不是被判重挡在门外、消息从此消失。它只撤销仍然属于 // msgID 的那一条认领,不碰后来的认领者。 func (o *Orchestrator) releaseClientMsg(sessionID, clientMsgID, msgID string) { o.mu.Lock() defer o.mu.Unlock() st := o.states[sessionID] if st == nil || st.clientMsgs[clientMsgID] != msgID { return } delete(st.clientMsgs, clientMsgID) for i, id := range st.clientMsgOrder { if id == clientMsgID { st.clientMsgOrder = append(st.clientMsgOrder[:i], st.clientMsgOrder[i+1:]...) return } } }