diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index e5b74929..68c1e3e4 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -274,17 +274,22 @@ export function streamCopilot( /** * 订阅一次运行的事件流。 * - * 先补历史再接实时由后端保证,这里只负责断线重连 —— 重连时带上已收到的最大 - * seq,所以不会重复也不会丢事件。 + * 先补历史再接实时由后端保证,这里负责断线重连——重连时带上已收到的最大 seq。 + * + * `after` 必须由调用方给:lastSeq 是这次调用的闭包局部量,跨调用会归零。 + * 审批恢复时会重新 attachRun,不传 after 就是 after=0,后端按 `seq > after` + * 把整条历史再推一遍,而消费方多半是无条件追加——事件就这么重复累积了。 + * 已经有事件在手的场景(重新接上一条运行)应当传当前最大 seq。 */ export function streamRun( runId: string, onEvent: (event: RunEvent) => void, onClose?: () => void, + after = 0, ): () => void { let socket: WebSocket | null = null let closed = false - let lastSeq = 0 + let lastSeq = after let retry = 0 const connect = () => { diff --git a/frontend/src/pages/ChatPage.tsx b/frontend/src/pages/ChatPage.tsx index 342cfc14..9a7d2937 100644 --- a/frontend/src/pages/ChatPage.tsx +++ b/frontend/src/pages/ChatPage.tsx @@ -186,15 +186,12 @@ function Approvals({ turn }: { turn: ChatTurn }) { const reattach = useChat((s) => s.reattach) const pending = approvals.filter((a) => a.run_id === turn.run?.id && a.status === 'pending') - // 审批处理完后运行会继续,重新接上事件流才能看到后续 - useEffect(() => { - if (turn.phase === 'waiting' && !pending.length) void reattach(turn.id) - }, [pending.length, turn.phase, turn.id, reattach]) - if (!pending.length) return null return (
- {pending.map((a) => )} + {pending.map((a) => ( + reattach(turn.id)} /> + ))}
) } diff --git a/frontend/src/run/RunPanel.tsx b/frontend/src/run/RunPanel.tsx index 3943d736..6d1f1852 100644 --- a/frontend/src/run/RunPanel.tsx +++ b/frontend/src/run/RunPanel.tsx @@ -172,7 +172,18 @@ function RunLauncher() { // ------------------------------------------------------------------------- -export function ApprovalCard({ approval }: { approval: any }) { +/** + * 人工介入卡片。 + * + * onResolved 是给非画布场景用的:这个组件被对话页和运行页复用,而默认的 + * "恢复后重新接上事件流"走的是 studio store——在对话页里用就会同时开两条 + * 订阅(一条 chat 的、一条 studio 的),两边各收一份事件。谁用谁负责重新 + * 接流,是唯一不会打架的分工。 + */ +export function ApprovalCard({ approval, onResolved }: { + approval: any + onResolved?: (runId: string) => void | Promise +}) { const toast = useToast() const refreshApprovals = useCatalog((s) => s.refreshApprovals) const attachRun = useStudio((s) => s.attachRun) @@ -192,7 +203,9 @@ export function ApprovalCard({ approval }: { approval: any }) { ...(approval.mode !== 'approve' ? { value } : {}), }) await refreshApprovals() - await attachRun(approval.run_id) // 恢复后重新接上事件流 + // 恢复后重新接上事件流:调用方给了 onResolved 就听它的 + if (onResolved) await onResolved(approval.run_id) + else await attachRun(approval.run_id) // 驳回在"补充输入/编辑草稿"模式下会终止整个运行(图上没有 rejected 那条 // 出口边),说清楚比笼统一句"已驳回"诚实 toast( diff --git a/frontend/src/store/chat.ts b/frontend/src/store/chat.ts index b1e7a12f..10b7f58f 100644 --- a/frontend/src/store/chat.ts +++ b/frontend/src/store/chat.ts @@ -41,6 +41,11 @@ export interface ChatTurn { output: Record | null error: string startedAt: number + /** 已收到的最大事件 seq。重新接流时要带上,否则后端把历史整条重推 */ + lastSeq: number + /** 这一轮自己的取消句柄。放 turn 上而不是 store 上:store 单值会被 + * 下一轮 ask 覆盖,上一轮的 WebSocket 就再也关不掉了 */ + cancel: (() => void) | null } interface ChatState { @@ -88,6 +93,7 @@ export const useChat = create((set, get) => ({ id, question, phase: 'planning', status: PHASE_TEXT.planning, thinking: '', graph: null, explanation: '', run: null, steps: [], output: null, error: '', startedAt: Date.now(), + lastSeq: 0, cancel: null, }, ], })) @@ -148,22 +154,31 @@ export const useChat = create((set, get) => ({ set({ busy: false, cancel: null }) }, ) + patch(() => ({ cancel: cancelCopilot })) set({ cancel: cancelCopilot }) }, stop: () => { + // 只停最后一轮。把所有非终态 turn 一律标成"已取消"会误杀正等人工介入的 + // 历史轮次——那些是等着用户回来处理的,不是卡住的。 + const turns = get().turns + const last = turns[turns.length - 1] + last?.cancel?.() get().cancel?.() set((s) => ({ busy: false, cancel: null, turns: s.turns.map((t) => - t.phase === 'done' || t.phase === 'error' ? t : { ...t, phase: 'error', error: '已取消' }, + t.id === last?.id && !['done', 'error', 'waiting'].includes(t.phase) + ? { ...t, phase: 'error', error: '已取消', cancel: null } + : t, ), })) }, clear: () => { get().cancel?.() + get().turns.forEach((t) => t.cancel?.()) // 每轮各有各的订阅,逐个关 set({ turns: [], busy: false, cancel: null }) }, @@ -174,7 +189,7 @@ export const useChat = create((set, get) => ({ set((s) => ({ turns: s.turns.map((t) => (t.id === turnId ? { ...t, ...fn(t) } : t)) })) patch(() => ({ phase: 'running', status: PHASE_TEXT.running })) set({ busy: true }) - watch(turn.run.id, turnId, patch, set) + watch(turn.run.id, turnId, patch, set, turn.lastSeq) }, })) @@ -202,9 +217,12 @@ function watch( turnId: string, patch: (fn: (t: ChatTurn) => Partial) => void, set: any, + after = 0, ) { const stop = streamRun(runId, (event: RunEvent) => { const d: any = event.data ?? {} + // 记下进度:审批恢复要重新接流,带上它才不会把历史再收一遍 + patch((t) => ({ lastSeq: Math.max(t.lastSeq ?? 0, event.seq ?? 0) })) switch (event.type) { case 'tool.start': { const tool = String(d.tool ?? '') @@ -261,7 +279,10 @@ function watch( void finish(runId, turnId, patch, set) break } - }) + }, undefined, after) + // 句柄同时挂在 turn 和 store 上:turn 上的用于精确关闭这一轮, + // store 上的是"当前活跃流"的快捷方式 + patch(() => ({ cancel: stop })) set({ cancel: stop }) } diff --git a/frontend/src/store/studio.ts b/frontend/src/store/studio.ts index 31bd3e03..9115205b 100644 --- a/frontend/src/store/studio.ts +++ b/frontend/src/store/studio.ts @@ -502,10 +502,17 @@ export const useStudio = create((set, get) => ({ attachRun: async (runId) => { get().unsubscribe?.() + // 手上已有的事件不要再收一遍。审批恢复会走到这里,而 applyEvent 是无条件 + // 追加——不传 after 的话后端把整条历史重推一遍,时间线上每条都出现两次。 + const current = get() + const after = current.run?.id === runId + ? current.events.reduce((max, e) => Math.max(max, e.seq ?? 0), 0) + : 0 const stop = streamRun( runId, (event) => get().applyEvent(event), () => set({ streaming: false }), + after, ) set({ unsubscribe: stop, streaming: true }) },