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 <cursoragent@cursor.com>
This commit is contained in:
weichen
@@ -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<typeof createEditorHydrationGate>;
|
||||
type TurnRuntime = ReturnType<typeof useTurnRuntime>;
|
||||
type TaskStream = ReturnType<typeof useTaskRunStream>;
|
||||
|
||||
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<DeliverableItem[]>;
|
||||
activeDocIdRef: MutableRefObject<string>;
|
||||
editorHydrationGateRef: MutableRefObject<Gate>;
|
||||
turnRuntime: TurnRuntime;
|
||||
taskRunStream: TaskStream;
|
||||
chatWindow: {
|
||||
replaceMessageServerId: (tempId: string, serverId: number) => void;
|
||||
refresh: () => Promise<void>;
|
||||
};
|
||||
setMessages: Dispatch<SetStateAction<ChatMessage[]>>;
|
||||
setDeliverables: Dispatch<SetStateAction<DeliverableItem[]>>;
|
||||
setDocCache: Dispatch<SetStateAction<Record<string, DocumentAST>>>;
|
||||
setActiveDocId: Dispatch<SetStateAction<string>>;
|
||||
setActiveCategory: Dispatch<SetStateAction<DeliverableCategory>>;
|
||||
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<ChatMessage>) => {
|
||||
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<ChatMessage> | ((m: ChatMessage) => Partial<ChatMessage>)) => {
|
||||
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 };
|
||||
}
|
||||
Reference in New Issue
Block a user