diff --git a/internal/capability/registry/policy_store_test.go b/internal/capability/registry/policy_store_test.go index 0f073dc..5f9a898 100644 --- a/internal/capability/registry/policy_store_test.go +++ b/internal/capability/registry/policy_store_test.go @@ -104,9 +104,12 @@ func TestToolPolicyMigration_UpDownUpOnDataDirCopy(t *testing.T) { t.Fatalf("insert into copy: %v", err) } - // down：只回滚最后一步（= 0012）。 - if err := db.MigrateDown(ctx, conn, 1, testStoreLogger()); err != nil { - t.Fatalf("MigrateDown(copy, 1): %v", err) + // down：回滚到 0012 之前 —— 按已应用的迁移现算步数，而不是写死「最后一步」。 + // 写死会让后续任何新迁移（0013 起）悄悄变成「回滚那个新迁移」，于是本测试再也 + // 测不到 0012 的 down（0013 落地时正是这样暴露的）。 + steps := migrationStepsPast(t, ctx, conn, "0012_tool_policy") + if err := db.MigrateDown(ctx, conn, steps, testStoreLogger()); err != nil { + t.Fatalf("MigrateDown(copy, %d): %v", steps, err) } if toolPoliciesTableExists(t, ctx, conn) { t.Fatal("tool_policies 应在 MigrateDown 后消失") @@ -138,14 +141,44 @@ func TestToolPolicyMigration_UpDownOnMemory(t *testing.T) { if !toolPoliciesTableExists(t, ctx, conn) { t.Fatal("tool_policies 应在 MigrateUp 后存在") } - if err := db.MigrateDown(ctx, conn, 1, testStoreLogger()); err != nil { - t.Fatalf("MigrateDown: %v", err) + steps := migrationStepsPast(t, ctx, conn, "0012_tool_policy") + if err := db.MigrateDown(ctx, conn, steps, testStoreLogger()); err != nil { + t.Fatalf("MigrateDown(%d): %v", steps, err) } if toolPoliciesTableExists(t, ctx, conn) { t.Fatal("tool_policies 应在 MigrateDown 后消失") } } +// migrationStepsPast 计算「回滚到 id 之前」需要走几步：MigrateDown 从最新往回 +// 走，因此 id 自己以及比它新的每一个各算一步。按 schema_migrations 现算，本测试 +// 因此只锁定 0012 的 down，不会被之后新增的迁移（0013 及以后）带偏。 +func migrationStepsPast(t *testing.T, ctx context.Context, conn *sql.DB, id string) int { + t.Helper() + rows, err := conn.QueryContext(ctx, `SELECT id FROM schema_migrations`) + if err != nil { + t.Fatalf("列出已应用迁移：%v", err) + } + defer func() { _ = rows.Close() }() + steps := 0 + for rows.Next() { + var got string + if err := rows.Scan(&got); err != nil { + t.Fatalf("读取迁移 id：%v", err) + } + if got >= id { + steps++ + } + } + if err := rows.Err(); err != nil { + t.Fatalf("遍历迁移：%v", err) + } + if steps == 0 { + t.Fatalf("已应用迁移里没有 %s，本测试的前提不成立", id) + } + return steps +} + // TestRegistry_PolicyStoreSurvivesRestart 锁定「重启后策略仍在」： // 第一次运行 Save，第二次运行（新连接 + 新注册表 + 新 store）Load 回来。 func TestRegistry_PolicyStoreSurvivesRestart(t *testing.T) { diff --git a/internal/db/migrations/0013_chat_client_msg_id.down.sql b/internal/db/migrations/0013_chat_client_msg_id.down.sql new file mode 100644 index 0000000..d1358b8 --- /dev/null +++ b/internal/db/migrations/0013_chat_client_msg_id.down.sql @@ -0,0 +1,10 @@ +-- 0013_chat_client_msg_id down: 回滚客户端幂等键。 +-- +-- 先删索引再删列：SQLite 的 DROP COLUMN 在列仍被索引引用时会失败 +-- （client_msg_id 只被 idx_chat_messages_client_msg_id 引用，无视图与约束 +-- 牵涉）。回滚后的状态是「服务端不认幂等键」：列里的值随列一起丢弃， +-- 客户端只能自行按消息 id 对齐（即加入本列之前的行为）。 + +DROP INDEX IF EXISTS idx_chat_messages_client_msg_id; + +ALTER TABLE chat_messages DROP COLUMN client_msg_id; diff --git a/internal/db/migrations/0013_chat_client_msg_id.up.sql b/internal/db/migrations/0013_chat_client_msg_id.up.sql new file mode 100644 index 0000000..5cdb4cd --- /dev/null +++ b/internal/db/migrations/0013_chat_client_msg_id.up.sql @@ -0,0 +1,27 @@ +-- 0013_chat_client_msg_id: 客户端消息幂等键（Android 协同 §G4）。 +-- +-- 背景：chat 全链路此前**没有幂等** —— 同一段文本发两次就是两条用户消息、 +-- 两个回合（history.insert 每次都新铸 id）。弱网/切后台下的客户端超时重发 +-- 因此既留下重复气泡，也白烧一次模型调用。 +-- +-- client_msg_id 是**客户端生成**的去重键（例如一条乐观气泡的本地 id）。 +-- 服务端据此在同一会话内只承认第一条：重复到达返回既有消息，不再落库、 +-- 不再起回合。它**可选**：'' 意味着「这个客户端不带幂等键」，行为与加入本列 +-- 之前逐字相同（老客户端零影响，历史行也全是 ''）。 +-- +-- 判重由**唯一索引原子裁决**，而不是「先查再插」——后者是两个并发重发都 +-- 查不到、都插入的经典竞态： +-- (session_id, client_msg_id) 上建**部分**唯一索引，只覆盖 +-- client_msg_id != '' 的行。部分索引是必须的：若约束覆盖全表，所有空值行 +-- 会在同一会话里互相冲突，等于禁止一个会话出现第二条消息。空值行因此 +-- 完全不参与约束。 +-- +-- 0003 的 chat_messages 表在此已存在，故 ALTER ADD COLUMN 是非破坏性的正确 +-- 一步；全新库由 internal/runtime/history.NewStore 幂等地前滚（含该列与该 +-- 索引）。索引命名沿用 idx_{表}_{列} 口径，表名复数（援引 AGENTS §10 已登记 +-- 的例外）。 + +ALTER TABLE chat_messages ADD COLUMN client_msg_id TEXT NOT NULL DEFAULT ''; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_chat_messages_client_msg_id + ON chat_messages (session_id, client_msg_id) WHERE client_msg_id != ''; diff --git a/internal/runtime/history/client_msg_id.go b/internal/runtime/history/client_msg_id.go new file mode 100644 index 0000000..7160319 --- /dev/null +++ b/internal/runtime/history/client_msg_id.go @@ -0,0 +1,71 @@ +package history + +import ( + "context" + "database/sql" + "errors" + "fmt" + "strings" +) + +// 本文件归属聊天消息的**客户端幂等键**（Android 协同 §G4）。 +// +// 它是与会话编排器（internal/services/gateway/session_orchestrator.go）分摊的 +// 一件事的两半：编排器决定「这一轮还要不要跑」，存储决定「这一行还要不要写」。 +// 存储这半边的核心是 (session_id, client_msg_id) 上的部分唯一索引（见 0013 +// 迁移与本文件的 createClientMsgIDIndex）—— 判重由数据库原子裁决，因此并发的 +// 两次重发只有一个赢家，不存在「先查再插」的竞态窗口。 +// +// client_msg_id 为空串表示「这条消息没有幂等键」（老客户端、IM 镜像、历史行）， +// 部分索引刻意把它排除在外：否则同一会话的第二条空键消息就会撞上唯一约束。 + +// NewMessageID 返回一个新的消息 id（128 位密码学随机、十六进制编码），与 +// insert 在调用方省略 ID 时分配的 id 同源。它被会话编排器用来**预分配**一条 +// 用户消息的 id：这样同一条 clientMsgId 的并发重发即使赶在第一条真正落库之前 +// 到达，也能立刻拿到一个确定且终将存在的 messageId（而不是空值）。 +func NewMessageID() (string, error) { + return newID() +} + +// FindByClientMsgID 按 (sessionID, clientMsgID) 读回既有消息，供幂等判重。 +// 没有该行时返回包装了 ErrMessageNotFound 的错误；clientMsgID 为空时同理 +// （空键不是一条可判重的消息）。查询走部分唯一索引，至多命中一行。 +func (s *Store) FindByClientMsgID(ctx context.Context, sessionID, clientMsgID string) (Message, error) { + if clientMsgID == "" { + return Message{}, fmt.Errorf("history: find client msg in %s: %w", sessionID, ErrMessageNotFound) + } + row := s.db.QueryRowContext(ctx, `SELECT `+messageColumns+` + FROM chat_messages WHERE session_id = ? AND client_msg_id = ?`, sessionID, clientMsgID) + m, err := scanMessage(row) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return Message{}, fmt.Errorf("history: find client msg in %s: %w", sessionID, ErrMessageNotFound) + } + return Message{}, fmt.Errorf("history: find client msg in %s: %w", sessionID, err) + } + return m, nil +} + +// createClientMsgIDIndex 建 (session_id, client_msg_id) 的**部分**唯一索引 +// （只覆盖非空键）。它同时是幂等写入的冲突裁决者：insert 带非空 +// ClientMsgID 时用 ON CONFLICT DO NOTHING，由该索引决定谁写下这一行。 +const createClientMsgIDIndex = ` +CREATE UNIQUE INDEX IF NOT EXISTS idx_chat_messages_client_msg_id + ON chat_messages (session_id, client_msg_id) WHERE client_msg_id != ''` + +// ensureClientMsgIDColumn 在 chat_messages 早于本列（0013 迁移）建表时把它 +// 前滚：缺列则补加，已有则忽略重复列错误。它是幂等的，因此 NewStore 可在全新 +// schema 或既有 schema 上调用（与 memory 包的 ensureProjectIDColumn 同一手法）。 +func ensureClientMsgIDColumn(ctx context.Context, db *sql.DB) error { + _, err := db.ExecContext(ctx, `ALTER TABLE chat_messages + ADD COLUMN client_msg_id TEXT NOT NULL DEFAULT ''`) + if err == nil { + return nil + } + // 列已存在时（由 createMessagesTable 建出，或上一次运行补过）SQLite 报 + // "duplicate column name: client_msg_id"。 + if strings.Contains(err.Error(), "duplicate column name") { + return nil + } + return fmt.Errorf("history: roll forward client_msg_id column: %w", err) +} diff --git a/internal/runtime/history/client_msg_id_test.go b/internal/runtime/history/client_msg_id_test.go new file mode 100644 index 0000000..f17000c --- /dev/null +++ b/internal/runtime/history/client_msg_id_test.go @@ -0,0 +1,111 @@ +package history + +import ( + "context" + "errors" + "testing" +) + +// 本文件锁定 Android 协同 §G4 的存储半边：客户端幂等键（client_msg_id）由 +// (session_id, client_msg_id) 的**部分唯一索引**原子裁决，因此判重不是「先查 +// 再插」的竞态写法，而是「写入被数据库拒绝，然后读回既有那一行」。 +// 部分（WHERE client_msg_id != ''）是必须的：否则同一会话的所有空键行会互相 +// 冲突，等于禁止一个会话出现第二条消息。 + +func TestAppendFullDedupesClientMsgID(t *testing.T) { + t.Parallel() + ctx := context.Background() + st := newStore(t) + + first, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "同一段", ClientMsgID: "cm-1"}) + if err != nil { + t.Fatalf("第一次写入：%v", err) + } + // 第二次：换一个 id（像客户端重发时服务端会另铸一个），同一个 clientMsgId。 + second, err := st.AppendFull(ctx, Message{ID: "重发时的新 id", SessionID: MainSessionID, Role: RoleUser, Content: "同一段", ClientMsgID: "cm-1"}) + if err != nil { + t.Fatalf("第二次写入：%v", err) + } + if second.ID != first.ID { + t.Fatalf("第二次返回的 id = %q，want 既有那条 %q（唯一索引裁决，不新写一行）", second.ID, first.ID) + } + if second.ClientMsgID != "cm-1" { + t.Fatalf("第二次返回的消息 = %+v，want 回填的 clientMsgId", second) + } + + msgs, err := st.List(ctx, MainSessionID, 0, 50, "") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(msgs) != 1 { + t.Fatalf("行数 = %d, want 1（同一个 clientMsgId 只落一条）", len(msgs)) + } + if msgs[0].ClientMsgID != "cm-1" { + t.Fatalf("读回的消息 = %+v，want clientMsgId=cm-1（读接口可见）", msgs[0]) + } + + // 省略幂等键（空串）不参与约束：同一会话的两条空键消息都必须写得进去。 + for i := 0; i < 2; i++ { + if _, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "无键"}); err != nil { + t.Fatalf("第 %d 条空键写入：%v", i+1, err) + } + } + msgs, err = st.List(ctx, MainSessionID, 0, 50, "") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(msgs) != 3 { + t.Fatalf("行数 = %d, want 3（空 clientMsgId 绝不受唯一索引约束）", len(msgs)) + } +} + +func TestAppendDedupesClientMsgIDPerSession(t *testing.T) { + t.Parallel() + ctx := context.Background() + st := newStore(t) + + // 幂等键只在**会话内**唯一：两个会话用同一个键互不影响 + // （client_msg_id 由客户端生成，不同会话之间没有可比性）。 + main, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "主会话", ClientMsgID: "cm-x"}) + if err != nil { + t.Fatalf("主会话写入：%v", err) + } + sub := MainSessionID + ":sub-1" + other, err := st.AppendFull(ctx, Message{SessionID: sub, Role: RoleUser, Content: "子会话", ClientMsgID: "cm-x"}) + if err != nil { + t.Fatalf("子会话写入：%v", err) + } + if other.ID == main.ID { + t.Fatalf("子会话返回了主会话的行 %q，want 各自一条", main.ID) + } +} + +func TestFindByClientMsgID(t *testing.T) { + t.Parallel() + ctx := context.Background() + st := newStore(t) + + full, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "定位我", ClientMsgID: "cm-find"}) + if err != nil { + t.Fatalf("写入：%v", err) + } + got, err := st.FindByClientMsgID(ctx, MainSessionID, "cm-find") + if err != nil { + t.Fatalf("按幂等键读取：%v", err) + } + if got.ID != full.ID || got.ClientMsgID != "cm-find" { + t.Fatalf("读回 = %+v，want id=%q clientMsgId=cm-find", got, full.ID) + } + + // 没见过的键与空键都必须是明确的 ErrMessageNotFound（编排器据此判定「不是 + // 重复」，而不是把查询错误当成重复）。 + for _, key := range []string{"cm-missing", ""} { + if _, err := st.FindByClientMsgID(ctx, MainSessionID, key); !errors.Is(err, ErrMessageNotFound) { + t.Fatalf("键 %q 的错误 = %v，want wrap of ErrMessageNotFound", key, err) + } + } + // 会话也必须匹配：同一个键在别的会话里不存在。 + if _, err := st.FindByClientMsgID(ctx, MainSessionID+":other", "cm-find"); !errors.Is(err, ErrMessageNotFound) { + t.Fatalf("跨会话查询的错误 = %v，want wrap of ErrMessageNotFound", err) + } +} diff --git a/internal/runtime/history/list_after.go b/internal/runtime/history/list_after.go index 2a70648..0ffcec6 100644 --- a/internal/runtime/history/list_after.go +++ b/internal/runtime/history/list_after.go @@ -13,7 +13,7 @@ import ( // 消息保持插入顺序。返回切片非 nil；limit 钳制到 [1, maxPageLimit]。 func (s *Store) ListAfter(ctx context.Context, sessionID string, after int64, limit int, platform string) ([]Message, error) { limit = clampLimit(limit) - query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, persona_id, created_at + query := `SELECT ` + messageColumns + ` FROM chat_messages WHERE session_id = ? AND created_at > ?` args := []any{sessionID, after} if platform != "" { diff --git a/internal/runtime/history/log_reads.go b/internal/runtime/history/log_reads.go new file mode 100644 index 0000000..39ccb7f --- /dev/null +++ b/internal/runtime/history/log_reads.go @@ -0,0 +1,160 @@ +package history + +import ( + "context" + "fmt" +) + +// 本文件归属 chat_messages 的**日志与窗口读取**：适配器日志页的三个粒度 +// （单适配器 / 单平台 / 全适配器）与 WS 回放/上下文尾部用的 Recent。它们与 +// store.go 的 List/Search 分开按职责分文件，并共用 store.go 的 messageColumns +// 列清单与 scanMessage，因此「一列都不许漏」的口径仍然只有一处。 + +// ListPlatformLog 按最新优先返回 platform 跨所有非主（隔离）会话的至多 limit +// 条消息。主会话消息被排除，因此只有该平台自己的消息流（群聊 + 陌生人私聊） +// 浮现到适配器日志页（design/16 §4.6 §4）。before 是毫秒游标；非零时只返回 +// 严格更旧的消息。返回切片非 nil。 +func (s *Store) ListPlatformLog(ctx context.Context, platform string, before int64, limit int) ([]Message, error) { + limit = clampLimit(limit) + query := `SELECT ` + messageColumns + ` + FROM chat_messages WHERE platform = ? AND session_id != ?` + args := []any{platform, MainSessionID} + if before > 0 { + query += ` AND created_at < ?` + args = append(args, before) + } + query += ` ORDER BY created_at DESC LIMIT ?` + args = append(args, limit) + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("history: platform log %s: %w", platform, err) + } + defer func() { _ = rows.Close() }() + + msgs := make([]Message, 0) + for rows.Next() { + m, err := scanMessage(rows) + if err != nil { + return nil, fmt.Errorf("history: scan platform log %s: %w", platform, err) + } + msgs = append(msgs, m) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("history: iterate platform log %s: %w", platform, err) + } + return msgs, nil +} + +// ListAdapterLog 按最新优先返回单个 IM 适配器（平台 + 账号注册名，例如 +// "kook:3bef..."）的至多 limit 条消息，供按适配器的日志页使用。它不排除主 Web +// 会话：主人的 IM 私聊住在主会话里但仍携带 adapter_id，因此按适配器的日志包含 +// 主人镜像。adapterID 为空时返回空（非 nil）切片。before 是毫秒游标；非零时只 +// 返回严格更旧的消息。返回切片非 nil。 +func (s *Store) ListAdapterLog(ctx context.Context, adapterID string, before int64, limit int) ([]Message, error) { + limit = clampLimit(limit) + if adapterID == "" { + return make([]Message, 0), nil + } + query := `SELECT ` + messageColumns + ` + FROM chat_messages WHERE adapter_id = ?` + args := []any{adapterID} + if before > 0 { + query += ` AND created_at < ?` + args = append(args, before) + } + query += ` ORDER BY created_at DESC LIMIT ?` + args = append(args, limit) + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("history: adapter log %s: %w", adapterID, err) + } + defer func() { _ = rows.Close() }() + + msgs := make([]Message, 0) + for rows.Next() { + m, err := scanMessage(rows) + if err != nil { + return nil, fmt.Errorf("history: scan adapter log %s: %w", adapterID, err) + } + msgs = append(msgs, m) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("history: iterate adapter log %s: %w", adapterID, err) + } + return msgs, nil +} + +// ListAllLog 按最新优先返回跨所有 IM 适配器来源（adapter_id 非空）的至多 limit +// 条消息，供适配器总览页使用。before 是毫秒游标；非零时只返回严格更旧的消息。 +// 返回切片非 nil。 +func (s *Store) ListAllLog(ctx context.Context, before int64, limit int) ([]Message, error) { + limit = clampLimit(limit) + query := `SELECT ` + messageColumns + ` + FROM chat_messages WHERE adapter_id != ''` + var args []any + if before > 0 { + query += ` AND created_at < ?` + args = append(args, before) + } + query += ` ORDER BY created_at DESC LIMIT ?` + args = append(args, limit) + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("history: all adapter log: %w", err) + } + defer func() { _ = rows.Close() }() + + msgs := make([]Message, 0) + for rows.Next() { + m, err := scanMessage(rows) + if err != nil { + return nil, fmt.Errorf("history: scan all adapter log: %w", err) + } + msgs = append(msgs, m) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("history: iterate all adapter log: %w", err) + } + return msgs, nil +} + +// Recent 按最旧优先返回 sessionID 最近的 n 条消息（最靠尾部的那段窗口），供 +// WS 回放 / 上下文尾部使用。返回切片非 nil。 +// +// 并列打破（裁决 13.4.5 的上下文「最新性/一致性」）：同一毫秒写入的多条消息 +// 在 created_at 上并列，而 Windows 上 time.Now() 的步长本就约 0.44ms —— 用户 +// 连发多条（design/15 §2.1 的忙碌续接场景）时并列是常态。`rowid DESC` 让 LIMIT +// 取到的是**最后插入**的那几条（否则窗口会留下更旧的、丢掉刚发出的），而反转后 +// 同毫秒内即回到插入顺序，使续接窗口的时序与真实对话一致。该口径与 ListAfter +// 的「rowid 打破 created_at 的并列」相同。 +func (s *Store) Recent(ctx context.Context, sessionID string, n int) ([]Message, error) { + if n < 1 { + n = defaultPageLimit + } + rows, err := s.db.QueryContext(ctx, `SELECT `+messageColumns+` + FROM chat_messages WHERE session_id = ? ORDER BY created_at DESC, rowid DESC LIMIT ?`, sessionID, n) + if err != nil { + return nil, fmt.Errorf("history: recent %s: %w", sessionID, err) + } + defer func() { _ = rows.Close() }() + + // Recent 让最新的排在最后，使窗口处于时间顺序；先按查询顺序构建再原地反转。 + msgs := make([]Message, 0) + for rows.Next() { + m, err := scanMessage(rows) + if err != nil { + return nil, fmt.Errorf("history: scan recent %s: %w", sessionID, err) + } + msgs = append(msgs, m) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("history: iterate recent %s: %w", sessionID, err) + } + for i, j := 0, len(msgs)-1; i < j; i, j = i+1, j-1 { + msgs[i], msgs[j] = msgs[j], msgs[i] + } + return msgs, nil +} diff --git a/internal/runtime/history/message_ops.go b/internal/runtime/history/message_ops.go index a6c36b8..9411178 100644 --- a/internal/runtime/history/message_ops.go +++ b/internal/runtime/history/message_ops.go @@ -25,7 +25,7 @@ func (s *Store) Search(ctx context.Context, sessionID, query string, before int6 if query == "" { return make([]Message, 0), nil } - sqlText := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, persona_id, created_at + sqlText := `SELECT ` + messageColumns + ` FROM chat_messages WHERE session_id = ? AND content LIKE ? ESCAPE '\'` args := []any{sessionID, "%" + escapeLike(query) + "%"} if before > 0 { @@ -98,7 +98,7 @@ func (s *Store) GetMessage(ctx context.Context, id string) (Message, error) { // messageByID 按 id 读取一条消息，行不存在时返回 ErrMessageNotFound。 func (s *Store) messageByID(ctx context.Context, id string) (Message, error) { - row := s.db.QueryRowContext(ctx, `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, persona_id, created_at + row := s.db.QueryRowContext(ctx, `SELECT `+messageColumns+` FROM chat_messages WHERE id = ?`, id) m, err := scanMessage(row) if err != nil { diff --git a/internal/runtime/history/store.go b/internal/runtime/history/store.go index 878d41c..120d75f 100644 --- a/internal/runtime/history/store.go +++ b/internal/runtime/history/store.go @@ -106,7 +106,14 @@ type Message struct { // 一致。它只是展示/归属元数据：身份与权限由渠道 actor 元数据决定 // （项目红线）。 PersonaID string - CreatedAt int64 + // ClientMsgID 是客户端为这条消息生成的幂等键（Android 协同 §G4）—— 通常是 + // 一条乐观气泡的本地 id，也正因此它必须原样回填并可见：客户端用它把本地 + // 气泡与服务端真相对齐，而不是靠「文本相等」这类脆弱匹配。同一会话内同一 + // 非空键只允许一条消息（部分唯一索引裁决，见 client_msg_id.go）。空串意味着 + // 「这条消息没有幂等键」（老客户端、IM 镜像、历史行），行为与加入该列之前 + // 逐字相同。 + ClientMsgID string + CreatedAt int64 } // SessionMeta 是一个持久化会话面向线上的元数据。 @@ -221,18 +228,31 @@ type rowScanner interface { } // NewStore 在 db 上构建一个 Store，确保 chat_sessions/chat_messages schema 存在 -// 且固定主会话行在位。该 DDL 与 0003 迁移一致；它是幂等的，因此存储自足、可从 -// 任意 *sql.DB 使用，与 memory 和 session 包一致。db 必须可用；nil 返回 -// ErrNilDB。log 可以为 nil。 +// 且固定主会话行在位。该 DDL 与 0003 迁移一致（列清单随后续迁移同步增长，见 +// client_msg_id.go）；它是幂等的，因此存储自足、可从任意 *sql.DB 使用，与 +// memory 和 session 包一致。db 必须可用；nil 返回 ErrNilDB。log 可以为 nil。 func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) { if db == nil { return nil, ErrNilDB } log = logger.WithSource(discardLogger(log), "history") + // 顺序不能反：createMessagesTable 对全新库已含 client_msg_id，随后 + // ensureClientMsgIDColumn 把早于该列建出的既有表前滚（幂等），最后才建索引 + // —— 部分唯一索引引用 client_msg_id，列还不存在时建它必然失败。 for _, ddl := range []string{ createSessionsTable, createMessagesTable, + } { + if _, err := db.ExecContext(ctx, ddl); err != nil { + return nil, fmt.Errorf("history: apply schema: %w", err) + } + } + if err := ensureClientMsgIDColumn(ctx, db); err != nil { + return nil, err + } + for _, ddl := range []string{ createMessagesIndex, + createClientMsgIDIndex, } { if _, err := db.ExecContext(ctx, ddl); err != nil { return nil, fmt.Errorf("history: apply schema: %w", err) @@ -262,6 +282,12 @@ func (s *Store) AppendFull(ctx context.Context, msg Message) (Message, error) { // insert 是 Append/AppendFull 共用的主体。它在调用方把 id/CreatedAt 留空/留零时 // 分配新的 id 与现在的毫秒时间戳，然后插入该行并返回得到的消息。 +// +// msg.ClientMsgID 非空时（Android 协同 §G4，客户端幂等键）写入走 +// ON CONFLICT DO NOTHING，由 (session_id, client_msg_id) 的部分唯一索引**原子** +// 裁决谁写下了这一行；命中冲突时不写入，改读回既有那一行返回（因此调用方拿到 +// 的是权威消息，而不是一条并不存在的「刚插入的行」）。判重不在这里「先查再插」 +// —— 那是两个并发重发都能通过检查、然后双双插入的竞态。 func (s *Store) insert(ctx context.Context, msg Message) (Message, error) { if msg.ID == "" { id, err := newID() @@ -279,14 +305,27 @@ func (s *Store) insert(ctx context.Context, msg Message) (Message, error) { if msg.Role == RoleUser { s.ensureTitle(ctx, msg.SessionID, msg.Content) } - _, err := s.db.ExecContext(ctx, `INSERT INTO chat_messages - (id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, persona_id, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + query := `INSERT INTO chat_messages + (id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, persona_id, client_msg_id, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + if msg.ClientMsgID != "" { + query += ` ON CONFLICT DO NOTHING` + } + res, err := s.db.ExecContext(ctx, query, msg.ID, msg.SessionID, msg.Role, msg.Content, msg.Mode, msg.Mood, - msg.Image, msg.Thinking, msg.Platform, msg.DeviceID, msg.SenderName, msg.AdapterID, msg.PersonaID, msg.CreatedAt) + msg.Image, msg.Thinking, msg.Platform, msg.DeviceID, msg.SenderName, msg.AdapterID, msg.PersonaID, msg.ClientMsgID, msg.CreatedAt) if err != nil { return Message{}, fmt.Errorf("history: append %s to %s: %w", msg.Role, msg.SessionID, err) } + if msg.ClientMsgID != "" { + n, err := res.RowsAffected() + if err != nil { + return Message{}, fmt.Errorf("history: append %s to %s: %w", msg.Role, msg.SessionID, err) + } + if n == 0 { + return s.FindByClientMsgID(ctx, msg.SessionID, msg.ClientMsgID) + } + } return msg, nil } @@ -297,7 +336,7 @@ func (s *Store) insert(ctx context.Context, msg Message) (Message, error) { // 通道）。返回切片非 nil。limit 钳制到 [1, maxPageLimit]。 func (s *Store) List(ctx context.Context, sessionID string, before int64, limit int, platform string) ([]Message, error) { limit = clampLimit(limit) - query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, persona_id, created_at + query := `SELECT ` + messageColumns + ` FROM chat_messages WHERE session_id = ?` args := []any{sessionID} if before > 0 { @@ -331,155 +370,6 @@ func (s *Store) List(ctx context.Context, sessionID string, before int64, limit return msgs, nil } -// ListPlatformLog 按最新优先返回 platform 跨所有非主（隔离）会话的至多 limit -// 条消息。主会话消息被排除，因此只有该平台自己的消息流（群聊 + 陌生人私聊） -// 浮现到适配器日志页（design/16 §4.6 §4）。before 是毫秒游标；非零时只返回 -// 严格更旧的消息。返回切片非 nil。 -func (s *Store) ListPlatformLog(ctx context.Context, platform string, before int64, limit int) ([]Message, error) { - limit = clampLimit(limit) - query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, persona_id, created_at - FROM chat_messages WHERE platform = ? AND session_id != ?` - args := []any{platform, MainSessionID} - if before > 0 { - query += ` AND created_at < ?` - args = append(args, before) - } - query += ` ORDER BY created_at DESC LIMIT ?` - args = append(args, limit) - - rows, err := s.db.QueryContext(ctx, query, args...) - if err != nil { - return nil, fmt.Errorf("history: platform log %s: %w", platform, err) - } - defer func() { _ = rows.Close() }() - - msgs := make([]Message, 0) - for rows.Next() { - m, err := scanMessage(rows) - if err != nil { - return nil, fmt.Errorf("history: scan platform log %s: %w", platform, err) - } - msgs = append(msgs, m) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("history: iterate platform log %s: %w", platform, err) - } - return msgs, nil -} - -// ListAdapterLog 按最新优先返回单个 IM 适配器（平台 + 账号注册名，例如 -// "kook:3bef..."）的至多 limit 条消息，供按适配器的日志页使用。它不排除主 Web -// 会话：主人的 IM 私聊住在主会话里但仍携带 adapter_id，因此按适配器的日志包含 -// 主人镜像。adapterID 为空时返回空（非 nil）切片。before 是毫秒游标；非零时只 -// 返回严格更旧的消息。返回切片非 nil。 -func (s *Store) ListAdapterLog(ctx context.Context, adapterID string, before int64, limit int) ([]Message, error) { - limit = clampLimit(limit) - if adapterID == "" { - return make([]Message, 0), nil - } - query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, persona_id, created_at - FROM chat_messages WHERE adapter_id = ?` - args := []any{adapterID} - if before > 0 { - query += ` AND created_at < ?` - args = append(args, before) - } - query += ` ORDER BY created_at DESC LIMIT ?` - args = append(args, limit) - - rows, err := s.db.QueryContext(ctx, query, args...) - if err != nil { - return nil, fmt.Errorf("history: adapter log %s: %w", adapterID, err) - } - defer func() { _ = rows.Close() }() - - msgs := make([]Message, 0) - for rows.Next() { - m, err := scanMessage(rows) - if err != nil { - return nil, fmt.Errorf("history: scan adapter log %s: %w", adapterID, err) - } - msgs = append(msgs, m) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("history: iterate adapter log %s: %w", adapterID, err) - } - return msgs, nil -} - -// ListAllLog 按最新优先返回跨所有 IM 适配器来源（adapter_id 非空）的至多 limit -// 条消息，供适配器总览页使用。before 是毫秒游标；非零时只返回严格更旧的消息。 -// 返回切片非 nil。 -func (s *Store) ListAllLog(ctx context.Context, before int64, limit int) ([]Message, error) { - limit = clampLimit(limit) - query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, persona_id, created_at - FROM chat_messages WHERE adapter_id != ''` - var args []any - if before > 0 { - query += ` AND created_at < ?` - args = append(args, before) - } - query += ` ORDER BY created_at DESC LIMIT ?` - args = append(args, limit) - - rows, err := s.db.QueryContext(ctx, query, args...) - if err != nil { - return nil, fmt.Errorf("history: all adapter log: %w", err) - } - defer func() { _ = rows.Close() }() - - msgs := make([]Message, 0) - for rows.Next() { - m, err := scanMessage(rows) - if err != nil { - return nil, fmt.Errorf("history: scan all adapter log: %w", err) - } - msgs = append(msgs, m) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("history: iterate all adapter log: %w", err) - } - return msgs, nil -} - -// Recent 按最旧优先返回 sessionID 最近的 n 条消息（最靠尾部的那段窗口），供 -// WS 回放 / 上下文尾部使用。返回切片非 nil。 -// -// 并列打破（裁决 13.4.5 的上下文「最新性/一致性」）：同一毫秒写入的多条消息 -// 在 created_at 上并列，而 Windows 上 time.Now() 的步长本就约 0.44ms —— 用户 -// 连发多条（design/15 §2.1 的忙碌续接场景）时并列是常态。`rowid DESC` 让 LIMIT -// 取到的是**最后插入**的那几条（否则窗口会留下更旧的、丢掉刚发出的），而反转后 -// 同毫秒内即回到插入顺序，使续接窗口的时序与真实对话一致。该口径与 ListAfter -// 的「rowid 打破 created_at 的并列」相同。 -func (s *Store) Recent(ctx context.Context, sessionID string, n int) ([]Message, error) { - if n < 1 { - n = defaultPageLimit - } - rows, err := s.db.QueryContext(ctx, `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, persona_id, created_at - FROM chat_messages WHERE session_id = ? ORDER BY created_at DESC, rowid DESC LIMIT ?`, sessionID, n) - if err != nil { - return nil, fmt.Errorf("history: recent %s: %w", sessionID, err) - } - defer func() { _ = rows.Close() }() - - // Recent 让最新的排在最后，使窗口处于时间顺序；先按查询顺序构建再原地反转。 - msgs := make([]Message, 0) - for rows.Next() { - m, err := scanMessage(rows) - if err != nil { - return nil, fmt.Errorf("history: scan recent %s: %w", sessionID, err) - } - msgs = append(msgs, m) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("history: iterate recent %s: %w", sessionID, err) - } - for i, j := 0, len(msgs)-1; i < j; i, j = i+1, j-1 { - msgs[i], msgs[j] = msgs[j], msgs[i] - } - return msgs, nil -} - // DeleteSession 在一个事务中删除一个子会话及其全部消息。固定主会话不可变， // 返回 ErrImmutableMain（design/16 §6）。 func (s *Store) DeleteSession(ctx context.Context, sessionID string) error { @@ -523,11 +413,18 @@ func (s *Store) ensureMainSession(ctx context.Context) error { return nil } -// scanMessage 读取一行 chat_messages。消息各列按位置扫描。 +// messageColumns 是 scanMessage 期望的 chat_messages 列清单，也是每条读取 +// 语句（List/ListAfter/Recent/Search/messageByID/FindByClientMsgID）共用的 +// **唯一**一份列清单（位置顺序必须与 scanMessage 的 Scan 完全一致）。 +// 它刻意只写一次：此前七个查询各抄一份，加一列就得改七处，漏一处就只会在 +// 运行时以「扫描到的列数与列不清单不符」炸掉；共用常量让漏改成为不可能。 +const messageColumns = `id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, persona_id, client_msg_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.Thinking, &platform, &deviceID, &senderName, &adapterID, &m.PersonaID, &m.CreatedAt); err != nil { + if err := sc.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &mode, &mood, &m.Image, &m.Thinking, &platform, &deviceID, &senderName, &adapterID, &m.PersonaID, &m.ClientMsgID, &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 @@ -592,6 +489,7 @@ CREATE TABLE IF NOT EXISTS chat_messages ( 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 )` diff --git a/internal/runtime/loop/memory_writeback_test.go b/internal/runtime/loop/memory_writeback_test.go new file mode 100644 index 0000000..a2bc632 --- /dev/null +++ b/internal/runtime/loop/memory_writeback_test.go @@ -0,0 +1,228 @@ +package loop + +import ( + "context" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/restrict" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/capacity" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// TestRun_WorkingMemory_WriteOnSettle_回合落定写入工作记忆 是审计 P0-4 的 +// 主断言：跑完一个回合后，该会话的工作记忆窗口非空、摘要非空，且 +// assembleContext 的读点（loop/memory.go）真的把它作为前导块并入 —— +// 接线之前生产代码从不调用 Add，窗口恒空、块永不出现。 +func TestRun_WorkingMemory_WriteOnSettle_回合落定写入工作记忆(t *testing.T) { + t.Parallel() + store := newTestMemoryStore(t) + wm := memory.NewWorkingMemory(0) + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "好的"}}}} + a := newTestAgent(t, ad, nil) + l := New(Options{ + Agent: a, Memory: store, WorkingMemory: wm, + PersonaID: "persona-a", MaxRounds: 8, + }) + + ctx := WithSessionEnv(context.Background(), OwnerEnv()) + msgs := []model.Message{model.TextMessage(model.RoleUser, "帮我查一下资料")} + if _, err := l.Run(ctx, "web:wm", msgs); err != nil { + t.Fatalf("Run: %v", err) + } + + win := wm.SessionWindow("web:wm") + if len(win) == 0 { + t.Fatal("工作记忆会话窗口仍为空：回合落定没有写入（P0-4 未修）") + } + if win[0].Content[0].Text != "帮我查一下资料" { + t.Fatalf("窗口首条 = %q, want 本轮用户消息", win[0].Content[0].Text) + } + if got := wm.SummarizeSession("web:wm"); got == "" { + t.Fatal("SummarizeSession 仍为 \"\"：读点拿不到内容") + } + + // 读点：下一次组装把工作记忆并入前导块（此前该块永不出现）。 + // 组装在生产路径上总带着该 Run 的会话 id（Run 顶部 withLoopSession）， + // 这里同样把会话挂上，免得读到别的会话的窗口。 + view := l.assembleContext(withLoopSession(ctx, "web:wm"), msgs, 1) + if len(view) != 2 { + t.Fatalf("assemble = %d 条, want 2（工作记忆前导块 + 对话）", len(view)) + } + if head := messageText(view[0]); !strings.Contains(head, "工作记忆") { + t.Fatalf("assemble 首条不是工作记忆块: %q", head) + } + + // 会话恢复：同一会话再来一个回合（携带完整历史）时窗口被回填， + // 且不与其他会话/默认窗口混同。 + wm.RecordSession("web:other", []model.Message{model.TextMessage(model.RoleUser, "另一个会话")}) + if got := wm.SummarizeSession("web:wm"); strings.Contains(got, "另一个会话") { + t.Fatalf("会话隔离被破坏: %q", got) + } + if got := wm.Summarize(); got != "" { + t.Fatalf("默认窗口被回合写入污染: %q", got) + } +} + +// TestRun_IntentWriteback_OwnerLearns_闭环生效 是审计 P0-17 的主断言： +// owner 会话跑完一个任务回合后，L4 判定被**真实写回记忆**，并且这条 +// 记忆能被 L4 读回 —— 一个自身没有强信号的后续表达因此被唤醒。 +// 对照（变异反证）：不喂召回的同一判定保持静息。 +func TestRun_IntentWriteback_OwnerLearns_闭环生效(t *testing.T) { + t.Parallel() + store := newTestMemoryStore(t) + dec := capacity.New(capacity.Config{Enabled: true, MinConfidence: 0.6}) + ret := memory.NewRetriever(store) + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "好的"}}}} + a := newTestAgent(t, ad, nil) + l := New(Options{ + Agent: a, Memory: store, Retriever: ret, Capacity: dec, + PersonaID: "persona-a", MaxRounds: 8, + }) + + ctx := WithSessionEnv(context.Background(), OwnerEnv()) + if _, err := l.Run(ctx, "web:l4", []model.Message{model.TextMessage(model.RoleUser, "帮我搞一下那个部署")}); err != nil { + t.Fatalf("Run: %v", err) + } + + items, err := store.List(context.Background()) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(items) != 1 { + t.Fatalf("回合落定产出 %d 条记忆, want 1（L4 判定回写，P0-17 未修）", len(items)) + } + record := items[0] + if !strings.HasPrefix(record.Content, "intent:") { + t.Fatalf("回写内容 = %q, want capacity.RecordIntent 的规范格式", record.Content) + } + if record.Scope != memory.ScopeUser { + t.Fatalf("回写作用域 = %q, want user（用户习惯域，L4 的读域）", record.Scope) + } + if record.PersonaID != "persona-a" { + t.Fatalf("回写角色 = %q, want persona-a", record.PersonaID) + } + if _, ok := memory.ParseSourceConversation(record.SourceConversation); !ok { + t.Fatalf("回写溯源 = %q, want 结构化 session/turn 溯源", record.SourceConversation) + } + + // 闭环：这条记忆被真实召回后，L4 唤醒一个自身无强信号的表达。 + const probe = "那个部署怎么样了" + recall := l.capacityRecall(ctx, []model.Message{model.TextMessage(model.RoleUser, probe)}) + if len(recall) == 0 { + t.Fatal("写回的记忆没有被检索召回，闭环不成立") + } + without := dec.Decide(ctx, capacity.Input{Messages: []string{probe}}) + if without.State != capacity.StateQuiet { + t.Fatalf("前提不成立：探针文本自身已有强信号（state=%q）", without.State) + } + with := dec.Decide(ctx, capacity.Input{Messages: []string{probe}, Recall: recall}) + if with.State == capacity.StateQuiet { + t.Fatalf("L4 未从写回的记忆中学到习惯：state=%q（want 唤醒）", with.State) + } +} + +// TestRun_MemoryHooks_RestrictedSessionsSkip_受限会话零写入 是 P0-4/P0-17 +// 的红线反证：群聊与陌生人会话（MemWritable=false）既不写共享记忆、 +// 也不落任何工作记忆，且别的主人的会话窗口绝不泄露到这些上下文里。 +func TestRun_MemoryHooks_RestrictedSessionsSkip_受限会话零写入(t *testing.T) { + t.Parallel() + store := newTestMemoryStore(t) + wm := memory.NewWorkingMemory(0) + dec := capacity.New(capacity.Config{Enabled: true, MinConfidence: 0.6}) + ad := &mockAdapter{} + a := newTestAgent(t, ad, nil) + l := New(Options{ + Agent: a, Memory: store, Retriever: memory.NewRetriever(store), Capacity: dec, + WorkingMemory: wm, PersonaID: "persona-a", MaxRounds: 8, + }) + + // owner 会话先写出一条意图记录与一个工作记忆窗口。 + ownerCtx := WithSessionEnv(context.Background(), OwnerEnv()) + if _, err := l.Run(ownerCtx, "web:owner", []model.Message{model.TextMessage(model.RoleUser, "帮我搞一下那个部署")}); err != nil { + t.Fatalf("owner Run: %v", err) + } + ownerItems, err := store.List(context.Background()) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(ownerItems) != 1 { + t.Fatalf("owner 回合写入 %d 条, want 1", len(ownerItems)) + } + + restricted := []struct { + name string + env SessionEnv + }{ + {"group", SessionEnv{ActorEnv: restrict.ActorGroupMember, MemWritable: false}}, + {"stranger", SessionEnv{ActorEnv: restrict.ActorStranger, MemWritable: false}}, + } + // 主人私有的意图记录（含原话）落在共享的用户习惯域以便 L4 闭环， + // 但绝不能被群聊/陌生人读走：受限块里只允许出现非敏感共享记忆。 + seedIntentAndHabit(t, store) + before, err := store.List(context.Background()) + if err != nil { + t.Fatalf("List: %v", err) + } + for _, tc := range restricted { + ctx := WithSessionEnv(context.Background(), tc.env) + if _, err := l.Run(ctx, "sess-"+tc.name, []model.Message{model.TextMessage(model.RoleUser, "帮我搞一下那个部署")}); err != nil { + t.Fatalf("%s Run: %v", tc.name, err) + } + if got := len(wm.SessionWindow("sess-" + tc.name)); got != 0 { + t.Fatalf("%s 会话写了工作记忆: %d 条（只读隔离被破坏）", tc.name, got) + } + // 主人会话的窗口绝不进入受限上下文的前导块。 + view := l.assembleContext(ctx, []model.Message{model.TextMessage(model.RoleUser, "go")}, 1) + // 正向对照：非敏感共享记忆（用户习惯）确实注入，证明这个块不是 + // 空的 —— 因此下面的「不出现」是真过滤，而不是什么都没注入。 + block := "" + for _, m := range view { + block += messageText(m) + "\n" + } + if !strings.Contains(block, "喜欢手冲咖啡") { + t.Fatalf("%s 上下文缺少非敏感共享记忆（测试前提不成立）:\n%s", tc.name, block) + } + for _, m := range view { + text := messageText(m) + if strings.Contains(text, "工作记忆") { + t.Fatalf("%s 上下文泄露了工作记忆块: %q", tc.name, text) + } + if strings.Contains(text, "intent:") { + t.Fatalf("%s 上下文泄露了主人的 L4 意图记录（原话）: %q", tc.name, text) + } + if strings.Contains(text, "帮我搞一下那个部署") { + t.Fatalf("%s 上下文泄露了主人原话: %q", tc.name, text) + } + } + } + after, err := store.List(context.Background()) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(after) != len(before) { + t.Fatalf("受限会话写入了记忆: %d 条 -> %d 条（只读隔离被破坏）", len(before), len(after)) + } +} + +// seedIntentAndHabit 往共享域里放一条 L4 意图记录（主人原话）与一条 +// 普通用户习惯记忆：受限上下文只允许后者出现。 +func seedIntentAndHabit(t *testing.T, store *memory.Store) { + t.Helper() + seeds := []memory.MemoryItem{ + { + PersonaID: "persona-a", Scope: memory.ScopeUser, Kind: memory.KindRelationship, + Content: "intent:isTask=true;taskKind=ops;confidence=0.9;text=帮我搞一下那个部署", + }, + { + PersonaID: "persona-a", Scope: memory.ScopeUser, Kind: memory.KindRelationship, + Content: "用户习惯：喜欢手冲咖啡", + }, + } + for _, it := range seeds { + if _, err := store.Add(context.Background(), it); err != nil { + t.Fatalf("seed memory: %v", err) + } + } +} diff --git a/internal/services/gateway/chat_idempotency_test.go b/internal/services/gateway/chat_idempotency_test.go new file mode 100644 index 0000000..db3b4b0 --- /dev/null +++ b/internal/services/gateway/chat_idempotency_test.go @@ -0,0 +1,412 @@ +package gateway + +import ( + "encoding/json" + "net/http" + "testing" + "time" + + "github.com/gorilla/websocket" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" +) + +// 本文件锁定 Android 协同 §G 的两条：G2（REST `stream:true` 不得「静默成功」） +// 与 G4（chat 全链路幂等 —— clientMsgId）。 +// +// G2 的核心断言：**总线饱和（或不投递）时，经 REST `stream:true` 发起的对话 +// 仍然落库并产生帧**。它与 ws_route_chat_test.go 的 +// TestWSChatFrameSurvivesSaturatedBus 是同一条原则在另一条入口上的落地：会话 +// 消息不得依赖可丢弃的通道（design/15 §2.1）。 +// +// G4 的核心断言：同一 clientMsgId 发两次 → 只有一条用户消息、只有一个回合，且 +// 第二次的响应明确（既有 messageId）；省略该字段时行为与今天逐字一致。 + +// postChat 以 owner token POST 一次 /api/v1/chat，返回响应与原始响应体。 +func postChat(t *testing.T, url, token string, body map[string]any) (*http.Response, []byte) { + t.Helper() + return doReq(t, http.MethodPost, url+"/api/v1/chat", body, token) +} + +// dataOf 解出 {data:...} 信封里的 data 对象。 +func dataOf(t *testing.T, body []byte) map[string]any { + t.Helper() + raw, ok := decodeBody(t, body)["data"] + if !ok { + t.Fatalf("响应没有 data 信封：%s", body) + } + var out map[string]any + if err := json.Unmarshal(raw, &out); err != nil { + t.Fatalf("解码 data 失败：%v（body=%s）", err, body) + } + return out +} + +// errorCodeOf 解出 {error:{code}} 里的 code。 +func errorCodeOf(t *testing.T, body []byte) string { + t.Helper() + var eobj struct { + Code string `json:"code"` + } + if err := json.Unmarshal(decodeBody(t, body)["error"], &eobj); err != nil { + t.Fatalf("解码 error 信封失败：%v（body=%s）", err, body) + } + return eobj.Code +} + +// frameTypes 汇总见过的帧类型，供失败信息使用（读帧超时时能一眼看出漏了什么）。 +func chatFrameTypes(frames []wsEvent) []string { + out := make([]string, 0, len(frames)) + for _, f := range frames { + out = append(out, f.Type) + } + return out +} + +// readFramesUntil 在读到一个 want 类型的帧之前持续读帧（最多 5 秒），返回它 +// 见过的**全部**帧。它带读超时，因此断言失败时是明确的报错，而不是挂死。 +func readChatFramesUntil(t *testing.T, conn *websocket.Conn, want string) []wsEvent { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + seen := make([]wsEvent, 0, 8) + for time.Now().Before(deadline) { + _ = conn.SetReadDeadline(deadline) + _, msg, err := conn.ReadMessage() + if err != nil { + t.Fatalf("等待 %q 时读 WS 帧失败：%v（已见 %v）", want, err, chatFrameTypes(seen)) + } + var ev wsEvent + if err := json.Unmarshal(msg, &ev); err != nil { + t.Fatalf("解码帧 %s：%v", msg, err) + } + seen = append(seen, ev) + if ev.Type == want { + return seen + } + } + t.Fatalf("等待 %q 超时（已见 %v）", want, chatFrameTypes(seen)) + return nil +} + +// hasFrame 报告帧列表里是否出现过某类型。 +func hasFrame(frames []wsEvent, typ string) bool { + for _, f := range frames { + if f.Type == typ { + return true + } + } + return false +} + +// TestChatStreamSurvivesSaturatedBus 是 G2 的主断言：总线**完全不投递**（饱和 +// 丢弃）时，REST `stream:true` 依然驱动回合、落库用户消息与助手回复，并产出下行 +// 帧。修复前这一轮是 publishChatStream → bus.Publish → hub 订阅：饱和时事件被 +// 丢，客户端却早就拿到了 200 {"stream":true} —— 不落库、没有任何帧。 +func TestChatStreamSurvivesSaturatedBus(t *testing.T) { + t.Parallel() + run := &recordingRunner{} + hist := newReplayHistoryStore() + sb := &saturatedBus{} + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: sb}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + conn := dialWS(t, ts.URL, tok) + defer conn.Close() + waitConnCount(t, srv, 1) + + resp, body := postChat(t, ts.URL, tok, map[string]any{"sessionId": "web", "text": "总线饱和也要到", "stream": true}) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200（body=%s）", resp.StatusCode, body) + } + if got := dataOf(t, body); got["stream"] != true { + t.Fatalf("data = %v, want stream:true", got) + } + + // 200 回给客户端时用户消息已经落库（handler 同步穿过提交与落库），助手回复 + // 随回合结算到达。 + waitUntil(t, "回合被驱动", func() bool { return run.count() == 1 }) + waitUntil(t, "用户+助手消息落库", func() bool { return len(hist.records()) == 2 }) + recs := hist.records() + if recs[0].Role != "user" || recs[0].Content != "总线饱和也要到" { + t.Fatalf("用户消息 = %+v，want 刚发的那条", recs[0]) + } + if recs[1].Role != "assistant" || recs[1].Content == "" { + t.Fatalf("助手回复 = %+v，want 非空", recs[1]) + } + + // 帧：用户消息的 message_persisted 必然在 stream_end 之前出现过。 + frames := readChatFramesUntil(t, conn, "stream_end") + if !hasFrame(frames, "message_persisted") { + t.Fatalf("帧 = %v，want 至少一条 message_persisted（客户端要靠它追加消息）", chatFrameTypes(frames)) + } + + // 会话消息绝不经可丢弃的通道：ws.chat 主题上不应有任何发布。 + for _, subject := range sb.subjects() { + if subject == wsChatSubject { + t.Fatalf("REST stream:true 仍把会话消息发布到可丢弃的 bus 主题 %q（design/15 §2.1 禁止）", subject) + } + } +} + +// TestChatStreamWithoutRunnerIsNotSilentlyAccepted 钉住诚实的退化：没接 loop 的 +// 网关（runner 为 nil）**不能**对 stream:true 回 200 —— 修复前它照样回 200， +// 而客户端永远等不到任何帧、也永远不会落库。这里与同步路径同口径回 500。 +func TestChatStreamWithoutRunnerIsNotSilentlyAccepted(t *testing.T) { + t.Parallel() + env := newEnv(t, nil, false) + tok := env.token(t, auth.RoleOwner) + resp, body := postChat(t, env.ts.URL, tok, map[string]any{"sessionId": "web", "text": "hi", "stream": true}) + if resp.StatusCode != http.StatusInternalServerError { + t.Fatalf("status = %d, want 500（body=%s）", resp.StatusCode, body) + } + if code := errorCodeOf(t, body); code != "chat_unavailable" { + t.Fatalf("code = %q, want chat_unavailable", code) + } +} + +// TestChatStreamAnswersBusyAndFull 锁定 G2 的顺带项：`stream:true` 也能给出 +// 「排队 / 满」这两个明确的答案，而不是一律 200。（修复前 handler 提前 return， +// 202 queued 与 409 chat_busy 在这条路径上永远不可能出现。） +func TestChatStreamAnswersBusyAndFull(t *testing.T) { + t.Parallel() + release := make(chan struct{}) + run := &blockingRunner{release: release} + hist := newReplayHistoryStore() + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: &saturatedBus{}}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + // 第一条：会话空闲 → 200 stream:true，回合在后台跑着。 + resp, body := postChat(t, ts.URL, tok, map[string]any{"sessionId": "web", "text": "A", "stream": true}) + if resp.StatusCode != http.StatusOK { + t.Fatalf("第一条 status = %d, want 200（body=%s）", resp.StatusCode, body) + } + waitUntil(t, "A 回合在飞", func() bool { return run.started() == 1 }) + + // 第二条：会话正忙 → 202 {stream:true, queued:true, queueDepth:1}，不阻塞。 + resp, body = postChat(t, ts.URL, tok, map[string]any{"sessionId": "web", "text": "B", "stream": true}) + if resp.StatusCode != http.StatusAccepted { + t.Fatalf("第二条 status = %d, want 202（body=%s）", resp.StatusCode, body) + } + got := dataOf(t, body) + if got["queued"] != true || got["stream"] != true || got["queueDepth"] != float64(1) { + t.Fatalf("第二条 data = %v, want {stream:true,queued:true,queueDepth:1}", got) + } + + // 填满待处理队列（默认上限 5；已入队 1 条），下一条必须是 409 chat_busy。 + for i := 0; i < 4; i++ { + resp, body = postChat(t, ts.URL, tok, map[string]any{"sessionId": "web", "text": "fill", "stream": true}) + if resp.StatusCode != http.StatusAccepted { + t.Fatalf("第 %d 条填充 status = %d, want 202（body=%s）", i+3, resp.StatusCode, body) + } + } + resp, body = postChat(t, ts.URL, tok, map[string]any{"sessionId": "web", "text": "溢出", "stream": true}) + if resp.StatusCode != http.StatusConflict { + t.Fatalf("溢出 status = %d, want 409（body=%s）", resp.StatusCode, body) + } + if code := errorCodeOf(t, body); code != "chat_busy" { + t.Fatalf("溢出 code = %q, want chat_busy", code) + } + + close(release) + waitUntil(t, "会话排空", func() bool { + return !srv.hub.orchestrator.Busy("web") && srv.hub.orchestrator.QueueDepth("web") == 0 + }) +} + +// TestChatClientMsgIDDedupesTurn 是 G4 的主断言（REST 同步路径）：同一个 +// clientMsgId 发两次 → 只有一条用户消息、只有一个回合；第二次的响应明确给出既有 +// messageId，客户端据此对齐本地乐观气泡，而不是靠「文本相等」猜。 +func TestChatClientMsgIDDedupesTurn(t *testing.T) { + t.Parallel() + run := &countingRunner{} + hist := &fakeHistory{} + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: &saturatedBus{}}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + body := map[string]any{"sessionId": "web", "text": "弱网重发的那条", "clientMsgId": "local-bubble-1"} + + resp, first := postChat(t, ts.URL, tok, body) + if resp.StatusCode != http.StatusOK { + t.Fatalf("第一次 status = %d, want 200（body=%s）", resp.StatusCode, first) + } + + // 第二次：逐字相同的 clientMsgId → 明确判重，不落库、不起回合。 + resp, second := postChat(t, ts.URL, tok, body) + if resp.StatusCode != http.StatusOK { + t.Fatalf("第二次 status = %d, want 200（body=%s）", resp.StatusCode, second) + } + dup := dataOf(t, second) + if dup["duplicate"] != true { + t.Fatalf("第二次 data = %v, want duplicate:true", dup) + } + msgID, _ := dup["messageId"].(string) + if msgID == "" { + t.Fatalf("第二次 data = %v, want 既有 messageId（客户端要用它对齐气泡）", dup) + } + + // 一个回合、一条用户消息：模型只被驱动一次，库里只有 user+assistant 两条。 + if got := run.count(); got != 1 { + t.Fatalf("回合数 = %d, want 1（重复的 clientMsgId 绝不起第二个回合）", got) + } + recs := hist.records() + if len(recs) != 2 { + t.Fatalf("落库条数 = %d, want 2（一条用户 + 一条助手）", len(recs)) + } + if recs[0].Role != "user" || recs[0].ClientMsgID != "local-bubble-1" { + t.Fatalf("用户消息 = %+v，want clientMsgId=local-bubble-1 被回填", recs[0]) + } + if recs[0].ID != msgID { + t.Fatalf("响应 messageId = %q，want 等于落库那条用户消息的 id %q", msgID, recs[0].ID) + } +} + +// TestChatClientMsgIDDuplicateWhilePending 锁定幂等在**排队**分支上的行为：会话 +// 忙时，同一个 clientMsgId 的第二条必须立刻判重，而不是又排一次队、又落一条用户 +// 消息（重复到达的判定与队列分支无关，两者都在状态锁下的同一道门里）。 +func TestChatClientMsgIDDuplicateWhilePending(t *testing.T) { + t.Parallel() + release := make(chan struct{}) + run := &blockingRunner{release: release} + hist := &fakeHistory{} + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: &saturatedBus{}}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + // 第一条（不带幂等键）占住会话：stream 路径立刻返回，回合在后台跑着。 + resp, body := postChat(t, ts.URL, tok, map[string]any{"sessionId": "web", "text": "占住会话", "stream": true}) + if resp.StatusCode != http.StatusOK { + t.Fatalf("第一条 status = %d, want 200（body=%s）", resp.StatusCode, body) + } + waitUntil(t, "会话变忙", func() bool { return run.started() == 1 }) + + dup := map[string]any{"sessionId": "web", "text": "排队的那条", "clientMsgId": "cm-pending"} + resp, body = postChat(t, ts.URL, tok, dup) + if resp.StatusCode != http.StatusAccepted { + t.Fatalf("排队那条 status = %d, want 202（body=%s）", resp.StatusCode, body) + } + // 同键再来一次：判重，队列深度不再增长。 + resp, body = postChat(t, ts.URL, tok, dup) + if resp.StatusCode != http.StatusOK { + t.Fatalf("重复那条 status = %d, want 200（body=%s）", resp.StatusCode, body) + } + if got := dataOf(t, body); got["duplicate"] != true { + t.Fatalf("重复那条 data = %v, want duplicate:true", got) + } + if depth := srv.hub.orchestrator.QueueDepth("web"); depth != 1 { + t.Fatalf("待处理深度 = %d, want 1（重复的那条不得再排队）", depth) + } + withKey := 0 + for _, m := range hist.records() { + if m.ClientMsgID == "cm-pending" { + withKey++ + } + } + if withKey != 1 { + t.Fatalf("带该 clientMsgId 的用户消息 = %d 条, want 1", withKey) + } + + close(release) + waitUntil(t, "会话排空", func() bool { + return !srv.hub.orchestrator.Busy("web") && srv.hub.orchestrator.QueueDepth("web") == 0 + }) +} + +// TestChatWithoutClientMsgIDIsUnchanged 钉住 G4 的兼容口径：省略 clientMsgId 时 +// 行为与今天逐字一致 —— 同一段文本发两次就是两条消息、两个回合，响应里也没有 +// 任何 duplicate/messageId 字段（老客户端零影响）。 +func TestChatWithoutClientMsgIDIsUnchanged(t *testing.T) { + t.Parallel() + run := &countingRunner{} + hist := &fakeHistory{} + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: &saturatedBus{}}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + body := map[string]any{"sessionId": "web", "text": "同一段文本"} + + for i := 0; i < 2; i++ { + resp, raw := postChat(t, ts.URL, tok, body) + if resp.StatusCode != http.StatusOK { + t.Fatalf("第 %d 次 status = %d, want 200（body=%s）", i+1, resp.StatusCode, raw) + } + got := dataOf(t, raw) + if _, ok := got["duplicate"]; ok { + t.Fatalf("第 %d 次 data = %v，未带幂等键时不该有 duplicate", i+1, got) + } + if got["reply"] == nil { + t.Fatalf("第 %d 次 data = %v, want 正常回复契约", i+1, got) + } + } + if got := run.count(); got != 2 { + t.Fatalf("回合数 = %d, want 2（无幂等键时逐字保持旧行为）", got) + } + recs := hist.records() + if len(recs) != 4 { + t.Fatalf("落库条数 = %d, want 4（两次各一条用户 + 一条助手）", len(recs)) + } + for _, m := range recs { + if m.ClientMsgID != "" { + t.Fatalf("落库消息 = %+v，未带幂等键时 clientMsgId 必须为空", m) + } + } +} + +// TestWSChatClientMsgIDDuplicateIsExplicit 锁定 WS 侧的重复语义：同一 +// clientMsgId 的两条 chat 帧**并发到达**（两条帧各自进 goroutine）时，只有一个 +// 赢家 —— 一个回合、一条用户消息；输掉的那条收到明确的 `chat_duplicate` 帧， +// 其 messageId 指向真正落库的那条用户消息。 +func TestWSChatClientMsgIDDuplicateIsExplicit(t *testing.T) { + t.Parallel() + run := &recordingRunner{} + hist := newReplayHistoryStore() + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: &saturatedBus{}}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + conn := dialWS(t, ts.URL, tok) + defer conn.Close() + waitConnCount(t, srv, 1) + + payload := json.RawMessage(`{"sessionId":"web","text":"同一条","clientMsgId":"ws-bubble-1"}`) + for i := 0; i < 2; i++ { + if err := conn.WriteJSON(wsInbound{Type: "chat", Data: payload}); err != nil { + t.Fatalf("第 %d 条 chat 帧写入失败：%v", i+1, err) + } + } + + waitUntil(t, "只驱动一个回合", func() bool { return run.count() == 1 }) + waitUntil(t, "用户+助手消息落库", func() bool { return len(hist.records()) == 2 }) + + var userID string + for _, m := range hist.records() { + if m.Role == "user" { + if m.ClientMsgID != "ws-bubble-1" { + t.Fatalf("用户消息 = %+v，want clientMsgId 被回填", m) + } + userID = m.ID + } + } + if userID == "" { + t.Fatal("没有找到落库的用户消息") + } + + frames := readChatFramesUntil(t, conn, EventChatDuplicate) + var dupFrame *wsEvent + for i := range frames { + if frames[i].Type == EventChatDuplicate { + dupFrame = &frames[i] + } + } + if dupFrame == nil { + t.Fatalf("帧 = %v，want 一条 %s", chatFrameTypes(frames), EventChatDuplicate) + } + data, ok := dupFrame.Data.(map[string]any) + if !ok { + t.Fatalf("chat_duplicate 载荷 = %#v，want 对象", dupFrame.Data) + } + if data["clientMsgId"] != "ws-bubble-1" { + t.Fatalf("chat_duplicate 载荷 = %v，want clientMsgId=ws-bubble-1", data) + } + if data["messageId"] != userID { + t.Fatalf("chat_duplicate messageId = %v，want 落库那条的 id %q", data["messageId"], userID) + } + // 重复的那条绝不产生第二个回合。 + if got := run.count(); got != 1 { + t.Fatalf("回合数 = %d, want 1", got) + } +} diff --git a/internal/services/gateway/chat_stream_dispatch.go b/internal/services/gateway/chat_stream_dispatch.go new file mode 100644 index 0000000..f9fae74 --- /dev/null +++ b/internal/services/gateway/chat_stream_dispatch.go @@ -0,0 +1,129 @@ +package gateway + +import ( + "context" + "fmt" + "net/http" + "time" + + apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" +) + +// 本文件归属 REST `POST /api/v1/chat` 的 `stream:true` 派发（Android 协同 §G2）。 +// +// 修复前的链路是 publishChatStream → bus.Publish → hub 订阅 → streamChat，三处 +// 叠加起来让「客户端拿到 200、却什么都没发生」成为可能： +// ① bus 的订阅缓冲满时会**丢事件**（bus 包文档明示的代价），而会话消息被丢就是 +// 用户消息消失； +// ② handler 在问编排器之前就回了 200 {"stream":true}； +// ③ Publish 的错误被丢弃，且传的是**请求作用域**的 ctx（请求返回后即被取消， +// Publish 直接返回）。 +// +// 修复后它与 WS 的 chat 帧走**同一条直达路径**（ws_route.go 的既有先例， +// design/15 §2.1「会话消息不得被 drop」）：接了编排器就直接提交，完全不经总线。 +// handler 因此在回 200 之前就知道这一轮真的被接收并落库。只有在没接编排器时才 +// 回退到总线发布，而发布失败不再被吞掉 —— 回 503 + 明确错误码，绝不假装成功。 + +// chatStreamTurnBudget 是 stream:true 派生回合上下文的上限。该上下文必须与请求 +// 作用域**脱钩**：handler 返回后 r.Context() 即被取消，若回合挂在它下面，流式 +// 回合会在客户端刚收到 200 时被连带取消（回合是提交之后在后台跑的）。因此基础 +// 用 context.WithoutCancel（保留 trace/审计等值、丢掉取消信号），再套一个上限。 +// +// 上限取一个远大于任何正常回合的值：工具循环上限 8 轮（loop.DefaultMaxRounds）， +// 正常一轮是秒级到分钟级，30 分钟只在回合真的挂死时兜底 —— 它是安全网，不是 +// 功能限制（与 WS 路径的 context.Background() 相比，只是多了一个上界）。 +const chatStreamTurnBudget = 30 * time.Minute + +// handleChatStream 是 POST /api/v1/chat 的 stream:true 分支。它先把这一轮交给 +// submitChatStream（与 WS chat 帧同一条直达路径），拿到判定后才回响应，因此 +// **200 意味着消息已被接住并落库**；派发用的 ctx 与请求作用域脱钩、带上限 +// （见 detachedTurnContext）。 +func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request, req chatRequest) { + if s.runner == nil { + // 与同步路径同一口径：没接 loop 就不能假装受理了一轮对话（修复前这里 + // 照样回 200，客户端永远等不到任何帧）。 + writeError(w, apperr.Wrap(apperr.ErrInternal, "chat_unavailable", "chat service is not configured")) + return + } + sid := req.SessionID + if sid == "" { + sid = defaultSessionID + } + sr, err := s.submitChatStream(detachedTurnContext(r.Context()), sid, req.Text, req.Image, req.DeviceID, req.Mode, req.ClientMsgID) + if err != nil { + // 派发没成功：明确报 503，而不是让客户端以为这一轮在跑（协同 G2）。 + s.log.Error("chat stream dispatch", "err", err, "session", sid) + writeError(w, apperr.Wrap(apperr.ErrTimeout, "chat_dispatch_failed", "流式通道不可用，本轮未提交；请重试")) + return + } + s.writeChatStreamAck(w, sr, sid) +} + +// submitChatStream 把一次 stream:true 的对话交给编排器，并把它对本次提交的判定 +// 交回 HTTP 层（协同 G2）。error 非 nil 表示这一轮**没有**被接收，此时调用方 +// 绝不能回 2xx。 +func (s *Server) submitChatStream(ctx context.Context, sessionID, text, image, deviceID, mode, clientMsgID string) (SubmitResult, error) { + payload := map[string]any{ + "sessionId": sessionID, + "text": text, + "image": image, + "deviceId": deviceID, + "mode": mode, + "clientMsgId": clientMsgID, + } + if s.hub == nil || s.hub.orchestrator == nil { + // 未接编排器：回退到既有的总线触发路径（未接线的 hub 保持完全相同的 + // ws.chat 契约），但发布失败与「没有总线」都不再是静默成功。 + return s.publishChatStream(ctx, payload) + } + // 直达编排器：streamChat 同步返回提交判定（真正的回合跑在它自己派生的 + // goroutine 上），因此这里不会阻塞在模型调用上。 + return s.hub.streamChat(ctx, payload), nil +} + +// publishChatStream 是 stream:true 的兜底派发：把载荷发布到 ws.chat 主题，由 hub +// 的中继消费。它只在未接编排器时被用到，并且**把错误交回调用方** —— 修复前它吞掉 +// 了 Publish 的错误，于是总线关闭/ctx 已取消时客户端照样拿到 200。 +func (s *Server) publishChatStream(ctx context.Context, payload map[string]any) (SubmitResult, error) { + if s.bus == nil { + return SubmitResult{}, fmt.Errorf("gateway: chat stream has neither an orchestrator nor a bus: %w", apperr.ErrInternal) + } + if err := s.bus.Publish(ctx, wsChatSubject, payload); err != nil { + return SubmitResult{}, fmt.Errorf("gateway: publish %s: %w", wsChatSubject, err) + } + // 回退路径只能确认「已交给总线」，不能确认编排器收下了：如实按既有契约 + // 当已接受处理（2xx），并且已由调用方记日志。 + return SubmitResult{Accepted: true, Immediate: true}, nil +} + +// detachedTurnContext 返回一个与请求作用域脱钩、带上限的回合上下文（见 +// chatStreamTurnBudget）。它刻意**不**把 cancel 交出去：回合活过这次 HTTP 请求， +// 在这里取消父 ctx 会连坐取消在途回合 —— cancel 只是「提前释放计时器」的优化， +// 上限到点自会释放，因此显式丢弃并写明原因（vet 的 lostcancel 检查也要求把 +// cancel 用在某个地方）。 +func detachedTurnContext(parent context.Context) context.Context { + ctx, cancel := context.WithTimeout(context.WithoutCancel(parent), chatStreamTurnBudget) + _ = cancel + return ctx +} + +// writeChatStreamAck 把 stream:true 的提交判定写成响应（协同 G2 顺带项 + G4）。 +// 此前它无条件回 200 {"stream":true}，于是「排队满」与「重复到达」都只能靠客户端 +// 猜；现在四种判定各有明确的状态码与载荷： +// - 重复到达：200 {duplicate:true, messageId}（没有新回合、没有新帧，客户端用 +// messageId 把本地乐观气泡对齐到既有消息） +// - 待处理队列满：409 {error:{code:"chat_busy"}}（与同步路径同一口径） +// - 会话正忙、已续接排队：202 {stream:true, queued:true, queueDepth} +// - 已开始：200 {stream:true} +func (s *Server) writeChatStreamAck(w http.ResponseWriter, sr SubmitResult, sessionID string) { + switch { + case sr.Duplicate: + writeJSON(w, http.StatusOK, map[string]any{"duplicate": true, "messageId": sr.MessageID, "sessionId": sessionID}) + case sr.PendingFull: + writeError(w, apperr.Wrap(apperr.ErrConflict, "chat_busy", "会话正忙，待处理消息已满")) + case sr.Pending: + writeJSON(w, http.StatusAccepted, map[string]any{"stream": true, "queued": true, "queueDepth": sr.QueueDepth}) + default: + writeJSON(w, http.StatusOK, map[string]any{"stream": true}) + } +} diff --git a/internal/services/gateway/extra_test.go b/internal/services/gateway/extra_test.go index 23e1ec3..ffb82fc 100644 --- a/internal/services/gateway/extra_test.go +++ b/internal/services/gateway/extra_test.go @@ -336,7 +336,10 @@ func TestChatBadBody(t *testing.T) { func TestChatStreamMode(t *testing.T) { t.Parallel() - env := newEnv(t, nil, false) + // 接了 loop 的网关：`stream:true` 与 WS 的 chat 帧走同一条直达编排器的 + // 路径（协同 G2），因此它照旧回 200 {"stream":true}。未接 loop 时回 + // 500 chat_unavailable —— 见 TestChatStreamWithoutRunnerIsNotSilentlyAccepted。 + env := newEnv(t, &stubRunner{res: &loop.Result{Text: "ok", StopReason: "stop"}}, false) tok := env.token(t, auth.RoleOwner) resp, body := doReq(t, http.MethodPost, env.ts.URL+"/api/v1/chat", map[string]any{"sessionId": "s1", "text": "hi", "stream": true}, tok) if resp.StatusCode != http.StatusOK { diff --git a/internal/services/gateway/history.go b/internal/services/gateway/history.go index e332160..0886024 100644 --- a/internal/services/gateway/history.go +++ b/internal/services/gateway/history.go @@ -78,7 +78,12 @@ type historyMessage struct { // 气泡。为空时省略，而空意味着「全局默认角色」—— // 多角色之前的行读起来与以往完全一致。 PersonaID string `json:"personaId,omitempty"` - CreatedAt int64 `json:"createdAt"` + // ClientMsgID 是客户端自带的幂等键（Android 协同 §G4），原样回填， + // 因此客户端能把本地乐观气泡与服务端真相对齐（而不再靠「文本相等」 + // 这类脆弱匹配），也能在重连后从读接口收敛。它为空时省略，而空意味着 + // 「这条消息没有幂等键」（老客户端、IM 镜像、历史行）。 + ClientMsgID string `json:"clientMsgId,omitempty"` + CreatedAt int64 `json:"createdAt"` } // handleHistorySessions 先返回主会话，再返回每个子会话（最旧 @@ -120,7 +125,7 @@ func toHistoryLastMessage(m *history.MessageSummary) *historyLastMessage { // 新会话可选绑定的角色（design/22 §4.5）。它是设计的 // 优先级中「为新会话播种角色」的那一半：此处显式给出的角色 // 成为会话级绑定（唯一事实来源），而省略它则 -// 保持 persona_id = ''，使会话继承全局默认。 +// 保持 persona_id 为空串，使会话继承全局默认。 type createSubSessionRequest struct { PersonaID string `json:"personaId"` } @@ -312,7 +317,7 @@ func (s *Server) handleChannelPlatformLog(w http.ResponseWriter, r *http.Request // handleChannelMessages 为管理适配器日志页返回至多 limit 条消息， // 最新优先，覆盖三种粒度之一（design/16 §4.6 §4）：单个 // 适配器配置（?adapter=kook:3bef...）、整个平台（?platform=kook），或 -// 全部 IM 适配器总览（二者都不给，即 adapter_id != ''）。before/limit 复用 +// 全部 IM 适配器总览（二者都不给，即 adapter_id 非空）。before/limit 复用 // 标准游标分页。主 Web 会话上的 owner 私有镜像在此 // 不被排除，因为它携带 adapter_id；每种粒度直接按 // adapter_id / platform 过滤。 @@ -403,7 +408,11 @@ func toHistoryMessage(m history.Message) historyMessage { SenderName: m.SenderName, AdapterID: m.AdapterID, PersonaID: m.PersonaID, - CreatedAt: m.CreatedAt, + // ClientMsgID 一并回填：它正是客户端拿来做乐观气泡对齐的键 + // （协同 G4），读接口（GET /sessions/{id}/messages）与 + // `message_persisted` 帧因此给出同一个答案。 + ClientMsgID: m.ClientMsgID, + CreatedAt: m.CreatedAt, } } diff --git a/internal/services/gateway/rest.go b/internal/services/gateway/rest.go index 39d57a2..60f28a1 100644 --- a/internal/services/gateway/rest.go +++ b/internal/services/gateway/rest.go @@ -110,13 +110,18 @@ type loginResponse struct { // Mode 是客户端发送该回合时的行为模式（design/01 §3、design/16 // §5.1）；它可选，并按消息持久化以供气泡渲染，省略时 // 回退到 "chat"。 +// ClientMsgID 是客户端自带的幂等键（Android 协同 §G4）：它可选，同一会话内 +// 同一个非空值只承认第一条 —— 重复到达返回既有消息 id（`{duplicate:true, +// messageId}`）而不再落库、不再起回合，客户端据此把本地乐观气泡对齐到服务端 +// 真相（替代「靠文本相等」的脆弱匹配）。省略它时行为与加入该字段之前逐字相同。 type chatRequest struct { - SessionID string `json:"sessionId"` - Text string `json:"text"` - Image string `json:"image,omitempty"` - DeviceID string `json:"deviceId,omitempty"` - Mode string `json:"mode,omitempty"` - Stream bool `json:"stream"` + SessionID string `json:"sessionId"` + Text string `json:"text"` + Image string `json:"image,omitempty"` + DeviceID string `json:"deviceId,omitempty"` + Mode string `json:"mode,omitempty"` + ClientMsgID string `json:"clientMsgId,omitempty"` + Stream bool `json:"stream"` } // chatResponse 是同步聊天回复的载荷。 @@ -236,8 +241,9 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { return } if req.Stream { - s.publishChatStream(r.Context(), req.SessionID, req.Text, req.Image, req.DeviceID, req.Mode) - writeJSON(w, http.StatusOK, map[string]any{"stream": true}) + // 协同 G2：不再「先回 200，再看有没有落地」。与 WS 的 chat 帧同一条 + // 直达编排器的路径，由 handleChatStream 按判定回明确的状态码。 + s.handleChatStream(w, r, req) return } if s.runner == nil { @@ -252,8 +258,16 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { // 来自会话的绑定（design/22 §4.5）。 msgs := userChatMessages(req.Text, req.Image) env := s.envResolver.owner(r.Context(), req.SessionID) + // 客户端的 deviceID、行为模式与可选幂等键（协同 G4）都经 ctx 串进编排器。 + ctx := withClientMsgHint(withMode(withDeviceID(r.Context(), req.DeviceID), req.Mode), clientMsgHint{ClientMsgID: req.ClientMsgID}) if s.hub.orchestrator != nil { - sr, res := s.hub.orchestrator.SubmitSync(withMode(withDeviceID(r.Context(), req.DeviceID), req.Mode), req.SessionID, msgs, env) + sr, res := s.hub.orchestrator.SubmitSync(ctx, req.SessionID, msgs, env) + // 重复到达（同一个 clientMsgId）：明确告知既有消息 id，而不是再跑一遍 + // 回合、再落一条用户消息（协同 G4）。 + if sr.Duplicate { + writeJSON(w, http.StatusOK, map[string]any{"duplicate": true, "messageId": sr.MessageID, "sessionId": req.SessionID}) + return + } if !sr.Accepted { if sr.PendingFull { writeError(w, apperr.Wrap(apperr.ErrConflict, "chat_busy", "会话正忙，待处理消息已满")) diff --git a/internal/services/gateway/session_orchestrator.go b/internal/services/gateway/session_orchestrator.go index 3d3b484..967373c 100644 --- a/internal/services/gateway/session_orchestrator.go +++ b/internal/services/gateway/session_orchestrator.go @@ -52,6 +52,13 @@ const ( EventSessionIdle = "session_idle" EventSessionInterrupt = "session_interrupted" EventPendingFull = "pending_full" + // EventChatDuplicate 是一次重复到达的明确应答（G4 幂等）：本条 clientMsgId + // 已有落库的消息，本次没有新回合、也不会有新帧。它的 data 携带 + // {sessionId, clientMsgId, messageId}，让客户端把本地乐观气泡对齐到既有 + // 消息；没有发过这个 clientMsgId 的其他客户端自行忽略即可。它不是状态机 + // 迁移，因此不在 design/11 §7.1 的「状态信号」白名单里 —— 帧门默认只把它 + // 投给 owner/admin（fail-closed）。 + EventChatDuplicate = "chat_duplicate" ) const ( @@ -82,6 +89,15 @@ type SubmitResult struct { PendingFull bool // Empty 报告回合因既无文本也无图片而被拒绝。 Empty bool + // Duplicate 报告该回合被判为**重复到达**（G4 幂等）：同一个 clientMsgId + // 此前已经在本会话落过库，因此本次既不落库、也不启动回合。它是「接住了、 + // 但什么都没做」的诚实答案，绝不是拒绝 —— 客户端据此把本地乐观气泡对齐到 + // MessageID，而不必重试、也不会看到第二个回合。 + Duplicate bool + // MessageID 是本次提交对应的那条用户消息 id：正常路径是刚落库的消息 + // （带幂等键时是编排器预分配的那一个），重复路径是既有那条。省略 + // clientMsgId 的老客户端为 ""。 + MessageID string // Err 在同步路径上于运行出错时设置；REST 调用方 // 用它返回 5xx，而不是编造回复。 Err error @@ -129,6 +145,14 @@ type sessionState struct { // 回合以它为先驱，等本回合真正返回后才开始 —— 因此中断后立即到达的 // 消息绝不会在被取消回合落定其 checkpoint 之前读到那份旧快照。 fence turnFence + // clientMsgs 记录本会话已认领的 clientMsgId → 该消息的 id（G4 幂等）。 + // 它只在状态锁下读写，且只是**快路径 + 同进程并发仲裁**：跨重启的判重由 + // chat_messages 上 (session_id, client_msg_id) 的部分唯一索引裁决 + // （单二进制单进程，不存在跨进程并发）。见 session_persist.go。 + clientMsgs map[string]string + // clientMsgOrder 是 clientMsgs 的插入顺序，用于有界淘汰 + // （maxClientMsgsPerSession）：淘汰只丢快路径，存储仍是权威。 + clientMsgOrder []string } // pendingMsg 是一条已排队用户回合。它携带自己的 env，使续跑 @@ -228,20 +252,30 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes } deviceID := deviceIDFrom(ctx) msgMode := modeFrom(ctx) + // 两道门（队列满 / clientMsgId 重复）由 admitLocked 在锁内判定；它不广播， + // 状态事件在这里解锁之后再发。 + a := o.admitLocked(ctx, st, sessionID) + if a.full { + o.mu.Unlock() + o.emitState(sessionID, EventPendingFull, map[string]any{"sessionId": sessionID, "mode": a.mode, "max": o.maxPending}) + return SubmitResult{PendingFull: true}, nil + } + if a.dup { + o.mu.Unlock() + o.emitState(sessionID, EventChatDuplicate, map[string]any{"sessionId": sessionID, "clientMsgId": a.hint.ClientMsgID, "messageId": a.hint.MessageID}) + return SubmitResult{Accepted: true, Duplicate: true, MessageID: a.hint.MessageID}, nil + } + // 认领成功：把预分配的消息 id 盖进 ctx。空 clientMsgId 时它是零值，ctx 与 + // 加入该字段之前逐字相同。 + ctx = withClientMsgHint(ctx, a.hint) if st.mode != ModeIdle { - if len(st.pending) >= o.maxPending { - mode := st.mode - o.mu.Unlock() - o.emitState(sessionID, EventPendingFull, map[string]any{"sessionId": sessionID, "mode": mode, "max": o.maxPending}) - return SubmitResult{PendingFull: true}, nil - } st.pending = append(st.pending, pendingMsg{messages: messages, env: env, deviceID: deviceID, mode: msgMode, at: time.Now()}) depth := len(st.pending) mode := st.mode o.mu.Unlock() o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID) o.emitState(sessionID, EventPendingAck, map[string]any{"sessionId": sessionID, "mode": mode, "queueDepth": depth}) - return SubmitResult{Accepted: true, Pending: true, QueueDepth: depth}, nil + return SubmitResult{Accepted: true, Pending: true, QueueDepth: depth, MessageID: a.hint.MessageID}, nil } st.gen++ @@ -271,17 +305,25 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID) if sync { - res, err := o.runTurnFenced(runCtx, sessionID, turn, env, fence) - fence.release() - if err != nil { - o.log.Error("orchestrator turn", "err", err, "session", sessionID) - } - _ = o.completeTurn(sessionID, gen, res, err, deviceID, msgMode, env.PersonaID) - return SubmitResult{Accepted: true, Immediate: true, Err: err}, res + return o.runImmediateSync(runCtx, sessionID, gen, turn, env, fence, deviceID, msgMode, a.hint.MessageID) } go o.goTurn(sessionID, gen, turn, env, runCtx, cancel, deviceID, msgMode, fence) - return SubmitResult{Accepted: true, Immediate: true}, nil + return SubmitResult{Accepted: true, Immediate: true, MessageID: a.hint.MessageID}, nil +} + +// runImmediateSync 在**调用方** goroutine 上跑完刚启动的那个回合（REST 同步 +// 路径），结算它并把回合结果一并交回调用方。它与 goTurn 是同一轮的两个驱动 +// 方式（前台等待 / 后台续跑），共享 runTurnFenced + completeTurn 这一对，因此 +// 谁驱动都产生同样的持久化与状态迁移。 +func (o *Orchestrator) runImmediateSync(ctx context.Context, sessionID string, gen uint64, turn []model.Message, env loop.SessionEnv, fence turnFence, deviceID, mode, msgID string) (SubmitResult, *loop.Result) { + res, err := o.runTurnFenced(ctx, sessionID, turn, env, fence) + fence.release() + if err != nil { + o.log.Error("orchestrator turn", "err", err, "session", sessionID) + } + _ = o.completeTurn(sessionID, gen, res, err, deviceID, mode, env.PersonaID) + return SubmitResult{Accepted: true, Immediate: true, MessageID: msgID, Err: err}, res } // goTurn 在后台 goroutine 上运行单个回合，并在它结束时 diff --git a/internal/services/gateway/session_persist.go b/internal/services/gateway/session_persist.go index c0338f8..dadde26 100644 --- a/internal/services/gateway/session_persist.go +++ b/internal/services/gateway/session_persist.go @@ -2,6 +2,7 @@ package gateway import ( "context" + "errors" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" @@ -55,10 +56,15 @@ func modeFrom(ctx context.Context) string { // 之后切换角色的会话仍保有准确的「谁说了这个」流。未接线 // 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 @@ -67,11 +73,28 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string if text == "" && image == "" { continue } - full, err := o.history.AppendFull(ctx, history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image, DeviceID: deviceID, Mode: modeOrDefault(mode), PersonaID: personaID}) + 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) } } @@ -111,3 +134,189 @@ func messageTextImage(m model.Message) (text, image string) { } 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 + } + } +} diff --git a/internal/services/gateway/ws_hub_turn.go b/internal/services/gateway/ws_hub_turn.go index 37b6f83..ea0a524 100644 --- a/internal/services/gateway/ws_hub_turn.go +++ b/internal/services/gateway/ws_hub_turn.go @@ -45,14 +45,19 @@ func (h *hub) relay(ctx context.Context, e bus.Event) error { // orchestrator 的 hub 保持完全相同的 stream_chunk/stream_end 契约。 // 它跑在各自的 goroutine 上，因此慢的模型调用不会阻塞 bus 订阅 // worker。 -func (h *hub) streamChat(ctx context.Context, payload any) { +// +// 它把编排器对本次提交的判定**交回调用方**（协同 G2）：REST 的 stream:true +// 路径与 WS 帧共用这一段，前者据此回明确的 2xx/非 2xx/202/409，而不是在问 +// 编排器之前就先回 200 —— 那样客户端可能拿到 200 却什么都没发生。异步调用点 +// （`go h.streamChat(...)`，WS 帧与 bus 中继）照常丢弃返回值。 +func (h *hub) streamChat(ctx context.Context, payload any) SubmitResult { var req chatStreamRequest if b, err := json.Marshal(payload); err == nil { _ = json.Unmarshal(b, &req) } if h.runner == nil || (strings.TrimSpace(req.Text) == "" && strings.TrimSpace(req.Image) == "") { h.broadcast(wsEvent{Type: "stream_end", TurnID: newTurnID(), Seq: int(h.streamSeq.Add(1)), SessionID: req.SessionID, Data: map[string]any{"stopReason": "idle"}}) - return + return SubmitResult{} } if req.SessionID == "" { req.SessionID = "web" @@ -62,15 +67,17 @@ func (h *hub) streamChat(ctx context.Context, payload any) { // 切到角色 Y 的会话在下一轮以 Y 应答，而未绑定的/既有的会话继续 // 解析为全局默认角色。 msgs := userChatMessages(req.Text, req.Image) - // 通过 ctx 把客户端的 deviceID 与行为模式串进 orchestrator，使 - // 持久化的用户轮次与助手回复都带上标记（design/16 §4.5、§5.1）。 - ctx = withMode(withDeviceID(ctx, req.DeviceID), req.Mode) + // 通过 ctx 把客户端的 deviceID、行为模式与可选幂等键串进 + // orchestrator，使持久化的用户轮次与助手回复都带上标记，并让重复到达的 + // clientMsgId 被判重（design/16 §4.5、§5.1；协同 G4）。空 clientMsgId 是 + // 空操作，因此老客户端的行为逐字不变。 + ctx = withClientMsgHint(withMode(withDeviceID(ctx, req.DeviceID), req.Mode), clientMsgHint{ClientMsgID: req.ClientMsgID}) env := h.envResolver.owner(ctx, req.SessionID) if h.orchestrator != nil { - _ = h.orchestrator.Submit(ctx, req.SessionID, msgs, env) - return + return h.orchestrator.Submit(ctx, req.SessionID, msgs, env) } _, _ = h.runTurn(ctx, req.SessionID, msgs, env) + return SubmitResult{Accepted: true, Immediate: true} } // runTurn 执行一个 agent 轮次并广播其输出流。它是 orchestrator（经其 @@ -194,6 +201,11 @@ type chatStreamRequest struct { // Mode 是客户端发送该轮时所用的行为模式（design/01 §3、 // design/16 §5.1）；它可选，按消息持久化，缺省时回退为 "chat"。 Mode string `json:"mode,omitempty"` + // ClientMsgID 是客户端自带的幂等键（协同 G4）：同一会话内同一个非空 + // clientMsgId 只承认第一条，重复到达不再落库、不再起回合，而是回一条 + // `chat_duplicate`（data 带既有 messageId）。它可选，省略时行为与加入该 + // 字段之前逐字相同。 + ClientMsgID string `json:"clientMsgId,omitempty"` } // newTurnID 返回新的随机十六进制标识，用于把一串流式 WS 事件关联到 @@ -206,14 +218,3 @@ func newTurnID() string { } return "turn-" + time.Now().UTC().Format("20060102150405.000000000") } - -// publishChatStream 发布一个 ws.chat 流式触发器（REST 流式模式）。 -// deviceID 是发送方的浏览器/设备 id，mode 是客户端发送该轮时所使用的 -// 行为模式，二者都随载荷携带，使 ws.chat 消费方能给持久化的轮次打 -// 标记（design/16 §4.5、§5.1）。 -func (s *Server) publishChatStream(ctx context.Context, sessionID, text, image, deviceID, mode string) { - if s.bus == nil { - return - } - _ = s.bus.Publish(ctx, wsChatSubject, map[string]any{"sessionId": sessionID, "text": text, "image": image, "deviceId": deviceID, "mode": mode}) -}