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

@@ -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 {

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 故障不阻断请求（限流是柔性行为，放行并记日志于调用方）
+		// 降级（KTD10）：不再无界放行；按实例保守限流并计数供观测告警。
-	}
+		r.degraded.Add(1)
-	if n == 1 {
+		return r.fallback.Allow(ctx, key, limit, window)
-		r.rdb.Expire(ctx, rkey, window+time.Second)
 	}
 	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() }

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")
+	}
+}

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)
+	}
+}

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

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))
+}

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 <id> 切换画像 · /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) })

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)
+	}
+}

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)
+	_, 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 (?,?,?,?,?,?,?,?,?,?)
+		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`),
+		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

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++ {
+	// 事务内收口（U3，KTD3/KTD7）：lockConversation（PG 事务级咨询锁；
-		_, err := s.db.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), nowISO(), requestID, m.ConversationID)
+	// 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 {

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`,
+			}}},
 	)
 }
 

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
+}

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)
+	}
+}

internal/store/store_test.go

@@ -90,8 +90,8 @@ func TestMigrationUpgradesExistingDatabase(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(got) != 18 {
+	if len(got) != len(migrations) {
-		t.Fatalf("expected versions [1..18], got %v", got)
+		t.Fatalf("expected versions [1..%d], got %v", len(migrations), got)
 	}
 	// 老库形态（v2 前）：events 全局自增、settings 单列主键。
 	if err := st.MigrateDown(1); err != nil {