fix(ai写作): 继续从checkpoint续跑(修resume_is_map) + 刷新=重置不重连

后端 worker(runtime/runs/worker.py): 空 command(无 resume/update/goto)不再构造
Command()(会因无挂起 interrupt 抛 'resume_is_map' 未绑定错误),改为 astream(None)
让 Pregel 续跑 checkpoint 里挂起的当前阶段节点。有 resume/update/goto 时(4个干预卡)
仍走 Command,行为不变。前端"继续"发空 command:{} 即从当前阶段续跑、不回 intent。

按用户要求把"断网/刷新自动重连续跑"整套移除(刷新=重置即可):
- 删 sessionStorage 持久化 + reconnectTargetRef + 恢复/持久化 effect + host rehydrate
- 删 buildRehydrateSnapshot/rehydrate action(保留 clear_streaming)
- 提交去掉 onDisconnect:'continue'/streamResumable;reconnectOnMount 改 false
- 卸载不再自动后台挂起;刷新断 SSE 后端默认 cancel 取消 run,页面回 idle

手动暂停/继续保留:暂停=stream.stop(),继续=clear_streaming+空command 从checkpoint续。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

frontend-web/src/open-canvas/components/ai-writing/AIWritingPanel.tsx

@@ -27,10 +27,9 @@ import {
  * 暂停 / 继续功能开关（现已开启）。
  *
  * 真·暂停：``pauseWriting`` → ``stream.stop()`` 取消当前 run（LangGraph 已完成的阶段
- * 已落 checkpoint，不丢）。真·继续：``resumeWriting`` → 从最近 checkpoint 重新 submit
+ * 已落 checkpoint，不丢）。真·继续：``resumeWriting`` → 带空 ``command:{}`` 重新 submit，
- * 驱动 graph 续跑（会重跑当前未完成的阶段，已完成阶段不重来）。配合 AIWritingContext
+ * 后端 worker 据此让 Pregel 从最近 checkpoint 续跑挂起的当前阶段节点（重跑当前未完成阶段，
- * 的 ``onDisconnect=continue`` + sessionStorage 持久化 + ``reconnectOnMount``，断网/刷新/
+ * 已完成阶段不重来；见 runtime/runs/worker.py）。刷新页面 = 重置（不自动重连）。
- * 切走后回到本页也会自动重连续跑到下一个确认点（详见 AIWritingContext 注释）。
  */
 const ENABLE_PAUSE_RESUME_UI = true;
 import { countArticleWords } from '../../utils/wordCount';
@@ -226,11 +225,10 @@ export function AIWritingPanel({
     })();
   }, [state.sessionId, reset, refresh, markBackground, clearBackground]);
 
-  // 离页 / 刷新 / 断网不再"自动转后台跑完"（那会自动替用户确认素材/大纲/审核）。
+  // 离页 / 刷新 = 重置（不自动重连、不自动转后台）：刷新会断开 SSE，后端默认
-  // 改为：运行带 on_disconnect=continue + streamResumable，断开后端不取消、继续跑到
+  // on_disconnect=cancel 取消该 run，页面重新挂载回到 idle。需要"后台自动跑完"
-  // 下一个确认点等待；活跃 sessionId 已持久化进 sessionStorage（见 AIWritingContext），
+  // （自动确认素材/大纲/审核到定稿、离页不中断）可手动点「后台挂起」。
-  // 回到本页/刷新后由 useStream(reconnectOnMount) 自动重连续跑。需要"后台自动跑完"
+  // 页内的「暂停/继续」从 checkpoint 续跑当前阶段，不受刷新重置影响（刷新即新开）。
-  // 仍可手动点「后台挂起」。run 万一在断开时被取消，最近 checkpoint 仍在，用「继续」续跑。
 
   // ── 侧边栏"最近写作"通信：两种路径 ────────────────────────────────────────
   // 用 ref 包住 handler，保证事件监听器始终能拿到最新版本

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<Dispatch<Action>>;
   sessionIdRef: RefObject<string | null>;
   streamRef: MutableRefObject<AIWritingStreamApi | null>;
   seenInterruptRef: MutableRefObject<string | null>;
-  /** 非空 = 本次挂载是"断网/刷新后重连"目标会话；重连成功 rehydrate 后清空。 */
-  reconnectTargetRef: MutableRefObject<string | null>;
 }) {
   const eventBatcherRef = useRef<ReturnType<typeof createProgressEventBatcher> | 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<AIWritingStreamApi | null>(null);
   const seenInterruptRef = useRef<string | null>(null);
-  // 断网/刷新重连目标：首帧同步读 sessionStorage 的活跃会话，rehydrate 后由 host 清空。
-  // **必须在渲染期完成**（早于任何 effect），否则下面"持久化"effect 会在 idle 首帧清掉它。
-  const reconnectTargetRef = useRef<string | null>(null);
-  const reconnectInitRef = useRef(false);
-  if (!reconnectInitRef.current) {
-    reconnectInitRef.current = true;
-    reconnectTargetRef.current = readActiveSession();
-  }
 
-  // 首帧即把"待重连会话"作为 threadId 交给 useStream，让 reconnectOnMount 在初次挂载就重连
+  // 刷新页面 = 重置：不持久化、不自动重连。provider 重新挂载即回到 idle 初始态
-  // （而不是等 restore effect 派发后再下一帧）；state.sessionId 落定后两者同值、无缝接管。
+  // （刷新会断开 SSE，后端默认 on_disconnect=cancel 取消该 run，不留孤儿）。
-  const threadId = state.sessionId ?? reconnectTargetRef.current ?? undefined;
+  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<string, unknown>,
           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}
     </AIWritingContext.Provider>

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<AIWritingState> }
   // 手动「继续」从检查点重跑当前阶段前，清空该阶段的流式缓冲（避免与重跑的新输出粘连）。
   | { 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<string, unknown>, 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<AIWritingState> {
-  const pkg = (values.material_package && typeof values.material_package === 'object')
-    ? (values.material_package as Record<string, unknown>) : null;
-  const materials = pkg && Array.isArray(pkg.materials)
-    ? (pkg.materials as Record<string, unknown>[]).map(normalizeMaterial) : [];
-  const skillSteps = pkg && Array.isArray(pkg.skill_search_steps ?? pkg.skillSearchSteps)
-    ? ((pkg.skill_search_steps ?? pkg.skillSearchSteps) as Record<string, unknown>[]).map(normalizeSkillStep) : [];
-  const draft = (values.current_draft && typeof values.current_draft === 'object')
-    ? (values.current_draft as Record<string, unknown>) : null;
-  const progressEvents = Array.isArray(values.progress_events)
-    ? (values.progress_events as Record<string, unknown>[]).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;

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(
+                stream_input = _LGCommand(
-                resume=command.get("resume"),
+                    resume=command.get("resume"),
-                update=command.get("update"),
+                    update=command.get("update"),
-                goto=command.get("goto"),
+                    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