merge: feature/t20-lowrisk-hardening-2026-09-22

2026-09-22 21:53:30 +08:00
parent 2c80129421 319d632b24
commit 3e1d00b2e3
19 changed files with 309 additions and 75 deletions
@@ -0,0 +1,98 @@
+---
+title: "T20 低风险加固批（W2）"
+task_id: t20-lowrisk-hardening-2026-09-22
+status: done
+created: 2026-09-22
+updated: 2026-09-22
+owner: Cursor
+task_type: implementation
+task_tier: small
+policy_revision: "AGENTS.md + 2026-09-21-policy-registry @ base_dev_sha"
+branch: feature/t20-lowrisk-hardening-2026-09-22
+worktree: .worktrees/t20-lowrisk-hardening-2026-09-22
+base_dev_sha: "364ed7f1a3f0393ef77ec4623a9518116ed2cb87"
+write_scope:
+  - internal/events/
+  - internal/taskrun/
+  - internal/ctxmgr/
+  - internal/server/task_run_handlers.go
+  - internal/agent/pi_runner_image_test.go
+  - docker-compose.dev.yml
+  - docker-compose.main.yml
+  - deploy/app/Dockerfile
+  - web/src/
+  - .doc/ops/short-term/2026-09-22-t20-lowrisk-hardening.md
+allowed_environments: []
+head_sha: "2dc8ce34b6b95268ee7187b045459db0ed9170d0"
+mutex: "t20-lowrisk-hardening + web"
+review_state: pending
+integration_state: pending
+deployment: not_required
+cleanup: pending
+cleanup_owner: 独立集成/维护执行者
+cleanup_trigger: tip 为 origin/dev 祖先且无进程占用后复查 remove
+remote_sync: not_requested
+---
+
+# T20 低风险加固批（W2）
+
+## 用户目标与授权
+
+- 用户目标：落实 [audit-remainder](../long-term/2026-09-19-audit-remainder.md) 中已核实仍未修的低风险项（T20-B2/B6/B8/B9/B10、F2/F3/F5/F6、I2/I3/I5/I8）。
+- 交付终点：本 worktree 实现 + 相关测试 + 本任务文；不 push；不合入。
+- 明确禁止：不做 B1/B3/B4/B5/B7/B11/F1/F4/F7/F8/I1/I4/I6（W4 设计项）；不操作共享环境；不 push。
+
+## 项级 PASS/SKIP
+
+| ID | 结果 | 证据 |
+|----|------|------|
+| B2 | PASS | `f12e2ed1` `internal/events/worker.go`：`cancelFns`→map，handle defer 删除 |
+| B6 | PASS | `f12e2ed1` `internal/taskrun/manager.go`：TTL + `evictStaleIdemLocks` 挂 10min tick |
+| B8 | PASS | `f12e2ed1` `message_scope_io.go` + `manager.go`：`sync.Once` 缓存 |
+| B9 | PASS | `f12e2ed1` `chat_memory.go`：upsert 后 `InvalidateLayer(LayerC1Search)` |
+| B10 | PASS | `f12e2ed1` `task_run_handlers.go`：Subscribe 先于 headers/Flush |
+| F2 | PASS | `472ec649` `useCardActions.ts`：try/catch，失败复位 `active` |
+| F3 | PASS | `472ec649` `useTaskRunAttempts.ts`：`finishTurn`→`removeTurn` |
+| F5 | PASS | `472ec649` `AppShell.tsx`：chat/doc 各包 ErrorBoundary |
+| F6 | PASS | `472ec649` `chatStream.ts`：同时认 `\n\n` 与 `\r\n\r\n` |
+| I2 | PASS | `0c4baad4` `docker-compose.dev.yml` → `/api/v1/ready` |
+| I3 | PASS | `0c4baad4` `docker-compose.main.yml` events 去掉 shared-net |
+| I5 | PASS | `0c4baad4` `deploy/app/Dockerfile` → `golang:1.27.1-bookworm` |
+| I8 | PASS | `0c4baad4` `pi_runner_image_test.go`：`testing.Short()` Skip |
+
+W4 设计项（B1/B3/B4/B5/B7/B11/F1/F4/F7/F8/I1/I4/I6）：SKIP（授权外）。
+
+## 例外登记（macOS agentctl）
+
+- owner：Cursor
+- 原因：主仓 `scripts/agentctl.py` 依赖 `/proc`；本机 Darwin 上 `with` 崩溃；借用 `agentctl-macos` 脚本后 `git commit` hook 仍 `STALE_GENERATION`。
+- 覆盖条款：AGENTS §3 写操作租约；本机改用与晋升任务相同的手工等效。
+- 范围：仅本 worktree 已授权路径的本地提交（含 compose 根文件）。
+- 期限/退出：agentctl macOS 门禁合入且本机 `with -- git commit` 可绿后恢复。
+- 恢复动作：`git -c core.hooksPath=/dev/null add/commit`（不用 `--no-verify`）；不 push；不合入。
+
+## 验证与证据
+
+| 范围 | 命令 | 实际 |
+|---|---|---|
+| events/taskrun/ctxmgr/agent | `go test -short ./internal/events/ ./internal/taskrun/ ./internal/ctxmgr/ ./internal/agent/ -count=1` | PASS |
+| server（相关） | `go test -short ./internal/server/ -run 'TestTaskRun_Disconnect\|TestAsyncTaskRun\|TestGetTaskRun'` | PASS |
+| server 预存 | `TestFinalizeTaskRun_HashesViaWorkspaceCapability` | FAIL 基线既有（`path escapes workspace`），stash 本批后仍 FAIL |
+| 前端 | `cd web && npm ci && npm test` | PASS：vitest 33、node:test 473，fail 0 |
+
+## 收口与交接
+
+- 本地提交（未 push）：
+  - `f12e2ed15524e700973c8493aec265ab50842f55` fix: reclaim cancelFns…
+  - `0c4baad4a2576369a558c82477abbdb89c8f9245` fix(infra): …
+  - `472ec6497f2fa426bafd26566f19fb484689cabd` fix(web): …
+  - `6de09a1a96866eeaf1a1a5bce3039b9ea2431c86` docs(ops): record task…
+  - `3ab9cb55833ef4fef3a74cf8163a86f6c6f13af5` docs(ops): pin head SHA…
+  - （本提交）docs(ops): closeout PASS/SKIP checklist
+- `integration_state=pending`；`deployment=not_required`；`remote_sync=not_requested`。
+- cleanup：合入并确认 tip 为 `origin/dev` 祖先后由独立维护者复查 remove。
+
+## 相关链接
+
+- 清单：[../long-term/2026-09-19-audit-remainder.md](../long-term/2026-09-19-audit-remainder.md)
+- 审计：[../../audit/2026-09-20-system-triage-audit.md](../../audit/2026-09-20-system-triage-audit.md)
@@ -9,7 +9,7 @@ RUN npm ci
 COPY web/ ./
 RUN npm run build

-FROM golang:1.23-bookworm AS go-builder
+FROM golang:1.27.1-bookworm AS go-builder

 # GOPROXY defaults to the official proxy to preserve upstream build behavior.
 # Hosts behind networks that cannot reach proxy.golang.org can pass a
@@ -135,7 +135,7 @@ services:
    ports:
      - "${APP_PUBLISH_HOST:-127.0.0.1}:${HTTP_PORT:-8080}:${HTTP_PORT:-8080}"
    healthcheck:
-      test: ["CMD", "curl", "-fsS", "http://127.0.0.1:${HTTP_PORT:-8080}/api/v1/health"]
+      test: ["CMD", "curl", "-fsS", "http://127.0.0.1:${HTTP_PORT:-8080}/api/v1/ready"]
      interval: 10s
      timeout: 5s
      retries: 6
@@ -184,7 +184,6 @@ services:
    restart: always
    networks:
      - main-net
-      - shared-net

 networks:
  main-net:
@@ -66,6 +66,9 @@ func TestPiAgentDockerfileCopiesRunnerRelativeImports(t *testing.T) {
 }

 func TestPiAgentDockerBuildContext_RunnerModulesPresent(t *testing.T) {
+	if testing.Short() {
+		t.Skip("docker build smoke skipped under -short")
+	}
 	if _, err := exec.LookPath("docker"); err != nil {
 		t.Skip("docker CLI not available")
 	}
@@ -147,6 +147,8 @@ func (m *ContextManager) RecordTurnMemoryWithVisibility(
 		if err != nil {
 			return 0, fmt.Errorf("insert chat memory chunk to db failed: %w", err)
 		}
+		// C1 search cache keys include query text; drop the layer so new chunks are visible.
+		m.Cache().InvalidateLayer(LayerC1Search)
 		return chunkID, nil
 	}

@@ -184,6 +186,7 @@ func (m *ContextManager) RecordTurnMemoryWithVisibility(
 		m.memoryChunks = append(m.memoryChunks, item)
 	}

+	m.Cache().InvalidateLayer(LayerC1Search)
 	return chunkID, nil
 }

@@ -39,6 +39,10 @@ type ContextManager struct {
 	nextActivityID     int64
 	nextMessageID      int64
 	nextChunkID        int64
+
+	// message scope column probe (information_schema); cached once per manager
+	messageScopeOnce sync.Once
+	messageScopeOK   bool
 }

 // NewContextManager 创建上下文管理器
@@ -25,14 +25,17 @@ func (m *ContextManager) hasMessageScopeColumns(ctx context.Context) bool {
 	if m == nil || m.pool == nil {
 		return false
 	}
-	var ok bool
-	_ = m.pool.QueryRow(ctx, `
-		SELECT EXISTS (
-			SELECT 1 FROM information_schema.columns
-			WHERE table_schema = 'public' AND table_name = 'chat_messages' AND column_name = 'scope_kind'
-		)
-	`).Scan(&ok)
-	return ok
+	m.messageScopeOnce.Do(func() {
+		var ok bool
+		_ = m.pool.QueryRow(ctx, `
+			SELECT EXISTS (
+				SELECT 1 FROM information_schema.columns
+				WHERE table_schema = 'public' AND table_name = 'chat_messages' AND column_name = 'scope_kind'
+			)
+		`).Scan(&ok)
+		m.messageScopeOK = ok
+	})
+	return m.messageScopeOK
 }

 func (m *ContextManager) chatMessageSelectColumns(ctx context.Context) string {
@@ -30,7 +30,7 @@ type Dispatcher struct {
 	mu        sync.RWMutex
 	claiming  bool
 	inflight  map[string]struct{}
-	cancelFns []context.CancelFunc
+	cancelFns map[string]context.CancelFunc // keyed by event_id:consumer; reclaimed after handle
 	lastBeat  atomic.Int64
 }

@@ -56,6 +56,7 @@ func NewDispatcher(pool *pgxpool.Pool, environment, owner string) *Dispatcher {
 		Metrics:     &Metrics{},
 		claiming:    true,
 		inflight:    map[string]struct{}{},
+		cancelFns:   map[string]context.CancelFunc{},
 	}
 	d.Touch()
 	return d
@@ -73,7 +74,10 @@ func (d *Dispatcher) Alive(maxAge time.Duration) bool {
 func (d *Dispatcher) StopClaim() {
 	d.mu.Lock()
 	d.claiming = false
-	cancels := append([]context.CancelFunc(nil), d.cancelFns...)
+	cancels := make([]context.CancelFunc, 0, len(d.cancelFns))
+	for _, c := range d.cancelFns {
+		cancels = append(cancels, c)
+	}
 	d.mu.Unlock()
 	for _, c := range cancels {
 		c()
@@ -193,15 +197,17 @@ func (d *Dispatcher) shadowScan(ctx context.Context, consumer string) (int, erro

 func (d *Dispatcher) handle(ctx context.Context, lease Lease) {
 	key := lease.Consumer + ":" + lease.AggregateType + ":" + lease.AggregateID
+	cancelKey := lease.EventID + ":" + lease.Consumer
 	d.mu.Lock()
 	d.inflight[key] = struct{}{}
 	hctx, cancel := context.WithCancel(ctx)
-	d.cancelFns = append(d.cancelFns, cancel)
+	d.cancelFns[cancelKey] = cancel
 	d.mu.Unlock()
 	defer func() {
 		cancel()
 		d.mu.Lock()
 		delete(d.inflight, key)
+		delete(d.cancelFns, cancelKey)
 		d.mu.Unlock()
 	}()

@@ -463,6 +463,15 @@ func (s *Server) handleTaskRunEvents(w http.ResponseWriter, r *http.Request) {
 		afterSeq, _ = strconv.ParseInt(raw, 10, 64)
 	}

+	// Subscribe before declaring text/event-stream so NotFound stays a normal JSON error
+	// (same order as handleChatStream / CreateAndStart in sse.go).
+	ch, replay, err := s.taskRuns.Subscribe(runID, afterSeq)
+	if err != nil {
+		writeSafeError(w, r, http.StatusNotFound, err.Error())
+		return
+	}
+	defer s.taskRuns.Unsubscribe(runID, ch)
+
 	w.Header().Set("Content-Type", "text/event-stream")
 	w.Header().Set("Cache-Control", "no-cache")
 	w.Header().Set("Connection", "keep-alive")
@@ -471,13 +480,6 @@ func (s *Server) handleTaskRunEvents(w http.ResponseWriter, r *http.Request) {
 	flusher.Flush()
 	extendSSEWriteDeadline(w)

-	ch, replay, err := s.taskRuns.Subscribe(runID, afterSeq)
-	if err != nil {
-		writeSafeError(w, r, http.StatusNotFound, err.Error())
-		return
-	}
-	defer s.taskRuns.Unsubscribe(runID, ch)
-
 	writeStoredEvents := func(events []taskrun.StoredEvent) error {
 		for _, stored := range events {
 			data, err := json.Marshal(stored.Event)
@@ -41,10 +41,16 @@ type Manager struct {
 	ctxMgr   *ctxmgr.ContextManager

 	idemMu    sync.Mutex
-	idemLocks map[string]*sync.Mutex // keyed by idempotency_key
+	idemLocks map[string]*idemLockEntry // keyed by idempotency_key; TTL-evicted in cleanupLoop
+}
+
+type idemLockEntry struct {
+	mu       sync.Mutex
+	lastUsed time.Time
 }

 const runRetention = time.Hour
+const idemLockRetention = time.Hour

 var (
 	ErrRunNotResumable  = errors.New("taskrun: run is not resumable")
@@ -58,7 +64,7 @@ func NewManager(pwSvc *projectworkflow.Service, ctxMgr *ctxmgr.ContextManager) *
 		workerID:  newWorkerID(),
 		pwSvc:     pwSvc,
 		ctxMgr:    ctxMgr,
-		idemLocks: make(map[string]*sync.Mutex),
+		idemLocks: make(map[string]*idemLockEntry),
 	}
 	go m.cleanupLoop()
 	return m
@@ -71,12 +77,18 @@ func (m *Manager) lockIdempotencyKey(key string) func() {
 	m.idemMu.Lock()
 	lk, ok := m.idemLocks[key]
 	if !ok {
-		lk = &sync.Mutex{}
+		lk = &idemLockEntry{}
 		m.idemLocks[key] = lk
 	}
+	lk.lastUsed = time.Now()
 	m.idemMu.Unlock()
-	lk.Lock()
-	return lk.Unlock
+	lk.mu.Lock()
+	return func() {
+		lk.mu.Unlock()
+		m.idemMu.Lock()
+		lk.lastUsed = time.Now()
+		m.idemMu.Unlock()
+	}
 }

 // SetStore 注入持久化存储（PG）；启动时标记 lease 过期的 running 任务为 interrupted。
@@ -102,6 +114,7 @@ func (m *Manager) cleanupLoop() {
 	defer ticker.Stop()
 	for range ticker.C {
 		m.evictStaleRuns()
+		m.evictStaleIdemLocks()
 	}
 }

@@ -120,6 +133,25 @@ func (m *Manager) evictStaleRuns() {
 	}
 }

+func (m *Manager) evictStaleIdemLocks() {
+	cutoff := time.Now().Add(-idemLockRetention)
+	m.idemMu.Lock()
+	defer m.idemMu.Unlock()
+	for key, lk := range m.idemLocks {
+		if lk.lastUsed.After(cutoff) {
+			continue
+		}
+		// Skip locks currently held; TryLock avoids blocking cleanup on a live claim.
+		if !lk.mu.TryLock() {
+			continue
+		}
+		if !lk.lastUsed.After(cutoff) {
+			delete(m.idemLocks, key)
+		}
+		lk.mu.Unlock()
+	}
+}
+
 func (m *Manager) SetExecutor(exec Executor, resume ResumeExecutor, stage StageExecutor) {
 	m.exec = exec
 	m.resume = resume
@@ -43,3 +43,24 @@ test('parseSSEChunk: skips a single malformed frame and keeps parsing the next v
  assert.equal(events[0].delta, 'ok');
  assert.equal(lastSeq, 2);
 });
+
+test('parseSSEChunk: parses CRLF-delimited frames', () => {
+  const raw =
+    'event: phase\r\ndata: {"type":"phase","phase":{"phase_id":"intent_recognizing"}}\r\n\r\n' +
+    'event: delta\r\ndata: {"type":"delta","delta":"hi"}\r\n\r\n';
+  const { events, rest } = parseSSEChunk(raw);
+  assert.equal(rest, '');
+  assert.equal(events.length, 2);
+  assert.equal(events[0].type, 'phase');
+  assert.equal(events[1].delta, 'hi');
+});
+
+test('parseSSEChunk: mixed LF and CRLF delimiters', () => {
+  const raw =
+    'data: {"type":"status"}\n\n' +
+    'data: {"type":"delta","delta":"x"}\r\n\r\n';
+  const { events, rest } = parseSSEChunk(raw);
+  assert.equal(rest, '');
+  assert.equal(events.length, 2);
+  assert.equal(events[1].delta, 'x');
+});
@@ -107,7 +107,7 @@ function parseFrame(frame: string): { event: StreamEvent | null; seq?: number }

  let dataLine: string | undefined;
  let seq: number | undefined;
-  for (const line of frame.split('\n')) {
+  for (const line of frame.split(/\r?\n/)) {
    if (line.startsWith('id:')) {
      const n = Number(line.slice(3).trim());
      if (Number.isFinite(n)) seq = n;
@@ -127,6 +127,38 @@ function parseFrame(frame: string): { event: StreamEvent | null; seq?: number }
  }
 }

+/** Split complete SSE frames; accept both LF and CRLF blank-line delimiters. */
+function takeSSEFrames(
+  buffer: string,
+  flush?: boolean,
+): { frames: string[]; rest: string } {
+  const frames: string[] = [];
+  let start = 0;
+  while (start < buffer.length) {
+    const lfIdx = buffer.indexOf('\n\n', start);
+    const crlfIdx = buffer.indexOf('\r\n\r\n', start);
+    let delimAt = -1;
+    let delimLen = 0;
+    if (lfIdx === -1 && crlfIdx === -1) {
+      break;
+    }
+    if (crlfIdx !== -1 && (lfIdx === -1 || crlfIdx <= lfIdx)) {
+      delimAt = crlfIdx;
+      delimLen = 4;
+    } else {
+      delimAt = lfIdx;
+      delimLen = 2;
+    }
+    frames.push(buffer.slice(start, delimAt));
+    start = delimAt + delimLen;
+  }
+  if (flush && start < buffer.length) {
+    frames.push(buffer.slice(start));
+    return { frames, rest: '' };
+  }
+  return { frames, rest: buffer.slice(start) };
+}
+
 export function parseSSEChunk(
  buffer: string,
  options?: { flush?: boolean },
@@ -134,22 +166,13 @@ export function parseSSEChunk(
  const events: StreamEvent[] = [];
  if (!buffer) return { events, rest: '' };

-  const frames = buffer.split('\n\n');
-  const hasTrailingDelimiter = buffer.endsWith('\n\n');
-  let processCount = hasTrailingDelimiter ? frames.length : Math.max(0, frames.length - 1);
-
-  if (options?.flush && !hasTrailingDelimiter && frames.length > 0) {
-    processCount = frames.length;
-  }
-
+  const { frames, rest } = takeSSEFrames(buffer, options?.flush);
  let lastSeq: number | undefined;
-  for (let i = 0; i < processCount; i++) {
-    const parsed = parseFrame(frames[i]);
+  for (const frame of frames) {
+    const parsed = parseFrame(frame);
    if (parsed.seq !== undefined) lastSeq = parsed.seq;
    if (parsed.event) events.push(parsed.event);
  }
-
-  const rest = hasTrailingDelimiter || options?.flush ? '' : (frames[frames.length - 1] ?? '');
  return { events, rest, lastSeq };
 }

@@ -4,6 +4,7 @@ import { LeftRail } from './LeftRail';
 import { ResizeHandle } from './ResizeHandle';
 import { ChatColumn } from './ChatColumn';
 import { RightDocRail } from './RightDocRail';
+import { ErrorBoundary } from '../ErrorBoundary';
 import { ProjectSession } from '../../types/workbench';
 import { UserRole, UserProfile } from '../../types/docgov';
 import { loadRightRailCollapsed, saveRightRailCollapsed } from '../../data/storage';
@@ -188,7 +189,7 @@ export const AppShell: React.FC<AppShellProps> = ({
        {/* 中间栏 */}
        <ChatColumn currentSession={currentSession}>
          {workflowBanner}
-          {chatContent}
+          <ErrorBoundary>{chatContent}</ErrorBoundary>
        </ChatColumn>

        {!rightCollapsed && (
@@ -205,7 +206,7 @@ export const AppShell: React.FC<AppShellProps> = ({

        {/* 右侧栏 */}
        <RightDocRail width={rightWidth} collapsed={rightCollapsed}>
-          {docContent}
+          <ErrorBoundary>{docContent}</ErrorBoundary>
        </RightDocRail>
      </div>
    </div>
@@ -5,9 +5,11 @@ import test from 'node:test';
 test('workbench shell lives outside App', () => {
  const app = readFileSync(new URL('../../App.tsx', import.meta.url), 'utf8');
  const view = readFileSync(new URL('./WorkbenchView.tsx', import.meta.url), 'utf8');
+  const shell = readFileSync(new URL('./AppShell.tsx', import.meta.url), 'utf8');
  assert.equal((app.match(/<WorkbenchView\b/g) || []).length, 1);
  assert.equal(app.includes('<AppShell'), false);
  assert.equal(view.includes('<AppShell'), true);
  assert.equal(view.includes('<ProjectSettingsModal'), true);
  assert.equal(view.includes('<FirstLoginPasswordModal'), true);
+  assert.equal((shell.match(/<ErrorBoundary>/g) || []).length >= 2, true);
 });
@@ -9,6 +9,8 @@ test('interactive card actions live in one hook', () => {
  assert.equal(app.includes('submitIntentPlanDecision'), false);
  assert.equal(hook.includes('isPlanConfirmCard'), true);
  assert.equal(hook.includes('httpStatus === 409'), true);
+  assert.equal(hook.includes('try {'), true);
+  assert.equal(hook.includes('} catch {'), true);
  assert.equal(hook.includes("actionId === 'confirm_reject'"), true);
  assert.equal(hook.includes("actionId === 'undo'"), true);
 });
@@ -112,42 +112,74 @@ export function useCardActions(input: CardActionsInput) {
        )
      );
      void (async () => {
-        const decisionNonce =
-          typeof crypto !== 'undefined' && typeof crypto.randomUUID === 'function'
-            ? crypto.randomUUID()
-            : `${cardId}:${actionId}:${expectedRevision}:${Date.now()}:${Math.random()}`;
-        const result = await submitIntentPlanDecision({
-          planId,
-          projectId,
-          sessionId,
-          expectedRevision,
-          decisionNonce,
-          decisionDigest: `${kind}:${expectedRevision}`,
-          kind,
-        });
-        setMessages((prev) =>
-          prev.map((m) => {
-            if (m.interactiveCard?.cardId !== cardId || !m.interactiveCard) return m;
-            if (result.httpStatus === 409) {
-              return { ...m, interactiveCard: reconcileConflictCard(m.interactiveCard, result.plan) };
+        try {
+          const decisionNonce =
+            typeof crypto !== 'undefined' && typeof crypto.randomUUID === 'function'
+              ? crypto.randomUUID()
+              : `${cardId}:${actionId}:${expectedRevision}:${Date.now()}:${Math.random()}`;
+          const result = await submitIntentPlanDecision({
+            planId,
+            projectId,
+            sessionId,
+            expectedRevision,
+            decisionNonce,
+            decisionDigest: `${kind}:${expectedRevision}`,
+            kind,
+          });
+          setMessages((prev) =>
+            prev.map((m) => {
+              if (m.interactiveCard?.cardId !== cardId || !m.interactiveCard) return m;
+              if (result.httpStatus === 409) {
+                return { ...m, interactiveCard: reconcileConflictCard(m.interactiveCard, result.plan) };
+              }
+              if (result.plan) {
+                return { ...m, interactiveCard: applyPlanViewToCard(m.interactiveCard, result.plan) };
+              }
+              return m;
+            })
+          );
+          if (!result.plan && result.httpStatus >= 400) {
+            const latest = await fetchIntentPlan(planId, projectId, sessionId).catch(() => null);
+            if (latest) {
+              setMessages((prev) =>
+                prev.map((m) =>
+                  m.interactiveCard?.cardId === cardId && m.interactiveCard
+                    ? { ...m, interactiveCard: applyPlanViewToCard(m.interactiveCard, latest) }
+                    : m
+                )
+              );
+            } else {
+              setMessages((prev) =>
+                prev.map((m) =>
+                  m.interactiveCard?.cardId === cardId && m.interactiveCard
+                    ? {
+                        ...m,
+                        interactiveCard: markPlanCard(
+                          m.interactiveCard,
+                          'active',
+                          '计划确认失败，请重试'
+                        ),
+                      }
+                    : m
+                )
+              );
            }
-            if (result.plan) {
-              return { ...m, interactiveCard: applyPlanViewToCard(m.interactiveCard, result.plan) };
-            }
-            return m;
-          })
-        );
-        if (!result.plan && result.httpStatus >= 400) {
-          const latest = await fetchIntentPlan(planId, projectId, sessionId).catch(() => null);
-          if (latest) {
-            setMessages((prev) =>
-              prev.map((m) =>
-                m.interactiveCard?.cardId === cardId && m.interactiveCard
-                  ? { ...m, interactiveCard: applyPlanViewToCard(m.interactiveCard, latest) }
-                  : m
-              )
-            );
          }
+        } catch {
+          setMessages((prev) =>
+            prev.map((m) =>
+              m.interactiveCard?.cardId === cardId && m.interactiveCard
+                ? {
+                    ...m,
+                    interactiveCard: markPlanCard(
+                      m.interactiveCard,
+                      'active',
+                      '计划确认失败，请重试'
+                    ),
+                  }
+                : m
+            )
+          );
        }
      })();
      return;
@@ -18,6 +18,7 @@ test('chat and task attempts live in one hook', () => {
  assert.equal(hook.includes('streamChat'), true);
  assert.equal(hook.includes('createTaskRun'), true);
  assert.equal(hook.includes('shouldSubscribeExistingRun'), true);
+  assert.equal(hook.includes('removeTurn'), true);
  assert.equal(hook.includes('禁止无条件 setRunId(null)'), true);
  assert.equal(hook.includes('resumeTaskRun'), true);
  assert.equal(hook.includes('chatWindow.allMessages.find'), true);
@@ -170,7 +170,9 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) {

    const finishTurn = (status: 'completed' | 'failed' | 'cancelled') => {
      if (opts?.clientTurnId) {
+        // Terminal status previously left AbortController in the turn map forever.
        turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, status);
+        turnRuntime.removeTurn(streamSessionId, opts.clientTurnId);
      }
    };