From 7466a15319f35b457a325e5f3061028c5a44e7b3 Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 13 Sep 2026 13:54:49 +0000 Subject: [PATCH] fix(docgov): harden UI-03 sync poll, reconnectRunId, and E2E status attrs Use deliverablesRef + limited poll for post-task server docs; pass reconnectRunId across auto-retry; GetRun reads idempotency_key; StartClaimedRun fails closed on UpdateRun; MessageBubble/E2E scope this-turn assistant; correct audit retry/schema19 wording. Co-authored-by: Cursor --- ...6-09-13-controlled-doc-client-sync-ui03.md | 24 ++-- internal/taskrun/claim_idempotency_test.go | 42 ++++++ internal/taskrun/manager.go | 4 +- internal/taskrun/store.go | 12 +- web/e2e/document-revision-runtime.spec.ts | 90 ++++++------- web/src/App.tsx | 120 ++++++++++-------- web/src/components/chat/MessageBubble.tsx | 2 + web/src/lib/postTaskDeliverableSync.test.ts | 65 ++++++++++ web/src/lib/postTaskDeliverableSync.ts | 62 +++++++++ web/src/lib/taskRunReconnect.test.ts | 42 ++++++ web/src/lib/taskRunReconnect.ts | 38 ++++++ 11 files changed, 385 insertions(+), 116 deletions(-) create mode 100644 web/src/lib/taskRunReconnect.test.ts create mode 100644 web/src/lib/taskRunReconnect.ts diff --git a/.doc/audit/2026-09-13-controlled-doc-client-sync-ui03.md b/.doc/audit/2026-09-13-controlled-doc-client-sync-ui03.md index 024e1d5e..62e7c3dd 100644 --- a/.doc/audit/2026-09-13-controlled-doc-client-sync-ui03.md +++ b/.doc/audit/2026-09-13-controlled-doc-client-sync-ui03.md @@ -46,19 +46,19 @@ severity: critical 3. **真实 UI 为下拉/抽屉交互**： `SlimTabBar.tsx` 为面包屑下拉（`button.dropdown-trigger` -> `.dropdown-menu`）或大盘抽屉（`DeckDrawer`），页面不存在水平文档 Tab。 -### C. Playwright 串行 Retry 现象与 DB 约束异常暴露 +### C. 前端断线自动重试与 DB 约束异常暴露 -1. **Playwright 串行重试导致任务重发**： - Trace 与聊天现场显示：同一项目下，13:02 成功执行并完成首轮任务后，因重试机制或步骤重入，13:05 再次向同一个会话发送了相同的首轮指令： +1. **前端断线自动重试导致同 turn 二次触发（非 Playwright retries）**： + `web/playwright.config.ts` 中 **`retries: 0`**。现场 13:02 完成首轮后，13:05 再次出现同 prompt 处理中，根因是前端 `runTaskAttempt` 在连接异常路径上自动重试，且重试开头 `setRunId(null)` / 未显式传递 `reconnectRunId`，导致重复 `POST /tasks/runs` 而非订阅既有 run： - 13:02：`TRIGGER_TASK 请基于本项目编制首轮...` → `《用户需求表》已生成。当前状态：已完成` - - 13:05：`TRIGGER_TASK 请基于本项目编制首轮...` → 触发二次运行，助手再次进入处理中 + - 13:05：同 turn 自动重试路径再次进入处理中（工程修复：`reconnectRunId` 跨 attempt + Claim 幂等） 2. **暴露 `task_runs` 状态约束缺陷**： 在二次运行与高风险候选流转路径中，后台服务尝试将 `task_runs.status` 更新为 `pending_author_confirmation` 时，暴露了数据库底层约束拦截： - `task_runs` 表定义中的 `chk_task_run_status` 仅允许： + goose **`00019`** 原始 `chk_task_run_status` 仅允许： ```sql - CHECK (status IN ('queued', 'running', 'completed', 'failed', 'cancelled', 'quality_failed')) + CHECK (status IN ('pending', 'running', 'completed', 'failed', 'cancelled', 'finalizing_failed', 'quality_failed', 'interrupted')) ``` - 该枚举**未包含** `pending_author_confirmation`，导致写入状态时触发 PostgreSQL check constraint 违规。 + 该枚举**未包含** `pending_author_confirmation`，导致写入状态时触发 PostgreSQL check constraint 违规（由 goose `00030` 补齐）。 --- @@ -97,10 +97,10 @@ severity: critical | 根因 | 修复 | 门禁 | |---|---|---| -| A 任务完成后未 merge 受控文档 | `postTaskDeliverableSync` + `App.tsx` done/pending/reconnect 后 `listProjectWorkflowDocuments` → merge → 切到 `doc_proj*` | `npm test` 232 PASS；`npm run build` PASS | -| A′ intake vs URS 误匹配 | `templateMatch`：**intake 标记优先**于裸「用户需求」 | `templateMatch` + sync 单测（`npx tsx --test` 18/18） | -| B UI-03 选择器命中 Banner | E2E 改为 SlimTabBar/DeckDrawer 精确选择器；要求服务端 `doc_proj*` | 待真实 UI 重跑 | -| C `pending_author_confirmation` 约束 | goose `00030` 扩展 `chk_task_run_status` | **真实现卷迁移由合入前独立验证**（本提交不跑复杂 PG 集成测） | -| C′ 重试重复建 run / 重复写 chat | `ClaimRun` 先占 `idempotency_key`；非 owner 直接 202；FE reconnect 同 `run_id` 订阅 | `go test -short ./internal/taskrun ./internal/server` PASS | +| A 任务完成后未 merge 受控文档 | `pollUntilPostTaskServerDoc` + `deliverablesRef` 显式快照（禁止 setState updater 取 mergedSnap）+ 有限重试至 `doc_proj*` | focused + `npm test` / `build` | +| A′ intake vs URS 误匹配 | `templateMatch`：**intake 标记优先**于裸「用户需求」 | 单测 PASS | +| B UI-03 选择器命中 Banner | SlimTabBar/DeckDrawer；`data-chat-role/status` + 本轮 assistant sibling 等待；UI-11 禁止全局 div | 待真实 UI 重跑 | +| C `pending_author_confirmation` 约束 | goose `00030` 扩展 `chk_task_run_status`（相对 **00019** 枚举） | 真实现卷迁移合入前独立验证 | +| C′ 断线自动重试重复 POST | `reconnectRunId` 显式跨 retry；`ClaimRun`；`GetRun` 读回 `idempotency_key`；`StartClaimedRun` UpdateRun 失败不启 worker | `go test -short ./internal/taskrun` | **未做**：占槽 rebuild、真实 UI UI-03 重验、schema30 对现有重复 `(session_id,client_turn_id)` 卷的迁移实测。 diff --git a/internal/taskrun/claim_idempotency_test.go b/internal/taskrun/claim_idempotency_test.go index d143966d..1cc97d9e 100644 --- a/internal/taskrun/claim_idempotency_test.go +++ b/internal/taskrun/claim_idempotency_test.go @@ -220,3 +220,45 @@ func TestClaimThenStart_Concurrent_OneExecutor(t *testing.T) { t.Fatalf("want 1 persisted run, got %d", len(store.byID)) } } + +type failUpdateStore struct { + *memIdemStore + updateErr error +} + +func (f *failUpdateStore) UpdateRun(ctx context.Context, rec *RunRecord) error { + if f.updateErr != nil { + return f.updateErr + } + return f.memIdemStore.UpdateRun(ctx, rec) +} + +func TestStartClaimedRun_UpdateRunFailureDoesNotStartWorker(t *testing.T) { + base := newMemIdemStore() + store := &failUpdateStore{memIdemStore: base, updateErr: errors.New("persist boom")} + mgr := NewManager(nil, nil) + mgr.SetStore(store) + + rec := &RunRecord{ + SessionID: "sess-upd-fail", + ClientTurnID: "turn-upd-fail", + IdempotencyKey: BuildTaskRunIdempotencyKey("sess-upd-fail", "turn-upd-fail"), + UserID: 1, + TaskKey: "intake", + Status: "pending", + } + claimed, created, err := mgr.ClaimRun(context.Background(), rec) + if err != nil || !created { + t.Fatalf("claim: created=%v err=%v", created, err) + } + _, err = mgr.StartClaimedRun(context.Background(), claimed, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, "") + if err == nil { + t.Fatal("expected StartClaimedRun to fail when UpdateRun fails") + } + mgr.mu.Lock() + _, live := mgr.runs[claimed.RunID] + mgr.mu.Unlock() + if live { + t.Fatal("worker must not start when UpdateRun persistence fails") + } +} diff --git a/internal/taskrun/manager.go b/internal/taskrun/manager.go index d0845dc2..2efbd448 100644 --- a/internal/taskrun/manager.go +++ b/internal/taskrun/manager.go @@ -243,7 +243,9 @@ func (m *Manager) StartClaimedRun(parent context.Context, record *RunRecord, tas record.LeaseExpiresAt = &leaseUntil if m.store != nil && m.store.Available() { - _ = m.store.UpdateRun(parent, record) + if err := m.store.UpdateRun(parent, record); err != nil { + return nil, fmt.Errorf("taskrun: persist running status before start: %w", err) + } if err := m.store.AcquireLease(parent, record.RunID, m.workerID, defaultLeaseTTL); err != nil { slog.Warn("taskrun_lease_acquire_failed", "run_id", record.RunID, "error", err) } diff --git a/internal/taskrun/store.go b/internal/taskrun/store.go index 5ef80854..13abb602 100644 --- a/internal/taskrun/store.go +++ b/internal/taskrun/store.go @@ -191,15 +191,16 @@ func (s *PGStore) GetRun(ctx context.Context, runID string) (*RunRecord, error) } var rec RunRecord var projectID, assistantMsgID *int64 + var idempotencyKey *string if s.hasLeaseColumns(ctx) { err := s.pool.QueryRow(ctx, ` - SELECT run_id, session_id, project_id, user_id, client_turn_id, + SELECT run_id, session_id, project_id, user_id, client_turn_id, idempotency_key, task_key, stage, status, workspace_id, assistant_message_id, error_message, worker_id, lease_expires_at, attempt, created_at, updated_at FROM task_runs WHERE run_id = $1 `, runID).Scan( - &rec.RunID, &rec.SessionID, &projectID, &rec.UserID, &rec.ClientTurnID, + &rec.RunID, &rec.SessionID, &projectID, &rec.UserID, &rec.ClientTurnID, &idempotencyKey, &rec.TaskKey, &rec.Stage, &rec.Status, &rec.WorkspaceID, &assistantMsgID, &rec.ErrorMessage, &rec.WorkerID, &rec.LeaseExpiresAt, &rec.Attempt, &rec.CreatedAt, &rec.UpdatedAt, @@ -212,13 +213,13 @@ func (s *PGStore) GetRun(ctx context.Context, runID string) (*RunRecord, error) } } else { err := s.pool.QueryRow(ctx, ` - SELECT run_id, session_id, project_id, user_id, client_turn_id, + SELECT run_id, session_id, project_id, user_id, client_turn_id, idempotency_key, task_key, stage, status, workspace_id, assistant_message_id, error_message, created_at, updated_at FROM task_runs WHERE run_id = $1 `, runID).Scan( - &rec.RunID, &rec.SessionID, &projectID, &rec.UserID, &rec.ClientTurnID, + &rec.RunID, &rec.SessionID, &projectID, &rec.UserID, &rec.ClientTurnID, &idempotencyKey, &rec.TaskKey, &rec.Stage, &rec.Status, &rec.WorkspaceID, &assistantMsgID, &rec.ErrorMessage, &rec.CreatedAt, &rec.UpdatedAt, ) @@ -235,6 +236,9 @@ func (s *PGStore) GetRun(ctx context.Context, runID string) (*RunRecord, error) if assistantMsgID != nil { rec.AssistantMsgID = *assistantMsgID } + if idempotencyKey != nil { + rec.IdempotencyKey = *idempotencyKey + } return &rec, nil } diff --git a/web/e2e/document-revision-runtime.spec.ts b/web/e2e/document-revision-runtime.spec.ts index 71a21d1f..630cd2f3 100644 --- a/web/e2e/document-revision-runtime.spec.ts +++ b/web/e2e/document-revision-runtime.spec.ts @@ -222,40 +222,60 @@ async function adminCreateUser( } /** - * Squad wait: MUST observe start (user bubble + running/thinking) before terminal. - * failed status → FAIL. + * Squad wait: observe THIS turn's user bubble, then its following assistant sibling + * running → terminal. Never match a prior completed bubble globally. */ async function waitForSquadCompletion(page: Page, userPromptSnippet: string, timeoutMs = 420_000) { const startedAt = Date.now(); - const userBubble = page.locator('div, article, li').filter({ hasText: userPromptSnippet }).first(); + const userBubble = page + .locator('[data-chat-role="user"]') + .filter({ hasText: userPromptSnippet }) + .last(); await expect(userBubble).toBeVisible({ timeout: 30_000 }); - const running = page - .locator('text=正在思考') - .or(page.locator('text=生成中')) - .or(page.locator('.agent-pulse-active')) - .or(page.locator('text=执行中')) - .or(page.locator('text=编排中')); - await expect(running.first()).toBeVisible({ timeout: 60_000 }); + const assistant = userBubble.locator( + 'xpath=following-sibling::*[@data-chat-role="assistant"][1]', + ); + await expect(assistant).toBeVisible({ timeout: 30_000 }); - const terminal = page - .locator('text=已完成') - .or(page.locator('text=待编制人确认')) - .or(page.locator('text=失败')) - .or(page.locator('[data-status="completed"]')) - .or(page.locator('[data-status="pending_author_confirmation"]')) - .or(page.locator('[data-status="failed"]')); + await expect(assistant).toHaveAttribute( + 'data-chat-status', + /pending|streaming_status|auto_retrying/, + { timeout: 60_000 }, + ); - await expect(terminal.first()).toBeVisible({ timeout: timeoutMs }); + await expect(assistant).toHaveAttribute( + 'data-chat-status', + /completed|pending_author_confirmation|failed|cancelled|quality_failed/, + { timeout: timeoutMs }, + ); - if (await page.locator('text=失败').or(page.locator('[data-status="failed"]')).first().isVisible().catch(() => false)) { + const terminal = (await assistant.getAttribute('data-chat-status')) || ''; + if (terminal === 'failed') { fail('squad terminal status is failed'); } - // Ensure running indicator has cleared. - await expect(page.locator('text=正在思考，请稍候...')).toBeHidden({ timeout: 60_000 }).catch(() => undefined); return Date.now() - startedAt; } +async function openWorkflowDocByLabel(page: Page, label: RegExp) { + const trigger = page.locator('button.dropdown-trigger').first(); + if ((await trigger.count()) > 0) { + await trigger.click(); + const item = page + .locator('.dropdown-menu button, .dropdown-menu [role="menuitem"]') + .filter({ hasText: label }) + .first(); + if ((await item.count()) === 0) fail(`workflow slot not in SlimTabBar dropdown: ${label}`); + await item.click(); + return; + } + const deck = page.locator('[data-testid="deck-drawer"], .deck-drawer').first(); + if ((await deck.count()) > 0) await deck.click(); + const card = page.locator('[data-doc-id^="doc_proj"]').filter({ hasText: label }).first(); + if ((await card.count()) === 0) fail(`no server-backed workflow card for ${label}`); + await card.click(); +} + async function sendChat(page: Page, text: string) { const box = page.getByPlaceholder(/向项目助手提问/); await expect(box).toBeVisible({ timeout: 15_000 }); @@ -535,25 +555,7 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { const promptSlice = '请基于本项目编制首轮'; await waitForSquadCompletion(page, promptSlice); // 精确打开工作流文档槽：SlimTabBar 下拉 / DeckDrawer，禁止全局 div 泛匹配。 - const trigger = page.locator('button.dropdown-trigger').first(); - if ((await trigger.count()) > 0) { - await trigger.click(); - const item = page.locator('.dropdown-menu button, .dropdown-menu [role="menuitem"]').filter({ - hasText: /用户初始需求|需求摸底|intake/i, - }).first(); - if ((await item.count()) === 0) fail('UI-03: no intake slot in SlimTabBar dropdown'); - await item.click(); - } else { - const deck = page.locator('[data-testid="deck-drawer"], .deck-drawer').first(); - if ((await deck.count()) > 0) await deck.click(); - const card = page.locator('[data-doc-id^="doc_proj"], [data-doc-id*="intake"]').filter({ - hasText: /初始需求|intake/i, - }).first(); - if ((await card.count()) === 0) { - fail('UI-03: no server-backed intake deliverable card (doc_proj* / intake)'); - } - await card.click(); - } + await openWorkflowDocByLabel(page, /用户初始需求|需求摸底|intake/i); await expect(page.locator('[data-testid="doc-rev-version"]')).toBeVisible({ timeout: 60_000 }); await expect(page.getByTestId('doc-rev-template-hint')).toHaveCount(0); await expect(page.locator('text=版本账本未就绪')).toHaveCount(0); @@ -794,15 +796,13 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { '请另行生成一份《系统设计输入》受控文档（第二份），不要覆盖上一份需求文档，并给出可打开的交付物。'; await sendChat(page, prompt); await waitForSquadCompletion(page, '系统设计输入'); - const secondTab = page.locator('[role="tab"], button, div').filter({ hasText: /设计输入|系统设计/ }).first(); - if ((await secondTab.count()) === 0) fail('second controlled document tab not found after chat'); - await secondTab.click(); - secondDocTitle = (await secondTab.textContent())?.trim() || ''; + await openWorkflowDocByLabel(page, /设计输入|系统设计/); + secondDocTitle = '系统设计输入'; await expect(page.locator('[data-testid="doc-rev-version"]')).toBeVisible({ timeout: 45_000 }); const states: string[] = []; for (const label of [controlledTitle, secondDocTitle]) { if (!label) continue; - await page.getByText(label, { exact: false }).first().click(); + await openWorkflowDocByLabel(page, new RegExp(label.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'))); const { v, r } = await readVersionBadge(page); states.push(`${label}:${v}/${r}`); } diff --git a/web/src/App.tsx b/web/src/App.tsx index 2ed6abe0..87ea8eb6 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -49,9 +49,15 @@ import { } from './lib/projectMemberAdapter'; import { mergeWorkflowDocumentsFromServer } from './lib/deliverableSync'; import { - resolvePostTaskActiveDocId, + pollUntilPostTaskServerDoc, shouldSwitchTemplateSlotToServer, } from './lib/postTaskDeliverableSync'; +import { + isRemoteTaskTerminal, + nextAttemptOpts, + resolveReconnectRunId, + shouldSubscribeExistingRun, +} from './lib/taskRunReconnect'; import { buildControlledLoadErrorSkeleton, buildControlledLoadingSkeleton, @@ -584,6 +590,8 @@ export default function App() { const [deliverables, setDeliverables] = useState(() => loadDeliverables(ALL_INITIAL_DELIVERABLES) ); + const deliverablesRef = useRef(deliverables); + deliverablesRef.current = deliverables; // 自动持久化 deliverables（序列化时剔除不可持久化的 File 对象，见 storage.ts） useEffect(() => { @@ -2461,11 +2469,20 @@ export default function App() { asstId: string, mode: 'run' | 'resume' | 'restart', attemptIndex: number, - opts?: { phaseEvents?: PhaseEvent[]; workDir?: string } + opts?: { phaseEvents?: PhaseEvent[]; workDir?: string; reconnectRunId?: string } ) => { const sessionId = task.sessionId || activeSessionId; + let retainedRunId = resolveReconnectRunId({ + attemptIndex, + optsReconnectRunId: opts?.reconnectRunId, + }); taskRunStream.abortSession(sessionId); - taskRunStream.setRunId(sessionId, null); + // 重连保留显式 runId；禁止无条件 setRunId(null) 导致自动重试丢失 + if (retainedRunId) { + taskRunStream.setRunId(sessionId, retainedRunId); + } else { + taskRunStream.setRunId(sessionId, null); + } const controller = new AbortController(); taskRunStream.setController(sessionId, controller); @@ -2485,8 +2502,7 @@ export default function App() { taskRunStream.setThinking(sessionId, false); return; } - // 重试必须携带既有 run_id，禁止无条件新 POST - runTaskAttempt(task, asstId, mode, attemptIndex + 1, opts); + runTaskAttempt(task, asstId, mode, attemptIndex + 1, nextAttemptOpts(opts, retainedRunId)); }, delayMs); }; @@ -2507,49 +2523,44 @@ export default function App() { const syncControlledDocsAfterTask = async (taskKey: string, preferredDocId?: string | null) => { if (!task.projectId || task.projectId <= 0 || !taskKey) return; - const workflowDocs = await listProjectWorkflowDocuments(task.projectId).catch(() => []); - if (workflowDocs.length === 0) return; - let mergedSnap: DeliverableItem[] = []; - setDeliverables((prev) => { - const merged = mergeWorkflowDocumentsFromServer( - prev, - workflowDocs, - task.projectId!, - currentUser, - ); - mergedSnap = merged; - const serverBackedIds = merged - .filter((d) => d.projectId === task.projectId && shouldLoadControlledContent(d)) - .map((d) => d.docId); - if (serverBackedIds.length > 0) { - setDocCache((cache) => { - const next = { ...cache }; - serverBackedIds.forEach((docId) => { - delete next[docId]; - }); - return next; - }); - setControlledContentReady((ready) => { - const next = { ...ready }; - serverBackedIds.forEach((docId) => { - delete next[docId]; - }); - return next; - }); - } - return merged; - }); - const serverDocId = resolvePostTaskActiveDocId({ - deliverables: mergedSnap, + const { merged, serverDocId } = await pollUntilPostTaskServerDoc({ + listDocs: () => listProjectWorkflowDocuments(task.projectId!).catch(() => []), + getDeliverables: () => deliverablesRef.current, + merge: (prev, docs) => + mergeWorkflowDocumentsFromServer(prev, docs, task.projectId!, currentUser), projectId: task.projectId, taskKey, preferredDocId, + maxAttempts: 10, + delayMs: 800, + onMerged: (next) => { + deliverablesRef.current = next; + setDeliverables(next); + const serverBackedIds = next + .filter((d) => d.projectId === task.projectId && shouldLoadControlledContent(d)) + .map((d) => d.docId); + if (serverBackedIds.length === 0) return; + setDocCache((cache) => { + const cleared = { ...cache }; + serverBackedIds.forEach((docId) => { + delete cleared[docId]; + }); + return cleared; + }); + setControlledContentReady((ready) => { + const cleared = { ...ready }; + serverBackedIds.forEach((docId) => { + delete cleared[docId]; + }); + return cleared; + }); + }, }); if ( serverDocId && shouldSwitchTemplateSlotToServer({ activeDocId, - deliverables: mergedSnap, + deliverables: merged, projectId: task.projectId, taskKey, serverDocId, @@ -2632,7 +2643,7 @@ export default function App() { } case 'done': { settled = true; - const finishedRunId = taskRunStream.activeTaskRunId; + const finishedRunId = retainedRunId || taskRunStream.activeTaskRunId; taskRunStream.setRunId(sessionId, null); taskRunStream.setThinking(sessionId, false); void (async () => { @@ -2649,6 +2660,7 @@ export default function App() { status: nextStatus, livePhaseLabel: undefined, taskWorkspaceId: ev.done?.workspace_id, + ...(finishedRunId ? { taskRunId: finishedRunId } : {}), }); await syncControlledDocsAfterTask(task.taskKey || 'intake', preferredDocId); await chatWindow.refresh().catch(() => undefined); @@ -2667,25 +2679,21 @@ export default function App() { : buildTaskRequest(task); if (task.sessionId && task.projectId && task.projectId > 0 && mode === 'run') { - // 断线重试：若已有 run 仍在执行，只订阅，禁止新 POST - let existingRunId: string | undefined; - setMessages((prev) => { - const asst = prev.find((m) => m.id === asstId); - existingRunId = asst?.taskRunId || undefined; - return prev; + // 断线重试：显式 reconnectRunId 保留；禁止 setMessages updater 取 runId + const existingRunId = resolveReconnectRunId({ + attemptIndex, + optsReconnectRunId: retainedRunId || opts?.reconnectRunId, }); - if (!existingRunId) { - existingRunId = taskRunStream.activeTaskRunId || undefined; - } if (attemptIndex > 0 && existingRunId) { const remote = await getTaskRun(existingRunId).catch(() => null); - if (remote && (remote.status === 'running' || remote.status === 'pending')) { + if (remote && shouldSubscribeExistingRun(remote.status)) { + retainedRunId = existingRunId; taskRunStream.setRunId(sessionId, existingRunId); patchAsst({ taskRunId: existingRunId, livePhaseLabel: '正在重连任务…' }); await subscribeTaskRunEvents(existingRunId, { onEvent }, controller.signal); return; } - if (remote?.status === 'pending_author_confirmation' || remote?.status === 'completed') { + if (remote && isRemoteTaskTerminal(remote.status)) { settled = true; const mapped = mapTaskStatusToMessageStatus(remote.status, 'assistant') || 'completed'; patchAsst({ @@ -2708,6 +2716,7 @@ export default function App() { client_turn_id: task.clientTurnId, user_message: task.userMessage, }); + retainedRunId = created.run_id; taskRunStream.setRunId(sessionId, created.run_id); patchAsst({ taskRunId: created.run_id }); await subscribeTaskRunEvents(created.run_id, { onEvent }, controller.signal); @@ -2724,7 +2733,7 @@ export default function App() { taskRunStream.setThinking(sessionId, false); return; } - const runId = taskRunStream.activeTaskRunId; + const runId = retainedRunId || taskRunStream.activeTaskRunId || undefined; if (runId) { try { const remote = await getTaskRun(runId); @@ -2734,6 +2743,7 @@ export default function App() { status: 'completed', livePhaseLabel: undefined, taskWorkspaceId: remote.workspace_id, + taskRunId: runId, }); taskRunStream.setThinking(sessionId, false); await syncControlledDocsAfterTask(task.taskKey || 'intake'); @@ -2745,15 +2755,17 @@ export default function App() { status: 'pending_author_confirmation', livePhaseLabel: undefined, taskWorkspaceId: remote.workspace_id, + taskRunId: runId, }); taskRunStream.setThinking(sessionId, false); await syncControlledDocsAfterTask(task.taskKey || 'intake'); await chatWindow.refresh().catch(() => undefined); return; } - if (remote.status === 'running' || remote.status === 'pending') { + if (shouldSubscribeExistingRun(remote.status)) { // 原 run 仍在跑：重订阅，不 POST 新 run settled = false; + retainedRunId = runId; patchAsst({ livePhaseLabel: '正在重连任务…', taskRunId: runId }); await subscribeTaskRunEvents(runId, { onEvent }, controller.signal); return; diff --git a/web/src/components/chat/MessageBubble.tsx b/web/src/components/chat/MessageBubble.tsx index afcc5df9..4ff483a4 100644 --- a/web/src/components/chat/MessageBubble.tsx +++ b/web/src/components/chat/MessageBubble.tsx @@ -91,6 +91,8 @@ const MessageBubbleInner: React.FC = ({ id={message.serverId ? `chat-anchor-${message.serverId}` : (message.id ? `chat-anchor-${message.id}` : undefined)} data-server-id={message.serverId} data-category={message.category} + data-chat-role={message.role} + data-chat-status={status || (isUser ? 'user' : 'unknown')} className={isCategoryDimmed ? 'chat-category-dimmed' : undefined} onMouseEnter={() => setIsHovered(true)} onMouseLeave={() => setIsHovered(false)} diff --git a/web/src/lib/postTaskDeliverableSync.test.ts b/web/src/lib/postTaskDeliverableSync.test.ts index a6a2d363..e944fc4c 100644 --- a/web/src/lib/postTaskDeliverableSync.test.ts +++ b/web/src/lib/postTaskDeliverableSync.test.ts @@ -3,8 +3,10 @@ import assert from 'node:assert/strict'; import { resolvePostTaskActiveDocId, shouldSwitchTemplateSlotToServer, + pollUntilPostTaskServerDoc, } from './postTaskDeliverableSync'; import type { DeliverableItem } from '../types/workbench'; +import type { WorkflowDocumentSummary } from '../api/projectDocuments'; const base = (over: Partial): DeliverableItem => ({ docId: 'doc-x', @@ -88,3 +90,66 @@ test('shouldSwitchTemplateSlotToServer when active is local template slot', () = false, ); }); + +test('pollUntilPostTaskServerDoc retries until same taskKey server doc appears', async () => { + const local = [base({ docId: 'doc-p6089-intake', isTemplateOnly: true })]; + let calls = 0; + const snaps: DeliverableItem[][] = []; + const result = await pollUntilPostTaskServerDoc({ + listDocs: async () => { + calls += 1; + if (calls < 3) return []; + return [ + { + doc_id: 'doc_proj6089_intake', + project_id: 6089, + task_key: 'intake', + title: '用户初始需求收集表', + lifecycle_state: 'drafting', + is_template_only: false, + } satisfies WorkflowDocumentSummary, + ]; + }, + getDeliverables: () => local, + merge: (prev, docs) => { + if (docs.length === 0) return prev; + return [ + ...prev.filter((d) => d.docId !== 'doc_proj6089_intake'), + base({ + docId: docs[0].doc_id, + isTemplateOnly: false, + taskKey: docs[0].task_key, + }), + ]; + }, + projectId: 6089, + taskKey: 'intake', + maxAttempts: 5, + delayMs: 1, + sleep: async () => undefined, + onMerged: (merged) => { + snaps.push(merged); + }, + }); + assert.equal(result.serverDocId, 'doc_proj6089_intake'); + assert.equal(result.attempts, 3); + assert.ok(snaps.length >= 3); + assert.ok(snaps[snaps.length - 1].some((d) => d.docId === 'doc_proj6089_intake')); +}); + +test('pollUntilPostTaskServerDoc returns null after maxAttempts without relying on setState', async () => { + const local = [base({ docId: 'doc-p6089-intake', isTemplateOnly: true })]; + const result = await pollUntilPostTaskServerDoc({ + listDocs: async () => [], + getDeliverables: () => local, + merge: (prev) => prev, + projectId: 6089, + taskKey: 'intake', + maxAttempts: 3, + delayMs: 1, + sleep: async () => undefined, + }); + assert.equal(result.serverDocId, null); + assert.equal(result.attempts, 3); + assert.equal(result.merged[0]?.docId, 'doc-p6089-intake'); +}); diff --git a/web/src/lib/postTaskDeliverableSync.ts b/web/src/lib/postTaskDeliverableSync.ts index 545bb754..f49a040d 100644 --- a/web/src/lib/postTaskDeliverableSync.ts +++ b/web/src/lib/postTaskDeliverableSync.ts @@ -1,4 +1,5 @@ import type { DeliverableItem } from '../types/workbench'; +import type { WorkflowDocumentSummary } from '../api/projectDocuments'; /** * 任务完成后：在已 merge 的 deliverables 中定位同 taskKey 的服务端受控文档槽， @@ -63,3 +64,64 @@ export function shouldSwitchTemplateSlotToServer(params: { if (active.projectId !== projectId || active.taskKey !== taskKey) return false; return Boolean(active.isTemplateOnly) || active.docId.startsWith('doc-p'); } + +export type PollUntilPostTaskServerDocParams = { + listDocs: () => Promise; + /** 每次 attempt 读取当前本地快照（应用侧用 ref.current，禁止依赖 setState updater 回写）。 */ + getDeliverables: () => DeliverableItem[]; + merge: (prev: DeliverableItem[], docs: WorkflowDocumentSummary[]) => DeliverableItem[]; + projectId: number; + taskKey: string; + preferredDocId?: string | null; + maxAttempts?: number; + delayMs?: number; + sleep?: (ms: number) => Promise; + /** 每轮 merge 后回调（应用侧同步写 ref + setDeliverables(merged)）。 */ + onMerged?: (merged: DeliverableItem[]) => void; +}; + +/** + * Finalize 后文档可能尚未立即可见：有限重试 list+merge，直到同 taskKey 出现非模板 server doc。 + * 返回最后一次 merged 快照与解析到的 serverDocId（未出现则为 null）。 + */ +export async function pollUntilPostTaskServerDoc( + params: PollUntilPostTaskServerDocParams, +): Promise<{ merged: DeliverableItem[]; serverDocId: string | null; attempts: number }> { + const { + listDocs, + getDeliverables, + merge, + projectId, + taskKey, + preferredDocId, + maxAttempts = 10, + delayMs = 800, + sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)), + onMerged, + } = params; + + let merged = getDeliverables(); + let serverDocId: string | null = null; + let attempts = 0; + + for (let i = 0; i < maxAttempts; i++) { + attempts = i + 1; + const docs = await listDocs(); + merged = merge(getDeliverables(), docs); + onMerged?.(merged); + serverDocId = resolvePostTaskActiveDocId({ + deliverables: merged, + projectId, + taskKey, + preferredDocId, + }); + if (serverDocId) { + return { merged, serverDocId, attempts }; + } + if (i + 1 < maxAttempts) { + await sleep(delayMs); + } + } + + return { merged, serverDocId: null, attempts }; +} diff --git a/web/src/lib/taskRunReconnect.test.ts b/web/src/lib/taskRunReconnect.test.ts new file mode 100644 index 00000000..8bfb8212 --- /dev/null +++ b/web/src/lib/taskRunReconnect.test.ts @@ -0,0 +1,42 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { + resolveReconnectRunId, + shouldSubscribeExistingRun, + isRemoteTaskTerminal, + nextAttemptOpts, +} from './taskRunReconnect'; + +test('resolveReconnectRunId prefers explicit opts and ignores attempt without opts', () => { + assert.equal( + resolveReconnectRunId({ attemptIndex: 1, optsReconnectRunId: 'run_abc' }), + 'run_abc', + ); + assert.equal(resolveReconnectRunId({ attemptIndex: 0, optsReconnectRunId: null }), undefined); + assert.equal(resolveReconnectRunId({ attemptIndex: 2, optsReconnectRunId: ' ' }), undefined); + assert.equal(resolveReconnectRunId({ attemptIndex: 3 }), undefined); +}); + +test('shouldSubscribeExistingRun only for in-flight statuses', () => { + assert.equal(shouldSubscribeExistingRun('running'), true); + assert.equal(shouldSubscribeExistingRun('pending'), true); + assert.equal(shouldSubscribeExistingRun('completed'), false); + assert.equal(shouldSubscribeExistingRun('failed'), false); +}); + +test('isRemoteTaskTerminal covers confirmation and quality_failed', () => { + assert.equal(isRemoteTaskTerminal('pending_author_confirmation'), true); + assert.equal(isRemoteTaskTerminal('completed'), true); + assert.equal(isRemoteTaskTerminal('quality_failed'), true); + assert.equal(isRemoteTaskTerminal('running'), false); +}); + +test('nextAttemptOpts carries retained runId for reconnect without new POST', () => { + const next = nextAttemptOpts<{ reconnectRunId?: string; workDir?: string }>( + { workDir: '/w' }, + 'run_keep', + ); + assert.equal(next.reconnectRunId, 'run_keep'); + assert.equal(next.workDir, '/w'); + assert.equal(nextAttemptOpts(undefined, undefined).reconnectRunId, undefined); +}); diff --git a/web/src/lib/taskRunReconnect.ts b/web/src/lib/taskRunReconnect.ts new file mode 100644 index 00000000..237d28e5 --- /dev/null +++ b/web/src/lib/taskRunReconnect.ts @@ -0,0 +1,38 @@ +/** + * 长任务断线自动重试：用显式 reconnectRunId 跨 attempt 保留， + * 禁止依赖 setState updater / 已被清空的 stream.runId。 + */ + +export function resolveReconnectRunId(input: { + attemptIndex: number; + optsReconnectRunId?: string | null; +}): string | undefined { + const fromOpts = input.optsReconnectRunId?.trim(); + if (fromOpts) return fromOpts; + // attempt 0 且无显式 id → 允许新 POST；重试必须靠 opts 传入 + return undefined; +} + +export function shouldSubscribeExistingRun(status: string | null | undefined): boolean { + return status === 'running' || status === 'pending'; +} + +export function isRemoteTaskTerminal(status: string | null | undefined): boolean { + return ( + status === 'completed' || + status === 'pending_author_confirmation' || + status === 'quality_failed' + ); +} + +/** 自动重试调度时把已知 runId 写入下一轮 opts，避免 abort 后丢失。 */ +export function nextAttemptOpts( + opts: T | undefined, + retainedRunId: string | undefined, +): T & { reconnectRunId?: string } { + const base = { ...(opts || ({} as T)) }; + if (retainedRunId) { + return { ...base, reconnectRunId: retainedRunId }; + } + return base; +}