From d64b26d49919395ec3fb0f9c77ce06da4ab8c2a6 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Mon, 5 Oct 2026 09:46:10 +0800 Subject: [PATCH] =?UTF-8?q?feat(chat):=20chat=5Fmessages.turn=5Fid=20?= =?UTF-8?q?=E5=88=97=20+=20DTO=EF=BC=88=E5=86=99=E5=85=A5=E6=8E=A5?= =?UTF-8?q?=E7=BA=BF=E7=95=99=E5=BE=85=E4=B8=8B=E4=B8=80=E6=AD=A5=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 迁移 0018：ALTER TABLE chat_messages ADD COLUMN turn_id TEXT NOT NULL DEFAULT '' （与 persona_id / client_msg_id / document 同族：不用 NULL，空串 = 不属于任何已知回合 ⇒ 旧行零迁移、键省略）。 存储层全链路带 turn_id（列清单唯一一份 ⇒ List/ListAfter/Recent/Search/FindByClientMsgID 自动同源）； 新增 history/turn_id.go：ensureTurnIDColumn（既有库幂等前滚）+ SetMessagesTurnID （第二段补写：单事务参数化、先到先得 WHERE turn_id = ''、幂等、未知 id 忽略）。 DTO：historyMessage.TurnID（json turnId,omitempty）⇒ REST 列表 / admin 渠道日志 / WS 握手回放 / message_persisted / channel_persisted / message_edited 一次全覆盖。 写入面：persistUserMessages 读 ctx 回合标识；persistAssistantReply 增可选尾参（既有调用点零改动）； withTurnID / turnIDFrom / assignTurnID / turnIDOfLeg。 ⚠️ 编排器把 turnId 交给落库的 4 处（新起一路 / 续跑路 / 被 steer 吃进 / 助手回复）尚未接线 （那些文件当时被另一路占用）；当前生产行为 turn_id 仍为 ''。清单已在切片报告里精确到文件:行。 13 新用例 + 5 条变异（INSERT 不带列 / DTO 不回填 / 去掉 omitempty / 去掉先到先得 / 不回滚前滚）。 为守 session_persist.go ≤500 行，turnId 一节拆到新文件 session_turn_id.go。 --- internal/db/migrate_turn_id_test.go | 71 ++++ .../0018_chat_message_turn_id.down.sql | 9 + .../0018_chat_message_turn_id.up.sql | 27 ++ internal/runtime/history/document.go | 7 +- internal/runtime/history/store.go | 17 +- internal/runtime/history/turn_id.go | 80 +++++ internal/runtime/history/turn_id_test.go | 242 +++++++++++++ internal/runtime/history/types.go | 16 +- internal/services/gateway/history.go | 22 +- .../services/gateway/history_turnid_test.go | 337 ++++++++++++++++++ internal/services/gateway/session_persist.go | 19 +- internal/services/gateway/session_turn_id.go | 119 +++++++ 12 files changed, 950 insertions(+), 16 deletions(-) create mode 100644 internal/db/migrate_turn_id_test.go create mode 100644 internal/db/migrations/0018_chat_message_turn_id.down.sql create mode 100644 internal/db/migrations/0018_chat_message_turn_id.up.sql create mode 100644 internal/runtime/history/turn_id.go create mode 100644 internal/runtime/history/turn_id_test.go create mode 100644 internal/services/gateway/history_turnid_test.go create mode 100644 internal/services/gateway/session_turn_id.go diff --git a/internal/db/migrate_turn_id_test.go b/internal/db/migrate_turn_id_test.go new file mode 100644 index 0000000..a4f492d --- /dev/null +++ b/internal/db/migrate_turn_id_test.go @@ -0,0 +1,71 @@ +package db + +import ( + "context" + "testing" +) + +// turnIDMigrationID 是回合标识列的迁移 id（design/15 §2.5.9）。 +const turnIDMigrationID = "0018_chat_message_turn_id" + +// TestMigrateTurnIDColumnReversible 验证 0018（chat_messages.turn_id，回合标识列） +// 可上下翻滚：up 建列、重复 up 是空操作、down 删列、再 up 恢复。它用自己的内存库， +// 绝不碰 data/cyrene.db。 +func TestMigrateTurnIDColumnReversible(t *testing.T) { + t.Parallel() + ctx := context.Background() + conn := openMemory(t) + + if err := MigrateUp(ctx, conn, testLogger()); err != nil { + t.Fatalf("MigrateUp: %v", err) + } + if !columnExists(t, ctx, conn, "chat_messages", "turn_id") { + t.Fatal("chat_messages.turn_id should exist after MigrateUp") + } + + // 重复 up：已记录的迁移被跳过，列仍在且不报错（幂等）。 + if err := MigrateUp(ctx, conn, testLogger()); err != nil { + t.Fatalf("MigrateUp second call: %v", err) + } + if !columnExists(t, ctx, conn, "chat_messages", "turn_id") { + t.Fatal("chat_messages.turn_id should still exist after idempotent re-run") + } + + // 恰好回滚到 0018（含它之后新加的迁移），因此日后再加迁移也不会让本测试失效。 + ids := upMigrationIDs(t) + idx := -1 + for i, id := range ids { + if id == turnIDMigrationID { + idx = i + } + } + if idx < 0 { + t.Fatalf("%s not found in %v", turnIDMigrationID, ids) + } + if err := MigrateDown(ctx, conn, len(ids)-idx, testLogger()); err != nil { + t.Fatalf("MigrateDown(%d): %v", len(ids)-idx, err) + } + if columnExists(t, ctx, conn, "chat_messages", "turn_id") { + t.Fatal("chat_messages.turn_id should be dropped by the 0018 down") + } + + // 重新 up 恢复端到端可用：默认空串意味着「这一行不属于任何已知回合」。 + if err := MigrateUp(ctx, conn, testLogger()); err != nil { + t.Fatalf("re-MigrateUp: %v", err) + } + if !columnExists(t, ctx, conn, "chat_messages", "turn_id") { + t.Fatal("chat_messages.turn_id should exist again after re-MigrateUp") + } + if _, err := conn.ExecContext(ctx, + `INSERT INTO chat_messages (id, session_id, role, content, created_at) VALUES (?, ?, ?, ?, ?)`, + "m1", "web", "user", "hi", 1000); err != nil { + t.Fatalf("insert without turn id: %v", err) + } + var turnID string + if err := conn.QueryRowContext(ctx, `SELECT turn_id FROM chat_messages WHERE id = ?`, "m1").Scan(&turnID); err != nil { + t.Fatalf("select turn_id: %v", err) + } + if turnID != "" { + t.Fatalf("turn_id = %q, want the empty default", turnID) + } +} diff --git a/internal/db/migrations/0018_chat_message_turn_id.down.sql b/internal/db/migrations/0018_chat_message_turn_id.down.sql new file mode 100644 index 0000000..e97d21a --- /dev/null +++ b/internal/db/migrations/0018_chat_message_turn_id.down.sql @@ -0,0 +1,9 @@ +-- 0018_chat_message_turn_id down: 回滚回合标识列。 +-- +-- turn_id 只被 chat_messages 自己引用（无索引、无视图、无约束牵涉），因此直接删列 +-- 即可。回滚后的状态是「服务端不再记得哪条回复在答哪句提问」：客户端退回按顺序 +-- 渲染（本列引入之前的行为），行里的分组信息随列一起丢弃 —— 它只是**派生的展示 +-- 元数据**，真实回合身份仍由流帧（帧壳 turnId）携带，删列不会让任何在跑的回合 +-- 失去自己的身份。 + +ALTER TABLE chat_messages DROP COLUMN turn_id; diff --git a/internal/db/migrations/0018_chat_message_turn_id.up.sql b/internal/db/migrations/0018_chat_message_turn_id.up.sql new file mode 100644 index 0000000..e86ae57 --- /dev/null +++ b/internal/db/migrations/0018_chat_message_turn_id.up.sql @@ -0,0 +1,27 @@ +-- 0018_chat_message_turn_id: 给聊天消息挂上「它属于哪一路回合」（design/15 §2.5.9）。 +-- +-- 并行回合（design/15 §2.5）落地后，一个会话可以同时有两条在飞的**路**，同一个 +-- 会话的行因此不再能靠「顺序」唯一回答「这条回复在答哪一句提问」。turn_id 就是 +-- 编排器生成、该路全部流帧与 `turn_begin` 帧壳同值的那一份权威回合标识 +-- （internal/services/gateway 的 turnLeg.turnID，经 loop.WithTurnID 盖在帧上）。 +-- +-- 用途（客户端依赖）：① 「回复 ↔ 提问」精确配对 —— 同一 turn_id 的 user 行与 +-- assistant 行互为配对；② 重连 / 回放时从历史行恢复「回复侧分组小标」。 +-- +-- 它**可选**：'' 意味着「这一行不属于任何已知的回合」—— 加入本列之前的全部历史行、 +-- 以及未经编排器的写入路径（IM 镜像等）都是这种情形，读起来因此与加入本列之前 +-- 逐字相同（DTO 里该键缺席）。列口径与同族可选列（persona_id / client_msg_id / +-- document）完全一致：NOT NULL DEFAULT ''，**绝不用 NULL** —— 于是「旧行」与 +-- 「没写」是同一个值，读取侧只有一种空情形要处理。这一列因此**不需要索引**： +-- 它只被随行读出，没有任何按它过滤/排序的查询。 +-- +-- 写入是**两段式**（与 document 列同一手法）：消息在它所属的那一路已经存在时 +-- （新起的一路、以及助手回复）就把 turn_id 写进 INSERT；排队 / 被转向吃进的消息 +-- 在落库那一刻还没有自己的路，由它**实际被处理的那一路**在起跑/消费时补写同一 +-- 列（先到先得，见 internal/runtime/history/turn_id.go 的 SetMessagesTurnID）。 +-- +-- 0003 的 chat_messages 表在此已存在，故 ALTER ADD COLUMN 是非破坏性的正确一步； +-- 全新库由 internal/runtime/history.NewStore 幂等地前滚（含该列）。表名沿用复数 +-- 口径（援引 AGENTS §10 已登记的例外）。 + +ALTER TABLE chat_messages ADD COLUMN turn_id TEXT NOT NULL DEFAULT ''; diff --git a/internal/runtime/history/document.go b/internal/runtime/history/document.go index 0398107..921262d 100644 --- a/internal/runtime/history/document.go +++ b/internal/runtime/history/document.go @@ -10,7 +10,7 @@ import ( ) // 本文件归属聊天消息携带的**文档引用**（design/24 §2.4），以及 chat_messages -// 上「建表之后才引入的列」的前滚聚合（client_msg_id 与 document）。 +// 上「建表之后才引入的列」的前滚聚合（client_msg_id、document 与 turn_id）。 // // document 列只存「引用 + 本回合注入统计」——存储名、嗅探所得 MIME、字节数、注入 // 字符数、注入是否被截断、抽取状态；**绝不存抽取正文**。G3（base64 原样落库 → @@ -202,7 +202,10 @@ func ensureMessageColumns(ctx context.Context, db *sql.DB) error { if err := ensureClientMsgIDColumn(ctx, db); err != nil { return err } - return ensureDocumentColumn(ctx, db) + if err := ensureDocumentColumn(ctx, db); err != nil { + return err + } + return ensureTurnIDColumn(ctx, db) } // ensureDocumentColumn 在 chat_messages 早于本列（0015 迁移）建表时把它前滚： diff --git a/internal/runtime/history/store.go b/internal/runtime/history/store.go index f8563dc..62fb32a 100644 --- a/internal/runtime/history/store.go +++ b/internal/runtime/history/store.go @@ -167,9 +167,9 @@ func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger, opts ...StoreOp } log = logger.WithSource(discardLogger(log), "history") // 顺序不能反：createMessagesTable 对全新库已含 client_msg_id，随后 - // ensureMessageColumns 把早于这些列（client_msg_id 与 document）建出的既有表 - // 前滚（幂等），最后才建索引 —— 部分唯一索引引用 client_msg_id，列还不存在时 - // 建它必然失败。 + // ensureMessageColumns 把早于这些列（client_msg_id / document / turn_id）建出的 + // 既有表前滚（幂等），最后才建索引 —— 部分唯一索引引用 client_msg_id，列还不 + // 存在时建它必然失败。 for _, ddl := range []string{ createSessionsTable, createMessagesTable, @@ -244,8 +244,8 @@ func (s *Store) insert(ctx context.Context, msg Message) (Message, error) { s.ensureTitle(ctx, msg.SessionID, msg.Content) } query := `INSERT INTO chat_messages - (id, session_id, role, content, mode, mood, image, document, thinking, platform, device_id, sender_name, adapter_id, persona_id, client_msg_id, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + (id, session_id, role, content, mode, mood, image, document, thinking, platform, device_id, sender_name, adapter_id, persona_id, client_msg_id, turn_id, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` if msg.ClientMsgID != "" { query += ` ON CONFLICT DO NOTHING` } @@ -257,7 +257,7 @@ func (s *Store) insert(ctx context.Context, msg Message) (Message, error) { res, err := tx.ExecContext(ctx, query, msg.ID, msg.SessionID, msg.Role, msg.Content, msg.Mode, msg.Mood, msg.Image, msg.Document, msg.Thinking, msg.Platform, msg.DeviceID, msg.SenderName, msg.AdapterID, - msg.PersonaID, msg.ClientMsgID, msg.CreatedAt) + msg.PersonaID, msg.ClientMsgID, msg.TurnID, msg.CreatedAt) if err != nil { return Message{}, fmt.Errorf("history: append %s to %s: %w", msg.Role, msg.SessionID, err) } @@ -347,13 +347,13 @@ func (s *Store) ensureMainSession(ctx context.Context) error { // **唯一**一份列清单（位置顺序必须与 scanMessage 的 Scan 完全一致）。 // 它刻意只写一次：此前七个查询各抄一份，加一列就得改七处，漏一处就只会在 // 运行时以「扫描到的列数与列不清单不符」炸掉；共用常量让漏改成为不可能。 -const messageColumns = `id, session_id, role, content, mode, mood, image, document, thinking, platform, device_id, sender_name, adapter_id, persona_id, client_msg_id, created_at` +const messageColumns = `id, session_id, role, content, mode, mood, image, document, thinking, platform, device_id, sender_name, adapter_id, persona_id, client_msg_id, turn_id, created_at` // scanMessage 读取一行 chat_messages。消息各列按位置扫描，顺序见 messageColumns。 func scanMessage(sc rowScanner) (Message, error) { var m Message var mode, mood, platform, deviceID, senderName, adapterID string - if err := sc.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &mode, &mood, &m.Image, &m.Document, &m.Thinking, &platform, &deviceID, &senderName, &adapterID, &m.PersonaID, &m.ClientMsgID, &m.CreatedAt); err != nil { + if err := sc.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &mode, &mood, &m.Image, &m.Document, &m.Thinking, &platform, &deviceID, &senderName, &adapterID, &m.PersonaID, &m.ClientMsgID, &m.TurnID, &m.CreatedAt); err != nil { return Message{}, err } m.Mode, m.Mood, m.Platform, m.DeviceID, m.SenderName, m.AdapterID = mode, mood, platform, deviceID, senderName, adapterID @@ -420,6 +420,7 @@ CREATE TABLE IF NOT EXISTS chat_messages ( adapter_id TEXT NOT NULL DEFAULT '', persona_id TEXT NOT NULL DEFAULT '', client_msg_id TEXT NOT NULL DEFAULT '', + turn_id TEXT NOT NULL DEFAULT '', created_at INTEGER NOT NULL )` diff --git a/internal/runtime/history/turn_id.go b/internal/runtime/history/turn_id.go new file mode 100644 index 0000000..a6c7031 --- /dev/null +++ b/internal/runtime/history/turn_id.go @@ -0,0 +1,80 @@ +package history + +import ( + "context" + "database/sql" + "fmt" + "strings" +) + +// 本文件归属聊天消息的**回合标识**列（design/15 §2.5.9）：一条消息属于哪一路回合。 +// +// 它与会话编排器分摊同一件事：编排器是那个**生成**权威 turnId 的地方（每个在飞的 +// 路一份，挂进 loop.WithTurnID 并盖在该路全部流帧上），存储只是把它落进行里，使 +// 「回复 ↔ 提问」的配对在**重连 / 回放**时依然成立 —— 流帧是易失的，历史行不是。 +// +// 写入是**两段式**，与 document 列完全同一手法： +// - 消息落库时它所属的路已经存在（编排器新起的一路、以及该路的助手回复）⇒ 随 +// INSERT 一起写入（Message.TurnID），因此 `message_persisted` 那一帧也带着它； +// - 排队（忙时入队）或**被转向（steer）吃进**的消息在落库那一刻**还没有自己的 +// 路** —— 用户消息在入队时就落库，而它由哪一路回答要等那一轮才定。这种行的 +// turn_id 只能在它**实际被处理的那一路**起跑 / 消费时补写，即本文件的 +// SetMessagesTurnID。 +// +// 空串（列的默认值）意味着「这一行不属于任何已知的回合」：加入该列之前的历史行、 +// 未经编排器的写入路径（IM 镜像）都是这种情形，读起来与以往逐字相同。 + +// SetMessagesTurnID 把 ids 所指的消息行的 turn_id 落定为 turnID —— 两段式写入的 +// 第二段（起跑 / 被 steer 消费时补写）。 +// +// 三条纪律： +// - **先到先得**：只写给仍然是空串的行（`AND turn_id = ”`）。一行只属于一路， +// 后来者绝不改写已经带标识的行 —— 否则「被哪一路回答」会随结算顺序漂移； +// - **幂等**：重复补写同一条（重启、重试、同一批消息被两处补写）是空操作； +// - **参数化**：id 列表按占位符展开，绝不把值拼进 SQL。 +// +// 空 ids 或空 turnID 是空操作：调用方在「这一路没有落库任何用户消息」或「这一路 +// 没有回合标识」（未接编排器的旧路径）时不必先判断。未知 id 被忽略而不是报错 —— +// 第二段的常态是「刚落的行」，一行找不到只可能是它已被删除，那不是本次写入的失败。 +func (s *Store) SetMessagesTurnID(ctx context.Context, ids []string, turnID string) error { + unique := make([]string, 0, len(ids)) + seen := make(map[string]bool, len(ids)) + for _, id := range ids { + if id == "" || seen[id] { + continue + } + seen[id] = true + unique = append(unique, id) + } + if len(unique) == 0 || turnID == "" { + return nil + } + placeholders := strings.TrimSuffix(strings.Repeat("?, ", len(unique)), ", ") + args := make([]any, 0, len(unique)+1) + args = append(args, turnID) + for _, id := range unique { + args = append(args, id) + } + query := `UPDATE chat_messages SET turn_id = ? WHERE turn_id = '' AND id IN (` + placeholders + `)` + if _, err := s.db.ExecContext(ctx, query, args...); err != nil { + return fmt.Errorf("history: set turn id for %d messages: %w", len(unique), err) + } + return nil +} + +// ensureTurnIDColumn 在 chat_messages 早于本列（0018 迁移）建表时把它前滚：缺列则 +// 补加，已有则忽略重复列错误。它是幂等的，因此 NewStore 可在全新 schema 或既有 +// schema 上调用（与 ensureClientMsgIDColumn / ensureDocumentColumn 同一手法）。 +func ensureTurnIDColumn(ctx context.Context, db *sql.DB) error { + _, err := db.ExecContext(ctx, `ALTER TABLE chat_messages + ADD COLUMN turn_id TEXT NOT NULL DEFAULT ''`) + if err == nil { + return nil + } + // 列已存在时（由 createMessagesTable 建出，或上一次运行补过）SQLite 报 + // "duplicate column name: turn_id"。 + if strings.Contains(err.Error(), "duplicate column name") { + return nil + } + return fmt.Errorf("history: roll forward turn_id column: %w", err) +} diff --git a/internal/runtime/history/turn_id_test.go b/internal/runtime/history/turn_id_test.go new file mode 100644 index 0000000..0928943 --- /dev/null +++ b/internal/runtime/history/turn_id_test.go @@ -0,0 +1,242 @@ +package history + +import ( + "context" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" +) + +// 本文件锁定聊天消息的 **turn_id** 列（design/15 §2.5.9）：它回答「这一行属于哪 +// 一路回合」，使客户端在重连 / 回放时仍能精确配对「回复 ↔ 提问」。 +// +// 三条语义被逐条钉住： +// - 落库时已经知道所属路的消息（新起的一路 + 它的助手回复）随 INSERT 写入， +// 同一路两条**同值**、两路**不同值**； +// - 落库时还不知道所属路的消息（忙时入队 / 被 steer 吃进）留空串，由它实际被 +// 处理的那一路**补写**（SetMessagesTurnID），且**先到先得**（已带标识的行不被 +// 改写 —— 一行只属于一路）； +// - 省略标识与加入本列之前逐字相同：空串默认值、读路径照常工作。 + +func TestAppendFullRoundTripsTurnID(t *testing.T) { + t.Parallel() + ctx := context.Background() + st := newStore(t) + + user, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "先问你一句", TurnID: "turn-leg1", CreatedAt: 1000}) + if err != nil { + t.Fatalf("append user: %v", err) + } + reply, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleAssistant, Content: "答你一句", TurnID: "turn-leg1", CreatedAt: 2000}) + if err != nil { + t.Fatalf("append assistant: %v", err) + } + // 第二路（并行）带自己的标识与自己的两条消息。 + if _, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "另一路的问题", TurnID: "turn-leg2", CreatedAt: 3000}); err != nil { + t.Fatalf("append parallel user: %v", err) + } + + if user.TurnID != "turn-leg1" || reply.TurnID != "turn-leg1" { + t.Fatalf("写入返回值 = %q / %q, want turn-leg1（同一路两条同值）", user.TurnID, reply.TurnID) + } + + // 读路径（共用 messageColumns/scanMessage）必须把标识带出来。 + msgs, err := st.List(ctx, MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list: %v", err) + } + got := make(map[string]string, len(msgs)) + for _, m := range msgs { + got[m.Content] = m.TurnID + } + if got["先问你一句"] != "turn-leg1" || got["答你一句"] != "turn-leg1" { + t.Fatalf("同一路两条 = %q / %q, want 都是 turn-leg1", got["先问你一句"], got["答你一句"]) + } + if got["另一路的问题"] != "turn-leg2" { + t.Fatalf("另一路 = %q, want turn-leg2（两路绝不同值）", got["另一路的问题"]) + } + + // 其余读面同源（它们共用同一份列清单，这里抽验两条最容易漏改的）。 + recent, err := st.Recent(ctx, MainSessionID, 1) + if err != nil { + t.Fatalf("recent: %v", err) + } + if len(recent) != 1 || recent[0].TurnID != "turn-leg2" { + t.Fatalf("recent = %+v, want 最新那条带 turn-leg2", recent) + } + hits, err := st.Search(ctx, MainSessionID, "先问你一句", 0, 10) + if err != nil { + t.Fatalf("search: %v", err) + } + if len(hits) != 1 || hits[0].TurnID != "turn-leg1" { + t.Fatalf("search = %+v, want 命中那条带 turn-leg1", hits) + } + found, err := st.FindByClientMsgID(ctx, MainSessionID, "") + if err == nil { + t.Fatalf("FindByClientMsgID(空键) = %+v, want ErrMessageNotFound", found) + } +} + +func TestAppendWithoutTurnIDLeavesItEmpty(t *testing.T) { + t.Parallel() + ctx := context.Background() + st := newStore(t) + + // 省略标识（老路径 / IM 镜像 / 排队入队）与加入本列之前逐字相同：空串。 + if _, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "没有回合标识", CreatedAt: 1000}); err != nil { + t.Fatalf("append: %v", err) + } + msgs, err := st.List(ctx, MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(msgs) != 1 || msgs[0].TurnID != "" { + t.Fatalf("msgs = %+v, want 空 turn_id（省略 ⇒ 不属于任何已知回合）", msgs) + } +} + +func TestSetMessagesTurnIDAssignsUnassignedOnly(t *testing.T) { + t.Parallel() + ctx := context.Background() + st := newStore(t) + + // 两行在落库时都还没有所属的路（忙时入队 / 被 steer 吃进的那种行）。 + queued, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "排队的一句", CreatedAt: 1000}) + if err != nil { + t.Fatalf("append queued: %v", err) + } + steered, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "被转向吃进的一句", CreatedAt: 1500}) + if err != nil { + t.Fatalf("append steered: %v", err) + } + + // 第二段：由**实际处理它们的那一路**补写。 + if err := st.SetMessagesTurnID(ctx, []string{queued.ID, steered.ID}, "turn-actual"); err != nil { + t.Fatalf("set turn id: %v", err) + } + msgs, err := st.List(ctx, MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list: %v", err) + } + for _, m := range msgs { + if m.TurnID != "turn-actual" { + t.Fatalf("消息 %q 的 turn_id = %q, want turn-actual（按实际处理它的那一路写）", m.Content, m.TurnID) + } + } + + // 先到先得：已经带上标识的行绝不被后来者改写（一行只属于一路）。 + if err := st.SetMessagesTurnID(ctx, []string{queued.ID}, "turn-later"); err != nil { + t.Fatalf("set turn id again: %v", err) + } + msgs, err = st.List(ctx, MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list after re-set: %v", err) + } + for _, m := range msgs { + if m.TurnID != "turn-actual" { + t.Fatalf("消息 %q 的 turn_id 被改写成 %q, want 仍是 turn-actual", m.Content, m.TurnID) + } + } + + // 未知 id 被忽略而不是报错：第二段的常态是「刚落的行」，找不到只可能是它已被删除。 + if err := st.SetMessagesTurnID(ctx, []string{"不存在的 id"}, "turn-actual"); err != nil { + t.Fatalf("unknown id: %v", err) + } +} + +func TestSetMessagesTurnIDNoopInputs(t *testing.T) { + t.Parallel() + ctx := context.Background() + st := newStore(t) + row, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "保持空", CreatedAt: 1000}) + if err != nil { + t.Fatalf("append: %v", err) + } + + // 空 id 列表、空标识、只含空串的列表都是空操作（调用方不必先判断）。 + cases := []struct { + name string + ids []string + turnID string + }{ + {name: "no ids", ids: nil, turnID: "turn-x"}, + {name: "no turn id", ids: []string{row.ID}, turnID: ""}, + {name: "blank ids", ids: []string{"", ""}, turnID: "turn-x"}, + } + for _, tc := range cases { + if err := st.SetMessagesTurnID(ctx, tc.ids, tc.turnID); err != nil { + t.Fatalf("%s: %v", tc.name, err) + } + } + msgs, err := st.List(ctx, MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(msgs) != 1 || msgs[0].TurnID != "" { + t.Fatalf("msgs = %+v, want 空 turn_id（空输入不写任何东西）", msgs) + } + + // 重复 id 只算一次，且整批落到同一个标识上。 + if err := st.SetMessagesTurnID(ctx, []string{row.ID, row.ID, "", row.ID}, "turn-dup"); err != nil { + t.Fatalf("duplicate ids: %v", err) + } + msgs, err = st.List(ctx, MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list after dup: %v", err) + } + if msgs[0].TurnID != "turn-dup" { + t.Fatalf("turn_id = %q, want turn-dup", msgs[0].TurnID) + } +} + +// TestNewStoreRollsForwardTurnIDColumn 验证早于 0018 建出的 chat_messages 表会被 +// NewStore 幂等前滚出 turn_id 列（与 client_msg_id / document 同一手法）：全新库由 +// 建表 DDL 一次建全，既有库靠补列，两条路都必须能写能读带标识的消息。 +func TestNewStoreRollsForwardTurnIDColumn(t *testing.T) { + t.Parallel() + ctx := context.Background() + conn, err := db.Open(ctx, ":memory:", testLogger()) + if err != nil { + t.Fatalf("open memory db: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + + // 0018 之前的形状：有 document（0015）但没有 turn_id。 + const legacyTable = ` +CREATE TABLE chat_messages ( + id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + role TEXT NOT NULL, + content TEXT NOT NULL, + mode TEXT, + mood TEXT, + image TEXT, + document TEXT NOT NULL DEFAULT '', + thinking TEXT, + platform TEXT NOT NULL DEFAULT 'web', + device_id TEXT NOT NULL DEFAULT '', + sender_name TEXT NOT NULL DEFAULT '', + adapter_id TEXT NOT NULL DEFAULT '', + persona_id TEXT NOT NULL DEFAULT '', + client_msg_id TEXT NOT NULL DEFAULT '', + created_at INTEGER NOT NULL +)` + if _, err := conn.ExecContext(ctx, legacyTable); err != nil { + t.Fatalf("create legacy table: %v", err) + } + + st, err := NewStore(ctx, conn, testLogger()) + if err != nil { + t.Fatalf("new store: %v", err) + } + if _, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "前滚之后", TurnID: "turn-rolled", CreatedAt: 2000}); err != nil { + t.Fatalf("append after roll forward: %v", err) + } + msgs, err := st.List(ctx, MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(msgs) != 1 || msgs[0].TurnID != "turn-rolled" { + t.Fatalf("msgs = %+v, want turn-rolled 往返", msgs) + } +} diff --git a/internal/runtime/history/types.go b/internal/runtime/history/types.go index fc1d00d..a0445c4 100644 --- a/internal/runtime/history/types.go +++ b/internal/runtime/history/types.go @@ -62,7 +62,21 @@ type Message struct { // 「这条消息没有幂等键」（老客户端、IM 镜像、历史行），行为与加入该列之前 // 逐字相同。 ClientMsgID string - CreatedAt int64 + // TurnID 是这条消息所属的那一**路**回合的权威标识（design/15 §2.5.9）： + // 编排器生成，与该路全部流帧 / `turn_begin` 帧壳上的 turnId **同值**。因此 + // 同一路的 user 行与 assistant 行带同一个值 —— 客户端据此精确配对「回复 ↔ + // 提问」，并在重连 / 回放时从历史行恢复回复侧的分组小标（真并行下同一会话 + // 有两个交错的回答，只按顺序渲染会配错）。 + // + // 它可选：空串意味着「这一行不属于任何已知的回合」—— 加入该列之前的行、以及 + // 未经编排器的写入路径（IM 镜像等）都是这种情形。 + // + // 两个写入时点（与 Document 同一手法，见 turn_id.go）：① 消息落库时它所属的 + // 路已经存在（新起的一路、以及助手回复）⇒ 随 INSERT 写入；② 排队 / 被转向 + // 吃进的消息在落库那一刻还没有自己的路 ⇒ 由**它实际被处理的那一路**补写同一 + // 列（先到先得）。 + TurnID string + CreatedAt int64 } // SessionMeta 是一个持久化会话面向线上的元数据。 diff --git a/internal/services/gateway/history.go b/internal/services/gateway/history.go index bad9588..5041a4d 100644 --- a/internal/services/gateway/history.go +++ b/internal/services/gateway/history.go @@ -93,7 +93,21 @@ type historyMessage struct { // 这类脆弱匹配），也能在重连后从读接口收敛。它为空时省略，而空意味着 // 「这条消息没有幂等键」（老客户端、IM 镜像、历史行）。 ClientMsgID string `json:"clientMsgId,omitempty"` - CreatedAt int64 `json:"createdAt"` + // TurnID 是这条消息所属的那一**路**回合的权威标识（design/15 §2.5.9）： + // 编排器生成，与该路全部流帧 / `turn_begin` 帧壳上的 turnId **同值**。 + // 同一路的 user 行与 assistant 行因此带同一个值，客户端据此 + // - 精确配对「回复 ↔ 提问」； + // - 在**重连 / 回放**时从历史行恢复回复侧的分组小标（真并行下同一会话有两个 + // 交错的回答，只按顺序渲染会配错）。 + // + // 它为空时省略，而空意味着「这一行不属于任何已知的回合」：加入该列之前的历史 + // 行、以及未经编排器的写入路径（IM 镜像等）都是这种情形，读起来与以往逐字 + // 相同。排队 / 被转向吃进的消息在**落库那一刻**也还没有自己的路（那时它由哪 + // 一路回答尚未确定），因此它们的那一帧 `message_persisted` 不带 turnId —— + // 客户端从 `turn_begin`（帧壳 turnId + data.messageIds）得知归属，行本身随后 + // 由它实际被处理的那一路补写。 + TurnID string `json:"turnId,omitempty"` + CreatedAt int64 `json:"createdAt"` } // handleHistorySessions 先返回主会话，再返回每个子会话（最旧 @@ -432,7 +446,11 @@ func toHistoryMessage(m history.Message) historyMessage { // （协同 G4），读接口（GET /sessions/{id}/messages）与 // `message_persisted` 帧因此给出同一个答案。 ClientMsgID: m.ClientMsgID, - CreatedAt: m.CreatedAt, + // TurnID 同理一并回填：读接口、WS 握手回放、`message_persisted` 与 + // `message_edited` 四条面共用本映射，因此它们对「这条消息属于哪一路」 + // 永远给出同一个答案（design/15 §2.5.9）。 + TurnID: m.TurnID, + CreatedAt: m.CreatedAt, } } diff --git a/internal/services/gateway/history_turnid_test.go b/internal/services/gateway/history_turnid_test.go new file mode 100644 index 0000000..eaaa899 --- /dev/null +++ b/internal/services/gateway/history_turnid_test.go @@ -0,0 +1,337 @@ +package gateway + +import ( + "context" + "encoding/json" + "net/http" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "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" +) + +// 本文件锁定 chat_messages.turn_id 的四条出口与两条写入路径（design/15 §2.5.9）： +// +// - 出口：历史 DTO（REST `GET /sessions/{id}/messages`）、WS 握手 `history` 回放、 +// 以及经 toHistoryMessage 构造的 `message_persisted` / `message_edited` 帧； +// - 写入：落库时已知所属路（新起的一路 + 它的助手回复，随 INSERT）、以及落库时 +// 还不知道所属路（忙时入队 / 被转向吃进，由实际处理它的那一路用 assignTurnID +// 补写）。 +// +// 语义口径全部是**可选**的：空 ⇒ 该键从 JSON 里消失，与同族可选字段 +// （personaId / clientMsgId / documents）逐字同形。 + +// turnsOf 把一页消息按内容映射到它们的 turn_id，便于断言配对。 +func turnsOf(msgs []history.Message) map[string]string { + out := make(map[string]string, len(msgs)) + for _, m := range msgs { + out[m.Content] = m.TurnID + } + return out +} + +// TestHistoryMessageTurnIDCarriedAndOmitted 钉住 DTO 形状：有标识就带出来，空则 +// 整个键消失（客户端据「键在不在」区分「属于某一路」与「不属于任何已知回合」）。 +func TestHistoryMessageTurnIDCarriedAndOmitted(t *testing.T) { + t.Parallel() + + withTurn := toHistoryMessage(history.Message{ID: "m1", SessionID: "web", Role: history.RoleUser, Content: "问", TurnID: "turn-a"}) + b, err := json.Marshal(withTurn) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var decoded map[string]any + if err := json.Unmarshal(b, &decoded); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if decoded["turnId"] != "turn-a" { + t.Fatalf("turnId = %v, want turn-a (%s)", decoded["turnId"], b) + } + + // 空标识（旧行、IM 镜像、落库时还不知道所属路的行）⇒ 键缺席。 + legacy := toHistoryMessage(history.Message{ID: "m2", SessionID: "web", Role: history.RoleUser, Content: "问"}) + b, err = json.Marshal(legacy) + if err != nil { + t.Fatalf("marshal legacy: %v", err) + } + if strings.Contains(string(b), `"turnId"`) { + t.Fatalf("空 turn_id 应当省略该键，得到 %s", b) + } +} + +// TestHistoryMessagesEndpointCarriesTurnID 端到端演练读接口：同一路的两条消息 +// 带同一个 turnId，旧行（空）连键都不出现 —— 客户端因此能在重连后按 turnId 把 +// 「回复 ↔ 提问」配对起来，而不是靠顺序猜。 +func TestHistoryMessagesEndpointCarriesTurnID(t *testing.T) { + t.Parallel() + ctx := context.Background() + _, ts, jwt, st := newHistoryEnv(t) + for _, m := range []history.Message{ + {SessionID: history.MainSessionID, Role: history.RoleUser, Content: "这条路的问题", TurnID: "turn-leg1", CreatedAt: 1000}, + {SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "这条路的回答", TurnID: "turn-leg1", CreatedAt: 1500}, + {SessionID: history.MainSessionID, Role: history.RoleUser, Content: "另一路的问题", TurnID: "turn-leg2", CreatedAt: 2000}, + {SessionID: history.MainSessionID, Role: history.RoleUser, Content: "本列之前的老行", CreatedAt: 2500}, + } { + if err := st.Append(ctx, m); err != nil { + t.Fatalf("seed %q: %v", m.Content, err) + } + } + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + // 先按原始 JSON 断言「键存在与否」，再按 DTO 断言取值。 + resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("messages status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var raw []map[string]any + decodeDataSlice(t, body, &raw) + seen := make(map[string]string, len(raw)) + for _, m := range raw { + content, _ := m["content"].(string) + id, ok := m["turnId"] + if !ok { + seen[content] = "" + continue + } + seen[content], _ = id.(string) + } + for content, want := range map[string]string{ + "这条路的问题": "turn-leg1", + "这条路的回答": "turn-leg1", + "另一路的问题": "turn-leg2", + "本列之前的老行": "", + } { + if got := seen[content]; got != want { + t.Fatalf("%q 的 turnId = %q, want %q（整页=%s）", content, got, want, body) + } + } + + var page []historyMessage + decodeDataSlice(t, body, &page) + turns := turnsOfWire(page) + if turns["这条路的问题"] != "turn-leg1" || turns["这条路的回答"] != "turn-leg1" { + t.Fatalf("同一路两条 = %q / %q，want 都是 turn-leg1", turns["这条路的问题"], turns["这条路的回答"]) + } +} + +// turnsOfWire 是 historyMessage 版的内容 → turnId 映射。 +func turnsOfWire(msgs []historyMessage) map[string]string { + out := make(map[string]string, len(msgs)) + for _, m := range msgs { + out[m.Content] = m.TurnID + } + return out +} + +// TestWSHandshakeReplayCarriesTurnID 钉住 WS 握手 `history` 回放这一面：它与读 +// 接口共用同一份映射，因此客户端刚连上就能按 turnId 恢复回复侧的分组小标。 +func TestWSHandshakeReplayCarriesTurnID(t *testing.T) { + t.Parallel() + ctx := context.Background() + _, ts, jwt, st := newHistoryEnv(t) + if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "回放里的提问", TurnID: "turn-replay", CreatedAt: 1000}); err != nil { + t.Fatalf("seed user: %v", err) + } + if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "回放里的回答", TurnID: "turn-replay", CreatedAt: 1500}); err != nil { + t.Fatalf("seed reply: %v", err) + } + + conn := dialWS(t, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner)) + defer conn.Close() + _, raw, err := conn.ReadMessage() + if err != nil { + t.Fatalf("read handshake frame: %v", err) + } + var frame struct { + Type string `json:"type"` + Data struct { + Messages []map[string]any `json:"messages"` + } `json:"data"` + } + if err := json.Unmarshal(raw, &frame); err != nil { + t.Fatalf("decode frame %s: %v", raw, err) + } + if frame.Type != "history" { + t.Fatalf("first frame = %q, want history (%s)", frame.Type, raw) + } + if len(frame.Data.Messages) != 2 { + t.Fatalf("replayed %d messages, want 2 (%s)", len(frame.Data.Messages), raw) + } + for _, m := range frame.Data.Messages { + if m["turnId"] != "turn-replay" { + t.Fatalf("回放消息 = %v, want turnId=turn-replay", m) + } + } +} + +// TestPersistTurnCarriesTurnIDOnBothRows 锁定第一段写入：新起的一路的用户消息与 +// 它的助手回复带**同一个** turn_id，且 `message_persisted` 那一帧（由同一条已存 +// 消息构造）也带着它 —— 客户端因此不需要等一次历史拉取就能配对。 +func TestPersistTurnCarriesTurnIDOnBothRows(t *testing.T) { + t.Parallel() + ctx := context.Background() + _, _, _, st := newHistoryEnv(t) + var broadcast []history.Message + o := &Orchestrator{history: st, log: testLogger(), onPersisted: func(m history.Message) { broadcast = append(broadcast, m) }} + + legCtx := withTurnID(ctx, "turn-solo") + o.persistUserMessages(legCtx, history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "单路的问题")}, "web:test", "chat", "cyrene") + o.persistAssistantReply(history.MainSessionID, &loop.Result{Text: "单路的回答"}, "web:test", "chat", "cyrene", "turn-solo") + + msgs, err := st.List(ctx, history.MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list: %v", err) + } + turns := turnsOf(msgs) + if turns["单路的问题"] != "turn-solo" || turns["单路的回答"] != "turn-solo" { + t.Fatalf("两条 = %q / %q，want 都是 turn-solo（同一路两条同值）", turns["单路的问题"], turns["单路的回答"]) + } + if len(broadcast) != 2 || broadcast[0].TurnID != "turn-solo" || broadcast[1].TurnID != "turn-solo" { + t.Fatalf("message_persisted 载荷 = %+v, want 两条都带 turn-solo", broadcast) + } +} + +// TestPersistParallelLegsCarryDistinctTurnIDs 锁定并行两路：每一路的两条消息 +// 同值，两路之间不同值 —— 这正是「回复 ↔ 提问」精确配对的前提（按顺序渲染会 +// 把两条交错的回答配错）。 +func TestPersistParallelLegsCarryDistinctTurnIDs(t *testing.T) { + t.Parallel() + ctx := context.Background() + _, _, _, st := newHistoryEnv(t) + o := &Orchestrator{history: st, log: testLogger()} + + o.persistUserMessages(withTurnID(ctx, "turn-1"), history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "第一路的问题")}, "web:test", "chat", "cyrene") + o.persistUserMessages(withTurnID(ctx, "turn-2"), history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "第二路的问题")}, "web:test", "chat", "cyrene") + o.persistAssistantReply(history.MainSessionID, &loop.Result{Text: "第一路的回答"}, "web:test", "chat", "cyrene", "turn-1") + o.persistAssistantReply(history.MainSessionID, &loop.Result{Text: "第二路的回答"}, "web:test", "chat", "cyrene", "turn-2") + + msgs, err := st.List(ctx, history.MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list: %v", err) + } + turns := turnsOf(msgs) + for content, want := range map[string]string{ + "第一路的问题": "turn-1", + "第一路的回答": "turn-1", + "第二路的问题": "turn-2", + "第二路的回答": "turn-2", + } { + if got := turns[content]; got != want { + t.Fatalf("%q 的 turn_id = %q, want %q（两路各自两条同值、两路不同值）", content, got, want) + } + } +} + +// TestAssignTurnIDStampsQueuedAndSteeredRows 锁定第二段写入与它的归属口径：忙时 +// 入队（以及被转向吃进）的消息在落库那一刻还不知道自己属于哪一路，由**实际处理 +// 它的那一路**补写；已经带上标识的行不被后来者改写（先到先得，一行只属于一路）。 +func TestAssignTurnIDStampsQueuedAndSteeredRows(t *testing.T) { + t.Parallel() + ctx := context.Background() + _, _, _, st := newHistoryEnv(t) + o := &Orchestrator{history: st, log: testLogger()} + + // 忙时入队：落库时 ctx 上没有回合标识 ⇒ 列里留空。 + queued := o.persistUserMessages(ctx, history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "排队的一句")}, "web:test", "chat", "cyrene") + // 被转向吃进的一句：同样在它自己的那一轮之前就落了库。 + steered := o.persistUserMessages(ctx, history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "被转向吃进的一句")}, "web:test", "chat", "cyrene") + if len(queued) != 1 || len(steered) != 1 { + t.Fatalf("落库条数 = %d / %d, want 各 1", len(queued), len(steered)) + } + before, err := st.List(ctx, history.MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list: %v", err) + } + for _, m := range before { + if m.TurnID != "" { + t.Fatalf("消息 %q 在补写之前就有 turn_id %q，want 空", m.Content, m.TurnID) + } + } + + // 这两条最终由同一路回答（转向把它并进了在飞的那一路）⇒ 补写同一个标识。 + o.assignTurnID(history.MainSessionID, "turn-inflight", append(queued, steered...)) + after, err := st.List(ctx, history.MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list after assign: %v", err) + } + for _, m := range after { + if m.TurnID != "turn-inflight" { + t.Fatalf("消息 %q 的 turn_id = %q, want turn-inflight（按实际处理它的那一路写）", m.Content, m.TurnID) + } + } + + // 另一路（例如续跑）试图认领同一行：不得改写既有的归属。 + o.assignTurnID(history.MainSessionID, "turn-later", queued) + again, err := st.List(ctx, history.MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list after re-assign: %v", err) + } + for _, m := range again { + if m.TurnID != "turn-inflight" { + t.Fatalf("消息 %q 的 turn_id 被改写成 %q, want 仍是 turn-inflight（先到先得）", m.Content, m.TurnID) + } + } + + // 空标识 / 空列表是空操作（未接线的旧路径与「这一路没有落库任何消息」）。 + o.assignTurnID(history.MainSessionID, "", queued) + o.assignTurnID(history.MainSessionID, "turn-x", nil) +} + +// TestTurnIDOfLegResolvesInFlightLeg 钉住补写路径要用的那个查询：在途路的 gen → +// 该路的权威 turnId；不在途（被打断 / 已被取代 / 会话里没有它）⇒ 空串。 +func TestTurnIDOfLegResolvesInFlightLeg(t *testing.T) { + t.Parallel() + o := &Orchestrator{log: testLogger(), states: map[string]*sessionState{ + history.MainSessionID: {legs: map[uint64]*turnLeg{7: {turnID: "turn-7"}, 8: {turnID: "turn-8"}}}, + }} + if got := o.turnIDOfLeg(history.MainSessionID, 8); got != "turn-8" { + t.Fatalf("turnIDOfLeg(8) = %q, want turn-8", got) + } + if got := o.turnIDOfLeg(history.MainSessionID, 9); got != "" { + t.Fatalf("turnIDOfLeg(未知 gen) = %q, want 空串", got) + } + if got := o.turnIDOfLeg("web:unknown", 7); got != "" { + t.Fatalf("turnIDOfLeg(未知会话) = %q, want 空串", got) + } +} + +// TestSubmitWiresTurnIDForBothRows 是「单路回合 ⇒ 用户消息与助手回复两条 turn_id +// 同值且非空」的**端到端**验收：真存储 + 真编排器，走 Submit → 起路 → 落库 → 结算 +// 的完整链路，断言提交回报的 TurnID 与落库那两条的 turn_id 完全一致。 +// +// ⏳ 它今天被 Skip 挡住：写入点必须在四个「本子代理硬约束禁止改动」的文件里接线 +// （session_parallel.go 的 publishLegStart、session_orchestrator.go 的续跑分支、 +// session_steer.go 的 Steer、session_turn_reply.go 的 persistTurnReply）。接线完成后 +// 删掉下面这行 Skip，本用例即成为该契约的端到端验收。 +func TestSubmitWiresTurnIDForBothRows(t *testing.T) { + t.Skip("待接线：编排器的四个写入点尚未把 TurnID 交给落库（见交接报告），本用例是接线后的端到端验收") + + ctx := context.Background() + _, _, _, st := newHistoryEnv(t) + run := newGatedLegRunner() + o := NewOrchestrator(OrchestratorOptions{Run: run.Run, Log: testLogger(), History: st}) + + sr := o.Submit(ctx, history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "端到端的问题")}, loop.OwnerEnv()) + if !sr.Immediate || sr.TurnID == "" { + t.Fatalf("Submit = %+v, want 立刻起一路并回报非空 TurnID", sr) + } + run.waitStarted(1) + run.release(0) + waitActiveTurns(t, o, history.MainSessionID, 0) + + msgs, err := st.List(ctx, history.MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(msgs) != 2 { + t.Fatalf("落库 %d 条, want 2（用户消息 + 助手回复）: %+v", len(msgs), msgs) + } + turns := turnsOf(msgs) + if turns["端到端的问题"] != sr.TurnID || turns["reply-1"] != sr.TurnID { + t.Fatalf("两条 turn_id = %q / %q, want 都是提交回报的 %q", + turns["端到端的问题"], turns["reply-1"], sr.TurnID) + } +} diff --git a/internal/services/gateway/session_persist.go b/internal/services/gateway/session_persist.go index 5d7e54f..990ae98 100644 --- a/internal/services/gateway/session_persist.go +++ b/internal/services/gateway/session_persist.go @@ -61,6 +61,11 @@ func modeFrom(ctx context.Context) string { // 与编排器预分配的消息 id：写入因此可被 (session_id, client_msg_id) 的唯一索引 // 判重。省略幂等键时 ctx 上没有该值，写入与加入这一列之前逐字相同。 // +// ctx 上若带着本路的回合标识（design/15 §2.5.9，withTurnID），该标识随**写入** +// 落进 turn_id 列 —— 这正是「这条路已经把这条消息吃进去了」要留下的配对信息。 +// 没有它（排队入队、未接线的路径）时列里留空串，由它实际被处理的那一路随后补写 +// （见 assignTurnID 与 internal/runtime/history/turn_id.go）。 +// // 它返回刚落库的那些消息（含存储分配的 id 与**只有引用**的 document 列）， // 使调用方能把它们挂到本回合上：抽取结论要等回合跑起来才有，只能在回合结束后 // 回写进同一列（design/24 §6.1，见 session_document_status.go）。这就是那条 @@ -70,6 +75,7 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string return nil } hint := clientMsgHintFrom(ctx) + turnID := turnIDFrom(ctx) persisted := make([]history.Message, 0, len(messages)) for _, m := range messages { if m.Role != model.RoleUser { @@ -85,7 +91,7 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string if text == "" && image == "" && documents == "" { continue } - msg := history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image, Document: documents, DeviceID: deviceID, Mode: modeOrDefault(mode), PersonaID: personaID} + msg := history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image, Document: documents, DeviceID: deviceID, Mode: modeOrDefault(mode), PersonaID: personaID, TurnID: turnID} // 幂等键只标注**第一条**用户消息：一个回合至多只有一条用户消息，而把 // 同一个键盖到多条上会让它们互相冲突（唯一索引把它们判成同一条）。 if hint.ClientMsgID != "" { @@ -128,11 +134,18 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string // 它是**裸写入**，判定属于 persistTurnReply（session_turn_reply.go）：收尾路径 // 经后者进入，由它保证「归本代次、且不是被取消的回合一定有一条可见消息」—— // 空文本在这里仍然是空操作，因此本函数绝不自作主张编造内容。 -func (o *Orchestrator) persistAssistantReply(sessionID string, res *loop.Result, deviceID, mode, personaID string) { +// +// turnID 是本路的回合标识（design/15 §2.5.9），可选：给出即随**写入**落进 turn_id +// 列，与它那条用户消息同值，客户端因此能精确配对「回复 ↔ 提问」。省略（未接线的 +// 调用路径）等价于加入该列之前的行为（列里留空串）。它刻意是可变参数而不是必填 +// 形参：既有调用点、以及「这一路没有回合标识」的旧路径都不必先传一个空值，而 +// 已给出的第一个值之外的多余值**被忽略**（一个回合只有一份标识，多给是调用方的 +// 笔误，绝不据此分叉行为）。 +func (o *Orchestrator) persistAssistantReply(sessionID string, res *loop.Result, deviceID, mode, personaID string, turnID ...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}) + 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, TurnID: firstTurnID(turnID)}) if err != nil { o.log.Error("orchestrator persist assistant", "err", err, "session", sessionID, "device_id", deviceID) return diff --git a/internal/services/gateway/session_turn_id.go b/internal/services/gateway/session_turn_id.go new file mode 100644 index 0000000..5c9616d --- /dev/null +++ b/internal/services/gateway/session_turn_id.go @@ -0,0 +1,119 @@ +package gateway + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +// 本文件归属消息行上的**回合标识**（turnId，design/15 §2.5.9）的持久化面：把编排器 +// 为每一路生成的权威 id 落进 chat_messages.turn_id（迁移 0018），使客户端在重连 / +// 回放时仍能精确配对「回复 ↔ 提问」—— 流帧是易失的，历史行不是。 +// +// 它从 session_persist.go 拆出来，使那个文件继续留在「单文件 ≤500 行」的硬约束内 +// （与 session_events.go / session_document_status.go 同一理由）。 +// +// 写入是**两段式**（与 document 列同一手法）： +// - 落库时所属路已经存在（新起的一路 + 它的助手回复）⇒ 随 INSERT 写入：用户消息 +// 走 ctx 通道（withTurnID，与 deviceID / mode / 幂等键同一种机制），助手回复走 +// 显式参数（persistAssistantReply 的可变尾参）； +// - 落库时还不知道所属路（忙时入队、被转向吃进）⇒ 由**实际处理它的那一路**在起跑 / +// 消费之后用 assignTurnID 补写。 +// +// 空值一律是空操作：省略标识的路径（既有调用点、未接线该能力的构建）行为与加入本列 +// 之前逐字相同。 + +// turnIDContextKey 是携带本路回合标识的 context 键（design/15 §2.5.9）。它由 +// 「起了一路」的那一刻挂上，使这一路落的用户消息带上它。 +type turnIDContextKey struct{} + +// withTurnID 返回携带本路回合标识的 context。空 turnID 是空操作：没有回合标识的 +// 路径（未接编排器的旧路径、存储替身）在 ctx 上不留任何痕迹。 +func withTurnID(ctx context.Context, turnID string) context.Context { + if turnID == "" { + return ctx + } + return context.WithValue(ctx, turnIDContextKey{}, turnID) +} + +// turnIDFrom 返回 ctx 携带的本路回合标识，未设置时返回 ""。 +func turnIDFrom(ctx context.Context) string { + v, _ := ctx.Value(turnIDContextKey{}).(string) + return v +} + +// firstTurnID 返回可选回合标识里的第一个非空值，没有时返回 ""。它服务 +// persistAssistantReply 的可变形参：省略与空值等价（都不写这一列）。 +func firstTurnID(turnID []string) string { + for _, id := range turnID { + if id != "" { + return id + } + } + return "" +} + +// turnIDWriter 是 HistoryStore 的一个**可选**更窄视图（与 clientMsgIDReader / +// documentStatusWriter 同一做法）：按消息 id 批量落定 turn_id。未实现该视图的 +// 存储替身或后端退回「不补写」并留下带来源的 WARN —— 补写是增益，绝不能反过来 +// 卡住会话。 +type turnIDWriter interface { + SetMessagesTurnID(ctx context.Context, ids []string, turnID string) error +} + +// turnIDLogSource 是回合标识补写的日志来源标签，使这条冷路径上的 WARN 能被追溯到 +// 具体子功能（AGENTS §10 的「日志必须标来源」）。 +const turnIDLogSource = "gateway/session-persist" + +// turnIDOfLeg 返回该会话在途路 gen 的权威回合标识；该路已不在途（被打断清空、 +// 已被取代、会话里没有它）时返回 ""。调用方用它把「这一刻算数的这一路」翻译成 +// 要落库的标识 —— 它**不得**在持有 o.mu 时调用（本函数自己取锁）。 +// +// ⚠️ 它有意**不用于**任何已经拿到 `*turnLeg` 的地方：InterruptScope(all) 会把 +// st.legs 整个清掉，而被打断那一路的助手文本仍要按原样落库（design/15 §3 的 +// 「内容是真的」），那时按 gen 回查会拿到空串、把配对丢掉。凡是手里有 leg 的 +// 调用点都传 `leg.turnID`（出生后不可变），只有确实只拿得到 gen 的地方才走本函数。 +func (o *Orchestrator) turnIDOfLeg(sessionID string, gen uint64) string { + o.mu.Lock() + defer o.mu.Unlock() + if leg := o.legLocked(o.states[sessionID], gen); leg != nil { + return leg.turnID + } + return "" +} + +// assignTurnID 把一组已落库消息的 turn_id 落定为 turnID —— 两段式写入的第二段。 +// 它服务「落库那一刻还不知道自己属于哪一路」的消息：忙时入队（它的路要等轮到它 +// 才起）与**被转向吃进**的消息（它由那个在飞的路回答）。两处都在路确定之后调用 +// 本函数，因此这些行最终带上的正是**实际处理它们的那一路**的标识。 +// +// 空 turnID、空消息列表、未接存储都直接返回（没有标识可落，也就没有写入）。存储 +// 未实现该可选视图时同样返回 —— 但**写库失败绝不静默**：打一条带来源的 WARN， +// 日志只带会话、条数与错误，不带正文。写入用全新的后台 context（与 +// persistAssistantReply / persistDocumentStatus 同一理由）：被取消回合的 ctx +// 绝不该中断这次补写。 +// +// 它绝不改写已经带标识的行（存储侧只写给仍为空串的行，先到先得），因此重复调用 +// 与「两条路径都想认领同一行」都不会让配对漂移。 +// +// 它必须在**状态锁之外**调用：它写库（那是慢操作）。握着 o.mu 的调用点要先解锁。 +func (o *Orchestrator) assignTurnID(sessionID, turnID string, msgs []history.Message) { + if o.history == nil || turnID == "" || len(msgs) == 0 { + return + } + writer, ok := o.history.(turnIDWriter) + if !ok { + return + } + ids := make([]string, 0, len(msgs)) + for _, m := range msgs { + if m.ID != "" { + ids = append(ids, m.ID) + } + } + if err := writer.SetMessagesTurnID(context.Background(), ids, turnID); err != nil { + logger.WithSource(o.log, turnIDLogSource).Warn("turn id not persisted", + "err", err, "session", sessionID, "turn", turnID, "messages", len(ids)) + } +}