From 5cd67be98203cc9e63ac34a210ca590f8c5b6877 Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 11 Oct 2026 04:37:50 +0000 Subject: [PATCH] Close request atomicity: locked message append, card versions, atomic rate limit, recovery holds (U3 slice 1) Why: the approved plan U3 (KTD3/KTD7/KTD10 plus minimal recovery isolation) requires concurrent submits to not silently overwrite, unknown-side-effect runs to never auto-resume, and the shared rate limiter to degrade conservatively instead of failing open. Evidence: full backend suite green twice-relevant paths against real PostgreSQL 18.6 and Redis (isolated instance) and on SQLite; new tests - card_version_test (concurrent single winner), recovery_hold_test + recovery_holds_test (skip/release/fail-closed CRUD), ratelimit atomic-window and degraded-fallback tests. Impact: AppendMessageWithRequest now allocates seq under the conversation advisory lock in one transaction; activity_cards gain version (v19) with SaveCardVersioned/GetCardVersion/CardVersion and cardsvc user mutations (status/pin/action/resolve-by-ref) return 409-mapped conflicts while system mirrors keep blind writes; recovery_holds (v20) with admin list/release endpoints and fail-closed resume skip; Redis limiter uses one Lua INCR+PEXPIRE script and falls back to a per-instance limit with a Degraded counter. Refs: proposals/2026-10-11-collaboration-storage-and-distribution.md U3 KTD3/KTD7/KTD10; R3/R6/R7/R8. --- internal/cardsvc/cardsvc.go | 30 ++++-- internal/ratelimit/ratelimit.go | 29 ++++-- internal/ratelimit/ratelimit_test.go | 47 ++++++++++ internal/runmgr/recovery_hold_test.go | 122 +++++++++++++++++++++++++ internal/runmgr/subagent_transcript.go | 10 ++ internal/server/accounts.go | 25 +++++ internal/server/server.go | 4 +- internal/store/card_version_test.go | 97 ++++++++++++++++++++ internal/store/cards.go | 62 ++++++++++++- internal/store/messages.go | 28 +++++- internal/store/migrate.go | 42 +++++++++ internal/store/recovery_holds.go | 85 +++++++++++++++++ internal/store/recovery_holds_test.go | 53 +++++++++++ internal/store/store_test.go | 4 +- 14 files changed, 612 insertions(+), 26 deletions(-) create mode 100644 internal/runmgr/recovery_hold_test.go create mode 100644 internal/store/card_version_test.go create mode 100644 internal/store/recovery_holds.go create mode 100644 internal/store/recovery_holds_test.go diff --git a/internal/cardsvc/cardsvc.go b/internal/cardsvc/cardsvc.go index 66f867b..a17b6dc 100644 --- a/internal/cardsvc/cardsvc.go +++ b/internal/cardsvc/cardsvc.go @@ -95,6 +95,7 @@ func (s *Service) List(owner, conversationID string) ([]core.ActivityCard, error } // Update 保存整卡变更（流式 Body 增量等）并发布 card.updated；ref 保持原值。 +// 系统镜像路径（同会话 run 的顺序行为）沿用盲写；用户态裁决走 UpdateVersioned。 func (s *Service) Update(owner string, card core.ActivityCard) error { card.UpdatedAt = s.now().UTC() if err := s.st.SaveCard(owner, &card, ""); err != nil { @@ -103,6 +104,16 @@ func (s *Service) Update(owner string, card core.ActivityCard) error { return s.emitCard(card, core.EvCardUpdated, core.CardPayload{Card: card}) } +// UpdateVersioned 乐观并发保存（U3，KTD7）：版本不匹配透传 store.ErrConflict， +// HTTP 层映射 409；客户端重读卡后重试。 +func (s *Service) UpdateVersioned(owner string, card core.ActivityCard, expectedVersion int) error { + card.UpdatedAt = s.now().UTC() + if err := s.st.SaveCardVersioned(owner, &card, "", expectedVersion); err != nil { + return err + } + return s.emitCard(card, core.EvCardUpdated, core.CardPayload{Card: card}) +} + // Append 流式追加正文（card_append 工具；D45）：文本并入 body 尾部 markdown // 节点（尾节点非 markdown 则新增），逐次持久化；card.updated 按 cardAppendWindow // 节流——窗口内的追加只落库不发布（投影语义：事件携带整卡，后续事件自然覆盖 @@ -141,7 +152,7 @@ func (s *Service) Append(owner, id, text string) error { // SetStatus 状态迁移：更新并发布 card.status。 func (s *Service) SetStatus(owner, id string, status core.CardStatus, tone, note string) (core.ActivityCard, error) { - c, err := s.st.GetCard(owner, id) + c, ver, err := s.st.GetCardVersion(owner, id) if err != nil { return c, err } @@ -149,7 +160,7 @@ func (s *Service) SetStatus(owner, id string, status core.CardStatus, tone, note if tone != "" { c.Tone = tone } - if err := s.Update(owner, c); err != nil { + if err := s.UpdateVersioned(owner, c, ver); err != nil { return c, err } return c, s.emitCard(c, core.EvCardStatus, core.CardStatusPayload{CardID: c.ID, Status: status, Tone: c.Tone, Note: note}) @@ -157,12 +168,12 @@ func (s *Service) SetStatus(owner, id string, status core.CardStatus, tone, note // SetPinned 置顶/取消（收件箱"不沉没"语义）。 func (s *Service) SetPinned(owner, id string, pinned bool) (core.ActivityCard, error) { - c, err := s.st.GetCard(owner, id) + c, ver, err := s.st.GetCardVersion(owner, id) if err != nil { return c, err } c.Pinned = pinned - return c, s.Update(owner, c) + return c, s.UpdateVersioned(owner, c, ver) } // Dismiss 关闭卡（非确认语义，留痕）。 @@ -173,7 +184,7 @@ func (s *Service) Dismiss(owner, id string) (core.ActivityCard, error) { // Action 处理卡上按钮点击：仅 action_required 卡可裁决；结果 resolved 并发布 // card.resolved（ActionID 即按钮语义）。 func (s *Service) Action(owner, id, actionID, freeText string) (core.ActivityCard, string, error) { - c, err := s.st.GetCard(owner, id) + c, ver, err := s.st.GetCardVersion(owner, id) if err != nil { return c, "", err } @@ -197,7 +208,7 @@ func (s *Service) Action(owner, id, actionID, freeText string) (core.ActivityCar return c, "", ErrNoAction } c.Status = core.CardResolved - if err := s.Update(owner, c); err != nil { + if err := s.UpdateVersioned(owner, c, ver); err != nil { return c, "", err } return c, label, s.emitCard(c, core.EvCardResolved, core.CardResolvedPayload{CardID: c.ID, ActionID: actionID, Answer: freeText}) @@ -250,7 +261,7 @@ func (s *Service) ResolveByRef(owner, ref, actionID, answer string) error { } return err } - c, err := s.st.GetCard(owner, id) + c, ver, err := s.st.GetCardVersion(owner, id) if err != nil { return err } @@ -258,7 +269,10 @@ func (s *Service) ResolveByRef(owner, ref, actionID, answer string) error { return nil // 已决（双现任一侧先到） } c.Status = core.CardResolved - if err := s.Update(owner, c); err != nil { + if err := s.UpdateVersioned(owner, c, ver); err != nil { + if errors.Is(err, store.ErrConflict) { + return nil // 双现任一侧先到（版本竞争败者）：不重试，卡以先到侧为准 + } return err } if c.ConversationID != "" && s.emit != nil { diff --git a/internal/ratelimit/ratelimit.go b/internal/ratelimit/ratelimit.go index a22331a..2636459 100644 --- a/internal/ratelimit/ratelimit.go +++ b/internal/ratelimit/ratelimit.go @@ -6,6 +6,7 @@ package ratelimit import ( "context" "sync" + "sync/atomic" "time" "github.com/redis/go-redis/v9" @@ -52,14 +53,26 @@ func (m *Memory) Allow(_ context.Context, key string, limit int, window time.Dur return true } -// ---- Redis 固定窗口（INCR + EXPIRE；首击设窗）---- +// ---- Redis 固定窗口（Lua 原子 INCR+PEXPIRE；故障降级进程内保守限流）---- + +// incrExpire 原子计数并只在首击设窗（协作存储 U3，KTD10）：两步 INCR/EXPIRE +// 在 EXPIRE 失败或进程于其间退出时会留下永不过期的键；Lua 单脚本消除该窗口。 +var incrExpire = redis.NewScript(` +local n = redis.call('INCR', KEYS[1]) +if n == 1 then redis.call('PEXPIRE', KEYS[1], ARGV[1]) end +return n +`) type Redis struct { rdb *redis.Client + // fallback：Redis 故障时的进程内保守限流（同限值、按实例计数——是共享 + // 限流的近似而非替代；昂贵入口的硬保护在 PG admission，不依赖本层）。 + fallback *Memory + degraded atomic.Int64 } func NewRedis(addr, password string, db int) *Redis { - return &Redis{rdb: redis.NewClient(&redis.Options{Addr: addr, Password: password, DB: db})} + return &Redis{rdb: redis.NewClient(&redis.Options{Addr: addr, Password: password, DB: db}), fallback: NewMemory()} } func (r *Redis) Allow(ctx context.Context, key string, limit int, window time.Duration) bool { @@ -67,16 +80,18 @@ func (r *Redis) Allow(ctx context.Context, key string, limit int, window time.Du return true } rkey := "rl:" + key + ":" + time.Now().UTC().Truncate(window).Format("150405") - n, err := r.rdb.Incr(ctx, rkey).Result() + n, err := incrExpire.Run(ctx, r.rdb, []string{rkey}, window.Milliseconds()).Int64() if err != nil { - return true // Redis 故障不阻断请求（限流是柔性行为，放行并记日志于调用方） - } - if n == 1 { - r.rdb.Expire(ctx, rkey, window+time.Second) + // 降级（KTD10）：不再无界放行；按实例保守限流并计数供观测告警。 + r.degraded.Add(1) + return r.fallback.Allow(ctx, key, limit, window) } return n <= int64(limit) } +// Degraded 返回累计降级次数（观测：非零说明 Redis 不可用，需运维介入）。 +func (r *Redis) Degraded() int64 { return r.degraded.Load() } + func (r *Redis) Ping(ctx context.Context) error { return r.rdb.Ping(ctx).Err() } func (r *Redis) Close() error { return r.rdb.Close() } diff --git a/internal/ratelimit/ratelimit_test.go b/internal/ratelimit/ratelimit_test.go index 94d6a95..f78fdf1 100644 --- a/internal/ratelimit/ratelimit_test.go +++ b/internal/ratelimit/ratelimit_test.go @@ -55,3 +55,50 @@ func TestRedisWindow(t *testing.T) { t.Fatal("3rd request must be limited") } } + +// Lua 原子窗口（U3，KTD10）：计数与设窗单脚本完成，键必然带 TTL—— +// 窗口翻转后重新放行（原子性回归：旧两步实现窗口语义相同，此处锚定 +// 「INCR 与 PEXPIRE 不可分」的行为面）。 +func TestRedisWindowAtomicScript(t *testing.T) { + addr := os.Getenv("AGENT_REDIS_TEST_ADDR") + if addr == "" { + t.Skip("AGENT_REDIS_TEST_ADDR not set") + } + r := NewRedis(addr, "", 15) + defer r.Close() + ctx := context.Background() + key := "atomic-" + time.Now().Format("150405.000000000") + window := 300 * time.Millisecond + for i := 1; i <= 3; i++ { + if !r.Allow(ctx, key, 3, window) { + t.Fatalf("request %d should pass", i) + } + } + if r.Allow(ctx, key, 3, window) { + t.Fatal("4th request must be limited") + } + time.Sleep(window + 50*time.Millisecond) + if !r.Allow(ctx, key, 3, window) { + t.Fatal("new window must allow again") + } +} + +// Redis 故障降级（KTD10）：不可达时不再无界放行——按实例保守限流； +// Degraded 计数供观测告警。 +func TestRedisDegradedFallback(t *testing.T) { + r := NewRedis("127.0.0.1:1", "", 15) // 不可达端口 + defer r.Close() + ctx := context.Background() + key := "degraded-" + time.Now().Format("150405.000000000") + for i := 1; i <= 2; i++ { + if !r.Allow(ctx, key, 2, time.Minute) { + t.Fatalf("degraded request %d should pass (local limit 2)", i) + } + } + if r.Allow(ctx, key, 2, time.Minute) { + t.Fatal("degraded 3rd request must be limited locally") + } + if r.Degraded() == 0 { + t.Fatal("degradation must be observable") + } +} diff --git a/internal/runmgr/recovery_hold_test.go b/internal/runmgr/recovery_hold_test.go new file mode 100644 index 0000000..866de78 --- /dev/null +++ b/internal/runmgr/recovery_hold_test.go @@ -0,0 +1,122 @@ +package runmgr + +import ( + "os" + "path/filepath" + "testing" + "time" + + "agent-system-backend/internal/store" + core "agent-system-core" +) + +// 协作存储 U3：恢复隔离（recovery hold）。未知外部副作用的 run 不得被启动 +// 续跑自动重派；人工核对后显式解除才恢复。判定失败同样跳过（fail-closed）。 + +// hold 中的子 run：续跑跳过；解除后续跑恢复。 +func TestRecoveryHoldBlocksResumeUntilReleased(t *testing.T) { + runner := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "续跑结果"}}} + c, st := newSpawnEnv(t, runner) + c.mgr.runner = noopRunner{} + ws := t.TempDir() + c.mgr.SetWorkspaceRoot(ws) + + parentRun := c.run + if err := st.CreateRun(&parentRun); err != nil { + t.Fatal(err) + } + child := core.Run{ + ID: store.NewID("run"), ParentRunID: c.run.ID, ConversationID: c.run.ConversationID, + ProjectID: c.run.ProjectID, Owner: c.run.Owner, Kind: core.RunSubagent, ProfileID: c.run.ProfileID, + Status: core.RunRunning, CreatedAt: time.Now().UTC(), + } + if err := st.CreateRun(&child); err != nil { + t.Fatal(err) + } + c.mgr.writeSubagentRequest(child, "agent.sub.explore", "检索", "原始任务", "原始提示", true) + if _, err := st.SealInterrupted(child.ID); err != nil { + t.Fatal(err) + } + + before, err := st.SubagentRunIDs(c.run.ConversationID) + if err != nil { + t.Fatal(err) + } + + // run 级 hold：跳过。 + if err := st.SetRecoveryHold(child.ID, child.Owner, string(child.ConversationID), "shell 命令超时，副作用结果未知"); err != nil { + t.Fatal(err) + } + c.mgr.ResumeInterruptedChildren() + ids, err := st.SubagentRunIDs(c.run.ConversationID) + if err != nil { + t.Fatal(err) + } + if len(ids) != len(before) { + t.Fatalf("hold 未解除时续跑应跳过：before=%v after=%v", before, ids) + } + // resumed.json 不得写出（下次解除后仍可续跑）。 + dir := c.mgr.subagentDir(child.Owner, child.ConversationID, child.ID) + if _, err := os.Stat(filepath.Join(dir, "resumed.json")); !os.IsNotExist(err) { + t.Fatalf("hold 跳过不应写 resumed.json：%v", err) + } + holds, err := st.ListOpenRecoveryHolds() + if err != nil || len(holds) != 1 || holds[0].RunID != child.ID { + t.Fatalf("open holds=%v err=%v", holds, err) + } + + // 解除（人工核对后）：续跑恢复。 + if err := st.ReleaseRecoveryHold(child.ID); err != nil { + t.Fatal(err) + } + c.mgr.ResumeInterruptedChildren() + ids, err = st.SubagentRunIDs(c.run.ConversationID) + if err != nil { + t.Fatal(err) + } + if len(ids) != len(before)+1 { + t.Fatalf("解除后续跑应新增子 run：before=%v after=%v", before, ids) + } + if held, herr := st.HasOpenRecoveryHold(child.ID, string(child.ConversationID)); held || herr != nil { + t.Fatalf("解除后不应再有 hold：held=%v err=%v", held, herr) + } +} + +// 会话级 hold：同会话的其他中断子 run 一并跳过（副作用疑点未核对前， +// 整个会话的自动重派都冻结）。 +func TestRecoveryHoldConversationScope(t *testing.T) { + runner := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "续跑结果"}}} + c, st := newSpawnEnv(t, runner) + c.mgr.runner = noopRunner{} + ws := t.TempDir() + c.mgr.SetWorkspaceRoot(ws) + + parentRun := c.run + if err := st.CreateRun(&parentRun); err != nil { + t.Fatal(err) + } + child := core.Run{ + ID: store.NewID("run"), ParentRunID: c.run.ID, ConversationID: c.run.ConversationID, + ProjectID: c.run.ProjectID, Owner: c.run.Owner, Kind: core.RunSubagent, ProfileID: c.run.ProfileID, + Status: core.RunRunning, CreatedAt: time.Now().UTC(), + } + if err := st.CreateRun(&child); err != nil { + t.Fatal(err) + } + c.mgr.writeSubagentRequest(child, "agent.sub.explore", "检索", "原始任务", "原始提示", true) + if _, err := st.SealInterrupted(child.ID); err != nil { + t.Fatal(err) + } + + before, _ := st.SubagentRunIDs(c.run.ConversationID) + // hold 挂在「另一个」run 上、但同会话。 + other := string(c.run.ConversationID) + if err := st.SetRecoveryHold("run_other", c.run.Owner, other, "会话存在未知副作用，整体冻结"); err != nil { + t.Fatal(err) + } + c.mgr.ResumeInterruptedChildren() + ids, _ := st.SubagentRunIDs(c.run.ConversationID) + if len(ids) != len(before) { + t.Fatalf("会话级 hold 应跳过续跑：before=%v after=%v", before, ids) + } +} diff --git a/internal/runmgr/subagent_transcript.go b/internal/runmgr/subagent_transcript.go index 6370ce2..d83a262 100644 --- a/internal/runmgr/subagent_transcript.go +++ b/internal/runmgr/subagent_transcript.go @@ -128,6 +128,16 @@ func (m *Manager) ResumeInterruptedChildren() { return } for _, run := range runs { + // 恢复隔离（协作存储 U3）：未知外部副作用的 run 不自动续跑——hold + // 未解除即跳过；判定出错同样跳过（fail-closed：宁可人工核对，不盲重派）。 + if held, herr := m.st.HasOpenRecoveryHold(string(run.ID), string(run.ConversationID)); herr != nil || held { + if herr != nil { + log.Printf("[runmgr] resume skip %s: recovery hold 判定失败（fail-closed）: %v", run.ID, herr) + } else { + log.Printf("[runmgr] resume skip %s: recovery hold 未解除（人工核对后显式释放才会续跑）", run.ID) + } + continue + } dir := m.subagentDir(run.Owner, run.ConversationID, run.ID) if dir == "" { continue diff --git a/internal/server/accounts.go b/internal/server/accounts.go index 92688db..6bd42b1 100644 --- a/internal/server/accounts.go +++ b/internal/server/accounts.go @@ -279,3 +279,28 @@ func RandomPassword() string { rand.Read(raw) return hex.EncodeToString(raw) } + +// ---- 管理员：恢复隔离的核对与解除（协作存储 U3）---- + +// listRecoveryHolds 列出未解除的 recovery hold（人工核对的运维入口）。 +func (s *Server) listRecoveryHolds(w http.ResponseWriter, r *http.Request) { + if _, ok := s.requireAdmin(w, r); !ok { + return + } + holds, err := s.st.ListOpenRecoveryHolds() + writeJSONOr(w, map[string]any{"holds": holds}, err) +} + +// releaseRecoveryHold 人工解除单条 hold（幂等）；解除后该 run 才可能被再次 +// 续跑/接管。副作用未核对就解除的风险由操作者承担——与 task-close 同级谨慎。 +func (s *Server) releaseRecoveryHold(w http.ResponseWriter, r *http.Request) { + if _, ok := s.requireAdmin(w, r); !ok { + return + } + runID := r.PathValue("runId") + if _, err := s.st.GetRecoveryHold(runID); err != nil { + writeStoreErr(w, err) + return + } + writeOKOr(w, s.st.ReleaseRecoveryHold(runID)) +} diff --git a/internal/server/server.go b/internal/server/server.go index ec00459..b8d9332 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -122,7 +122,7 @@ func (s *Server) WireBuiltinCommands() { return map[string]any{"reply": fmt.Sprintf("可用命令（注册表 %d 条）——/new 新会话 · /mock on|off 演示模式 · /model <名称> 设置主模型 · /profile 切换画像 · /help 帮助", len(b.List()))}, nil })) must(b.RegisterBuiltin(agentcommand.CommandSpec{ID: "system.intent", Title: "意图路由", - Description: "G2-2 自然语言入口：文本匹配已声明意图后经命令总线分发（同审计信封）；未命中返回候选清单", + Description: "G2-2 自然语言入口：文本匹配已声明意图后经命令总线分发（同审计信封）；未命中返回候选清单", ParamsSchema: json.RawMessage(`{"type":"object","required":["text"],"properties":{"text":{"type":"string"}}}`)}, "/intent", func(ctx context.Context, req cmdbus.Request) (any, error) { text, _ := req.Params["text"].(string) @@ -303,6 +303,8 @@ func (s *Server) Handler() http.Handler { // ---- 管理（嵌入式档在线备份）---- mux.HandleFunc("POST /api/admin/backup", s.backup) + mux.HandleFunc("GET /api/admin/recovery-holds", s.listRecoveryHolds) + mux.HandleFunc("POST /api/admin/recovery-holds/{runId}/release", s.releaseRecoveryHold) // API 未匹配路径一律 404（不回落 SPA，避免 /api/* 返回 HTML 200）。 mux.HandleFunc("/api/", func(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) }) diff --git a/internal/store/card_version_test.go b/internal/store/card_version_test.go new file mode 100644 index 0000000..7000af5 --- /dev/null +++ b/internal/store/card_version_test.go @@ -0,0 +1,97 @@ +package store + +import ( + "errors" + "sync" + "testing" + "time" + + core "agent-system-core" +) + +// 协作存储 U3（KTD7）：卡片乐观并发。同版本竞争写入最多一个成功， +// 其余明确 ErrConflict（HTTP 层映射 409）；盲写（系统镜像）递增版本。 + +func newVersionedCard(t *testing.T) (core.ActivityCard, string) { + t.Helper() + return core.ActivityCard{ + ID: NewID("card"), Title: "版本卡", Status: core.CardActive, + CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC(), + }, "o" +} + +func TestCardVersionRoundTrip(t *testing.T) { + st := openTestStore(t) + c, owner := newVersionedCard(t) + if err := st.SaveCard(owner, &c, ""); err != nil { + t.Fatal(err) + } + v, err := st.CardVersion(owner, c.ID) + if err != nil || v != 1 { + t.Fatalf("建卡版本应为 1：v=%d err=%v", v, err) + } + // 乐观更新：expected 一致 → 成功且递增。 + if err := st.SaveCardVersioned(owner, &c, "", 1); err != nil { + t.Fatal(err) + } + if v, _ = st.CardVersion(owner, c.ID); v != 2 { + t.Fatalf("版本应递增到 2：%d", v) + } + // 旧版本重放：明确冲突。 + err = st.SaveCardVersioned(owner, &c, "", 1) + if !errors.Is(err, ErrConflict) { + t.Fatalf("旧版本应 ErrConflict：%v", err) + } + // 盲写同样递增（系统镜像路径）。 + if err := st.SaveCard(owner, &c, ""); err != nil { + t.Fatal(err) + } + if v, _ = st.CardVersion(owner, c.ID); v != 3 { + t.Fatalf("盲写应递增到 3：%d", v) + } +} + +// 并发同版本竞争：N 个写者同 expected 提交，恰一个成功，其余 ErrConflict； +// 不存在静默覆盖（R3 验收）。 +func TestCardVersionConcurrentSingleWinner(t *testing.T) { + st := openTestStore(t) + c, owner := newVersionedCard(t) + if err := st.SaveCard(owner, &c, ""); err != nil { + t.Fatal(err) + } + const writers = 8 + var mu sync.Mutex + ok, conflict := 0, 0 + var wg sync.WaitGroup + for i := 0; i < writers; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + dup := c + dup.Title = "并发写" + err := st.SaveCardVersioned(owner, &dup, "", 1) + mu.Lock() + defer mu.Unlock() + switch { + case err == nil: + ok++ + case errors.Is(err, ErrConflict): + conflict++ + default: + t.Errorf("意外错误：%v", err) + } + }(i) + } + wg.Wait() + if ok != 1 || conflict != writers-1 { + t.Fatalf("应恰一个成功：ok=%d conflict=%d", ok, conflict) + } + if v, _ := st.CardVersion(owner, c.ID); v != 2 { + t.Fatalf("竞争后版本应为 2：%d", v) + } + // cardsvc 读侧：GetCardVersion 返回卡与版本。 + _, ver, err := st.GetCardVersion(owner, c.ID) + if err != nil || ver != 2 { + t.Fatalf("GetCardVersion：ver=%d err=%v", ver, err) + } +} diff --git a/internal/store/cards.go b/internal/store/cards.go index 53078f1..49d13f3 100644 --- a/internal/store/cards.go +++ b/internal/store/cards.go @@ -19,9 +19,9 @@ func (s *Store) SaveCard(owner string, c *core.ActivityCard, ref string) error { if err != nil { return fmt.Errorf("save card %s: %w", c.ID, err) } - _, err = s.db.Exec(s.q(`INSERT INTO activity_cards (id, project_id, conversation_id, ref, owner, payload, status, pinned, created_at, updated_at) - VALUES (?,?,?,?,?,?,?,?,?,?) - ON CONFLICT(id) DO UPDATE SET payload=excluded.payload, status=excluded.status, pinned=excluded.pinned, ref=CASE WHEN excluded.ref='' THEN activity_cards.ref ELSE excluded.ref END, updated_at=excluded.updated_at`), + _, err = s.db.Exec(s.q(`INSERT INTO activity_cards (id, project_id, conversation_id, ref, owner, payload, status, pinned, created_at, updated_at, version) + VALUES (?,?,?,?,?,?,?,?,?,?,1) + ON CONFLICT(id) DO UPDATE SET payload=excluded.payload, status=excluded.status, pinned=excluded.pinned, ref=CASE WHEN excluded.ref='' THEN activity_cards.ref ELSE excluded.ref END, updated_at=excluded.updated_at, version=activity_cards.version+1`), c.ID, c.ProjectID, c.ConversationID, ref, owner, string(payload), string(c.Status), boolInt(c.Pinned), c.CreatedAt.Format(time.RFC3339Nano), c.UpdatedAt.Format(time.RFC3339Nano)) if err != nil { return fmt.Errorf("save card %s: %w", c.ID, err) @@ -29,6 +29,62 @@ func (s *Store) SaveCard(owner string, c *core.ActivityCard, ref string) error { return nil } +// SaveCardVersioned 乐观并发保存（U3，KTD7）：行存在且 expected 与当前 +// version 一致才应用并递增，否则 ErrConflict（调用方回 409 让客户端重读 +// 重试）；行不存在时按新卡插入 version=1（expected==0 的建卡语义）。 +// 现产品没有卡片删除路径，expected>0 而行缺失不可达；若未来引入删除， +// 此处须补「插入回滚为冲突」的语义。 +func (s *Store) SaveCardVersioned(owner string, c *core.ActivityCard, ref string, expected int) error { + payload, err := json.Marshal(c) + if err != nil { + return fmt.Errorf("save card %s: %w", c.ID, err) + } + res, err := s.db.Exec(s.q(`INSERT INTO activity_cards (id, project_id, conversation_id, ref, owner, payload, status, pinned, created_at, updated_at, version) + VALUES (?,?,?,?,?,?,?,?,?,?,1) + ON CONFLICT(id) DO UPDATE SET payload=excluded.payload, status=excluded.status, pinned=excluded.pinned, ref=CASE WHEN excluded.ref='' THEN activity_cards.ref ELSE excluded.ref END, updated_at=excluded.updated_at, version=activity_cards.version+1 + WHERE activity_cards.version=?`), + c.ID, c.ProjectID, c.ConversationID, ref, owner, string(payload), string(c.Status), boolInt(c.Pinned), c.CreatedAt.Format(time.RFC3339Nano), c.UpdatedAt.Format(time.RFC3339Nano), expected) + if err != nil { + return fmt.Errorf("save card %s: %w", c.ID, err) + } + if n, _ := res.RowsAffected(); n == 0 { + // DO UPDATE 的 WHERE 未命中：行存在但版本不匹配 → 明确冲突（客户端重读）。 + return fmt.Errorf("save card %s: %w", c.ID, ErrConflict) + } + return nil +} + +// CardVersion 返回当前版本号（乐观并发的读侧）。 +func (s *Store) CardVersion(owner, id string) (int, error) { + var v int + err := s.db.QueryRow(s.q(`SELECT version FROM activity_cards WHERE id=? AND owner=?`), id, owner).Scan(&v) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return 0, ErrNotFound + } + return 0, err + } + return v, nil +} + +// GetCardVersion 读卡与当前版本（乐观并发的读侧；U3 KTD7）。 +func (s *Store) GetCardVersion(owner, id string) (core.ActivityCard, int, error) { + row := s.db.QueryRow(s.q(`SELECT payload, version FROM activity_cards WHERE id=? AND owner=?`), id, owner) + var payload string + var ver int + if err := row.Scan(&payload, &ver); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return core.ActivityCard{}, 0, ErrNotFound + } + return core.ActivityCard{}, 0, err + } + var c core.ActivityCard + if err := json.Unmarshal([]byte(payload), &c); err != nil { + return core.ActivityCard{}, 0, fmt.Errorf("card %s payload: %w", id, err) + } + return c, ver, nil +} + func (s *Store) GetCard(owner, id string) (core.ActivityCard, error) { row := s.db.QueryRow(s.q(`SELECT payload FROM activity_cards WHERE id=? AND owner=?`), id, owner) var payload string diff --git a/internal/store/messages.go b/internal/store/messages.go index a34cc26..7906e7f 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -40,8 +40,12 @@ func (s *Store) AppendMessageWithRequest(m *core.Message, requestID string) erro m.CreatedAt = m.CreatedAt.UTC() insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id) SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?` - for attempt := 0; attempt < 32; attempt++ { - _, err := s.db.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), nowISO(), requestID, m.ConversationID) + // 事务内收口（U3，KTD3/KTD7）：lockConversation（PG 事务级咨询锁； + // SQLite 单写连接天然串行）后 MAX(seq)+1 分配与 INSERT 同事务，序号 + // 竞态在锁下消失。唯一约束仍留作兜底——冲突只剩事务边界外的极端 + // 竞态（如手工写入），小退避重试即可收敛。 + for attempt := 0; attempt < 8; attempt++ { + err := s.appendMessageTx(m, requestID, insert) if err == nil { s.TouchConversation(m.ConversationID) return nil @@ -56,9 +60,6 @@ func (s *Store) AppendMessageWithRequest(m *core.Message, requestID string) erro return &DuplicateRequestError{Existing: existing} } } - // 并发同会话追加冲突（PG 多连接竞态）：平方退避（封顶 25ms）后重算 - // 序号。真库 24 路并发实测：8 次 × 0–7ms 线性退避会耗尽重试（真实 - // PG 门禁暴露的既有缺陷）；事务内会话锁的根治收口在 U3（KTD3/KTD7）。 backoff := attempt * attempt if backoff > 25 { backoff = 25 @@ -68,6 +69,23 @@ func (s *Store) AppendMessageWithRequest(m *core.Message, requestID string) erro return fmt.Errorf("store: message seq conflict retries exhausted (conversation %s)", m.ConversationID) } +// appendMessageTx 单事务完成同会话消息追加：会话锁 → 序号分配 → INSERT。 +// 请求幂等依赖 uq_messages_conv_request 唯一索引在事务内暴露冲突。 +func (s *Store) appendMessageTx(m *core.Message, requestID, insert string) error { + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + if err := s.dialect.lockConversation(tx, string(m.ConversationID)); err != nil { + return err + } + if _, err := tx.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), nowISO(), requestID, m.ConversationID); err != nil { + return err + } + return tx.Commit() +} + func (s *Store) ListMessages(conversationID string) ([]core.Message, error) { rows, err := s.db.Query(s.q(`SELECT id,conversation_id,run_id,role,parts,created_at,excluded,variant_group,variant_index FROM messages WHERE conversation_id=? ORDER BY seq`), conversationID) if err != nil { diff --git a/internal/store/migrate.go b/internal/store/migrate.go index c99b726..ee760d0 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -444,6 +444,48 @@ func init() { `DROP INDEX IF EXISTS idx_audit_owner`, `ALTER TABLE command_audit DROP COLUMN owner`, }}}, + migration{version: 19, name: "card_version", + // 卡片乐观并发（协作存储 U3，KTD7）：读改写带 expected version， + // 冲突返回 409；盲写（系统镜像/流式追加）同样递增版本。 + up: map[string][]string{"sqlite": { + `ALTER TABLE activity_cards ADD COLUMN version INTEGER NOT NULL DEFAULT 0`, + }, "postgres": { + `ALTER TABLE activity_cards ADD COLUMN version INTEGER NOT NULL DEFAULT 0`, + }}, + down: map[string][]string{"sqlite": { + `ALTER TABLE activity_cards DROP COLUMN version`, + }, "postgres": { + `ALTER TABLE activity_cards DROP COLUMN version`, + }}}, + migration{version: 20, name: "recovery_holds", + // 恢复隔离（协作存储 U3）：未知外部副作用的 run 记录持久 hold， + // 启动续跑入口必须跳过，人工核对后显式解除（方案 §5 U3）。 + up: map[string][]string{"sqlite": { + `CREATE TABLE IF NOT EXISTS recovery_holds ( + run_id TEXT PRIMARY KEY, + owner TEXT NOT NULL, + conversation_id TEXT NOT NULL DEFAULT '', + reason TEXT NOT NULL, + created_at TEXT NOT NULL, + released_at TEXT + )`, + `CREATE INDEX IF NOT EXISTS idx_recovery_holds_open ON recovery_holds(conversation_id) WHERE released_at IS NULL`, + }, "postgres": { + `CREATE TABLE IF NOT EXISTS recovery_holds ( + run_id TEXT PRIMARY KEY, + owner TEXT NOT NULL, + conversation_id TEXT NOT NULL DEFAULT '', + reason TEXT NOT NULL, + created_at TEXT NOT NULL, + released_at TEXT + )`, + `CREATE INDEX IF NOT EXISTS idx_recovery_holds_open ON recovery_holds(conversation_id) WHERE released_at IS NULL`, + }}, + down: map[string][]string{"sqlite": { + `DROP TABLE IF EXISTS recovery_holds`, + }, "postgres": { + `DROP TABLE IF EXISTS recovery_holds`, + }}}, ) } diff --git a/internal/store/recovery_holds.go b/internal/store/recovery_holds.go new file mode 100644 index 0000000..4ec9674 --- /dev/null +++ b/internal/store/recovery_holds.go @@ -0,0 +1,85 @@ +package store + +import ( + "database/sql" + "errors" + "fmt" +) + +// RecoveryHold 持久恢复隔离（协作存储 U3）：外部副作用结果未知的 run 记录 +// hold 后，启动续跑入口（ResumeInterruptedChildren 等）必须跳过，人工核对后 +// 显式解除（ReleaseRecoveryHold）。概念状态 recovering 在本增量由该独立隔离 +// 记录表达（方案 §5 U3）。 +type RecoveryHold struct { + RunID string + Owner string + ConversationID string + Reason string + CreatedAt string + ReleasedAt string // 空 = 未解除 +} + +// SetRecoveryHold 记录（或更新原因后重新生效）一条 hold：released_at 置空。 +// 幂等；同 run 重复记录只保留最后原因。 +func (s *Store) SetRecoveryHold(runID, owner, conversationID, reason string) error { + _, err := s.db.Exec(s.q(`INSERT INTO recovery_holds (run_id, owner, conversation_id, reason, created_at, released_at) + VALUES (?,?,?,?,?,NULL) + ON CONFLICT(run_id) DO UPDATE SET reason=excluded.reason, created_at=excluded.created_at, released_at=NULL`), + runID, owner, conversationID, reason, nowISO()) + if err != nil { + return fmt.Errorf("recovery hold %s: %w", runID, err) + } + return nil +} + +// GetRecoveryHold 读取单条（含已解除）；不存在返回 ErrNotFound。 +func (s *Store) GetRecoveryHold(runID string) (RecoveryHold, error) { + row := s.db.QueryRow(s.q(`SELECT run_id, owner, conversation_id, reason, created_at, COALESCE(released_at,'') FROM recovery_holds WHERE run_id=?`), runID) + var h RecoveryHold + if err := row.Scan(&h.RunID, &h.Owner, &h.ConversationID, &h.Reason, &h.CreatedAt, &h.ReleasedAt); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return RecoveryHold{}, ErrNotFound + } + return RecoveryHold{}, err + } + return h, nil +} + +// ListOpenRecoveryHolds 列出未解除的 hold（运维核对入口）。 +func (s *Store) ListOpenRecoveryHolds() ([]RecoveryHold, error) { + rows, err := s.db.Query(s.q(`SELECT run_id, owner, conversation_id, reason, created_at, COALESCE(released_at,'') FROM recovery_holds WHERE released_at IS NULL ORDER BY created_at`)) + if err != nil { + return nil, err + } + defer rows.Close() + var out []RecoveryHold + for rows.Next() { + var h RecoveryHold + if err := rows.Scan(&h.RunID, &h.Owner, &h.ConversationID, &h.Reason, &h.CreatedAt, &h.ReleasedAt); err != nil { + return nil, err + } + out = append(out, h) + } + return out, rows.Err() +} + +// HasOpenRecoveryHold 判定 run 或其会话是否存在未解除的 hold（续跑跳过的 +// 热路径）。查询出错时调用方必须按「有 hold」处理（fail-closed）。 +func (s *Store) HasOpenRecoveryHold(runID, conversationID string) (bool, error) { + var n int + err := s.db.QueryRow(s.q(`SELECT COUNT(*) FROM recovery_holds WHERE released_at IS NULL AND (run_id=? OR (conversation_id<>'' AND conversation_id=?))`), runID, conversationID).Scan(&n) + if err != nil { + return true, err + } + return n > 0, nil +} + +// ReleaseRecoveryHold 人工解除（幂等：不存在或已解除都视为成功）。 +func (s *Store) ReleaseRecoveryHold(runID string) error { + res, err := s.db.Exec(s.q(`UPDATE recovery_holds SET released_at=? WHERE run_id=? AND released_at IS NULL`), nowISO(), runID) + if err != nil { + return err + } + _ = res // 0 行 = 已解除或不存在：幂等成功 + return nil +} diff --git a/internal/store/recovery_holds_test.go b/internal/store/recovery_holds_test.go new file mode 100644 index 0000000..0572dbd --- /dev/null +++ b/internal/store/recovery_holds_test.go @@ -0,0 +1,53 @@ +package store + +import ( + "errors" + "testing" +) + +// 协作存储 U3：recovery hold 的持久 CRUD（双档；PG 路径由 CI 真库 job 覆盖 +// NULL 比较、COALESCE 与部分索引）。幂等解除与 fail-closed 判定是关键行为面。 +func TestRecoveryHoldsStoreCRUD(t *testing.T) { + st := openTestStore(t) + if err := st.SetRecoveryHold("run_a", "o", "cnv_1", "副作用未知"); err != nil { + t.Fatal(err) + } + // 重复记录：幂等更新原因并保持未解除。 + if err := st.SetRecoveryHold("run_a", "o", "cnv_1", "原因更新"); err != nil { + t.Fatal(err) + } + h, err := st.GetRecoveryHold("run_a") + if err != nil || h.Reason != "原因更新" || h.ReleasedAt != "" { + t.Fatalf("hold=%+v err=%v", h, err) + } + if held, _ := st.HasOpenRecoveryHold("run_a", "other"); !held { + t.Fatal("run 级判定应命中") + } + if held, _ := st.HasOpenRecoveryHold("run_b", "cnv_1"); !held { + t.Fatal("会话级判定应命中") + } + if held, _ := st.HasOpenRecoveryHold("run_b", "cnv_2"); held { + t.Fatal("无关 run/会话不应命中") + } + // 解除幂等。 + if err := st.ReleaseRecoveryHold("run_a"); err != nil { + t.Fatal(err) + } + if err := st.ReleaseRecoveryHold("run_a"); err != nil { + t.Fatalf("重复解除应幂等：%v", err) + } + if held, _ := st.HasOpenRecoveryHold("run_a", "cnv_1"); held { + t.Fatal("解除后不应命中") + } + if _, err := st.GetRecoveryHold("run_missing"); !errors.Is(err, ErrNotFound) { + t.Fatalf("缺失应 ErrNotFound：%v", err) + } + // 列表只含未解除。 + if err := st.SetRecoveryHold("run_c", "o", "", "另一条"); err != nil { + t.Fatal(err) + } + holds, err := st.ListOpenRecoveryHolds() + if err != nil || len(holds) != 1 || holds[0].RunID != "run_c" { + t.Fatalf("open holds=%v err=%v", holds, err) + } +} diff --git a/internal/store/store_test.go b/internal/store/store_test.go index d5d2e32..55edadf 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -90,8 +90,8 @@ func TestMigrationUpgradesExistingDatabase(t *testing.T) { if err != nil { t.Fatal(err) } - if len(got) != 18 { - t.Fatalf("expected versions [1..18], got %v", got) + if len(got) != len(migrations) { + t.Fatalf("expected versions [1..%d], got %v", len(migrations), got) } // 老库形态（v2 前）：events 全局自增、settings 单列主键。 if err := st.MigrateDown(1); err != nil { -- 2.54.0