Merge branch 'dhk-1' into 'master'

报错兜底

See merge request knowledge-base-bj/deerflow-server!26

frontend-web/src/open-canvas/api/ai-writing-sessions.ts

@@ -59,15 +59,29 @@ export async function deleteAIWritingSession(id: string): Promise<void> {
   if (!resp.ok) throw new Error(`删除失败: ${resp.status}`);
 }
 
-/** 保存会话的完整对话时间线，供历史回显出与实时对话一致的富时间线。 */
+/**
+ * 保存会话的完整对话时间线，供历史回显出与实时对话一致的富时间线。
+ *
+ * 该功能属于「锦上添花」：保存失败不应影响写作主流程，因此本函数永不抛错，
+ * 内部捕获所有异常（网络错误 / 接口 500 等），仅打印告警并返回是否成功。
+ */
 export async function saveAIWritingTranscript(
   id: string,
   transcript: AIWritingTranscript,
-): Promise<void> {
+): Promise<boolean> {
-  const resp = await apiFetch(`${BASE()}/${id}/transcript`, {
+  try {
-    method: 'PUT',
+    const resp = await apiFetch(`${BASE()}/${id}/transcript`, {
-    headers: { 'Content-Type': 'application/json' },
+      method: 'PUT',
-    body: JSON.stringify({ transcript }),
+      headers: { 'Content-Type': 'application/json' },
-  });
+      body: JSON.stringify({ transcript }),
-  if (!resp.ok) throw new Error(`保存对话时间线失败: ${resp.status}`);
+    });
+    if (!resp.ok) {
+      console.warn(`[ai-writing] 保存对话时间线失败: HTTP ${resp.status}（已忽略，不影响写作流程）`);
+      return false;
+    }
+    return true;
+  } catch (err) {
+    console.warn('[ai-writing] 保存对话时间线异常（已忽略，不影响写作流程）:', err);
+    return false;
+  }
 }

frontend-web/src/open-canvas/hooks/useAIWriting.ts

@@ -295,15 +295,19 @@ export function useAIWriting() {
     async (payload: UserInterventionPayload) => {
       if (!state.sessionId || !state.currentPause) return;
 
-      // 若处于暂停状态，先重连 SSE 再提交，实现"干预即续接"
+      // 提交干预前始终重连 SSE：
+      // - 正常流程下旧连接仍在，abort 后重建无副作用（后端通过 _session_pending_pause
+      //   重发 await_user，前端 handleEvent 对同一暂停点有去重）；
+      // - 若旧连接已因服务重启 / 反向代理空闲超时而断开，这一步能确保后端恢复出的
+      //   会话有消费者，否则 resume 后的进度事件会堆在队列里无人接收。
+      abortRef.current?.abort();
+      abortRef.current = new AbortController();
+      unsubRef.current = subscribeToProgress(
+        state.sessionId,
+        handleEvent,
+        abortRef.current.signal,
+      );
       if (state.isPaused) {
-        abortRef.current?.abort();
-        abortRef.current = new AbortController();
-        unsubRef.current = subscribeToProgress(
-          state.sessionId,
-          handleEvent,
-          abortRef.current.signal,
-        );
         setState((prev) => ({ ...prev, isPaused: false }));
       }
 

offline-backend-20260512/backend/CLAUDE.md

@@ -265,6 +265,7 @@ Custom agent metadata lives in the `agents` ORM table (`deerflow.persistence.age
 Two `TypeDecorator`s keep ORM columns correct across the `sqlite` / `postgres` / `mysql` backends. **All ORM timestamp/JSON columns must use these, not the raw SQLAlchemy types.**
 
 - **`PortableJSON`** — native `JSON` on SQLite/Postgres; `TEXT` (with manual `json.dumps`/`loads`) on MySQL, where older servers lack JSON support.
+- **`PortableLongText`** — a large text column: maps to MySQL `LONGTEXT` (4 GB) and plain `TEXT` on SQLite/Postgres (already unbounded). Use for big JSON/markdown blobs that would overflow MySQL's 64 KB `TEXT` cap — e.g. `ai_writing_sessions.transcript`.
 - **`BeijingDateTime`** — stores timestamps as Beijing (UTC+8) wall-clock time. The app records time with `datetime.now(UTC)`, but MySQL `DATETIME` / SQLite have no timezone and persist whatever wall clock the driver serializes — so a raw `DateTime` column would store UTC and read 8 hours behind Beijing. `BeijingDateTime` converts UTC→Beijing on bind and re-attaches `+08:00` on result, so the database holds Beijing time while Python still gets timezone-aware datetimes (instant comparisons stay correct, including in WHERE clauses). PostgreSQL `timestamptz` stores the instant unambiguously, so values pass through unchanged there. Uses a fixed UTC+8 offset (China has no DST), so no `tzdata` dependency.
 
 ### Sandbox System (`packages/harness/deerflow/sandbox/`)

offline-backend-20260512/backend/app/gateway/routers/ai_writing.py

@@ -149,6 +149,10 @@ _session_article_types: dict[str, list[dict]] = {}
 # so that a new SSE connection (e.g., after user clicks "继续") can re-inject it.
 _session_pending_pause: dict[str, tuple] = {}
 
+# Serializes session recovery so that a concurrent /stream + /resume on the same
+# lost session don't each rebuild a different queue object.
+_recovery_lock = asyncio.Lock()
+
 
 # ── 工具函数 ──────────────────────────────────────────────────────────────────
 
@@ -317,16 +321,26 @@ async def delete_session(session_id: str, request: Request):
 
 @router.put("/sessions/{session_id}/transcript")
 async def save_transcript(session_id: str, body: SaveTranscriptRequest, request: Request):
-    """保存会话的完整对话时间线，供历史会话回显出与实时对话一致的富时间线。"""
+    """保存会话的完整对话时间线，供历史会话回显出与实时对话一致的富时间线。
+
+    时间线持久化属于「锦上添花」功能：即便保存失败（如内网部署缺少 transcript
+    列、或 MySQL TEXT 列超长），也绝不能阻断写作流程。任何异常都吞掉并返回
+    success=False，由前端忽略即可。
+    """
     from app.gateway.deps import get_ai_writing_session_store, get_current_user
 
     repo = get_ai_writing_session_store(request)
     if repo is None:
-        return {"success": False}
+        return {"success": False, "reason": "store_unavailable"}
-    user_id = await get_current_user(request)
+    try:
-    updated = await repo.update(session_id, user_id=user_id, transcript=body.transcript)
+        user_id = await get_current_user(request)
+        updated = await repo.update(session_id, user_id=user_id, transcript=body.transcript)
+    except Exception:
+        logger.exception("ai_writing: failed to save transcript: session=%s", session_id)
+        return {"success": False, "reason": "save_failed"}
     if updated is None:
-        raise HTTPException(status_code=404, detail="Session not found")
+        # 会话不存在或不属于当前用户：同样不抛 500，返回 success=False 即可
+        return {"success": False, "reason": "session_not_found"}
     return {"success": True}
 
 
@@ -420,7 +434,17 @@ async def resume_writing(session_id: str, body: ResumeRequest, request: Request)
     from app.gateway.deps import get_current_user
 
     if session_id not in _event_queues:
-        raise HTTPException(status_code=404, detail="Session not found")
+        # 内存会话注册表丢失（进程内清理后重连等）。先尝试从 checkpointer 恢复，
+        # 让用户能在干预点继续，而不是直接 404 中断整个写作流程。
+        recovery = await _recover_session(session_id, request)
+        if recovery != "interrupted":
+            logger.warning(
+                "ai_writing: resume on unrecoverable session=%s (recovery=%s)", session_id, recovery
+            )
+            raise HTTPException(
+                status_code=404,
+                detail="会话状态已丢失或流程已结束，无法继续，请重新发起写作。",
+            )
 
     config = _make_config(session_id)
     intervention = {
@@ -623,6 +647,104 @@ def _event_type(evt: dict) -> str:
     return t
 
 
+async def _load_graph_snapshot(session_id: str):
+    """读取 AI 写作 Graph 在该 session 上的状态快照（失败/无状态返回 None）。
+
+    注意：Graph 用的是内存 checkpointer（MemorySaver），所以这里只能在
+    **进程存活**期间恢复；进程/容器重启后快照同样会丢失。
+    """
+    try:
+        from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH
+
+        snapshot = await AI_WRITING_GRAPH.aget_state(_make_config(session_id))
+    except Exception:
+        logger.exception("ai_writing: aget_state failed during recovery: session=%s", session_id)
+        return None
+    if not snapshot or not getattr(snapshot, "values", None):
+        return None
+    return snapshot
+
+
+def _extract_interrupt(snapshot) -> dict | None:
+    """从快照里取出当前挂起的 interrupt 负载（即 await_user 的数据）。"""
+    for task in getattr(snapshot, "tasks", None) or ():
+        for intr in getattr(task, "interrupts", None) or ():
+            value = getattr(intr, "value", None)
+            if isinstance(value, dict):
+                return value
+    return None
+
+
+async def _recover_session(session_id: str, request: Request | None = None) -> str:
+    """内存会话注册表丢失时，尝试从 checkpointer 重建会话。
+
+    返回:
+      - "interrupted"   —— 图正停在某个干预点，已重建队列并预置 await_user 事件，可继续 resume
+      - "finished"      —— 图已完成，已重建队列并预置 done 事件（+草稿）
+      - "unrecoverable" —— 无可恢复状态（进程/容器重启过，或 session 从未存在）
+
+    幂等：内存里已有队列时直接当作 "interrupted" 返回。整段在 `_recovery_lock`
+    下执行，避免并发的 /stream 与 /resume 各自重建出不同的队列对象。
+    """
+    async with _recovery_lock:
+        if session_id in _event_queues:
+            return "interrupted"
+
+        snapshot = await _load_graph_snapshot(session_id)
+        if snapshot is None:
+            return "unrecoverable"
+
+        queue: asyncio.Queue = asyncio.Queue()
+        _event_queues[session_id] = queue
+        _session_interventions.setdefault(session_id, [])
+        _session_model_names.setdefault(session_id, "")
+        _session_article_types.setdefault(session_id, [])
+
+        # 尽量恢复 DB 仓库引用，使后续 resume 仍能持久化草稿/审核结果
+        if request is not None and session_id not in _session_repos:
+            try:
+                from app.gateway.deps import get_ai_writing_session_store, get_current_user
+
+                repo = get_ai_writing_session_store(request)
+                if repo is not None:
+                    _session_repos[session_id] = repo
+                    _session_user_ids[session_id] = await get_current_user(request)
+            except Exception:
+                logger.exception("ai_writing: failed to restore repo during recovery: session=%s", session_id)
+
+        interrupt_val = _extract_interrupt(snapshot)
+        if interrupt_val is not None:
+            event = ("await_user", {**interrupt_val, "session_id": session_id})
+            await queue.put(event)
+            _session_pending_pause[session_id] = event
+            logger.info("ai_writing: recovered interrupted session from checkpointer: session=%s", session_id)
+            return "interrupted"
+
+        values = snapshot.values or {}
+        if not getattr(snapshot, "next", None) or values.get("status") == "done":
+            draft = values.get("current_draft") or {}
+            if draft.get("full_markdown"):
+                await queue.put(("draft_ready", {
+                    "session_id": session_id,
+                    "title": draft.get("title", ""),
+                    "markdown": draft.get("full_markdown", ""),
+                    "revision_count": draft.get("revision_count", 0),
+                }))
+            await queue.put(("done", {"session_id": session_id, "message": "写作已完成"}))
+            await queue.put(None)  # 哨兵：让 SSE 生成器读完即关闭
+            logger.info("ai_writing: recovered finished session from checkpointer: session=%s", session_id)
+            return "finished"
+
+        # 既不在干预点、也未结束：上次运行多半被中断在某个节点，无法续推
+        await queue.put(("error", {
+            "session_id": session_id,
+            "message": "写作流程已中断，请重新发起写作。",
+        }))
+        await queue.put(None)
+        logger.warning("ai_writing: session checkpoint stuck mid-flow, not resumable: session=%s", session_id)
+        return "finished"
+
+
 async def _sse_generator(session_id: str) -> AsyncGenerator[str, None]:
     """从队列消费事件，推送为 SSE 帧。"""
     # 等待队列创建（最多 5 秒）
@@ -631,10 +753,24 @@ async def _sse_generator(session_id: str) -> AsyncGenerator[str, None]:
             break
         await asyncio.sleep(0.1)
     else:
-        yield _sse("error", {"message": "Session not found"})
+        # 内存里没有这个会话队列。不要直接报「Session not found」——
-        return
+        # 先尝试从 checkpointer 恢复（图可能仍停在某个干预点 / 已完成）。
+        recovery = await _recover_session(session_id)
+        if recovery == "unrecoverable":
+            yield _sse("connected", {"session_id": session_id})
+            yield _sse("error", {
+                "session_id": session_id,
+                "message": "会话状态已丢失（服务可能已重启），请重新发起写作。",
+            })
+            return
+        # 恢复成功：_event_queues[session_id] 已重建并预置事件，继续走正常消费逻辑。
 
-    queue = _event_queues[session_id]
+    queue = _event_queues.get(session_id)
+    if queue is None:
+        # 极端兜底：恢复后又被并发清理
+        yield _sse("connected", {"session_id": session_id})
+        yield _sse("error", {"session_id": session_id, "message": "会话状态已丢失，请重新发起写作。"})
+        return
     yield _sse("connected", {"session_id": session_id})
 
     # Re-inject the saved await_user event if the queue is empty and a pending pause exists.

offline-backend-20260512/backend/config.yaml

@@ -41,6 +41,26 @@ models:
     temperature: 1.0
     supports_vision: true
     supports_thinking: true
+  - name: MiniMax-M2.7
+    display_name: MiniMax-M2.7
+    use: langchain_openai:ChatOpenAI
+    model: MiniMax-M2.7
+    api_key: sk-cp-mrsAGtzDpUl0TpOBPVoiRG_pWF1kDDN9_mDAnONBs6HIPcNUexSKtSbx2QOxfKBvJT6Xw6hDTV6hdkJYFWfNA8UtHQ08TiGJiosdRoW6q1sGb6Cb7D85NHk
+    base_url: https://api.minimaxi.com/v1
+    max_tokens: 4096
+    temperature: 1.0
+    supports_vision: true
+    supports_thinking: true
+  - name: minimax-m2.5
+    display_name: MiniMax M2.5
+    use: langchain_openai:ChatOpenAI
+    model: MiniMax-M2.5
+    api_key: sk-cp-mrsAGtzDpUl0TpOBPVoiRG_pWF1kDDN9_mDAnONBs6HIPcNUexSKtSbx2QOxfKBvJT6Xw6hDTV6hdkJYFWfNA8UtHQ08TiGJiosdRoW6q1sGb6Cb7D85NHk
+    base_url: https://api.minimaxi.com/v1
+    max_tokens: 4096
+    temperature: 1.0
+    supports_vision: true
+    supports_thinking: false
 
 title:
   enabled: true

offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/model.py

@@ -8,6 +8,7 @@ from sqlalchemy import DateTime, String, Text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from deerflow.persistence.base import Base
+from deerflow.persistence.types import PortableLongText
 
 
 class AIWritingSessionRow(Base):
@@ -22,7 +23,8 @@ class AIWritingSessionRow(Base):
     draft_markdown: Mapped[str | None] = mapped_column(Text, nullable=True)
     completed_interventions: Mapped[str | None] = mapped_column(Text, nullable=True)  # JSON array
     review_result: Mapped[str | None] = mapped_column(Text, nullable=True)  # JSON object
-    # 完整对话时间线（前端组装的 progressEvents + completedInterventions），用于历史回显
+    # 完整对话时间线（前端组装的 progressEvents + completedInterventions），用于历史回显。
-    transcript: Mapped[str | None] = mapped_column(Text, nullable=True)  # JSON object
+    # 用 PortableLongText：MySQL 上映射为 LONGTEXT，避免 TEXT 64KB 上限导致保存 500。
+    transcript: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)  # JSON object
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
     updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260519_02_ai_writing_transcript_longtext.py

@@ -0,0 +1,64 @@
+"""Widen ai_writing_sessions.transcript to LONGTEXT on MySQL.
+
+The ``transcript`` column stores the full AI-writing conversation timeline as a
+JSON blob. It was originally created as ``TEXT``, which on MySQL caps at 64 KB —
+a transcript with many progress events overflows it and the save endpoint
+returns HTTP 500. This revision widens the column to ``LONGTEXT`` (4 GB) on
+MySQL. SQLite/PostgreSQL ``TEXT`` is already unbounded, so they are left as-is.
+
+Revision ID: 20260519_02
+Revises: 20260519_01
+Create Date: 2026-05-19
+"""
+
+from __future__ import annotations
+
+from alembic import op
+from sqlalchemy import inspect
+from sqlalchemy.dialects.mysql import LONGTEXT
+
+
+revision = "20260519_02"
+down_revision = "20260519_01"
+branch_labels = None
+depends_on = None
+
+
+def _has_column(table_name: str, column_name: str) -> bool:
+    inspector = inspect(op.get_bind())
+    if table_name not in inspector.get_table_names():
+        return False
+    return any(col["name"] == column_name for col in inspector.get_columns(table_name))
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    # Only MySQL has the 64 KB TEXT limit; other backends need no change.
+    if bind.dialect.name != "mysql":
+        return
+    if not _has_column("ai_writing_sessions", "transcript"):
+        return
+    op.alter_column(
+        "ai_writing_sessions",
+        "transcript",
+        type_=LONGTEXT(),
+        existing_nullable=True,
+    )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    if bind.dialect.name != "mysql":
+        return
+    if not _has_column("ai_writing_sessions", "transcript"):
+        return
+    # Revert to TEXT. Rows longer than 64 KB would be truncated, but downgrade
+    # is a developer-only path.
+    import sqlalchemy as sa
+
+    op.alter_column(
+        "ai_writing_sessions",
+        "transcript",
+        type_=sa.Text(),
+        existing_nullable=True,
+    )

offline-backend-20260512/backend/packages/harness/deerflow/persistence/types.py

@@ -7,6 +7,7 @@ from datetime import datetime, timedelta, timezone
 from typing import Any
 
 from sqlalchemy import JSON, DateTime, Text
+from sqlalchemy.dialects.mysql import LONGTEXT
 from sqlalchemy.types import TypeDecorator
 
 # China Standard Time. China has observed UTC+8 with no DST since 1991, so a
@@ -39,6 +40,25 @@ class PortableJSON(TypeDecorator):
         return json.loads(value)
 
 
+class PortableLongText(TypeDecorator):
+    """A large text column.
+
+    MySQL's ``TEXT`` caps at 64 KB, which silently truncates — or, in STRICT
+    mode, raises ``Data too long`` (HTTP 500) — for large JSON blobs such as the
+    AI-writing conversation transcript. This decorator maps to MySQL ``LONGTEXT``
+    (4 GB) while staying plain ``TEXT`` on SQLite/PostgreSQL, where ``TEXT`` is
+    already unbounded.
+    """
+
+    impl = Text
+    cache_ok = True
+
+    def load_dialect_impl(self, dialect):
+        if dialect.name == "mysql":
+            return dialect.type_descriptor(LONGTEXT())
+        return dialect.type_descriptor(Text())
+
+
 class BeijingDateTime(TypeDecorator):
     """Store timestamps as Beijing (UTC+8) wall-clock time.
 

offline-backend-20260512/config.yaml

@@ -37,6 +37,26 @@ models:
   temperature: 1.0
   supports_vision: true
   supports_thinking: true
+- name: minimax-m2.7
+    display_name: MiniMax M2.7
+    use: langchain_openai:ChatOpenAI
+    model: MiniMax-M2.7
+    api_key: sk-cp-mrsAGtzDpUl0TpOBPVoiRG_pWF1kDDN9_mDAnONBs6HIPcNUexSKtSbx2QOxfKBvJT6Xw6hDTV6hdkJYFWfNA8UtHQ08TiGJiosdRoW6q1sGb6Cb7D85NHk
+    base_url: https://api.minimaxi.com
+    max_tokens: 4096
+    temperature: 1.0
+    supports_vision: true
+    supports_thinking: true
+- name: minimax-m2.5
+    display_name: MiniMax M2.5
+    use: langchain_openai:ChatOpenAI
+    model: MiniMax-M2.5
+    api_key: sk-cp-mrsAGtzDpUl0TpOBPVoiRG_pWF1kDDN9_mDAnONBs6HIPcNUexSKtSbx2QOxfKBvJT6Xw6hDTV6hdkJYFWfNA8UtHQ08TiGJiosdRoW6q1sGb6Cb7D85NHk
+    base_url: https://api.minimaxi.com
+    max_tokens: 4096
+    temperature: 1.0
+    supports_vision: true
+    supports_thinking: true
 tool_groups:
 - name: web
 - name: file:read