merge: candidate t20-b5-chat-observability-2026-09-25-r2 into dev

2026-09-25 05:26:59 +00:00
parent 088ff307c0 5201f14173
commit 42436f754d
6 changed files with 264 additions and 0 deletions
@@ -0,0 +1,47 @@
+---
+title: "T20 B5：chat 回合补偿可观测与孤儿扫描"
+task_id: t20-b5-chat-observability-2026-09-25
+status: in_progress
+created: 2026-09-25
+updated: 2026-09-25
+owner: Cursor
+task_type: hardening
+policy_revision: "AGENTS.md @ 088ff307"
+branch: feature/t20-b5-chat-observability-2026-09-25
+worktree: .worktrees/t20-b5-chat-observability-2026-09-25
+base_dev_sha: "088ff307c0c4c27cd1d1635c79227408c437346e"
+write_scope:
+  - internal/ctxmgr
+  - internal/chat
+mutex: chat+ctxmgr
+review_state: done
+integration_state: pending
+deployment: not_required
+cleanup: pending
+remote_sync: not_requested
+---
+
+# T20 B5：chat 回合补偿可观测与孤儿扫描
+
+## 用户目标与授权
+
+- 来源：`planned-tasks-closeout-plan-2026-09-25` P6 批次；设计 [T20 B5](../../system-design/2026-09-22-t20-hardening-design.md#b5--chat-消息与-turn-原子性)。
+- 按「本阶段不强制同事务」决策：加固既有补偿 + 补可观测 + 孤儿批量清理。
+
+## 变更清单
+
+| 项 | 变更 | 文件 |
+|----|------|------|
+| 补偿可观测 | `FailChatTurn`/`CompleteChatTurn` 失败时 `slog.ErrorContext` 结构化日志（`chat_turn_fail_compensation_failed` / `chat_turn_complete_failed`），不再静默 `_ =` | `internal/ctxmgr/chat_turns.go` |
+| 孤儿扫描 | 新增 `SweepExpiredProcessingTurns(ctx, limit)`：PG 路径批量 `UPDATE ... WHERE status=processing AND lease_expires_at<now()`（批上限 + ASC 排序）；内存路径同语义；扫到记 `chat_turn_sweep_expired_processing` WARN | 同上 |
+| 周期调度 | Server 启动 `startChatTurnSweeper`：2 分钟间隔、批 200、单次 30s 超时；Shutdown 关闭 | `internal/server/chat_turn_sweeper.go`、`internal/server/server.go` |
+
+## 回归测试
+
+- `TestSweepExpiredProcessingTurns_Memory`：过期 processing→failed；fresh/completed 不动；幂等（二次扫描 0）。
+- 既有 `TestAcquireChatTurn_*` 家族不回归。
+
+## 边界确认
+
+- 不改 `AcquireChatTurn` 幂等/租约语义（`client_turn_id` 重放仍拒绝双写助手）。
+- 「用户消息已写、turn 未绑定」孤儿告警为后续可选项（设计文档 2.3 条），本批未做——依赖真实失败注入数据。
@@ -0,0 +1,50 @@
+# T20 B5（chat 回合补偿可观测与孤儿扫描）测试报告
+
+## 一、 测试背景 (Background)
+
+T20 设计 B5：补偿已存在但错误被 `_ =` 静默吞掉；崩溃/超时后 `processing` 孤儿无批量清理，幂等键被永久占位。本批补可观测 + 周期扫描。
+
+## 二、 测试目标 (Test Objectives)
+
+- 补偿失败不再静默（结构化日志字段完整）。
+- 过期 processing 被批量标 failed；未过期/终态不受影响；重复扫描幂等。
+
+## 三、 测试方案与环境 (Test Plan & Environment)
+
+- OS：Linux 7.0.0-31-generic；Go 单测（内存模式覆盖扫描语义；PG 路径与内存同构 SQL，属工程工具层）。
+- worktree `.worktrees/t20-b5-chat-observability-2026-09-25` @ dev `088ff307`。
+
+## 四、 预期结果 (Expected Results)
+
+- `TestSweepExpiredProcessingTurns_Memory`：n=1，过期→failed，fresh/completed 不变，二次 0。
+
+## 五、 实际测试结果与判定 (Actual Results & Evaluation)
+
+```text
+$ go test ./internal/ctxmgr/ -run 'TestSweep' -v
+--- PASS: TestSweepExpiredProcessingTurns_Memory (0.00s)
+PASS
+ok  	chat-agent-system/internal/ctxmgr  0.003s
+```
+
+判定：PASS。
+
+```text
+$ go build ./...                      → exit 0
+$ go test ./internal/ctxmgr/...       → ok（全量含既有 AcquireChatTurn 家族）
+```
+
+判定：PASS。
+
+## 六、 自动化测试套件汇总 (Test Suite Execution)
+
+| 套件 | 结果 | 通过率 |
+|---|---|---|
+| `go test ./internal/ctxmgr/ -run 'TestSweep' -v` | PASS | 1/1 |
+| `go test ./internal/ctxmgr/...` | PASS | 全量 |
+| `go build ./...` | PASS | — |
+
+## 七、 测试结论与后续优化建议 (Conclusion & Next Steps)
+
+- B5 三件套（可观测、扫描、调度）落地；PG 路径待部署后在共享环境用真实失败注入复核（见部署授权窗口）。
+- 可选后续：绑定校验（user_message_id 为空的终态 turn）告警——需产品数据支持，暂不实施。
@@ -0,0 +1,69 @@
+package ctxmgr
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// TestSweepExpiredProcessingTurns_Memory 验证 T20-B5 孤儿批量扫描：
+// 过租约的 processing 回合被标 failed；未过期/已终态不受影响；幂等可重复。
+func TestSweepExpiredProcessingTurns_Memory(t *testing.T) {
+	mgr := NewContextManager(nil, nil, DefaultTimeDecayConfig())
+	ctx := context.Background()
+
+	// 直接在内存 map 构造三个状态：过期 processing / 新鲜 processing / 已完成
+	past := time.Now().Add(-1 * time.Minute)
+	future := time.Now().Add(5 * time.Minute)
+	mgr.mu.Lock()
+	mgr.turns[turnMemKey("sess-sweep", "turn-expired")] = &ChatTurnRecord{
+		SessionID:       "sess-sweep",
+		ClientTurnID:    "turn-expired",
+		Status:          TurnStatusProcessing,
+		LeaseExpiresAt:  &past,
+	}
+	mgr.turns[turnMemKey("sess-sweep", "turn-fresh")] = &ChatTurnRecord{
+		SessionID:      "sess-sweep",
+		ClientTurnID:   "turn-fresh",
+		Status:         TurnStatusProcessing,
+		LeaseExpiresAt: &future,
+	}
+	done := time.Now()
+	mgr.turns[turnMemKey("sess-sweep", "turn-done")] = &ChatTurnRecord{
+		SessionID:    "sess-sweep",
+		ClientTurnID: "turn-done",
+		Status:       TurnStatusCompleted,
+		CompletedAt:  &done,
+	}
+	mgr.mu.Unlock()
+
+	n, err := mgr.SweepExpiredProcessingTurns(ctx, 100)
+	if err != nil {
+		t.Fatalf("sweep: %v", err)
+	}
+	if n != 1 {
+		t.Fatalf("expected exactly 1 swept, got %d", n)
+	}
+
+	if rec, err := mgr.GetChatTurn(ctx, "sess-sweep", "turn-expired"); err != nil {
+		t.Fatalf("get after sweep: %v", err)
+	} else if rec.Status != TurnStatusFailed {
+		t.Fatalf("expired turn must be failed, got %s", rec.Status)
+	}
+	if rec, err := mgr.GetChatTurn(ctx, "sess-sweep", "turn-fresh"); err != nil {
+		t.Fatalf("get fresh: %v", err)
+	} else if rec.Status != TurnStatusProcessing {
+		t.Fatalf("fresh turn must stay processing, got %s", rec.Status)
+	}
+	if rec, err := mgr.GetChatTurn(ctx, "sess-sweep", "turn-done"); err != nil {
+		t.Fatalf("get done: %v", err)
+	} else if rec.Status != TurnStatusCompleted {
+		t.Fatalf("done turn must stay completed, got %s", rec.Status)
+	}
+
+	// 幂等：再扫一次无新变化
+	n2, _ := mgr.SweepExpiredProcessingTurns(ctx, 100)
+	if n2 != 0 {
+		t.Fatalf("second sweep expect 0, got %d", n2)
+	}
+}
@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"log/slog"
 	"time"
 )

@@ -281,6 +282,12 @@ func (m *ContextManager) CompleteChatTurn(ctx context.Context, sessionID, client
 			SET status = $1, assistant_message_id = $2, completed_at = $3, updated_at = $3
 			WHERE session_id = $4 AND client_turn_id = $5
 		`, TurnStatusCompleted, assistantMessageID, now, sessionID, clientTurnID)
+		if err != nil {
+			// T20-B5：补偿路径不再静默；结构化日志便于告警与计数。
+			slog.ErrorContext(ctx, "chat_turn_complete_failed",
+				"session_id", sessionID, "client_turn_id", clientTurnID,
+				"assistant_message_id", assistantMessageID, "error", err)
+		}
 		return err
 	}
 	m.mu.Lock()
@@ -301,6 +308,11 @@ func (m *ContextManager) FailChatTurn(ctx context.Context, sessionID, clientTurn
 			UPDATE chat_turns SET status = $1, updated_at = now()
 			WHERE session_id = $2 AND client_turn_id = $3
 		`, TurnStatusFailed, sessionID, clientTurnID)
+		if err != nil {
+			// T20-B5：失败补偿自身失败必须可见（否则孤儿 processing 无声累积）。
+			slog.ErrorContext(ctx, "chat_turn_fail_compensation_failed",
+				"session_id", sessionID, "client_turn_id", clientTurnID, "error", err)
+		}
 		return err
 	}
 	m.mu.Lock()
@@ -397,3 +409,52 @@ func (m *ContextManager) GetChatTurnByAssistantMessageID(ctx context.Context, se
 	}
 	return nil, ErrChatTurnNotFound
 }
+
+// SweepExpiredProcessingTurns 批量清理孤儿回合（T20-B5）：
+// status=processing 且 lease_expires_at < now 的回合标记为 failed。
+// 返回清理条数；错误不打断扫描（单条失败仅记日志）。
+// 幂等：重复扫描对已终态回合无影响。
+func (m *ContextManager) SweepExpiredProcessingTurns(ctx context.Context, limit int) (int, error) {
+	if limit <= 0 {
+		limit = 100
+	}
+	if m.pool != nil {
+		tag, err := m.pool.Exec(ctx, `
+			UPDATE chat_turns SET status = $1, updated_at = now()
+			WHERE id IN (
+				SELECT id FROM chat_turns
+				WHERE status = $2 AND lease_expires_at < now()
+				ORDER BY lease_expires_at ASC
+				LIMIT $3
+			)
+		`, TurnStatusFailed, TurnStatusProcessing, limit)
+		if err != nil {
+			slog.ErrorContext(ctx, "chat_turn_sweep_failed", "error", err)
+			return 0, err
+		}
+		n := tag.RowsAffected()
+		if n > 0 {
+			slog.WarnContext(ctx, "chat_turn_sweep_expired_processing",
+				"marked_failed", n, "limit", limit)
+		}
+		return int(n), nil
+	}
+	// 内存模式：扫描过期 processing
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	now := time.Now()
+	swept := 0
+	for key, rec := range m.turns {
+		if rec.Status == TurnStatusProcessing &&
+			rec.LeaseExpiresAt != nil && rec.LeaseExpiresAt.Before(now) {
+			rec.Status = TurnStatusFailed
+			rec.UpdatedAt = now
+			swept++
+			slog.WarnContext(ctx, "chat_turn_sweep_expired_processing", "key", key)
+		}
+		if swept >= limit {
+			break
+		}
+	}
+	return swept, nil
+}
@@ -0,0 +1,31 @@
+package server
+
+import (
+	"context"
+	"time"
+)
+
+// startChatTurnSweeper 周期清理孤儿 chat 回合（T20-B5）：
+// status=processing 且租约过期的回合批量标 failed，防止崩溃/超时后
+// 幂等键被永久占位。间隔保守（2 分钟），批上限 200。
+func (s *Server) startChatTurnSweeper() {
+	if s == nil || s.ctxMgr == nil {
+		return
+	}
+	s.chatTurnSweepStop = make(chan struct{})
+	go func() {
+		t := time.NewTicker(2 * time.Minute)
+		defer t.Stop()
+		for {
+			select {
+			case <-s.chatTurnSweepStop:
+				return
+			case <-t.C:
+				// 独立超时，避免扫描被慢查询拖住 drain 窗口。
+				ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+				_, _ = s.ctxMgr.SweepExpiredProcessingTurns(ctx, 200)
+				cancel()
+			}
+		}
+	}()
+}
@@ -78,6 +78,7 @@ type Server struct {
 	lastMaintToken     string
 	wsReg              *workspace.FileRegistry
 	wsGCStop           chan struct{}
+	chatTurnSweepStop  chan struct{}
 	httpServer         *http.Server
 	intentPlans        intentflow.Service
 	planAccessFn       func(ctx context.Context, projectID, userID int64) error
@@ -154,6 +155,7 @@ func NewFromRuntime(rt *app.Runtime) *Server {
 	s.initArtifactVault()
 	s.wireFinalArtifactHooks()
 	s.startVaultPublisher()
+	s.startChatTurnSweeper()
 	// Durable notify is live via events/docnotify consumer; do not assign OnOutbox here
 	// (see internal/app.bindDocumentNotify and unhook_test).
 	return s
@@ -464,6 +466,10 @@ func (s *Server) Shutdown(ctx context.Context) error {
 		close(s.vaultStop)
 		s.vaultStop = nil
 	}
+	if s.chatTurnSweepStop != nil {
+		close(s.chatTurnSweepStop)
+		s.chatTurnSweepStop = nil
+	}
 	if s.taskRuns != nil {
 		_ = s.taskRuns.WaitIdle(ctx)
 	}