diff --git a/.doc/ops/short-term/2026-09-25-t20-b5-chat-observability.md b/.doc/ops/short-term/2026-09-25-t20-b5-chat-observability.md new file mode 100644 index 00000000..cc4ebda1 --- /dev/null +++ b/.doc/ops/short-term/2026-09-25-t20-b5-chat-observability.md @@ -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 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 +} diff --git a/internal/server/chat_turn_sweeper.go b/internal/server/chat_turn_sweeper.go new file mode 100644 index 00000000..826a2d45 --- /dev/null +++ b/internal/server/chat_turn_sweeper.go @@ -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() + } + } + }() +} diff --git a/internal/server/server.go b/internal/server/server.go index f09c15b2..d02a74a6 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -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) }