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 a80bb281..a1b84bf9 100644 --- a/frontend-web/src/open-canvas/components/ai-writing/AIWritingPanel.tsx +++ b/frontend-web/src/open-canvas/components/ai-writing/AIWritingPanel.tsx @@ -1,6 +1,6 @@ import { useCallback, useEffect, useRef, useState } from 'react'; import { useNavigate, useSearchParams } from 'react-router-dom'; -import { ChevronsDown, CloudUpload, PanelRightClose, Play, Plus, RotateCcw, Settings2, Square, Trash2 } from 'lucide-react'; +import { ChevronsDown, CloudUpload, PanelRightClose, Pause, Play, Plus, RotateCcw, Settings2, Trash2 } from 'lucide-react'; import { toast } from 'sonner'; import { Button } from '@/components/ui/button'; @@ -21,26 +21,18 @@ import { adminCleanupAIWritingSessions, getAIWritingSession, suspendAIWritingToBackground, - suspendAIWritingToBackgroundKeepalive, } from '../../api/ai-writing-sessions'; /** - * 暂停 / 继续功能开关（默认 false = 关）。 + * 暂停 / 继续功能开关（现已开启）。 * - * 历史背景：legacy 路径下，为了让用户点"停止"后还能"继续"，引入了 per-worker - * 内存 dict（``_event_queues`` / ``_session_pending_pause``）。多 worker 部署下 - * 这些 dict 互相看不见 → SSE 重连失败 / "会话丢失" → 触发了整个 ai-writing-优化.md - * 的迁移计划。 - * - * 当前 LangGraph Server 路径下架构已无 per-worker 状态，但 ``stream.stop()`` 仅 - * 取消客户端 SSE 订阅，**不会真正暂停后端 run**（LLM 继续烧 token）；``resume`` - * 也无法重新订阅同一个 run。这是个产品语义没想清楚的功能，先关掉。 - * - * 后期想清楚怎么做"真·暂停"（``client.runs.cancel(run_id, action='interrupt')`` - * 保留 checkpoint）+ "真·继续"（``client.runs.joinStream`` 续上原 run）时， - * 把这个 flag 翻成 true 即可，UI / hook 全套代码都已就位。 + * 真·暂停：``pauseWriting`` → ``stream.stop()`` 取消当前 run（LangGraph 已完成的阶段 + * 已落 checkpoint，不丢）。真·继续：``resumeWriting`` → 从最近 checkpoint 重新 submit + * 驱动 graph 续跑（会重跑当前未完成的阶段，已完成阶段不重来）。配合 AIWritingContext + * 的 ``onDisconnect=continue`` + sessionStorage 持久化 + ``reconnectOnMount``，断网/刷新/ + * 切走后回到本页也会自动重连续跑到下一个确认点（详见 AIWritingContext 注释）。 */ -const ENABLE_PAUSE_RESUME_UI = false; +const ENABLE_PAUSE_RESUME_UI = true; import { countArticleWords } from '../../utils/wordCount'; import { AIWritingForm } from './AIWritingForm'; import { AIWritingTimeline } from './AIWritingTimeline'; @@ -234,26 +226,11 @@ export function AIWritingPanel({ })(); }, [state.sessionId, reset, refresh, markBackground, clearBackground]); - // 离页 / 刷新 / 关标签页时，若写作仍在进行中则自动转后台（与圆桌「自动挂起」一致）。 - // 用 ref 持有最新的「是否在跑 + 会话 id」，让卸载/卸载前回调读到即时值。 - const suspendRef = useRef<{ running: boolean; sid: string | null }>({ running: false, sid: null }); - useEffect(() => { - suspendRef.current = { running: isRunning, sid: state.sessionId ?? null }; - }, [isRunning, state.sessionId]); - - useEffect(() => { - const onBeforeUnload = () => { - const { running, sid } = suspendRef.current; - if (running && sid) suspendAIWritingToBackgroundKeepalive(sid); - }; - window.addEventListener('beforeunload', onBeforeUnload); - return () => { - window.removeEventListener('beforeunload', onBeforeUnload); - // 组件卸载（路由离开本页 / 切换工作区标签）：写作未完成则自动转后台。 - const { running, sid } = suspendRef.current; - if (running && sid) suspendAIWritingToBackgroundKeepalive(sid); - }; - }, []); + // 离页 / 刷新 / 断网不再"自动转后台跑完"（那会自动替用户确认素材/大纲/审核）。 + // 改为：运行带 on_disconnect=continue + streamResumable，断开后端不取消、继续跑到 + // 下一个确认点等待；活跃 sessionId 已持久化进 sessionStorage（见 AIWritingContext）， + // 回到本页/刷新后由 useStream(reconnectOnMount) 自动重连续跑。需要"后台自动跑完" + // 仍可手动点「后台挂起」。run 万一在断开时被取消，最近 checkpoint 仍在，用「继续」续跑。 // ── 侧边栏"最近写作"通信：两种路径 ──────────────────────────────────────── // 用 ref 包住 handler，保证事件监听器始终能拿到最新版本 @@ -384,8 +361,8 @@ export function AIWritingPanel({ 后台挂起 )} - {/* ⏸️/▶️ "停止 / 继续" 由 ENABLE_PAUSE_RESUME_UI 控制（默认 false 隐藏）。 - 后期把 flag 翻成 true 即可恢复，UI / hook 全套代码都已就位。 */} + {/* ⏸️/▶️ "暂停 / 继续"：暂停=stream.stop() 取消当前 run（已完成阶段已落 checkpoint）； + 继续=从 checkpoint 重新驱动 graph 续跑（重跑当前阶段，已完成阶段不丢）。 */} {ENABLE_PAUSE_RESUME_UI && isRunning && !state.isPaused && ( )} {ENABLE_PAUSE_RESUME_UI && state.isPaused && ( diff --git a/frontend-web/src/open-canvas/contexts/AIWritingContext.tsx b/frontend-web/src/open-canvas/contexts/AIWritingContext.tsx index 35c48bdd..92d797ab 100644 --- a/frontend-web/src/open-canvas/contexts/AIWritingContext.tsx +++ b/frontend-web/src/open-canvas/contexts/AIWritingContext.tsx @@ -26,6 +26,7 @@ import { saveAIWritingTranscript } from '../api/ai-writing-sessions'; import { countArticleWords } from '../utils/wordCount'; import { INITIAL_STATE, + buildRehydrateSnapshot, customToProgressEvent, interruptToEvent, nodeUpdatesToEvents, @@ -33,6 +34,37 @@ import { type Action, 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, AIWritingState, @@ -131,12 +163,15 @@ 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) { @@ -235,6 +270,33 @@ 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; @@ -249,8 +311,18 @@ 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(); + } - const threadId = state.sessionId ?? undefined; + // 首帧即把"待重连会话"作为 threadId 交给 useStream，让 reconnectOnMount 在初次挂载就重连 + // （而不是等 restore effect 派发后再下一帧）；state.sessionId 落定后两者同值、无缝接管。 + const threadId = state.sessionId ?? reconnectTargetRef.current ?? undefined; useEffect(() => { const s = state.status; @@ -263,8 +335,31 @@ 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; @@ -306,7 +401,8 @@ export function AIWritingProvider({ children }: { children: ReactNode }) { } as Partial, { config: { configurable: { model_name: request.modelName ?? '' } } as Record, - streamMode: ['values', 'updates', 'custom', 'messages-tuple'], + streamMode: [...WRITING_STREAM_MODE], + ...RESILIENT_RUN_OPTS, }, ); } catch (err) { @@ -410,7 +506,8 @@ export function AIWritingProvider({ children }: { children: ReactNode }) { const resumeModelName = state.request?.modelName; await stream.submit(undefined, { command: { resume: resumePayload }, - streamMode: ['values', 'updates', 'custom', 'messages-tuple'], + streamMode: [...WRITING_STREAM_MODE], + ...RESILIENT_RUN_OPTS, ...(resumeModelName ? { config: { configurable: { model_name: resumeModelName } } } : {}), }); } catch (err) { @@ -426,18 +523,46 @@ export function AIWritingProvider({ children }: { children: ReactNode }) { state.status]); const pauseWriting = useCallback(() => { + // 真停：取消当前 run（graph 已完成的阶段已落 checkpoint）。继续时从该 checkpoint 续跑。 void streamRef.current?.stop(); dispatch({ type: 'pause' }); }, []); + // 「继续」：清掉当前阶段流式缓冲后，从最近的 checkpoint 重新驱动 graph 续跑 + // （会重跑当前未完成的阶段，已完成阶段不丢）。同时用于断网后 run 已死的兜底恢复。 const resumeWriting = useCallback(() => { - dispatch({ type: 'resume_ui' }); - }, []); + const stream = streamRef.current; + if (!stream) { + dispatch({ type: 'resume_ui' }); + return; + } + dispatch({ type: 'clear_streaming' }); + const resumeModelName = state.request?.modelName; + void (async () => { + try { + await stream.submit(undefined, { + streamMode: [...WRITING_STREAM_MODE], + ...RESILIENT_RUN_OPTS, + ...(resumeModelName ? { config: { configurable: { model_name: resumeModelName } } } : {}), + }); + } catch (err) { + const message = getStreamErrorMessage(err); + if (isThreadNotAccessibleError(message)) { + notifyThreadAccessChangedAndReload(); + return; + } + dispatch({ type: 'set_error', message }); + } + })(); + }, [state.request]); const reset = useCallback(() => { void streamRef.current?.stop(); setArtifact(null); seenInterruptRef.current = null; + // 新建/重置：清掉重连目标与持久化，避免下次刷新误重连到这条已弃用的会话。 + reconnectTargetRef.current = null; + writeActiveSession(null); dispatch({ type: 'reset' }); }, [setArtifact]); @@ -454,6 +579,7 @@ 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 9c682bb3..f7145ba5 100644 --- a/frontend-web/src/open-canvas/hooks/useAIWritingStream.ts +++ b/frontend-web/src/open-canvas/hooks/useAIWritingStream.ts @@ -85,6 +85,10 @@ 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' } | { type: 'set_error'; message: string }; @@ -132,6 +136,23 @@ 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 { + ...prev, + isPaused: false, + intentStreamingText: '', intentThinkingText: '', + outlineStreamingText: '', outlineThinkingText: '', + streamingDraftSections: [], streamingDraftTitle: '', + streamingReviewItems: [], + skillSelectStreamingText: '', skillSelectThinkingText: '', + }; + case 'reset': return INITIAL_STATE; @@ -561,6 +582,39 @@ 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;