From 57873f6c84ea9c9316febe651bc8a5cfeabbfd85 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 19 Sep 2026 12:05:01 +0000 Subject: [PATCH] refactor(web): retry chat and task runs from one hook Keep reconnect on the existing run id so a dropped stream does not start a second task. Co-authored-by: Cursor --- .../long-term/2026-09-19-audit-remainder.md | 2 +- web/src/App.tsx | 754 +--------------- web/src/hooks/useTaskRunAttempts.test.ts | 15 + web/src/hooks/useTaskRunAttempts.ts | 805 ++++++++++++++++++ 4 files changed, 849 insertions(+), 727 deletions(-) create mode 100644 web/src/hooks/useTaskRunAttempts.test.ts create mode 100644 web/src/hooks/useTaskRunAttempts.ts diff --git a/.doc/ops/long-term/2026-09-19-audit-remainder.md b/.doc/ops/long-term/2026-09-19-audit-remainder.md index cbba840d..b819f382 100644 --- a/.doc/ops/long-term/2026-09-19-audit-remainder.md +++ b/.doc/ops/long-term/2026-09-19-audit-remainder.md @@ -46,7 +46,7 @@ updated: 2026-09-19 | FE 缓存 | done（版本字段）。窗口缓存写入 `{schema,data}`，无 schema 的旧载荷丢弃。`window cache: 无 schema` | | FE lucide | 关闭理由：`package-lock.json` 锁定 `lucide-react@1.42.0`，来源 `registry.npmjs.org`，带 integrity。`^1.16.0` 是合法的 1.x 范围，不是来源不明的包 | | M7 | done（出站意图 + 同键重放）。`document_content_projections` 与账本同一事务写入；投影成功才销账。`TestDocRevision_PatchOrder_RetryProjectsWithoutSecondLedgerCommit`。账本库与正文库仍不是一个 SQL 事务，进程退出后靠未投影行重放，不靠客户端记忆 | -| FE App.tsx | 部分。登录壳不再静态引用重模块。本地缓存、登录同步、文档补水、服务端文档同步、生命周期、发送消息和块编辑已分别抽出。立项、卡片动作和任务流仍在 `App.tsx` | +| FE App.tsx | 部分。登录壳不再静态引用重模块。本地缓存、登录同步、文档补水、服务端文档同步、生命周期、发送消息、块编辑，以及对话流/任务流重试已分别抽出。立项和卡片动作仍在 `App.tsx` | | H1-del | 关闭理由：`allowLegacyTestPrincipal` 仍被 `AllowInMemory && ContextDbHost==""` 的 dev/test 内存测试使用。删除会把这批测试变成未登录 401。保留告警，删除另开、先改测试夹具 | 仍开放：rdev P3 三样本真 PI、首次加载 U1/U6/U7、dev 容器重建与治理库 PG。公式引擎已离开入口包（`index-j0BIrjIR.js` 737.38 kB，br 170,451 B，包内无 katex），U3 仍缺浏览器瀑布，不关。P4 键名空间化已在本分支落地：新清单用带前缀的任务键，旧键与旧 doc_id 仍可读，`TestPublishedTaskKeysAreNamespacedAndDisjoint`。意图短语仍输出旧键，执行路径按别名接上。P2 规划界面已折叠长阶段、可批量纳入人工审批，并在选中 rdev 时提示 CDO 前置。U4 登录扇出收口仍有效，浏览器请求次数证据还没有。2026-09-19 在 `fix/audit-remainder-closeout` 跑过 `cd web && npm test && npm run build`：node:test 414 pass / 0 fail，vitest 32 pass；入口 `index-BFf6n8IA.js` 1,005.01 kB，br 235,285 B。CURRENT 文首 schema 52 与三套容器一致，00053 要等重建后才升。H2 批注/审计的项目成员检查在 Postgres 上走 `RequireProjectAccess`；内存授权器在无库时把任意正用户 id 视为成员，这是 `TestAuthorizeDocRevision_ReviewerAndNonMemberCannotWrite` 的隔离前提，不能改成默认拒绝。H2 的 HTTP 负例因此要在 `TEST_DATABASE_URL` 上补，本机未设置该变量。M7 的投影意图已随账本提交（goose `00053`）；容器 schema 仍是 52，要等占槽重建后才升到 53。 diff --git a/web/src/App.tsx b/web/src/App.tsx index 491341e7..09632ed6 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -17,7 +17,6 @@ import { deleteFreeformDoc, } from './api/freeformDoc'; import { upsertChatSession } from './api/chatSessions'; -import { listProjectWorkflowDocuments } from './api/projectDocuments'; import type { DiffResult, OverlayMode } from './features/documentRevision/types'; import { documentRevisionInactiveHint, @@ -43,25 +42,10 @@ import { type CreateProjectSubmitPhase, } from './lib/createProjectSubmitGuard'; import { mapApiMembersToProjectMembers } from './lib/projectMemberAdapter'; -import { mergeWorkflowDocumentsFromServer } from './lib/deliverableSync'; import { resolveCurrentDocRole } from './lib/currentDocRole'; import { decideModificationProposal, } from './api/docgov'; -import { - pollUntilPostTaskServerDoc, - shouldSwitchTemplateSlotToServer, -} from './lib/postTaskDeliverableSync'; -import { - isRemoteTaskTerminal, - nextAttemptOpts, - resolveReconnectRunId, - shouldSubscribeExistingRun, -} from './lib/taskRunReconnect'; -import { - GOVERNANCE_BLOCKED_HINT, - isBlockedGovernanceStatus, -} from './lib/blockedGovernanceCopy'; import { buildControlledLoadingSkeleton, controlledHydrateFingerprint, @@ -69,9 +53,7 @@ import { shouldLoadControlledContent, } from './lib/controlledDocHydrate'; import { - controlledHydrateCandidateIds, dropRecordKeys, - planControlledContentInvalidation, type ControlledInvalidationPlan, } from './lib/controlledContentInvalidation'; import { @@ -100,11 +82,6 @@ import { ShieldAlert } from 'lucide-react'; import { LoginView } from './components/auth/LoginView'; import { FirstLoginPasswordModal } from './components/auth/FirstLoginPasswordModal'; import { authClient } from './api/auth'; -import { streamChat, mapErrorInfo } from './api/chatStream'; -import { streamTask, buildTaskRequest, buildResumeRequest } from './api/taskStream'; -import type { StreamEvent } from './api/chatStream'; -import { shouldAutoRetryChat, shouldAutoRetryTask, autoRetryDelayMs, fallbackErrorClass } from './api/chatRetry'; -import { applyPhaseEvent } from './engine/phaseEventMapper'; import { LoginResult, SystemAuthConfig } from './types/auth'; import { markdownToBlockAST } from './engine/markdownToBlock'; import { blockASTToMarkdown } from './engine/blockToMarkdown'; @@ -112,12 +89,9 @@ import { DocumentAST, DocumentBlock } from './types/block'; import { ProjectSession, ChatMessage, - ChatErrorInfo, DeliverableItem, DeliverableCategory, InteractiveCardData, - PhaseEvent, - TaskRunContext, ProjectInfo, ProjectMember, ProjectMemberRole, @@ -158,7 +132,6 @@ import { getTemplateDefinitionByTaskKey, buildTemplateASTForProject, } from './engine/templateFactory'; -import { resolveTaskTemplateForMessage } from './engine/templateMatch'; import { loadSessions, loadActiveSessionId, @@ -176,12 +149,12 @@ import { useAuthenticatedWorkspaceSync } from './hooks/useAuthenticatedWorkspace import { useBlockEditing } from './hooks/useBlockEditing'; import { useControlledDocumentHydration } from './hooks/useControlledDocumentHydration'; import { useDocumentLifecycleActions } from './hooks/useDocumentLifecycleActions'; +import { useTaskRunAttempts } from './hooks/useTaskRunAttempts'; import { useSendMessage } from './hooks/useSendMessage'; import { useServerDocumentSync } from './hooks/useServerDocumentSync'; import { useWorkbenchPersistence } from './hooks/useWorkbenchPersistence'; import { useTurnRuntime } from './hooks/useTurnRuntime'; import { useTaskRunStream } from './hooks/useTaskRunStream'; -import { fetchChatTurnStatus } from './api/chatTurn'; import { mergeSessionUser } from './lib/mergeSessionUser'; import { RequestEpoch, shareInflight } from './data/initFanout'; import { @@ -194,10 +167,10 @@ import { DigestTimelineModal } from './components/history/DigestTimelineModal'; import { navigateToAnchor, parseAnchorHash, resolveTurnAnchorId } from './engine/anchorNavigator'; import { useUnreadMarks } from './hooks/useUnreadMarks'; import { CategoryDock } from './components/chat/CategoryDock'; -import { fetchChatCategories, fetchChatDates, mapTaskStatusToMessageStatus } from './api/chatWindow'; -import { ChatCategoryStat, ChatDateStat, QuotedMessage, AssistantMessageStatus } from './types/workbench'; +import { fetchChatCategories, fetchChatDates } from './api/chatWindow'; +import { ChatCategoryStat, ChatDateStat, QuotedMessage } from './types/workbench'; import { saveFile } from './data/fileStore'; -import { subscribeTaskRunEvents, createTaskRun, cancelTaskRun, getTaskRun } from './api/taskRuns'; +import { cancelTaskRun, getTaskRun } from './api/taskRuns'; export { ALL_INITIAL_DELIVERABLES, @@ -1688,701 +1661,30 @@ export default function App() { }; // 合成网络级异常的结构化错误信息 (fetch/stream 抛出而非后端 error 事件时使用) - const buildNetworkErrorInfo = ( - httpStatus?: number, - opts?: { errorCode?: string; detail?: string }, - ): ChatErrorInfo => { - const code = opts?.errorCode || ''; - const detail = opts?.detail || ''; - if (code === 'DOC_BUSY' || code === 'document_busy') { - return { - userMessage: detail || '该文档正在被其他用户或项目助手编辑，当前请求未创建新任务', - code: 'DOC_BUSY', - class: 'fatal', - actions: [], - }; - } - if (code === 'DOC_TASK_NOT_IN_ACTIVE_STAGE') { - return { - userMessage: detail || '该文档任务不在当前活动阶段，请先完成当前阶段工作流后再试', - code, - class: 'transient', - actions: ['retry'], - }; - } - if (code === 'DOC_DUTY_FORBIDDEN') { - return { - userMessage: detail || '当前账号不是该文档任务的职责人，无法执行', - code, - class: 'fatal', - actions: ['retry'], - }; - } - if (httpStatus === 401) { - return { - userMessage: '登录已失效，请重新登录后再试', - code: 'ERR_UNAUTHENTICATED', - class: 'fatal', - actions: ['retry'], - }; - } - if (httpStatus === 403) { - if (code === 'ERR_FORBIDDEN' || code === 'AUTHZ') { - return { - userMessage: detail || '无权访问该项目，请确认登录账号与项目成员身份', - code: code || 'ERR_FORBIDDEN', - class: 'fatal', - actions: ['retry'], - }; - } - return { - userMessage: detail || '无权执行该操作，请确认账号权限与工作流阶段', - code: code || 'ERR_FORBIDDEN', - class: 'fatal', - actions: ['retry'], - }; - } - return { - userMessage: '网络连接异常，请检查网络后重试', - code: 'NETWORK_ERROR', - class: fallbackErrorClass({ network: true }), - actions: ['retry'], - }; - }; - - // 驱动一次真实后端 SSE 流式对话尝试：气泡状态机 pending → streaming_status → completed/failed/cancelled。 - // 闲聊路径在无 client_turn_id 幂等前禁止自动重放 POST；失败仅展示手动「重试」。 - type StreamAttemptOpts = { - userMsgId?: string; - clientTurnId?: string; - quote?: QuotedMessage; - }; - - const runStreamAttempt = ( - userText: string, - asstId: string, - attemptIndex: number, - opts?: StreamAttemptOpts - ) => { - const quote = opts?.quote; - const controller = new AbortController(); - const streamSessionId = activeSessionId; - if (opts?.clientTurnId) { - turnRuntime.registerTurn({ - clientTurnId: opts.clientTurnId, - sessionId: streamSessionId, - userMsgId: opts.userMsgId || '', - asstId, - status: 'pending', - controller, - startedAt: Date.now(), - }); - } - - const currentProj = sessions.find((s) => s.sessionId === streamSessionId); - let accumulated = ''; - let settled = false; - - const updateAsst = (patch: Partial) => { - setMessages((prev) => prev.map((m) => (m.id === asstId ? { ...m, ...patch } : m))); - }; - - const scheduleRetry = (delayMs: number) => { - setTimeout(() => { - if (controller.signal.aborted) { - updateAsst({ status: 'cancelled', livePhaseLabel: undefined }); - if (opts?.clientTurnId) { - turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'cancelled'); - } - return; - } - runStreamAttempt(userText, asstId, attemptIndex + 1, opts); - }, delayMs); - }; - - const finishTurn = (status: 'completed' | 'failed' | 'cancelled') => { - if (opts?.clientTurnId) { - turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, status); - } - }; - - const failOrRetry = (info: ChatErrorInfo) => { - settled = true; - if (shouldAutoRetryChat(info, attemptIndex)) { - const delayMs = autoRetryDelayMs(attemptIndex); - updateAsst({ - status: 'auto_retrying', - livePhaseLabel: `连接异常，${Math.round(delayMs / 1000)}s 后自动重试…`, - }); - scheduleRetry(delayMs); - return; - } - updateAsst({ - status: 'failed', - errorInfo: { - ...info, - userMessage: - attemptIndex > 0 - ? `${info.userMessage}（请确认上一请求是否已在处理后再手动重试）` - : info.userMessage, - }, - livePhaseLabel: undefined, - }); - finishTurn('failed'); - }; - - updateAsst({ - status: 'pending', - livePhaseLabel: attemptIndex > 0 ? '正在手动重试…' : '助手正在处理…', - content: '', - errorInfo: undefined, - }); - if (opts?.clientTurnId) { - turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'pending'); - } - - const bindUserServerId = (serverId: number) => { - if (!opts?.userMsgId) return; - chatWindow.replaceMessageServerId(opts.userMsgId, serverId); - setMessages((prev) => - prev.map((m) => (m.id === opts.userMsgId ? { ...m, serverId } : m)) - ); - }; - - streamChat( - { - session_id: streamSessionId, - user_id: currentUser.id, - project_id: currentProj?.projectId || currentProjectId || 1, - project_name: currentProj?.title || '肌骨超声AI辅助诊断系统', - device_type: currentProj?.deviceType || '肌骨超声', - stage: currentProj?.stage || 'V0.5', - user_name: currentUser.name, - user_role: currentDocRole, - message: userText, - client_turn_id: opts?.clientTurnId, - quote: quote, - }, - { - onEvent: (ev) => { - if (controller.signal.aborted || settled) return; - switch (ev.type) { - case 'turn_ack': { - const userServerId = ev.turn?.user_message_id ?? ev.user_message_id; - if (typeof userServerId === 'number' && userServerId > 0) { - bindUserServerId(userServerId); - } - break; - } - case 'phase': - case 'status': { - const label = ev.phase?.message || ev.status?.message; - updateAsst({ status: 'streaming_status', livePhaseLabel: label || '助手正在处理…' }); - if (opts?.clientTurnId) { - turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'streaming'); - } - break; - } - case 'delta': { - accumulated += ev.delta || ''; - updateAsst({ status: 'streaming_status', content: accumulated }); - if (opts?.clientTurnId) { - turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'streaming'); - } - break; - } - case 'error': { - failOrRetry(mapErrorInfo(ev)); - break; - } - case 'done': { - settled = true; - const userServerId = ev.done?.user_message_id ?? ev.user_message_id; - const serverMsgId = ev.done?.message_id ?? ev.message_id ?? (ev as { messageId?: number }).messageId; - if (typeof userServerId === 'number' && userServerId > 0) { - bindUserServerId(userServerId); - } - if (typeof serverMsgId === 'number' && serverMsgId > 0) { - chatWindow.replaceMessageServerId(asstId, serverMsgId); - } - const exportCap = ev.export_capability || ev.done?.export_capability; - updateAsst({ - status: 'completed', - livePhaseLabel: undefined, - serverId: typeof serverMsgId === 'number' ? serverMsgId : undefined, - exportCapability: exportCap, - }); - finishTurn('completed'); - break; - } - default: - break; - } - }, - }, - controller.signal, - opts?.clientTurnId ? { idempotencyKey: opts.clientTurnId } : undefined - ).catch(async (_err) => { - if (settled) return; - if (controller.signal.aborted) { - updateAsst({ status: 'cancelled', livePhaseLabel: undefined }); - finishTurn('cancelled'); - return; - } - if (opts?.clientTurnId) { - try { - const turnStatus = await fetchChatTurnStatus(streamSessionId, opts.clientTurnId); - if (turnStatus?.status === 'completed' && turnStatus.assistantMessageId) { - settled = true; - if (turnStatus.userMessageId) bindUserServerId(turnStatus.userMessageId); - chatWindow.replaceMessageServerId(asstId, turnStatus.assistantMessageId); - updateAsst({ - status: 'completed', - livePhaseLabel: undefined, - serverId: turnStatus.assistantMessageId, - }); - finishTurn('completed'); - return; - } - if (turnStatus?.status === 'processing') { - updateAsst({ - status: 'streaming_status', - livePhaseLabel: '连接中断，服务端仍在处理…', - }); - if (opts?.clientTurnId) { - turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'streaming'); - } - return; - } - } catch { - // fall through to structured failure - } - } - const httpStatus = - typeof _err === 'object' && _err !== null && 'httpStatus' in _err - ? Number((_err as { httpStatus?: number }).httpStatus) - : undefined; - const errMeta = _err as { errorCode?: string; detail?: string; message?: string }; - failOrRetry( - buildNetworkErrorInfo(httpStatus, { - errorCode: errMeta.errorCode, - detail: errMeta.detail || errMeta.message?.split(': ').slice(1).join(': '), - }), - ); - }); - }; - - // P2：长任务尝试模式。'run' 新任务 | 'resume' 续跑失败工作区 | 'restart' 按原意图新工作区 - const runTaskAttempt = ( - task: TaskRunContext, - asstId: string, - mode: 'run' | 'resume' | 'restart', - attemptIndex: number, - opts?: { phaseEvents?: PhaseEvent[]; workDir?: string; reconnectRunId?: string } - ) => { - const sessionId = task.sessionId || activeSessionId; - let retainedRunId = resolveReconnectRunId({ - attemptIndex, - optsReconnectRunId: opts?.reconnectRunId, - }); - taskRunStream.abortSession(sessionId); - // 重连保留显式 runId；禁止无条件 setRunId(null) 导致自动重试丢失 - if (retainedRunId) { - taskRunStream.setRunId(sessionId, retainedRunId); - } else { - taskRunStream.setRunId(sessionId, null); - } - const controller = new AbortController(); - taskRunStream.setController(sessionId, controller); - - let accumulated = ''; - let settled = false; - - const patchAsst = (patch: Partial | ((m: ChatMessage) => Partial)) => { - setMessages((prev) => - prev.map((m) => (m.id === asstId ? { ...m, ...(typeof patch === 'function' ? patch(m) : patch) } : m)) - ); - }; - - const scheduleRetry = (delayMs: number) => { - setTimeout(() => { - if (controller.signal.aborted) { - patchAsst({ status: 'cancelled', livePhaseLabel: undefined }); - taskRunStream.setThinking(sessionId, false); - return; - } - runTaskAttempt(task, asstId, mode, attemptIndex + 1, nextAttemptOpts(opts, retainedRunId)); - }, delayMs); - }; - - const failOrRetry = (info: ChatErrorInfo) => { - settled = true; - if (shouldAutoRetryTask(info, attemptIndex)) { - const delayMs = autoRetryDelayMs(attemptIndex); - patchAsst({ - status: 'auto_retrying', - livePhaseLabel: `连接异常，${Math.round(delayMs / 1000)}s 后自动重试…`, - }); - scheduleRetry(delayMs); - return; - } - patchAsst({ status: 'failed', errorInfo: info, livePhaseLabel: undefined }); - taskRunStream.setThinking(sessionId, false); - }; - - const syncControlledDocsAfterTask = async (taskKey: string, preferredDocId?: string | null) => { - if (!task.projectId || task.projectId <= 0 || !taskKey) return; - 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); - applyControlledInvalidationPlan( - planControlledContentInvalidation({ - reason: 'post_task', - candidateIds: controlledHydrateCandidateIds(next, task.projectId), - activeDocId: activeDocIdRef.current, - editorOpen: editorHydrationGateRef.current.isEditorOpen(), - }), - ); - }, - }); - if ( - serverDocId && - shouldSwitchTemplateSlotToServer({ - activeDocId, - deliverables: merged, - projectId: task.projectId, - taskKey, - serverDocId, - }) - ) { - setActiveDocId(serverDocId); - setActiveCategory('workflow'); - } - }; - - // 等待态：resume 保留已有 phaseEvents 续接；run/restart 重置正文与步骤条 - patchAsst({ - status: 'pending', - livePhaseLabel: - mode === 'resume' ? '正在继续任务…' : mode === 'restart' ? '正在重新开始任务…' : '任务执行中…', - errorInfo: undefined, - ...(mode === 'resume' || attemptIndex > 0 - ? {} - : { content: '', phaseEvents: [], taskWorkspaceId: undefined }), - }); - taskRunStream.setThinking(sessionId, true); - - const onEvent = (ev: StreamEvent) => { - if (controller.signal.aborted || settled) return; - switch (ev.type) { - case 'phase': { - patchAsst((m) => ({ - status: 'streaming_status', - livePhaseLabel: ev.phase?.message || m.livePhaseLabel, - phaseEvents: applyPhaseEvent(opts?.phaseEvents || m.phaseEvents || [], ev.phase), - })); - break; - } - case 'status': { - if (ev.status?.message) { - patchAsst({ status: 'streaming_status', livePhaseLabel: ev.status.message }); - } - break; - } - case 'delta': { - accumulated += ev.delta || ''; - patchAsst({ status: 'streaming_status', content: accumulated }); - break; - } - case 'deliverable': { - const title = ev.deliverable?.title; - if (title) { - accumulated += (accumulated ? '\n\n' : '') + `📄 已生成文档：**${title}**`; - patchAsst({ status: 'streaming_status', content: accumulated }); - } - break; - } - case 'governance': { - const governance = ev.governance; - const docId = governance?.doc_id; - if (docId) { - const leaseReleased = governance.action === 'lease_released'; - const editLease = leaseReleased - ? undefined - : { holderType: 'squad' as const, holderId: taskRunStream.activeTaskRunId || undefined }; - setDeliverables((prev) => - prev.map((doc) => (doc.docId === docId ? { ...doc, editLease } : doc)), - ); - setDocCache((prev) => { - const ast = prev[docId]; - return ast ? { ...prev, [docId]: { ...ast, editLease } } : prev; - }); - } - if (governance?.notification_msg) { - patchAsst({ - status: 'streaming_status', - livePhaseLabel: governance.notification_msg, - }); - } - break; - } - case 'error': { - failOrRetry(mapErrorInfo(ev)); - break; - } - case 'done': { - settled = true; - const finishedRunId = retainedRunId || taskRunStream.activeTaskRunId; - taskRunStream.setRunId(sessionId, null); - taskRunStream.setThinking(sessionId, false); - void (async () => { - let nextStatus: AssistantMessageStatus = 'completed'; - let preferredDocId: string | null = null; - if (finishedRunId) { - const remote = await getTaskRun(finishedRunId).catch(() => null); - if (remote?.status) { - const mapped = mapTaskStatusToMessageStatus(remote.status, 'assistant'); - if (mapped) nextStatus = mapped; - } - } - patchAsst({ - status: nextStatus, - livePhaseLabel: undefined, - taskWorkspaceId: ev.done?.workspace_id, - ...(finishedRunId ? { taskRunId: finishedRunId } : {}), - }); - // quality_failed：未过门禁稿不得当作正式受控版本激活/同步切换 - if (nextStatus === 'completed' || nextStatus === 'pending_author_confirmation') { - await syncControlledDocsAfterTask(task.taskKey || 'intake', preferredDocId); - } - await chatWindow.refresh().catch(() => undefined); - })(); - break; - } - default: - break; - } - }; - - const startObserver = async () => { - const body = - mode === 'resume' && opts?.workDir - ? buildResumeRequest(task, opts.workDir) - : buildTaskRequest(task); - - if (task.sessionId && task.projectId && task.projectId > 0 && mode === 'run') { - // 断线重试：显式 reconnectRunId 保留；禁止 setMessages updater 取 runId - const existingRunId = resolveReconnectRunId({ - attemptIndex, - optsReconnectRunId: retainedRunId || opts?.reconnectRunId, - }); - if (attemptIndex > 0 && existingRunId) { - const remote = await getTaskRun(existingRunId).catch(() => null); - if (remote && shouldSubscribeExistingRun(remote.status)) { - retainedRunId = existingRunId; - taskRunStream.setRunId(sessionId, existingRunId); - patchAsst({ taskRunId: existingRunId, livePhaseLabel: '正在重连任务…' }); - await subscribeTaskRunEvents(existingRunId, { onEvent }, controller.signal); - return; - } - if (remote && isRemoteTaskTerminal(remote.status)) { - settled = true; - const mapped = mapTaskStatusToMessageStatus(remote.status, 'assistant') || 'failed'; - patchAsst({ - status: mapped, - livePhaseLabel: undefined, - taskWorkspaceId: remote.workspace_id, - taskRunId: existingRunId, - }); - taskRunStream.setThinking(sessionId, false); - if (mapped === 'completed' || mapped === 'pending_author_confirmation') { - await syncControlledDocsAfterTask(task.taskKey || 'intake'); - } - return; - } - } - - const created = await createTaskRun({ - ...body, - session_id: task.sessionId, - project_id: task.projectId, - user_id: task.userId, - client_turn_id: task.clientTurnId, - user_message: task.userMessage, - }); - retainedRunId = created.run_id; - taskRunStream.setRunId(sessionId, created.run_id); - patchAsst({ taskRunId: created.run_id }); - if (isBlockedGovernanceStatus(created.status)) { - settled = true; - patchAsst({ - status: 'blocked_pending_governance', - content: GOVERNANCE_BLOCKED_HINT, - livePhaseLabel: undefined, - taskRunId: created.run_id, - }); - taskRunStream.setRunId(sessionId, null); - taskRunStream.setThinking(sessionId, false); - await chatWindow.refresh().catch(() => undefined); - return; - } - await subscribeTaskRunEvents(created.run_id, { onEvent }, controller.signal); - return; - } - - await streamTask(body, { onEvent }, controller.signal); - }; - - void startObserver().catch(async (_err) => { - if (settled) return; - if (controller.signal.aborted) { - patchAsst({ status: 'cancelled', livePhaseLabel: undefined }); - taskRunStream.setThinking(sessionId, false); - return; - } - const runId = retainedRunId || taskRunStream.activeTaskRunId || undefined; - if (runId) { - try { - const remote = await getTaskRun(runId); - if (remote.status === 'quality_failed') { - settled = true; - patchAsst({ - status: 'quality_failed', - livePhaseLabel: undefined, - taskWorkspaceId: remote.workspace_id, - taskRunId: runId, - }); - taskRunStream.setThinking(sessionId, false); - return; - } - if (isBlockedGovernanceStatus(remote.status)) { - settled = true; - patchAsst({ - status: 'blocked_pending_governance', - content: GOVERNANCE_BLOCKED_HINT, - livePhaseLabel: undefined, - taskWorkspaceId: remote.workspace_id, - taskRunId: runId, - }); - taskRunStream.setThinking(sessionId, false); - return; - } - if (remote.status === 'completed') { - settled = true; - patchAsst({ - status: 'completed', - livePhaseLabel: undefined, - taskWorkspaceId: remote.workspace_id, - taskRunId: runId, - }); - taskRunStream.setThinking(sessionId, false); - await syncControlledDocsAfterTask(task.taskKey || 'intake'); - return; - } - if (remote.status === 'pending_author_confirmation') { - settled = true; - patchAsst({ - 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 (shouldSubscribeExistingRun(remote.status)) { - // 原 run 仍在跑：重订阅，不 POST 新 run - settled = false; - retainedRunId = runId; - patchAsst({ livePhaseLabel: '正在重连任务…', taskRunId: runId }); - await subscribeTaskRunEvents(runId, { onEvent }, controller.signal); - return; - } - if (remote.status === 'failed' || remote.status === 'cancelled') { - failOrRetry(buildNetworkErrorInfo()); - return; - } - } catch { - // fall through to network retry path - } - } - const httpStatus = - typeof _err === 'object' && _err !== null && 'httpStatus' in _err - ? Number((_err as { httpStatus?: number }).httpStatus) - : undefined; - const errMeta = _err as { errorCode?: string; detail?: string; message?: string }; - failOrRetry( - buildNetworkErrorInfo(httpStatus, { - errorCode: errMeta.errorCode, - detail: errMeta.detail || errMeta.message?.split(': ').slice(1).join(': '), - }), - ); - }); - }; - - // 结构化错误 / 已停止会话上的用户手动恢复动作 (P2：retry / continue / restart 分型恢复) - const handleErrorAction = (messageId: string, action: 'retry' | 'continue' | 'restart') => { - const target = messages.find((m) => m.id === messageId); - if (!target) return; - - // 长任务路径：按 taskRequest 重发 / 续跑工作区 / 按原意图重启 - if (target.taskRequest) { - if (action === 'retry') { - const userText = target.taskRequest.userMessage || ''; - const retryActive = - deliverables.find((d) => d.docId === activeDocId) || - visibleDeliverables.find((d) => d.docId === activeDocId); - const reResolved = resolveTaskTemplateForMessage( - userText, - currentProjectDeliverables, - retryActive, - ); - const retryTask = reResolved - ? { - ...target.taskRequest, - taskKey: reResolved.taskKey, - docId: reResolved.docId || target.taskRequest.docId, - } - : target.taskRequest; - runTaskAttempt(retryTask, messageId, 'run', 0); - } else if (action === 'continue') { - const wsId = target.errorInfo?.workspaceId || target.taskWorkspaceId; - if (!wsId) return; - runTaskAttempt(target.taskRequest, messageId, 'resume', 0, { - phaseEvents: target.phaseEvents, - workDir: wsId, - }); - } else { - if (!window.confirm('将放弃当前工作区并按原意图重新开始任务，确认？')) return; - runTaskAttempt(target.taskRequest, messageId, 'restart', 0); - } - return; - } - - // 闲聊路径（P1 行为不变：仅 retry） - if (action !== 'retry') return; - const sourceMsg = target.sourceUserMessageId - ? messages.find((m) => m.id === target.sourceUserMessageId) - : undefined; - const originalText = sourceMsg?.content?.trim(); - if (!originalText) return; - runStreamAttempt(originalText, messageId, 0, { - userMsgId: sourceMsg?.id, - clientTurnId: target.clientTurnId || sourceMsg?.clientTurnId, - }); - }; + const { runStreamAttempt, runTaskAttempt, handleErrorAction } = useTaskRunAttempts({ + activeSessionId, + activeDocId, + currentProjectId, + currentUser, + currentDocRole, + sessions, + messages, + deliverables, + visibleDeliverables, + currentProjectDeliverables, + deliverablesRef, + activeDocIdRef, + editorHydrationGateRef, + turnRuntime, + taskRunStream, + chatWindow, + setMessages, + setDeliverables, + setDocCache, + setActiveDocId, + setActiveCategory, + applyControlledInvalidationPlan, + }); // 用户主动停止：仅中止当前 session 的活跃 turn const handleStopStreaming = () => { diff --git a/web/src/hooks/useTaskRunAttempts.test.ts b/web/src/hooks/useTaskRunAttempts.test.ts new file mode 100644 index 00000000..7c4e1e0a --- /dev/null +++ b/web/src/hooks/useTaskRunAttempts.test.ts @@ -0,0 +1,15 @@ +import assert from 'node:assert/strict'; +import { readFileSync } from 'node:fs'; +import test from 'node:test'; + +test('chat and task attempts live in one hook', () => { + const app = readFileSync(new URL('../App.tsx', import.meta.url), 'utf8'); + const hook = readFileSync(new URL('./useTaskRunAttempts.ts', import.meta.url), 'utf8'); + assert.equal((app.match(/useTaskRunAttempts\(/g) || []).length, 1); + assert.equal(app.includes('streamChat('), false); + assert.equal(app.includes('createTaskRun('), false); + assert.equal(hook.includes('streamChat'), true); + assert.equal(hook.includes('createTaskRun'), true); + assert.equal(hook.includes('shouldSubscribeExistingRun'), true); + assert.equal(hook.includes('禁止无条件 setRunId(null)'), true); +}); diff --git a/web/src/hooks/useTaskRunAttempts.ts b/web/src/hooks/useTaskRunAttempts.ts new file mode 100644 index 00000000..c10d337c --- /dev/null +++ b/web/src/hooks/useTaskRunAttempts.ts @@ -0,0 +1,805 @@ +import type { Dispatch, MutableRefObject, SetStateAction } from 'react'; +import { + autoRetryDelayMs, + fallbackErrorClass, + shouldAutoRetryChat, + shouldAutoRetryTask, +} from '../api/chatRetry'; +import { mapErrorInfo, streamChat, type StreamEvent } from '../api/chatStream'; +import { fetchChatTurnStatus } from '../api/chatTurn'; +import { mapTaskStatusToMessageStatus } from '../api/chatWindow'; +import { listProjectWorkflowDocuments } from '../api/projectDocuments'; +import { createTaskRun, getTaskRun, subscribeTaskRunEvents } from '../api/taskRuns'; +import { buildResumeRequest, buildTaskRequest, streamTask } from '../api/taskStream'; +import { applyPhaseEvent } from '../engine/phaseEventMapper'; +import { resolveTaskTemplateForMessage } from '../engine/templateMatch'; +import { GOVERNANCE_BLOCKED_HINT, isBlockedGovernanceStatus } from '../lib/blockedGovernanceCopy'; +import { + controlledHydrateCandidateIds, + planControlledContentInvalidation, + type ControlledInvalidationPlan, +} from '../lib/controlledContentInvalidation'; +import { mergeWorkflowDocumentsFromServer } from '../lib/deliverableSync'; +import { createEditorHydrationGate } from '../lib/editorHydrationGate'; +import { + pollUntilPostTaskServerDoc, + shouldSwitchTemplateSlotToServer, +} from '../lib/postTaskDeliverableSync'; +import { + isRemoteTaskTerminal, + nextAttemptOpts, + resolveReconnectRunId, + shouldSubscribeExistingRun, +} from '../lib/taskRunReconnect'; +import type { DocumentAST } from '../types/block'; +import type { UserProfile, UserRole } from '../types/docgov'; +import type { + AssistantMessageStatus, + ChatErrorInfo, + ChatMessage, + DeliverableCategory, + DeliverableItem, + PhaseEvent, + ProjectSession, + QuotedMessage, + TaskRunContext, +} from '../types/workbench'; +import { useTaskRunStream } from './useTaskRunStream'; +import { useTurnRuntime } from './useTurnRuntime'; + +type Gate = ReturnType; +type TurnRuntime = ReturnType; +type TaskStream = ReturnType; + +export interface TaskRunAttemptsInput { + activeSessionId: string; + activeDocId: string; + currentProjectId: number; + currentUser: UserProfile; + currentDocRole: UserRole; + sessions: ProjectSession[]; + messages: ChatMessage[]; + deliverables: DeliverableItem[]; + visibleDeliverables: DeliverableItem[]; + currentProjectDeliverables: DeliverableItem[]; + deliverablesRef: MutableRefObject; + activeDocIdRef: MutableRefObject; + editorHydrationGateRef: MutableRefObject; + turnRuntime: TurnRuntime; + taskRunStream: TaskStream; + chatWindow: { + replaceMessageServerId: (tempId: string, serverId: number) => void; + refresh: () => Promise; + }; + setMessages: Dispatch>; + setDeliverables: Dispatch>; + setDocCache: Dispatch>>; + setActiveDocId: Dispatch>; + setActiveCategory: Dispatch>; + applyControlledInvalidationPlan: (plan: ControlledInvalidationPlan) => void; +} + +export function useTaskRunAttempts(input: TaskRunAttemptsInput) { + const { + activeSessionId, + activeDocId, + currentProjectId, + currentUser, + currentDocRole, + sessions, + messages, + deliverables, + visibleDeliverables, + currentProjectDeliverables, + deliverablesRef, + activeDocIdRef, + editorHydrationGateRef, + turnRuntime, + taskRunStream, + chatWindow, + setMessages, + setDeliverables, + setDocCache, + setActiveDocId, + setActiveCategory, + applyControlledInvalidationPlan, + } = input; + + const buildNetworkErrorInfo = ( + httpStatus?: number, + opts?: { errorCode?: string; detail?: string }, + ): ChatErrorInfo => { + const code = opts?.errorCode || ''; + const detail = opts?.detail || ''; + if (code === 'DOC_BUSY' || code === 'document_busy') { + return { + userMessage: detail || '该文档正在被其他用户或项目助手编辑，当前请求未创建新任务', + code: 'DOC_BUSY', + class: 'fatal', + actions: [], + }; + } + if (code === 'DOC_TASK_NOT_IN_ACTIVE_STAGE') { + return { + userMessage: detail || '该文档任务不在当前活动阶段，请先完成当前阶段工作流后再试', + code, + class: 'transient', + actions: ['retry'], + }; + } + if (code === 'DOC_DUTY_FORBIDDEN') { + return { + userMessage: detail || '当前账号不是该文档任务的职责人，无法执行', + code, + class: 'fatal', + actions: ['retry'], + }; + } + if (httpStatus === 401) { + return { + userMessage: '登录已失效，请重新登录后再试', + code: 'ERR_UNAUTHENTICATED', + class: 'fatal', + actions: ['retry'], + }; + } + if (httpStatus === 403) { + if (code === 'ERR_FORBIDDEN' || code === 'AUTHZ') { + return { + userMessage: detail || '无权访问该项目，请确认登录账号与项目成员身份', + code: code || 'ERR_FORBIDDEN', + class: 'fatal', + actions: ['retry'], + }; + } + return { + userMessage: detail || '无权执行该操作，请确认账号权限与工作流阶段', + code: code || 'ERR_FORBIDDEN', + class: 'fatal', + actions: ['retry'], + }; + } + return { + userMessage: '网络连接异常，请检查网络后重试', + code: 'NETWORK_ERROR', + class: fallbackErrorClass({ network: true }), + actions: ['retry'], + }; + }; + + // 驱动一次真实后端 SSE 流式对话尝试：气泡状态机 pending → streaming_status → completed/failed/cancelled。 + // 闲聊路径在无 client_turn_id 幂等前禁止自动重放 POST；失败仅展示手动「重试」。 + type StreamAttemptOpts = { + userMsgId?: string; + clientTurnId?: string; + quote?: QuotedMessage; + }; + + const runStreamAttempt = ( + userText: string, + asstId: string, + attemptIndex: number, + opts?: StreamAttemptOpts + ) => { + const quote = opts?.quote; + const controller = new AbortController(); + const streamSessionId = activeSessionId; + if (opts?.clientTurnId) { + turnRuntime.registerTurn({ + clientTurnId: opts.clientTurnId, + sessionId: streamSessionId, + userMsgId: opts.userMsgId || '', + asstId, + status: 'pending', + controller, + startedAt: Date.now(), + }); + } + + const currentProj = sessions.find((s) => s.sessionId === streamSessionId); + let accumulated = ''; + let settled = false; + + const updateAsst = (patch: Partial) => { + setMessages((prev) => prev.map((m) => (m.id === asstId ? { ...m, ...patch } : m))); + }; + + const scheduleRetry = (delayMs: number) => { + setTimeout(() => { + if (controller.signal.aborted) { + updateAsst({ status: 'cancelled', livePhaseLabel: undefined }); + if (opts?.clientTurnId) { + turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'cancelled'); + } + return; + } + runStreamAttempt(userText, asstId, attemptIndex + 1, opts); + }, delayMs); + }; + + const finishTurn = (status: 'completed' | 'failed' | 'cancelled') => { + if (opts?.clientTurnId) { + turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, status); + } + }; + + const failOrRetry = (info: ChatErrorInfo) => { + settled = true; + if (shouldAutoRetryChat(info, attemptIndex)) { + const delayMs = autoRetryDelayMs(attemptIndex); + updateAsst({ + status: 'auto_retrying', + livePhaseLabel: `连接异常，${Math.round(delayMs / 1000)}s 后自动重试…`, + }); + scheduleRetry(delayMs); + return; + } + updateAsst({ + status: 'failed', + errorInfo: { + ...info, + userMessage: + attemptIndex > 0 + ? `${info.userMessage}（请确认上一请求是否已在处理后再手动重试）` + : info.userMessage, + }, + livePhaseLabel: undefined, + }); + finishTurn('failed'); + }; + + updateAsst({ + status: 'pending', + livePhaseLabel: attemptIndex > 0 ? '正在手动重试…' : '助手正在处理…', + content: '', + errorInfo: undefined, + }); + if (opts?.clientTurnId) { + turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'pending'); + } + + const bindUserServerId = (serverId: number) => { + if (!opts?.userMsgId) return; + chatWindow.replaceMessageServerId(opts.userMsgId, serverId); + setMessages((prev) => + prev.map((m) => (m.id === opts.userMsgId ? { ...m, serverId } : m)) + ); + }; + + streamChat( + { + session_id: streamSessionId, + user_id: currentUser.id, + project_id: currentProj?.projectId || currentProjectId || 1, + project_name: currentProj?.title || '肌骨超声AI辅助诊断系统', + device_type: currentProj?.deviceType || '肌骨超声', + stage: currentProj?.stage || 'V0.5', + user_name: currentUser.name, + user_role: currentDocRole, + message: userText, + client_turn_id: opts?.clientTurnId, + quote: quote, + }, + { + onEvent: (ev) => { + if (controller.signal.aborted || settled) return; + switch (ev.type) { + case 'turn_ack': { + const userServerId = ev.turn?.user_message_id ?? ev.user_message_id; + if (typeof userServerId === 'number' && userServerId > 0) { + bindUserServerId(userServerId); + } + break; + } + case 'phase': + case 'status': { + const label = ev.phase?.message || ev.status?.message; + updateAsst({ status: 'streaming_status', livePhaseLabel: label || '助手正在处理…' }); + if (opts?.clientTurnId) { + turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'streaming'); + } + break; + } + case 'delta': { + accumulated += ev.delta || ''; + updateAsst({ status: 'streaming_status', content: accumulated }); + if (opts?.clientTurnId) { + turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'streaming'); + } + break; + } + case 'error': { + failOrRetry(mapErrorInfo(ev)); + break; + } + case 'done': { + settled = true; + const userServerId = ev.done?.user_message_id ?? ev.user_message_id; + const serverMsgId = ev.done?.message_id ?? ev.message_id ?? (ev as { messageId?: number }).messageId; + if (typeof userServerId === 'number' && userServerId > 0) { + bindUserServerId(userServerId); + } + if (typeof serverMsgId === 'number' && serverMsgId > 0) { + chatWindow.replaceMessageServerId(asstId, serverMsgId); + } + const exportCap = ev.export_capability || ev.done?.export_capability; + updateAsst({ + status: 'completed', + livePhaseLabel: undefined, + serverId: typeof serverMsgId === 'number' ? serverMsgId : undefined, + exportCapability: exportCap, + }); + finishTurn('completed'); + break; + } + default: + break; + } + }, + }, + controller.signal, + opts?.clientTurnId ? { idempotencyKey: opts.clientTurnId } : undefined + ).catch(async (_err) => { + if (settled) return; + if (controller.signal.aborted) { + updateAsst({ status: 'cancelled', livePhaseLabel: undefined }); + finishTurn('cancelled'); + return; + } + if (opts?.clientTurnId) { + try { + const turnStatus = await fetchChatTurnStatus(streamSessionId, opts.clientTurnId); + if (turnStatus?.status === 'completed' && turnStatus.assistantMessageId) { + settled = true; + if (turnStatus.userMessageId) bindUserServerId(turnStatus.userMessageId); + chatWindow.replaceMessageServerId(asstId, turnStatus.assistantMessageId); + updateAsst({ + status: 'completed', + livePhaseLabel: undefined, + serverId: turnStatus.assistantMessageId, + }); + finishTurn('completed'); + return; + } + if (turnStatus?.status === 'processing') { + updateAsst({ + status: 'streaming_status', + livePhaseLabel: '连接中断，服务端仍在处理…', + }); + if (opts?.clientTurnId) { + turnRuntime.updateTurnStatus(streamSessionId, opts.clientTurnId, 'streaming'); + } + return; + } + } catch { + // fall through to structured failure + } + } + const httpStatus = + typeof _err === 'object' && _err !== null && 'httpStatus' in _err + ? Number((_err as { httpStatus?: number }).httpStatus) + : undefined; + const errMeta = _err as { errorCode?: string; detail?: string; message?: string }; + failOrRetry( + buildNetworkErrorInfo(httpStatus, { + errorCode: errMeta.errorCode, + detail: errMeta.detail || errMeta.message?.split(': ').slice(1).join(': '), + }), + ); + }); + }; + + // P2：长任务尝试模式。'run' 新任务 | 'resume' 续跑失败工作区 | 'restart' 按原意图新工作区 + const runTaskAttempt = ( + task: TaskRunContext, + asstId: string, + mode: 'run' | 'resume' | 'restart', + attemptIndex: number, + opts?: { phaseEvents?: PhaseEvent[]; workDir?: string; reconnectRunId?: string } + ) => { + const sessionId = task.sessionId || activeSessionId; + let retainedRunId = resolveReconnectRunId({ + attemptIndex, + optsReconnectRunId: opts?.reconnectRunId, + }); + taskRunStream.abortSession(sessionId); + // 重连保留显式 runId；禁止无条件 setRunId(null) 导致自动重试丢失 + if (retainedRunId) { + taskRunStream.setRunId(sessionId, retainedRunId); + } else { + taskRunStream.setRunId(sessionId, null); + } + const controller = new AbortController(); + taskRunStream.setController(sessionId, controller); + + let accumulated = ''; + let settled = false; + + const patchAsst = (patch: Partial | ((m: ChatMessage) => Partial)) => { + setMessages((prev) => + prev.map((m) => (m.id === asstId ? { ...m, ...(typeof patch === 'function' ? patch(m) : patch) } : m)) + ); + }; + + const scheduleRetry = (delayMs: number) => { + setTimeout(() => { + if (controller.signal.aborted) { + patchAsst({ status: 'cancelled', livePhaseLabel: undefined }); + taskRunStream.setThinking(sessionId, false); + return; + } + runTaskAttempt(task, asstId, mode, attemptIndex + 1, nextAttemptOpts(opts, retainedRunId)); + }, delayMs); + }; + + const failOrRetry = (info: ChatErrorInfo) => { + settled = true; + if (shouldAutoRetryTask(info, attemptIndex)) { + const delayMs = autoRetryDelayMs(attemptIndex); + patchAsst({ + status: 'auto_retrying', + livePhaseLabel: `连接异常，${Math.round(delayMs / 1000)}s 后自动重试…`, + }); + scheduleRetry(delayMs); + return; + } + patchAsst({ status: 'failed', errorInfo: info, livePhaseLabel: undefined }); + taskRunStream.setThinking(sessionId, false); + }; + + const syncControlledDocsAfterTask = async (taskKey: string, preferredDocId?: string | null) => { + if (!task.projectId || task.projectId <= 0 || !taskKey) return; + 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); + applyControlledInvalidationPlan( + planControlledContentInvalidation({ + reason: 'post_task', + candidateIds: controlledHydrateCandidateIds(next, task.projectId), + activeDocId: activeDocIdRef.current, + editorOpen: editorHydrationGateRef.current.isEditorOpen(), + }), + ); + }, + }); + if ( + serverDocId && + shouldSwitchTemplateSlotToServer({ + activeDocId, + deliverables: merged, + projectId: task.projectId, + taskKey, + serverDocId, + }) + ) { + setActiveDocId(serverDocId); + setActiveCategory('workflow'); + } + }; + + // 等待态：resume 保留已有 phaseEvents 续接；run/restart 重置正文与步骤条 + patchAsst({ + status: 'pending', + livePhaseLabel: + mode === 'resume' ? '正在继续任务…' : mode === 'restart' ? '正在重新开始任务…' : '任务执行中…', + errorInfo: undefined, + ...(mode === 'resume' || attemptIndex > 0 + ? {} + : { content: '', phaseEvents: [], taskWorkspaceId: undefined }), + }); + taskRunStream.setThinking(sessionId, true); + + const onEvent = (ev: StreamEvent) => { + if (controller.signal.aborted || settled) return; + switch (ev.type) { + case 'phase': { + patchAsst((m) => ({ + status: 'streaming_status', + livePhaseLabel: ev.phase?.message || m.livePhaseLabel, + phaseEvents: applyPhaseEvent(opts?.phaseEvents || m.phaseEvents || [], ev.phase), + })); + break; + } + case 'status': { + if (ev.status?.message) { + patchAsst({ status: 'streaming_status', livePhaseLabel: ev.status.message }); + } + break; + } + case 'delta': { + accumulated += ev.delta || ''; + patchAsst({ status: 'streaming_status', content: accumulated }); + break; + } + case 'deliverable': { + const title = ev.deliverable?.title; + if (title) { + accumulated += (accumulated ? '\n\n' : '') + `📄 已生成文档：**${title}**`; + patchAsst({ status: 'streaming_status', content: accumulated }); + } + break; + } + case 'governance': { + const governance = ev.governance; + const docId = governance?.doc_id; + if (docId) { + const leaseReleased = governance.action === 'lease_released'; + const editLease = leaseReleased + ? undefined + : { holderType: 'squad' as const, holderId: taskRunStream.activeTaskRunId || undefined }; + setDeliverables((prev) => + prev.map((doc) => (doc.docId === docId ? { ...doc, editLease } : doc)), + ); + setDocCache((prev) => { + const ast = prev[docId]; + return ast ? { ...prev, [docId]: { ...ast, editLease } } : prev; + }); + } + if (governance?.notification_msg) { + patchAsst({ + status: 'streaming_status', + livePhaseLabel: governance.notification_msg, + }); + } + break; + } + case 'error': { + failOrRetry(mapErrorInfo(ev)); + break; + } + case 'done': { + settled = true; + const finishedRunId = retainedRunId || taskRunStream.activeTaskRunId; + taskRunStream.setRunId(sessionId, null); + taskRunStream.setThinking(sessionId, false); + void (async () => { + let nextStatus: AssistantMessageStatus = 'completed'; + let preferredDocId: string | null = null; + if (finishedRunId) { + const remote = await getTaskRun(finishedRunId).catch(() => null); + if (remote?.status) { + const mapped = mapTaskStatusToMessageStatus(remote.status, 'assistant'); + if (mapped) nextStatus = mapped; + } + } + patchAsst({ + status: nextStatus, + livePhaseLabel: undefined, + taskWorkspaceId: ev.done?.workspace_id, + ...(finishedRunId ? { taskRunId: finishedRunId } : {}), + }); + // quality_failed：未过门禁稿不得当作正式受控版本激活/同步切换 + if (nextStatus === 'completed' || nextStatus === 'pending_author_confirmation') { + await syncControlledDocsAfterTask(task.taskKey || 'intake', preferredDocId); + } + await chatWindow.refresh().catch(() => undefined); + })(); + break; + } + default: + break; + } + }; + + const startObserver = async () => { + const body = + mode === 'resume' && opts?.workDir + ? buildResumeRequest(task, opts.workDir) + : buildTaskRequest(task); + + if (task.sessionId && task.projectId && task.projectId > 0 && mode === 'run') { + // 断线重试：显式 reconnectRunId 保留；禁止 setMessages updater 取 runId + const existingRunId = resolveReconnectRunId({ + attemptIndex, + optsReconnectRunId: retainedRunId || opts?.reconnectRunId, + }); + if (attemptIndex > 0 && existingRunId) { + const remote = await getTaskRun(existingRunId).catch(() => null); + if (remote && shouldSubscribeExistingRun(remote.status)) { + retainedRunId = existingRunId; + taskRunStream.setRunId(sessionId, existingRunId); + patchAsst({ taskRunId: existingRunId, livePhaseLabel: '正在重连任务…' }); + await subscribeTaskRunEvents(existingRunId, { onEvent }, controller.signal); + return; + } + if (remote && isRemoteTaskTerminal(remote.status)) { + settled = true; + const mapped = mapTaskStatusToMessageStatus(remote.status, 'assistant') || 'failed'; + patchAsst({ + status: mapped, + livePhaseLabel: undefined, + taskWorkspaceId: remote.workspace_id, + taskRunId: existingRunId, + }); + taskRunStream.setThinking(sessionId, false); + if (mapped === 'completed' || mapped === 'pending_author_confirmation') { + await syncControlledDocsAfterTask(task.taskKey || 'intake'); + } + return; + } + } + + const created = await createTaskRun({ + ...body, + session_id: task.sessionId, + project_id: task.projectId, + user_id: task.userId, + client_turn_id: task.clientTurnId, + user_message: task.userMessage, + }); + retainedRunId = created.run_id; + taskRunStream.setRunId(sessionId, created.run_id); + patchAsst({ taskRunId: created.run_id }); + if (isBlockedGovernanceStatus(created.status)) { + settled = true; + patchAsst({ + status: 'blocked_pending_governance', + content: GOVERNANCE_BLOCKED_HINT, + livePhaseLabel: undefined, + taskRunId: created.run_id, + }); + taskRunStream.setRunId(sessionId, null); + taskRunStream.setThinking(sessionId, false); + await chatWindow.refresh().catch(() => undefined); + return; + } + await subscribeTaskRunEvents(created.run_id, { onEvent }, controller.signal); + return; + } + + await streamTask(body, { onEvent }, controller.signal); + }; + + void startObserver().catch(async (_err) => { + if (settled) return; + if (controller.signal.aborted) { + patchAsst({ status: 'cancelled', livePhaseLabel: undefined }); + taskRunStream.setThinking(sessionId, false); + return; + } + const runId = retainedRunId || taskRunStream.activeTaskRunId || undefined; + if (runId) { + try { + const remote = await getTaskRun(runId); + if (remote.status === 'quality_failed') { + settled = true; + patchAsst({ + status: 'quality_failed', + livePhaseLabel: undefined, + taskWorkspaceId: remote.workspace_id, + taskRunId: runId, + }); + taskRunStream.setThinking(sessionId, false); + return; + } + if (isBlockedGovernanceStatus(remote.status)) { + settled = true; + patchAsst({ + status: 'blocked_pending_governance', + content: GOVERNANCE_BLOCKED_HINT, + livePhaseLabel: undefined, + taskWorkspaceId: remote.workspace_id, + taskRunId: runId, + }); + taskRunStream.setThinking(sessionId, false); + return; + } + if (remote.status === 'completed') { + settled = true; + patchAsst({ + status: 'completed', + livePhaseLabel: undefined, + taskWorkspaceId: remote.workspace_id, + taskRunId: runId, + }); + taskRunStream.setThinking(sessionId, false); + await syncControlledDocsAfterTask(task.taskKey || 'intake'); + return; + } + if (remote.status === 'pending_author_confirmation') { + settled = true; + patchAsst({ + 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 (shouldSubscribeExistingRun(remote.status)) { + // 原 run 仍在跑：重订阅，不 POST 新 run + settled = false; + retainedRunId = runId; + patchAsst({ livePhaseLabel: '正在重连任务…', taskRunId: runId }); + await subscribeTaskRunEvents(runId, { onEvent }, controller.signal); + return; + } + if (remote.status === 'failed' || remote.status === 'cancelled') { + failOrRetry(buildNetworkErrorInfo()); + return; + } + } catch { + // fall through to network retry path + } + } + const httpStatus = + typeof _err === 'object' && _err !== null && 'httpStatus' in _err + ? Number((_err as { httpStatus?: number }).httpStatus) + : undefined; + const errMeta = _err as { errorCode?: string; detail?: string; message?: string }; + failOrRetry( + buildNetworkErrorInfo(httpStatus, { + errorCode: errMeta.errorCode, + detail: errMeta.detail || errMeta.message?.split(': ').slice(1).join(': '), + }), + ); + }); + }; + + // 结构化错误 / 已停止会话上的用户手动恢复动作 (P2：retry / continue / restart 分型恢复) + const handleErrorAction = (messageId: string, action: 'retry' | 'continue' | 'restart') => { + const target = messages.find((m) => m.id === messageId); + if (!target) return; + + // 长任务路径：按 taskRequest 重发 / 续跑工作区 / 按原意图重启 + if (target.taskRequest) { + if (action === 'retry') { + const userText = target.taskRequest.userMessage || ''; + const retryActive = + deliverables.find((d) => d.docId === activeDocId) || + visibleDeliverables.find((d) => d.docId === activeDocId); + const reResolved = resolveTaskTemplateForMessage( + userText, + currentProjectDeliverables, + retryActive, + ); + const retryTask = reResolved + ? { + ...target.taskRequest, + taskKey: reResolved.taskKey, + docId: reResolved.docId || target.taskRequest.docId, + } + : target.taskRequest; + runTaskAttempt(retryTask, messageId, 'run', 0); + } else if (action === 'continue') { + const wsId = target.errorInfo?.workspaceId || target.taskWorkspaceId; + if (!wsId) return; + runTaskAttempt(target.taskRequest, messageId, 'resume', 0, { + phaseEvents: target.phaseEvents, + workDir: wsId, + }); + } else { + if (!window.confirm('将放弃当前工作区并按原意图重新开始任务，确认？')) return; + runTaskAttempt(target.taskRequest, messageId, 'restart', 0); + } + return; + } + + // 闲聊路径（P1 行为不变：仅 retry） + if (action !== 'retry') return; + const sourceMsg = target.sourceUserMessageId + ? messages.find((m) => m.id === target.sourceUserMessageId) + : undefined; + const originalText = sourceMsg?.content?.trim(); + if (!originalText) return; + runStreamAttempt(originalText, messageId, 0, { + userMsgId: sourceMsg?.id, + clientTurnId: target.clientTurnId || sourceMsg?.clientTurnId, + }); + }; + + return { runStreamAttempt, runTaskAttempt, handleErrorAction }; +}