diff --git a/frontend-web/src/open-canvas/components/ai-writing/AIWritingPanel.tsx b/frontend-web/src/open-canvas/components/ai-writing/AIWritingPanel.tsx index a1b84bf9..2a87e7d6 100644 --- a/frontend-web/src/open-canvas/components/ai-writing/AIWritingPanel.tsx +++ b/frontend-web/src/open-canvas/components/ai-writing/AIWritingPanel.tsx @@ -27,10 +27,9 @@ import { * 暂停 / 继续功能开关（现已开启）。 * * 真·暂停：``pauseWriting`` → ``stream.stop()`` 取消当前 run（LangGraph 已完成的阶段 - * 已落 checkpoint，不丢）。真·继续：``resumeWriting`` → 从最近 checkpoint 重新 submit - * 驱动 graph 续跑（会重跑当前未完成的阶段，已完成阶段不重来）。配合 AIWritingContext - * 的 ``onDisconnect=continue`` + sessionStorage 持久化 + ``reconnectOnMount``，断网/刷新/ - * 切走后回到本页也会自动重连续跑到下一个确认点（详见 AIWritingContext 注释）。 + * 已落 checkpoint，不丢）。真·继续：``resumeWriting`` → 带空 ``command:{}`` 重新 submit， + * 后端 worker 据此让 Pregel 从最近 checkpoint 续跑挂起的当前阶段节点（重跑当前未完成阶段， + * 已完成阶段不重来；见 runtime/runs/worker.py）。刷新页面 = 重置（不自动重连）。 */ const ENABLE_PAUSE_RESUME_UI = true; import { countArticleWords } from '../../utils/wordCount'; @@ -226,11 +225,10 @@ export function AIWritingPanel({ })(); }, [state.sessionId, reset, refresh, markBackground, clearBackground]); - // 离页 / 刷新 / 断网不再"自动转后台跑完"（那会自动替用户确认素材/大纲/审核）。 - // 改为：运行带 on_disconnect=continue + streamResumable，断开后端不取消、继续跑到 - // 下一个确认点等待；活跃 sessionId 已持久化进 sessionStorage（见 AIWritingContext）， - // 回到本页/刷新后由 useStream(reconnectOnMount) 自动重连续跑。需要"后台自动跑完" - // 仍可手动点「后台挂起」。run 万一在断开时被取消，最近 checkpoint 仍在，用「继续」续跑。 + // 离页 / 刷新 = 重置（不自动重连、不自动转后台）：刷新会断开 SSE，后端默认 + // on_disconnect=cancel 取消该 run，页面重新挂载回到 idle。需要"后台自动跑完" + // （自动确认素材/大纲/审核到定稿、离页不中断）可手动点「后台挂起」。 + // 页内的「暂停/继续」从 checkpoint 续跑当前阶段，不受刷新重置影响（刷新即新开）。 // ── 侧边栏"最近写作"通信：两种路径 ──────────────────────────────────────── // 用 ref 包住 handler，保证事件监听器始终能拿到最新版本 diff --git a/frontend-web/src/open-canvas/contexts/AIWritingContext.tsx b/frontend-web/src/open-canvas/contexts/AIWritingContext.tsx index 0b6834c4..464b8138 100644 --- a/frontend-web/src/open-canvas/contexts/AIWritingContext.tsx +++ b/frontend-web/src/open-canvas/contexts/AIWritingContext.tsx @@ -26,7 +26,6 @@ import { saveAIWritingTranscript } from '../api/ai-writing-sessions'; import { countArticleWords } from '../utils/wordCount'; import { INITIAL_STATE, - buildRehydrateSnapshot, customToProgressEvent, interruptToEvent, nodeUpdatesToEvents, @@ -35,35 +34,6 @@ import { type AIWritingValues, } from '../hooks/useAIWritingStream'; -// ── 活跃写作会话持久化（断网/刷新后自动重连续跑）────────────────────────────── -// 把"正在进行中"的写作 sessionId 存进 sessionStorage（与 LangGraph SDK 自身的 -// reconnectOnMount run 元数据同一存储域，跨刷新/页内导航存活、按标签页隔离）， -// 刷新后据此恢复 threadId → useStream 重连该 run。done/error/idle 即清除。 -const ACTIVE_SESSION_KEY = 'ai-writing:active-session'; - -function readActiveSession(): string | null { - try { - return typeof sessionStorage !== 'undefined' ? sessionStorage.getItem(ACTIVE_SESSION_KEY) : null; - } catch { - return null; - } -} - -function writeActiveSession(id: string | null): void { - try { - if (typeof sessionStorage === 'undefined') return; - if (id) sessionStorage.setItem(ACTIVE_SESSION_KEY, id); - else sessionStorage.removeItem(ACTIVE_SESSION_KEY); - } catch { - /* ignore */ - } -} - -/** 提交运行时统一带上"断开保活"：网断/离页后端不取消该 run，继续跑到下一个确认点等待。 - * 不开 streamResumable —— 重连后的历史时间线由 buildRehydrateSnapshot 从 graph values 快照 - * 权威重建（不依赖 gateway 是否实现事件重放），开了反而会和 rehydrate 重复累积时间线。 */ -const RESILIENT_RUN_OPTS = { onDisconnect: 'continue' as const }; - const WRITING_STREAM_MODE = ['values', 'updates', 'custom', 'messages-tuple'] as const; import type { AIWritingRequest, @@ -163,15 +133,12 @@ const AIWritingStreamHost = memo(function AIWritingStreamHost({ sessionIdRef, streamRef, seenInterruptRef, - reconnectTargetRef, }: { threadId: string | undefined; dispatchRef: RefObject>; sessionIdRef: RefObject; streamRef: MutableRefObject; seenInterruptRef: MutableRefObject; - /** 非空 = 本次挂载是"断网/刷新后重连"目标会话；重连成功 rehydrate 后清空。 */ - reconnectTargetRef: MutableRefObject; }) { const eventBatcherRef = useRef | null>(null); if (!eventBatcherRef.current) { @@ -190,7 +157,8 @@ const AIWritingStreamHost = memo(function AIWritingStreamHost({ client: getAPIClient(), assistantId: 'ai_writing', threadId, - reconnectOnMount: true, + // 刷新页面 = 重置：不在挂载时自动重连已有 run（避免刷新后接回旧会话）。 + reconnectOnMount: false, onCreated(meta) { dispatchRef.current?.({ type: 'set_session', sessionId: meta.thread_id }); }, @@ -270,33 +238,6 @@ const AIWritingStreamHost = memo(function AIWritingStreamHost({ if (ev) emitProgressEvent(ev); }, [stream.interrupt, emitProgressEvent, seenInterruptRef, sessionIdRef]); - // ── 断网/刷新后重连：用 graph checkpoint 的 values 快照整体重建状态（只做一次）── - // 流程：刷新 → provider 从 sessionStorage 恢复 sessionId → useStream(reconnectOnMount) - // 重连该 run 并 fetchStateHistory 拉到 stream.values → 这里把 values 重建成前端状态， - // 让素材/大纲/草稿/审核/时间线立刻回显；run 若仍存活会继续流式跑到下一个确认点， - // run 若已结束/中断则由 stream.interrupt 既有 effect 设好确认点、或用户用「继续」续跑。 - const rehydratedRef = useRef(false); - useEffect(() => { - if (rehydratedRef.current) return; - const target = reconnectTargetRef.current; - if (!target || target !== threadId) return; - const values = stream.values as AIWritingValues | undefined; - // 等线程状态真正拉到（status 落定）再重建，避免拿空快照覆盖 - if (!values || !values.status) return; - rehydratedRef.current = true; - reconnectTargetRef.current = null; - const snapshot = buildRehydrateSnapshot(values, threadId ?? ''); - dispatchRef.current?.({ type: 'rehydrate', snapshot }); - // 若重连时正处于一个确认点（interrupt 待用户决策），紧接着补一条 await_user 设好 currentPause， - // 并对齐 seenInterruptRef 防止下方 interrupt effect 重复派发。 - const intr = stream.interrupt; - if (intr) { - const ev = interruptToEvent(intr.value, threadId ?? ''); - if (ev) dispatchRef.current?.({ type: 'handle', event: ev }); - seenInterruptRef.current = intr.id ?? JSON.stringify(intr.value); - } - }, [stream.values, stream.interrupt, threadId, reconnectTargetRef, dispatchRef, seenInterruptRef]); - useEffect(() => () => { eventBatcherRef.current?.cancel(); }, []); return null; @@ -311,18 +252,10 @@ export function AIWritingProvider({ children }: { children: ReactNode }) { sessionIdRef.current = state.sessionId; const streamRef = useRef(null); const seenInterruptRef = useRef(null); - // 断网/刷新重连目标：首帧同步读 sessionStorage 的活跃会话，rehydrate 后由 host 清空。 - // **必须在渲染期完成**（早于任何 effect），否则下面"持久化"effect 会在 idle 首帧清掉它。 - const reconnectTargetRef = useRef(null); - const reconnectInitRef = useRef(false); - if (!reconnectInitRef.current) { - reconnectInitRef.current = true; - reconnectTargetRef.current = readActiveSession(); - } - // 首帧即把"待重连会话"作为 threadId 交给 useStream，让 reconnectOnMount 在初次挂载就重连 - // （而不是等 restore effect 派发后再下一帧）；state.sessionId 落定后两者同值、无缝接管。 - const threadId = state.sessionId ?? reconnectTargetRef.current ?? undefined; + // 刷新页面 = 重置：不持久化、不自动重连。provider 重新挂载即回到 idle 初始态 + // （刷新会断开 SSE，后端默认 on_disconnect=cancel 取消该 run，不留孤儿）。 + const threadId = state.sessionId ?? undefined; useEffect(() => { const s = state.status; @@ -335,31 +268,8 @@ export function AIWritingProvider({ children }: { children: ReactNode }) { // eslint-disable-next-line react-hooks/exhaustive-deps }, [state.status, state.sessionId]); - // 持久化"活跃中"的写作会话（进行中 + 各确认点），供刷新/断网后自动重连； - // done/error/idle 即清除。重连恢复尚未完成（reconnectTargetRef 仍非空）时不清， - // 避免恢复过程中的瞬时 idle 把它抹掉。配合 useStream(reconnectOnMount) 完成"自动续跑"。 - useEffect(() => { - if (reconnectTargetRef.current) return; - const s = state.status; - const active = s !== 'idle' && s !== 'done' && s !== 'error'; - if (active && state.sessionId) writeActiveSession(state.sessionId); - else if (s === 'idle' || s === 'done' || s === 'error') writeActiveSession(null); - }, [state.status, state.sessionId]); - - // 挂载（含刷新后首帧）：若 reducer 还是初始态、但首帧已读到持久化的活跃会话， - // 恢复其 sessionId → threadId 落定 → useStream 重连该 run；host 在 values 到位后 rehydrate。 - useEffect(() => { - const target = reconnectTargetRef.current; - if (target && !state.sessionId) { - dispatch({ type: 'set_session', sessionId: target }); - } - // eslint-disable-next-line react-hooks/exhaustive-deps - }, []); - const startWriting = useCallback(async (request: AIWritingRequest) => { if (state.status !== 'idle' && state.status !== 'done' && state.status !== 'error') return; - // 开新会话：作废任何遗留的重连目标（避免 host 误把新线程当重连对象 rehydrate）。 - reconnectTargetRef.current = null; dispatch({ type: 'start', request }); const stream = streamRef.current; if (!stream) return; @@ -402,7 +312,6 @@ export function AIWritingProvider({ children }: { children: ReactNode }) { { config: { configurable: { model_name: request.modelName ?? '' } } as Record, streamMode: [...WRITING_STREAM_MODE], - ...RESILIENT_RUN_OPTS, }, ); } catch (err) { @@ -507,7 +416,6 @@ export function AIWritingProvider({ children }: { children: ReactNode }) { await stream.submit(undefined, { command: { resume: resumePayload }, streamMode: [...WRITING_STREAM_MODE], - ...RESILIENT_RUN_OPTS, ...(resumeModelName ? { config: { configurable: { model_name: resumeModelName } } } : {}), }); } catch (err) { @@ -549,7 +457,6 @@ export function AIWritingProvider({ children }: { children: ReactNode }) { await stream.submit(undefined, { command: {}, streamMode: [...WRITING_STREAM_MODE], - ...RESILIENT_RUN_OPTS, ...(resumeModelName ? { config: { configurable: { model_name: resumeModelName } } } : {}), }); } catch (err) { @@ -567,9 +474,6 @@ export function AIWritingProvider({ children }: { children: ReactNode }) { void streamRef.current?.stop(); setArtifact(null); seenInterruptRef.current = null; - // 新建/重置：清掉重连目标与持久化，避免下次刷新误重连到这条已弃用的会话。 - reconnectTargetRef.current = null; - writeActiveSession(null); dispatch({ type: 'reset' }); }, [setArtifact]); @@ -586,7 +490,6 @@ export function AIWritingProvider({ children }: { children: ReactNode }) { sessionIdRef={sessionIdRef} streamRef={streamRef} seenInterruptRef={seenInterruptRef} - reconnectTargetRef={reconnectTargetRef} /> {children} diff --git a/frontend-web/src/open-canvas/hooks/useAIWritingStream.ts b/frontend-web/src/open-canvas/hooks/useAIWritingStream.ts index f7145ba5..d3b3d188 100644 --- a/frontend-web/src/open-canvas/hooks/useAIWritingStream.ts +++ b/frontend-web/src/open-canvas/hooks/useAIWritingStream.ts @@ -85,8 +85,6 @@ export type Action = | { type: 'intervention_submitted'; record: CompletedIntervention; nextStatus: WritingStatus; pendingRevisionContext: AIWritingState['pendingRevisionContext'] } | { type: 'pause' } | { type: 'resume_ui' } - // 断网/刷新后重连：用 graph checkpoint 的 values 快照整体重建状态（见 buildRehydrateSnapshot）。 - | { type: 'rehydrate'; snapshot: Partial } // 手动「继续」从检查点重跑当前阶段前，清空该阶段的流式缓冲（避免与重跑的新输出粘连）。 | { type: 'clear_streaming' } | { type: 'reset' } @@ -136,11 +134,6 @@ export function reducer(prev: AIWritingState, action: Action): AIWritingState { case 'resume_ui': return { ...prev, isPaused: false }; - case 'rehydrate': - // 重连：以 INITIAL_STATE 为底，叠加快照（materials/outline/draft/review/status/ - // progressEvents/sessionId 等）。currentPause 由 stream.interrupt 的既有 effect 另行设置。 - return { ...INITIAL_STATE, ...action.snapshot, isPaused: false }; - case 'clear_streaming': // 「继续」重跑当前阶段：清掉各阶段流式缓冲，重跑的输出从干净状态累积。 return { @@ -582,39 +575,6 @@ export function normalizeProgressDict(parsed: Record, sessionId return out as unknown as ProgressEvent; } -/** 断网/刷新重连：把 graph checkpoint 的 values 快照重建成前端状态快照。 - * - * 用于 ``rehydrate`` action（以 INITIAL_STATE 为底叠加）。``status`` 后端与前端共用 - * 同一套字面量（state.py WritingStatus ↔ types/ai-writing.ts），可直接透传；timeline 由 - * 累积的 ``progress_events``（operator.add 全量）重放，素材/大纲/草稿/审核读各自快照字段。 - * pause 点（currentPause）由 ``stream.interrupt`` 的既有 effect 设置，这里不处理。 */ -export function buildRehydrateSnapshot(values: AIWritingValues, sessionId: string): Partial { - const pkg = (values.material_package && typeof values.material_package === 'object') - ? (values.material_package as Record) : null; - const materials = pkg && Array.isArray(pkg.materials) - ? (pkg.materials as Record[]).map(normalizeMaterial) : []; - const skillSteps = pkg && Array.isArray(pkg.skill_search_steps ?? pkg.skillSearchSteps) - ? ((pkg.skill_search_steps ?? pkg.skillSearchSteps) as Record[]).map(normalizeSkillStep) : []; - const draft = (values.current_draft && typeof values.current_draft === 'object') - ? (values.current_draft as Record) : null; - const progressEvents = Array.isArray(values.progress_events) - ? (values.progress_events as Record[]).map((e) => normalizeProgressDict(e, sessionId)) - : []; - - return { - sessionId, - status: (values.status as AIWritingState['status']) || 'researching', - materials, - skillSearchSteps: skillSteps, - outline: normalizeOutline(values.current_outline), - currentDraftMarkdown: String(draft?.full_markdown ?? draft?.fullMarkdown ?? ''), - currentDraftTitle: String(draft?.title ?? ''), - revisionCount: Number(draft?.revision_count ?? draft?.revisionCount ?? 0), - reviewResult: normalizeReviewResult(values.review_result), - progressEvents, - }; -} - /** custom 通道事件 → ProgressEvent。事件 ``type`` 字段决定是哪类。 */ export function customToProgressEvent(raw: unknown, sessionId: string): ProgressEvent | null { if (!raw || typeof raw !== 'object') return null; diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/worker.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/worker.py index 025fd95d..7135784a 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -282,13 +282,25 @@ async def run_agent( # 起新一轮。Pregel 检查到 Command 会从最近 checkpoint 处恢复，而不是重头跑。 stream_input: Any = graph_input if command is not None: - from langgraph.types import Command as _LGCommand + # 有 resume/update/goto（如 AI 写作 4 个干预卡的 Command(resume=user_input)）→ + # 构造 langgraph.types.Command，Pregel 从 interrupt 处恢复。 + # **空 command（无任何指令）= "从最近 checkpoint 继续执行挂起的任务"**：AI 写作 + # "暂停后继续" —— 暂停时硬取消了活跃节点（非 interrupt），continue 要让该阶段节点 + # 从 checkpoint 续跑。此时必须用 **None 输入**让 Pregel resume pending tasks； + # 既不能用 graph_input={}（空 dict 被当新一轮输入 → 从图入口 intent 重跑），也不能 + # 构造 Command()（没有挂起的 interrupt 可消费，LangGraph 内部会抛 + # "cannot access local variable 'resume_is_map'"）。 + has_directive = any(command.get(k) is not None for k in ("resume", "update", "goto")) + if has_directive: + from langgraph.types import Command as _LGCommand - stream_input = _LGCommand( - resume=command.get("resume"), - update=command.get("update"), - goto=command.get("goto"), - ) + stream_input = _LGCommand( + resume=command.get("resume"), + update=command.get("update"), + goto=command.get("goto"), + ) + else: + stream_input = None if len(lg_modes) == 1 and not stream_subgraphs: # Single mode, no subgraphs: astream yields raw chunks