容错处理

frontend-web/src/open-canvas/pages/AIWritingPage.tsx

@@ -1,3 +1,86 @@
+/**
+ * AI 写作入口页（左右双栏：聊天面板 + 草稿预览）。
+ *
+ * ──────────────────────────────────────────────────────────────────────────
+ *  Docker 内网部署后「会话丢失 / transcript 报 500 / resume 报 400」排查指南
+ * ──────────────────────────────────────────────────────────────────────────
+ *
+ *  后端涉及接口（前缀 /api/ai-writing）：
+ *    POST   /start                          启动写作流程
+ *    GET    /{session_id}/stream            SSE 进度流
+ *    POST   /{session_id}/resume            提交用户干预 → 继续流程
+ *    PUT    /sessions/{session_id}/transcript  保存对话时间线
+ *
+ *  ── 1) 确认 alembic 迁移已跑过（修 transcript 500）─────────────────────
+ *     transcript 列在 MySQL 上原是 TEXT(64 KB 上限)，长时间线会溢出抛 500。
+ *     迁移 20260519_02 把它改成 LONGTEXT。容器启动会自动跑，验证方法：
+ *
+ *       docker exec <gateway> sh -c "cd /app/backend \
+ *         && PYTHONPATH=. uv run alembic current"
+ *       docker exec <mysql>  mysql -uroot -p<pwd> deerflow -e \
+ *         "SHOW COLUMNS FROM ai_writing_sessions LIKE 'transcript';"
+ *
+ *     列类型应是 longtext。若仍是 text，手动执行：
+ *       docker exec <gateway> sh -c "cd /app/backend \
+ *         && PYTHONPATH=. uv run alembic upgrade head"
+ *
+ *  ── 2) 确认 .deer-flow 数据卷已挂出（修「容器没重启也丢会话」）──────
+ *     AI 写作 graph 的 checkpointer 已从 MemorySaver 换成 AsyncSqliteSaver，
+ *     文件落到 .deer-flow/data/checkpoints.db。docker-compose 必须把该目录
+ *     挂出来，否则容器重启数据照样丢：
+ *
+ *       services:
+ *         gateway:
+ *           volumes:
+ *             - ./data:/app/.deer-flow    # 关键：挂出 .deer-flow
+ *
+ *  ── 3) 确认 worker 数 = 1（关键嫌疑：多 worker 内存隔离）────────────
+ *     _event_queues / _session_pending_pause 都是模块级 dict，每个 worker
+ *     一份。多 worker 下，/start 命中 worker A、/resume 命中 worker B，B
+ *     的 dict 没有这个 session → 404/「会话丢失」。
+ *
+ *       docker exec <gateway> ps -ef | grep -E 'uvicorn|gunicorn'
+ *
+ *     若 > 1：要么把启动参数改成 --workers 1，要么在 nginx 加 sticky
+ *     session（ip_hash / cookie），要么把队列搬 Redis。
+ *
+ *  ── 4) 排查节点异常（嫌疑：graph 节点抛错触发清理分支）─────────────
+ *     _stream_graph_segment 的 except 分支会清掉 _event_queues[session_id]，
+ *     用户下一次 resume 就 404。日志里搜：
+ *
+ *       docker logs <gateway> 2>&1 | grep -E \
+ *         'ai_writing graph segment failed|ai_writing.*error' | tail -50
+ *
+ *  ── 5) 排查反向代理掐 SSE 长连接（嫌疑：proxy_read_timeout 太短）───
+ *     SSE 默认连接挂半个小时以上，nginx 默认 60s 会断。检查内网反代：
+ *
+ *       grep -rE 'proxy_read_timeout|sse|keepalive' /etc/nginx/
+ *
+ *     需在 SSE 路由上加：proxy_read_timeout 3600s; proxy_buffering off;
+ *
+ *  ── 定期清理（每天 03:00 跑）────────────────────────────────────────
+ *     config.yaml -> ai_writing.cleanup 段（retention_days: 7）。联动清理
+ *     MySQL 业务表 + SQLite checkpointer 里的 thread 状态。手动验证：
+ *
+ *       docker exec <mysql> mysql -uroot -p<pwd> deerflow -e \
+ *         "SELECT COUNT(*) FROM ai_writing_sessions \
+ *          WHERE updated_at < NOW() - INTERVAL 7 DAY;"
+ *
+ *  ── 不想等凌晨 3 点？手动触发一次清理（admin only）──────────────────
+ *       POST /api/ai-writing/admin/cleanup
+ *       Body（全部可选，留空走 config 默认）：
+ *         { "retention_days": 7, "only_finished": false, "batch_size": 500 }
+ *       返回：
+ *         { "deleted_db": N, "deleted_checkpoints": N,
+ *           "elapsed_seconds": X, "params": {...} }
+ *
+ *     例：清掉所有 1 小时前的会话（测试用）：
+ *       curl -X POST http://<gateway>/api/ai-writing/admin/cleanup \
+ *         -H 'Content-Type: application/json' \
+ *         -H 'Cookie: <admin session cookie>' \
+ *         -d '{"retention_days": 0}'
+ * ──────────────────────────────────────────────────────────────────────────
+ */
 import { useRef } from 'react';
 import { TooltipProvider } from '@/components/ui/tooltip';
 import type { GroupImperativeHandle } from 'react-resizable-panels';

offline-backend-20260512/backend/app/gateway/ai_writing_cleanup.py

@@ -0,0 +1,300 @@
+"""AI 写作会话定期清理。
+
+按 ``config.yaml`` 的 ``ai_writing.cleanup`` 段做联动清理：
+
+  1. 找出 MySQL ``ai_writing_sessions`` 表里 ``updated_at < now - retention_days``
+     的会话 id 列表（分批，避免大事务）
+  2. 对每个 id 调用 LangGraph checkpointer 的 ``adelete_thread()``，删掉
+     SQLite checkpointer 里对应的 thread 状态
+  3. 再把 MySQL 行批量删掉
+
+整个流程「先删 checkpointer 再删 DB」是有意为之 —— 即便清理被中途中断，
+也只会出现「checkpointer 删了但 DB 没删」的孤儿，下一轮清理会发现这些
+DB 行仍然过期，重新尝试清理（checkpointer 删一个不存在的 thread 是 idempotent）。
+反过来如果先删 DB，万一 checkpointer 删失败，文件里就永久留着不可达的状态。
+
+多 worker 下用文件锁做 leader election，只有抢到锁的 worker 才会真正跑清理。
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import sys
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+from croniter import croniter
+
+logger = logging.getLogger(__name__)
+
+
+# ── 文件锁 ────────────────────────────────────────────────────────────────────
+
+
+def _acquire_lock(lock_path: Path):
+    """跨平台非阻塞独占文件锁，多 worker 下做 leader election。
+
+    成功 → 返回持有锁的文件句柄（调用方需保持其存活直到主动释放）；
+    抢不到 / 打开失败 → 返回 None。进程退出时 OS 自动释放。
+    """
+    try:
+        lock_path.parent.mkdir(parents=True, exist_ok=True)
+        handle = open(lock_path, "a+")
+    except OSError:
+        logger.exception("ai_writing_cleanup: 打开锁文件失败 path=%s", lock_path)
+        return None
+    try:
+        if sys.platform == "win32":
+            import msvcrt
+
+            handle.seek(0)
+            msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
+        else:
+            import fcntl
+
+            fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except OSError:
+        handle.close()
+        return None
+    return handle
+
+
+def _release_lock(handle) -> None:
+    try:
+        if sys.platform == "win32":
+            import msvcrt
+
+            handle.seek(0)
+            msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
+        else:
+            import fcntl
+
+            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
+    except OSError:
+        pass
+    finally:
+        try:
+            handle.close()
+        except OSError:
+            pass
+
+
+# ── 清理执行 ──────────────────────────────────────────────────────────────────
+
+
+async def _delete_checkpoint_thread(checkpointer, thread_id: str) -> None:
+    """删 checkpointer 里 thread_id 对应的全部状态。失败只记日志，不抛。"""
+    if checkpointer is None:
+        return
+    # LangGraph 异步删法首选 adelete_thread；老版本只有同步 delete_thread。
+    delete_async = getattr(checkpointer, "adelete_thread", None)
+    if delete_async is not None:
+        try:
+            await delete_async(thread_id)
+            return
+        except Exception:
+            logger.warning(
+                "ai_writing_cleanup: adelete_thread failed thread_id=%s", thread_id, exc_info=True
+            )
+            return
+    delete_sync = getattr(checkpointer, "delete_thread", None)
+    if delete_sync is not None:
+        try:
+            # 同步方法跑在线程池里，避免堵 event loop
+            await asyncio.to_thread(delete_sync, thread_id)
+        except Exception:
+            logger.warning(
+                "ai_writing_cleanup: delete_thread failed thread_id=%s", thread_id, exc_info=True
+            )
+
+
+def _get_active_session_ids() -> set[str]:
+    """返回内存里**当前正在跑**的 session_id 集合。
+
+    这些会话的写作流程还活着 —— 任何清理都不能动它们，否则用户「正在写」
+    的稿子会被秒删，体验灾难。延迟到下一轮清理（24 小时后）再处理也比误
+    删强。
+    """
+    try:
+        # 延迟 import 避免循环依赖：cleanup 模块在 lifespan 启动时被 import，
+        # routers/ai_writing 也被 include_router 主动 import，两者拓扑无环但
+        # 还是放函数里更稳。
+        from app.gateway.routers.ai_writing import _event_queues
+
+        return set(_event_queues.keys())
+    except Exception:
+        # 兜底：拿不到就当作没有活动会话（最差就是清掉一个刚跑完的会话，
+        # 比误删正在写的要好得多 —— 用户至少能从 DB 历史里看到结果）
+        logger.warning("ai_writing_cleanup: 读取活动会话列表失败", exc_info=True)
+        return set()
+
+
+async def run_cleanup_once(
+    *,
+    retention_days: int,
+    only_finished: bool = False,
+    batch_size: int = 500,
+) -> dict:
+    """执行一轮清理。返回统计信息。
+
+    **安全保证**：清理永远跳过内存里仍在跑的会话（``_event_queues`` 里有
+    对应 key 的）—— 这条规则即使在 admin 手动触发清理时也强制生效，防止
+    把用户「正在写」的稿子误删。
+
+    幂等 + 可重入 —— 多次调用对结果只会越扫越干净。即便清理途中被中断，下次
+    再跑会重新把 DB 里仍然过期的行扫到（checkpointer 删一个已不存在的 thread
+    是 no-op）。
+    """
+    from deerflow.agents.ai_writing.graph import get_ai_writing_checkpointer
+    from deerflow.persistence.ai_writing_sessions import AIWritingSessionRepository
+    from deerflow.persistence.engine import get_session_factory
+
+    sf = get_session_factory()
+    if sf is None:
+        logger.info("ai_writing_cleanup: persistence backend=memory，跳过清理")
+        return {"deleted_db": 0, "deleted_checkpoints": 0, "skipped_active": 0}
+
+    repo = AIWritingSessionRepository(sf)
+    checkpointer = get_ai_writing_checkpointer()
+    active_ids = _get_active_session_ids()
+
+    cutoff = datetime.now(UTC) - timedelta(days=retention_days)
+    logger.info(
+        "ai_writing_cleanup: 开始清理 cutoff=%s only_finished=%s batch_size=%d active=%d",
+        cutoff.isoformat(),
+        only_finished,
+        batch_size,
+        len(active_ids),
+    )
+
+    total_db = 0
+    total_cp = 0
+    total_skipped = 0
+    while True:
+        ids = await repo.list_older_than(cutoff, only_finished=only_finished, limit=batch_size)
+        if not ids:
+            break
+
+        # 关键保护：跳过任何仍在内存里跑的会话
+        ids_to_delete = [sid for sid in ids if sid not in active_ids]
+        skipped = len(ids) - len(ids_to_delete)
+        if skipped:
+            logger.warning(
+                "ai_writing_cleanup: 跳过 %d 个仍在跑的会话: %s",
+                skipped,
+                [sid for sid in ids if sid in active_ids],
+            )
+            total_skipped += skipped
+
+        if not ids_to_delete:
+            # 本批全是活动会话 —— 但 list_older_than 是按 updated_at 升序的，
+            # 下一批拿出来的更旧，所以这种情况只可能整个 cutoff 区间都被活动
+            # 会话占满，不用继续往下扫，break
+            break
+
+        # 先删 checkpointer，再删 DB —— 保证「DB 删了 → checkpointer 一定也清干净」
+        for sid in ids_to_delete:
+            await _delete_checkpoint_thread(checkpointer, sid)
+            total_cp += 1
+
+        deleted = await repo.delete_by_ids(ids_to_delete)
+        total_db += deleted
+        logger.info(
+            "ai_writing_cleanup: 本批删 %d 行（DB），%d 个 thread（checkpointer），跳过 %d",
+            deleted,
+            len(ids_to_delete),
+            skipped,
+        )
+
+        # 整批比 batch_size 小 → 没有更多了
+        if len(ids) < batch_size:
+            break
+
+    logger.info(
+        "ai_writing_cleanup: 清理完成 total_db=%d total_checkpoints=%d skipped_active=%d",
+        total_db,
+        total_cp,
+        total_skipped,
+    )
+    return {
+        "deleted_db": total_db,
+        "deleted_checkpoints": total_cp,
+        "skipped_active": total_skipped,
+    }
+
+
+# ── 调度循环 ──────────────────────────────────────────────────────────────────
+
+
+def _next_cron_delay_seconds(cron_expr: str, base: datetime | None = None) -> float:
+    """根据 cron 表达式算到下一次触发的秒数。
+
+    用 ``datetime.now(UTC)`` 做基准。croniter 接受 naive 时间，这里转成 naive
+    UTC 喂进去；当然如果用户希望按本地时区跑（凌晨 3 点），需要在外层调整 ——
+    我们用本地时区基准，等价于 cron 在本地时区跑。
+    """
+    base = base or datetime.now()
+    nxt = croniter(cron_expr, base).get_next(datetime)
+    return max(1.0, (nxt - base).total_seconds())
+
+
+async def cleanup_scheduler_loop() -> None:
+    """常驻后台任务：按 cron 调度跑清理。
+
+    多 worker 部署下文件锁做 leader election，没拿到锁的 worker 直接退出。
+    被 cancel 时 CancelledError 不会被 except Exception 吞掉，能干净退出。
+    """
+    from deerflow.config.app_config import get_app_config
+    from deerflow.config.runtime_paths import runtime_home
+
+    try:
+        cfg = get_app_config()
+    except Exception:
+        logger.exception("ai_writing_cleanup: 加载 app config 失败，调度器退出")
+        return
+
+    raw = getattr(cfg, "ai_writing", None) or {}
+    cleanup_cfg = (raw.get("cleanup") if isinstance(raw, dict) else None) or {}
+    if not cleanup_cfg.get("enabled", False):
+        logger.info("ai_writing_cleanup: cleanup.enabled=false，调度器不启动")
+        return
+
+    cron_expr = str(cleanup_cfg.get("cron", "0 3 * * *"))
+    retention_days = int(cleanup_cfg.get("retention_days", 7))
+    only_finished = bool(cleanup_cfg.get("delete_only_finished", False))
+    batch_size = int(cleanup_cfg.get("batch_size", 500))
+
+    # 提前校验 cron 表达式，配错就早 fail 早暴露
+    try:
+        croniter(cron_expr, datetime.now())
+    except Exception:
+        logger.exception("ai_writing_cleanup: 非法 cron 表达式 %r，调度器退出", cron_expr)
+        return
+
+    lock_handle = _acquire_lock(runtime_home() / ".ai_writing_cleanup.lock")
+    if lock_handle is None:
+        logger.info("ai_writing_cleanup: 另一个 worker 已持有调度锁，本 worker 跳过")
+        return
+
+    logger.info(
+        "ai_writing_cleanup: 已启动 cron=%r retention_days=%d only_finished=%s batch_size=%d",
+        cron_expr,
+        retention_days,
+        only_finished,
+        batch_size,
+    )
+    try:
+        while True:
+            delay = _next_cron_delay_seconds(cron_expr)
+            logger.info("ai_writing_cleanup: 下一次清理 %.0f 秒后", delay)
+            await asyncio.sleep(delay)
+            try:
+                await run_cleanup_once(
+                    retention_days=retention_days,
+                    only_finished=only_finished,
+                    batch_size=batch_size,
+                )
+            except Exception:
+                logger.exception("ai_writing_cleanup: 本轮清理失败（不影响下一轮）")
+    finally:
+        _release_lock(lock_handle)

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

@@ -70,6 +70,12 @@ logger = logging.getLogger(__name__)
 # firing signals into a worker that is stuck waiting for shutdown cleanup.
 _SHUTDOWN_HOOK_TIMEOUT_SECONDS = 5.0
 
+# AI 写作 in-flight task 收尾的最大等待时间。一个 graph 片段通常在几秒内
+# 跑到下一个 interrupt 或结束，但单段草稿生成走完整个 LLM 可能耗时较久，
+# 给 30 秒兜底；超时则 cancel 强制结束，已落盘的 checkpoint 仍可用于重启
+# 后恢复，丢的只是最后一段 in-flight 但未 checkpoint 的进度。
+_AI_WRITING_DRAIN_TIMEOUT_SECONDS = 30.0
+
 
 async def _ensure_admin_user(app: FastAPI) -> None:
     """Startup hook: handle first boot and migrate orphan threads otherwise.
@@ -361,10 +367,68 @@ async def _curator_scheduler_loop() -> None:
         _release_singleton_lock(lock_handle)
 
 
+def _warn_if_multi_worker() -> None:
+    """检测 uvicorn / gunicorn 多 worker 部署并打 ERROR 日志。
+
+    AI 写作 + 主 agent 流式响应都依赖**进程内**的内存状态字典
+    （``_event_queues``、``_session_pending_pause`` 等）。多 worker 下，
+    用户的 /start /stream /resume 会被 OS 随机分配到不同 worker，
+    per-worker 的内存字典互相看不见 → SSE 事件丢失、会话「凭空消失」、
+    resume 走不通。
+
+    所以这里在 startup 打 ERROR 强提醒。**不抛异常 fail-fast** —— 万一
+    运维确实知道自己在干嘛（比如配了 sticky session），不应该阻止启动。
+    """
+    # uvicorn / gunicorn 都不会主动暴露 worker 数到代码层。可以从环境变量
+    # 或者通过观察 sibling 进程数粗略判断。优先用 WEB_CONCURRENCY（gunicorn
+    # / uvicorn 都认）。
+    web_concurrency = os.environ.get("WEB_CONCURRENCY", "").strip()
+    if web_concurrency and web_concurrency.isdigit() and int(web_concurrency) > 1:
+        logger.error(
+            "⚠ 检测到 WEB_CONCURRENCY=%s（多 worker 部署）。AI 写作 / 主 agent "
+            "流式响应依赖进程内内存状态，多 worker 下会出现「会话丢失 / "
+            "SSE 断流 / resume 400」等问题。强烈建议改为 WEB_CONCURRENCY=1 "
+            "或在反向代理上启用 sticky session。详见 AIWritingPage.tsx 的注释。",
+            web_concurrency,
+        )
+        return
+
+    # 兜底：扫 /proc 里同名 uvicorn 进程数（只在 Linux 容器里能用）
+    if not sys.platform.startswith("linux"):
+        return
+    try:
+        from pathlib import Path
+
+        peers = 0
+        for entry in Path("/proc").iterdir():
+            if not entry.name.isdigit():
+                continue
+            try:
+                cmdline = (entry / "cmdline").read_bytes().replace(b"\x00", b" ").decode("utf-8", "ignore")
+            except OSError:
+                continue
+            if "uvicorn" in cmdline and "app.gateway.app" in cmdline:
+                peers += 1
+        if peers > 1:
+            logger.error(
+                "⚠ 检测到 %d 个同名 uvicorn 进程。多 worker 部署会导致 AI 写作 "
+                "「会话丢失 / SSE 断流」等问题，请改为单 worker 或在反向代理上 "
+                "启用 sticky session。",
+                peers,
+            )
+    except Exception:
+        # 检测本身失败不算事 —— 这只是个提醒
+        pass
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     """Application lifespan handler."""
 
+    # 早一点检测多 worker 部署 —— 这是 AI 写作 / 主 agent 流式响应的常见
+    # 配置陷阱，每个 worker 启动时都会打一条 ERROR 强提醒运维
+    _warn_if_multi_worker()
+
     # Load config and check necessary environment variables at startup
     try:
         app.state.config = get_app_config()
@@ -402,6 +466,17 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         except Exception:
             logger.exception("Skill curator scheduler failed to start (non-fatal)")
 
+        # 启动 AI 写作会话定期清理:按 config.yaml 的 ai_writing.cleanup 段调度,
+        # 默认每天凌晨 3 点扫一次,删 retention_days(默认 7 天)以前的会话,联动
+        # 清理 MySQL 业务表 + SQLite checkpointer 里的 thread 状态。多 worker
+        # 下用文件锁选 leader,只有抢到锁的 worker 真正跑清理。
+        try:
+            from app.gateway.ai_writing_cleanup import cleanup_scheduler_loop
+
+            app.state.ai_writing_cleanup_task = asyncio.create_task(cleanup_scheduler_loop())
+        except Exception:
+            logger.exception("AI writing cleanup scheduler failed to start (non-fatal)")
+
         # Start IM channel service if any channels are configured
         try:
             from app.channels.service import start_channel_service
@@ -439,6 +514,44 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
             except Exception:
                 logger.exception("Curator scheduler shutdown error")
 
+        # 停止 AI 写作清理调度器
+        cleanup_task = getattr(app.state, "ai_writing_cleanup_task", None)
+        if cleanup_task is not None:
+            cleanup_task.cancel()
+            try:
+                await cleanup_task
+            except asyncio.CancelledError:
+                pass
+            except Exception:
+                logger.exception("AI writing cleanup scheduler shutdown error")
+
+        # 等待所有 in-flight 写作 graph task 跑到下一个 interrupt / 结束。
+        # 这是「写作流程能完整继续」的关键 —— 如果直接关 checkpointer
+        # connection，正在跑 astream 的 task 会在写 checkpoint 时抛错，
+        # 已经生成但还没落盘的进度全丢。
+        try:
+            from app.gateway.routers.ai_writing import wait_for_inflight_graph_tasks
+
+            await wait_for_inflight_graph_tasks(timeout=_AI_WRITING_DRAIN_TIMEOUT_SECONDS)
+        except Exception:
+            logger.exception("Wait for in-flight AI writing tasks failed (non-fatal)")
+
+        # 关闭 AI 写作 graph 持有的 checkpointer 连接（aiosqlite Connection / pg pool）
+        try:
+            from deerflow.agents.ai_writing.graph import shutdown_ai_writing_graph
+
+            await asyncio.wait_for(
+                shutdown_ai_writing_graph(),
+                timeout=_SHUTDOWN_HOOK_TIMEOUT_SECONDS,
+            )
+        except TimeoutError:
+            logger.warning(
+                "AI writing checkpointer shutdown exceeded %.1fs; proceeding with worker exit.",
+                _SHUTDOWN_HOOK_TIMEOUT_SECONDS,
+            )
+        except Exception:
+            logger.exception("Failed to shutdown AI writing checkpointer")
+
         try:
             scheduler_service = getattr(app.state, "scheduled_task_service", None)
             if scheduler_service is not None:

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

@@ -70,6 +70,28 @@ class RenameSessionRequest(BaseModel):
     title: str
 
 
+class AdminCleanupRequest(BaseModel):
+    """手动清理接口的可选覆盖参数。
+
+    全部留空 → 用 config.yaml 的 ai_writing.cleanup 段默认值；显式传值
+    → 覆盖配置（一般用于运维想验证一个临时阈值，例如清掉「最近 1 小时
+    以前的」会话来模拟整轮调度）。
+    """
+    retention_days: Optional[int] = Field(default=None, ge=0, description="保留多少天以内的会话；0 = 不保留")
+    only_finished: Optional[bool] = Field(default=None, description="True 时只清 done/error 状态")
+    batch_size: Optional[int] = Field(default=None, ge=1, le=10000, description="单批最多删多少行")
+
+
+class AdminCleanupResponse(BaseModel):
+    deleted_db: int
+    deleted_checkpoints: int
+    # 因「仍在跑」被保护掉的会话数。永远跳过 _event_queues 里活跃的会话，
+    # 即使 retention_days=0 也不会秒杀用户正在写的稿。
+    skipped_active: int = 0
+    elapsed_seconds: float
+    params: dict
+
+
 class SaveTranscriptRequest(BaseModel):
     # 前端组装的完整对话时间线（progressEvents + completedInterventions）
     transcript: dict
@@ -153,11 +175,40 @@ _session_pending_pause: dict[str, tuple] = {}
 # lost session don't each rebuild a different queue object.
 _recovery_lock = asyncio.Lock()
 
+# 在飞的 graph 推进任务（每次 /start 和 /resume 都会塞一个进来）。
+# Lifespan shutdown 时会等这些 task 跑到下一个 interrupt / 结束，再关闭
+# checkpointer 连接 —— 否则中途关 connection，task 内部的 sqlite 写入会抛
+# 错，导致 LLM 已经生成但还没落 checkpoint 的进度丢失。
+_inflight_graph_tasks: set[asyncio.Task] = set()
+
 
 # ── 工具函数 ──────────────────────────────────────────────────────────────────
 
 def _sse(event: str, data: dict) -> str:
-    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
+    """组装一帧 SSE。
+
+    JSON 序列化对 progress_events 来说理论上不会失败（节点里塞的都是 str/dict
+    /list/int 这类基础类型），但一旦塞进了 datetime / bytes / 自定义对象，
+    ``json.dumps`` 会抛 TypeError → SSE 生成器爆掉 → 客户端再也收不到任何
+    事件，包括关键的 await_user 和 done。
+
+    这里用 ``default=str`` 兜底任何「带 __str__」的对象；如果连 ``str()``
+    都炸了，最后还兜一次手写一条 error 帧推出去 —— 写作流程「能继续」比
+    「漂亮地序列化」更重要。
+    """
+    try:
+        payload = json.dumps(data, ensure_ascii=False, default=str)
+    except Exception as exc:
+        logger.exception("ai_writing: SSE payload 序列化失败 event=%s", event)
+        payload = json.dumps(
+            {
+                "session_id": data.get("session_id") if isinstance(data, dict) else None,
+                "message": f"事件序列化失败（已降级）: {type(exc).__name__}",
+            },
+            ensure_ascii=False,
+        )
+        event = "notice"  # 不要硬塞回原 event，万一是 error 会污染语义
+    return f"event: {event}\ndata: {payload}\n\n"
 
 
 def _ai_writing_config() -> dict:
@@ -193,14 +244,81 @@ def _truncate_title(text: str, max_len: int = 50) -> str:
     return text[:max_len] + ("…" if len(text) > max_len else "")
 
 
+def _spawn_graph_task(coro) -> asyncio.Task:
+    """启一个 graph 推进 task 并登记到 _inflight_graph_tasks。
+
+    用 ``add_done_callback`` 在完成时自动反登记 —— 不管 task 正常结束、抛
+    异常、还是被 cancel，都会清理；这样 _inflight_graph_tasks 不会泄漏。
+    """
+    task = asyncio.create_task(coro)
+    _inflight_graph_tasks.add(task)
+    task.add_done_callback(_inflight_graph_tasks.discard)
+    return task
+
+
+async def wait_for_inflight_graph_tasks(timeout: float) -> tuple[int, int]:
+    """Lifespan shutdown 入口：阻塞等待所有 in-flight graph task 收尾。
+
+    每个 ``_stream_graph_segment`` 跑到下一个 ``interrupt()`` 或 ``END`` 就
+    会自然返回；通常只需要等几秒到几十秒。超时后剩下的会被 ``cancel()``，
+    但 cancel 完仍要等它们 ack 退出（带极短的 grace），否则下一步关
+    checkpointer 时还会撞上活的协程。
+
+    Returns:
+      (drained, cancelled): 自然结束的数量、被 cancel 强制结束的数量
+    """
+    if not _inflight_graph_tasks:
+        return 0, 0
+
+    pending = set(_inflight_graph_tasks)
+    initial = len(pending)
+    logger.info("ai_writing: 等待 %d 个 in-flight graph task 收尾（timeout=%.1fs）", initial, timeout)
+
+    done, still_running = await asyncio.wait(pending, timeout=timeout)
+    drained = len(done)
+
+    cancelled = 0
+    if still_running:
+        logger.warning(
+            "ai_writing: 仍有 %d 个 task 超时未结束，强制 cancel", len(still_running)
+        )
+        for t in still_running:
+            t.cancel()
+            cancelled += 1
+        # 给 cancel 一点 ack 时间，避免下游关连接时撞活协程
+        await asyncio.wait(still_running, timeout=2.0)
+
+    logger.info("ai_writing: in-flight 收尾完成 drained=%d cancelled=%d", drained, cancelled)
+    return drained, cancelled
+
+
 # ── 路由 ──────────────────────────────────────────────────────────────────────
 
 @router.post("/start", response_model=StartWritingResponse)
 async def start_writing(body: StartWritingRequest, request: Request):
-    """启动写作流程，返回 session_id。"""
-    from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH
+    """启动写作流程，返回 session_id。
+
+    **失败兜底契约**：本接口除非 Pydantic 校验 body 失败（422），其他任何
+    后端组件（DB / checkpointer / 文章类型仓库 / 用户解析）失败都不会导致
+    /start 返回 5xx。哪怕兜底走的是「极简内存模式」也要先把 session_id
+    交给前端，让它能起 SSE 接到 error 事件，避免 UI 卡死在 loading 状态。
+    """
+    from deerflow.agents.ai_writing.graph import get_ai_writing_graph
     from app.gateway.deps import get_ai_writing_session_store, get_article_type_store, get_current_user
 
+    # get_ai_writing_graph() 内部已对 sqlite/postgres 失败做了 InMemorySaver
+    # 降级，理论上永不抛。但 build_ai_writing_graph 拼图结构也可能因下游
+    # 节点 import 失败抛错 —— 这种情况 graph 完全不可用，必须早 fail，前端
+    # 弹错处理。
+    try:
+        graph = await get_ai_writing_graph()
+    except Exception as e:
+        logger.exception("ai_writing: graph 构建彻底失败，无法启动")
+        raise HTTPException(
+            status_code=503,
+            detail=f"AI 写作服务暂不可用: {type(e).__name__}",
+        ) from e
+
     session_id = str(uuid.uuid4())
     queue: asyncio.Queue = asyncio.Queue()
     _event_queues[session_id] = queue
@@ -258,7 +376,7 @@ async def start_writing(body: StartWritingRequest, request: Request):
     }
     config = _make_config(session_id, body.model_name, valid_article_types=valid_article_types)
 
-    asyncio.create_task(_stream_graph_segment(session_id, initial_state, config, AI_WRITING_GRAPH))
+    _spawn_graph_task(_stream_graph_segment(session_id, initial_state, config, graph))
     return StartWritingResponse(session_id=session_id)
 
 
@@ -430,9 +548,11 @@ async def stream_progress(session_id: str):
 @router.post("/{session_id}/resume")
 async def resume_writing(session_id: str, body: ResumeRequest, request: Request):
     """用户提交干预决策，恢复暂停的流程。"""
-    from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH
+    from deerflow.agents.ai_writing.graph import get_ai_writing_graph
     from app.gateway.deps import get_current_user
 
+    graph = await get_ai_writing_graph()
+
     if session_id not in _event_queues:
         # 内存会话注册表丢失（进程内清理后重连等）。先尝试从 checkpointer 恢复，
         # 让用户能在干预点继续，而不是直接 404 中断整个写作流程。
@@ -482,8 +602,8 @@ async def resume_writing(session_id: str, body: ResumeRequest, request: Request)
     # Clear pending pause so SSE reconnect won't re-inject the stale await_user event
     _session_pending_pause.pop(session_id, None)
 
-    asyncio.create_task(
-        _stream_graph_segment(session_id, Command(resume=intervention), config, AI_WRITING_GRAPH)
+    _spawn_graph_task(
+        _stream_graph_segment(session_id, Command(resume=intervention), config, graph)
     )
     return {"message": "已提交，流程恢复中"}
 
@@ -491,15 +611,112 @@ async def resume_writing(session_id: str, body: ResumeRequest, request: Request)
 @router.get("/{session_id}/state")
 async def get_state(session_id: str):
     """查询当前流程状态快照。"""
-    from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH
+    from deerflow.agents.ai_writing.graph import get_ai_writing_graph
 
+    graph = await get_ai_writing_graph()
     config = _make_config(session_id)
-    snapshot = await AI_WRITING_GRAPH.aget_state(config)
+    snapshot = await graph.aget_state(config)
     if not snapshot or not snapshot.values:
         raise HTTPException(status_code=404, detail="Session not found")
     return snapshot.values
 
 
+# ── 管理接口 ──────────────────────────────────────────────────────────────────
+
+async def _require_admin(request: Request) -> None:
+    """放行 admin / auth-disabled，其余 403。
+
+    与 scheduled_tasks._require_admin 同套语义：auth_disabled 模式下
+    get_optional_user_from_request 返回 None，此时放行（开发环境一致体验）。
+    """
+    from app.gateway.deps import get_optional_user_from_request
+
+    user = await get_optional_user_from_request(request)
+    if user is None:
+        return
+    if getattr(user, "system_role", None) != "admin":
+        raise HTTPException(status_code=403, detail="AI 写作清理管理接口仅限管理员")
+
+
+def _resolve_cleanup_params(override: AdminCleanupRequest) -> dict:
+    """合并 config.yaml 默认值 + 请求显式覆盖。
+
+    显式传 0 / False 也算覆盖（用 ``is not None`` 判定），方便测「保留 0 天 =
+    全删」这种极端场景。
+    """
+    from deerflow.config import get_app_config
+
+    raw = getattr(get_app_config(), "ai_writing", None) or {}
+    cfg = (raw.get("cleanup") if isinstance(raw, dict) else None) or {}
+
+    return {
+        "retention_days": (
+            override.retention_days if override.retention_days is not None
+            else int(cfg.get("retention_days", 7))
+        ),
+        "only_finished": (
+            override.only_finished if override.only_finished is not None
+            else bool(cfg.get("delete_only_finished", False))
+        ),
+        "batch_size": (
+            override.batch_size if override.batch_size is not None
+            else int(cfg.get("batch_size", 500))
+        ),
+    }
+
+
+@router.post("/admin/cleanup", response_model=AdminCleanupResponse)
+async def admin_cleanup(
+    body: AdminCleanupRequest,
+    request: Request,
+) -> AdminCleanupResponse:
+    """**仅管理员** 手动触发一次会话清理（不等凌晨 3 点定时调度）。
+
+    用途：
+      - 验证 cleanup 配置改动是否生效
+      - 内网紧急清理（磁盘满 / MySQL 行数过多）
+      - 调试：传 ``retention_days: 0`` 一刷拉清空，方便重新跑回归
+
+    所有参数留空 → 沿用 config.yaml 的 ``ai_writing.cleanup`` 段默认。显式
+    传值会覆盖配置但**不修改 config.yaml**（一次性参数，下次定时调度还是用
+    配置文件的值）。
+    """
+    import time
+
+    from app.gateway.ai_writing_cleanup import run_cleanup_once
+
+    await _require_admin(request)
+    params = _resolve_cleanup_params(body)
+
+    started = time.monotonic()
+    try:
+        result = await run_cleanup_once(**params)
+    except Exception as e:
+        logger.exception("ai_writing admin cleanup failed: params=%s", params)
+        # 抛 500 是合理的 —— 管理员触发清理失败需要看到失败原因，跟「业务
+        # 接口被失败拖累」是两回事
+        raise HTTPException(
+            status_code=500,
+            detail=f"清理执行失败: {type(e).__name__}: {e}",
+        ) from e
+
+    elapsed = time.monotonic() - started
+    logger.info(
+        "ai_writing admin cleanup done: params=%s deleted_db=%d deleted_cp=%d elapsed=%.2fs",
+        params,
+        result["deleted_db"],
+        result["deleted_checkpoints"],
+        elapsed,
+    )
+    return AdminCleanupResponse(
+        deleted_db=result["deleted_db"],
+        deleted_checkpoints=result["deleted_checkpoints"],
+        skipped_active=result.get("skipped_active", 0),
+        elapsed_seconds=round(elapsed, 3),
+        params=params,
+    )
+
+
 # ── 内部辅助 ──────────────────────────────────────────────────────────────────
 
 async def _stream_graph_segment(
@@ -650,13 +867,14 @@ def _event_type(evt: dict) -> str:
 async def _load_graph_snapshot(session_id: str):
     """读取 AI 写作 Graph 在该 session 上的状态快照（失败/无状态返回 None）。
 
-    注意：Graph 用的是内存 checkpointer（MemorySaver），所以这里只能在
-    **进程存活**期间恢复；进程/容器重启后快照同样会丢失。
+    Graph 的 checkpointer 现在按 config.yaml 持久化到磁盘（默认 SQLite），
+    所以容器重启后只要数据卷还在，仍然能恢复出来。
     """
     try:
-        from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH
+        from deerflow.agents.ai_writing.graph import get_ai_writing_graph
 
-        snapshot = await AI_WRITING_GRAPH.aget_state(_make_config(session_id))
+        graph = await get_ai_writing_graph()
+        snapshot = await graph.aget_state(_make_config(session_id))
     except Exception:
         logger.exception("ai_writing: aget_state failed during recovery: session=%s", session_id)
         return None

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/__init__.py

@@ -1,3 +1,15 @@
-from .graph import AI_WRITING_GRAPH, build_ai_writing_graph
+from .graph import (
+    AI_WRITING_GRAPH,
+    build_ai_writing_graph,
+    get_ai_writing_checkpointer,
+    get_ai_writing_graph,
+    shutdown_ai_writing_graph,
+)
 
-__all__ = ["AI_WRITING_GRAPH", "build_ai_writing_graph"]
+__all__ = [
+    "AI_WRITING_GRAPH",
+    "build_ai_writing_graph",
+    "get_ai_writing_graph",
+    "get_ai_writing_checkpointer",
+    "shutdown_ai_writing_graph",
+]

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/graph.py

@@ -1,7 +1,8 @@
 """AI 写作 LangGraph Graph，含 4 个用户干预点。"""
 from __future__ import annotations
 
-from langgraph.checkpoint.memory import MemorySaver
+import logging
+
 from langgraph.graph import END, StateGraph
 from langgraph.types import interrupt
 
@@ -12,6 +13,8 @@ from .nodes.writer_outline import writer_outline_node
 from .nodes.writer_draft import writer_draft_node
 from .nodes.editor import editor_node
 
+logger = logging.getLogger(__name__)
+
 # ── 干预节点 ──────────────────────────────────────────────────────────────────
 
 async def pause_material_confirm(state: AIWritingState) -> dict:
@@ -298,7 +301,16 @@ def after_review_pause(state: AIWritingState) -> str:
 
 # ── 组装 Graph ────────────────────────────────────────────────────────────────
 
-def build_ai_writing_graph():
+def build_ai_writing_graph(checkpointer=None):
+    """构建并 compile AI 写作 graph。
+
+    Args:
+      checkpointer: 可选的 LangGraph checkpointer。
+        - ``None`` → 退回 ``MemorySaver`` —— 仅供测试 / 开发使用。
+          进程或容器重启后会话状态全部丢失，前端拿着旧 session_id 调
+          ``/resume`` 会得到 404。
+        - 其他 → 传入的 (Async)BaseCheckpointSaver 实例，用于持久化。
+    """
     g = StateGraph(AIWritingState)
 
     g.add_node("intent_parser",   intent_parser_node)
@@ -339,8 +351,188 @@ def build_ai_writing_graph():
     g.add_conditional_edges("pause_review", after_review_pause,
                              {"end": END, "writer_draft": "writer_draft"})
 
-    checkpointer = MemorySaver()
+    if checkpointer is None:
+        # MemorySaver 仅作为 fallback —— 提示一下，正式部署应通过
+        # get_ai_writing_graph() 拿到带持久化 checkpointer 的 graph。
+        from langgraph.checkpoint.memory import MemorySaver
+
+        checkpointer = MemorySaver()
     return g.compile(checkpointer=checkpointer)
 
 
-AI_WRITING_GRAPH = build_ai_writing_graph()
+# ── 持久化 checkpointer 单例（懒加载）─────────────────────────────────────────
+
+# 模块级 graph 槽位。import 时为 None；首次 ``get_ai_writing_graph()`` 调用时
+# 按 config.yaml 的 checkpointer 段懒构建出真正的 graph。
+# 测试可通过 ``monkeypatch.setattr(graph_mod, "AI_WRITING_GRAPH", stub)``
+# 直接注入桩；本模块所有访问点都走 ``get_ai_writing_graph()`` 这一入口，会优先
+# 返回 module attribute 的值，保留测试桩注入路径。
+AI_WRITING_GRAPH = None
+
+# 持有 async checkpointer（底层 aiosqlite Connection 或 psycopg pool 的所有者）。
+# 进程生命周期内常驻；shutdown 时由 ``shutdown_ai_writing_graph`` 显式关闭。
+_async_checkpointer_singleton = None
+
+
+def _fallback_to_memory_saver(reason: str):
+    """退回 InMemorySaver 的统一兜底入口。
+
+    把 reason 打到 warning 日志里方便运维定位。InMemorySaver 进程重启会丢
+    状态，但**当次写作能跑完** —— 我们的首要目标是「写作流程能继续」，
+    持久化是锦上添花。
+    """
+    from langgraph.checkpoint.memory import InMemorySaver
+
+    logger.warning(
+        "ai_writing: 持久化 checkpointer 不可用，退回 InMemorySaver。原因: %s",
+        reason,
+    )
+    return InMemorySaver()
+
+
+async def _get_async_checkpointer():
+    """按 config.yaml 的 checkpointer 段返回一个 async checkpointer 实例。
+
+    **绝不抛异常** —— 任何失败一律退回 InMemorySaver。这是为了保证「写作
+    流程能继续」这个首要目标：持久化坏了用户至少还能写完当次稿，重启会丢
+    会话但不影响当下创作。
+
+    优先级：
+      1. ``checkpointer:`` 段（``type=sqlite|postgres|memory``）
+      2. fallback 到 ``InMemorySaver`` 并打 warning
+
+    返回的 saver 与主 agent 公用同一 sqlite 文件 —— LangGraph 的 schema
+    用 ``(thread_id, checkpoint_ns)`` 做主键，AI 写作的 thread_id 是
+    会话 uuid，跟主 agent 的 LangGraph thread_id 不会撞。
+    """
+    global _async_checkpointer_singleton
+    if _async_checkpointer_singleton is not None:
+        return _async_checkpointer_singleton
+
+    try:
+        from deerflow.config.checkpointer_config import get_checkpointer_config
+
+        cp_cfg = get_checkpointer_config()
+    except Exception:
+        logger.exception("ai_writing: 加载 checkpointer 配置失败")
+        _async_checkpointer_singleton = _fallback_to_memory_saver("加载 config 抛异常")
+        return _async_checkpointer_singleton
+
+    if cp_cfg is None:
+        _async_checkpointer_singleton = _fallback_to_memory_saver(
+            "config.yaml 未配置 checkpointer 段"
+        )
+        return _async_checkpointer_singleton
+
+    if cp_cfg.type == "memory":
+        from langgraph.checkpoint.memory import InMemorySaver
+
+        _async_checkpointer_singleton = InMemorySaver()
+        logger.info("ai_writing: 配置 type=memory，使用 InMemorySaver")
+        return _async_checkpointer_singleton
+
+    if cp_cfg.type == "sqlite":
+        try:
+            import aiosqlite
+            from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+            from deerflow.runtime.store._sqlite_utils import (
+                ensure_sqlite_parent_dir,
+                resolve_sqlite_conn_str,
+            )
+
+            conn_str = resolve_sqlite_conn_str(
+                cp_cfg.connection_string or "ai_writing_checkpoints.db"
+            )
+            ensure_sqlite_parent_dir(conn_str)
+            # 直接持有 aiosqlite.Connection —— 进程生命周期，shutdown 时显式关。
+            conn = await aiosqlite.connect(conn_str)
+            saver = AsyncSqliteSaver(conn)
+            await saver.setup()
+            _async_checkpointer_singleton = saver
+            logger.info("ai_writing: 使用 AsyncSqliteSaver (%s)", conn_str)
+            return saver
+        except Exception as exc:
+            # sqlite 失败原因可能是：依赖未装 / 磁盘满 / 文件权限 / 路径不存在 /
+            # 文件被外部进程独占锁。任何一个都不能让 AI 写作整个不可用。
+            logger.exception(
+                "ai_writing: AsyncSqliteSaver 初始化失败，conn_str=%r",
+                getattr(cp_cfg, "connection_string", None),
+            )
+            _async_checkpointer_singleton = _fallback_to_memory_saver(
+                f"sqlite 初始化失败: {type(exc).__name__}: {exc}"
+            )
+            return _async_checkpointer_singleton
+
+    if cp_cfg.type == "postgres":
+        try:
+            from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
+            from psycopg_pool import AsyncConnectionPool
+
+            if not cp_cfg.connection_string:
+                raise ValueError("checkpointer.connection_string 是 postgres 必填项")
+            pool = AsyncConnectionPool(cp_cfg.connection_string, open=False)
+            await pool.open()
+            saver = AsyncPostgresSaver(pool)
+            await saver.setup()
+            _async_checkpointer_singleton = saver
+            logger.info("ai_writing: 使用 AsyncPostgresSaver")
+            return saver
+        except Exception as exc:
+            logger.exception("ai_writing: AsyncPostgresSaver 初始化失败")
+            _async_checkpointer_singleton = _fallback_to_memory_saver(
+                f"postgres 初始化失败: {type(exc).__name__}: {exc}"
+            )
+            return _async_checkpointer_singleton
+
+    # 未知 type —— config 配错也不能让服务挂
+    _async_checkpointer_singleton = _fallback_to_memory_saver(
+        f"未知 checkpointer.type={cp_cfg.type!r}"
+    )
+    return _async_checkpointer_singleton
+
+
+async def get_ai_writing_graph():
+    """懒加载并返回 AI 写作 graph 单例。
+
+    第一次调用按 config 选定的 checkpointer 构建 graph 并缓存；后续调用直接
+    返回缓存。测试可在调用前通过 ``monkeypatch.setattr`` 直接替换模块级
+    ``AI_WRITING_GRAPH`` —— 本函数会优先返回已设的值。
+    """
+    global AI_WRITING_GRAPH
+    if AI_WRITING_GRAPH is not None:
+        return AI_WRITING_GRAPH
+    checkpointer = await _get_async_checkpointer()
+    AI_WRITING_GRAPH = build_ai_writing_graph(checkpointer=checkpointer)
+    return AI_WRITING_GRAPH
+
+
+def get_ai_writing_checkpointer():
+    """返回当前已实例化的 checkpointer（清理任务用它来删 thread）。
+
+    若尚未懒构建过则返回 None。
+    """
+    return _async_checkpointer_singleton
+
+
+async def shutdown_ai_writing_graph() -> None:
+    """关闭 saver 持有的底层连接，并清空 graph 单例。Lifespan shutdown 调用。"""
+    global AI_WRITING_GRAPH, _async_checkpointer_singleton
+    saver = _async_checkpointer_singleton
+    _async_checkpointer_singleton = None
+    AI_WRITING_GRAPH = None
+    if saver is None:
+        return
+    # aiosqlite saver: saver.conn 是 aiosqlite.Connection
+    conn = getattr(saver, "conn", None)
+    if conn is not None and hasattr(conn, "close"):
+        try:
+            await conn.close()
+        except Exception:
+            logger.warning("ai_writing: 关闭 checkpointer connection 失败", exc_info=True)
+    # postgres saver: saver.pool 是 AsyncConnectionPool
+    pool = getattr(saver, "pool", None)
+    if pool is not None and hasattr(pool, "close"):
+        try:
+            await pool.close()
+        except Exception:
+            logger.warning("ai_writing: 关闭 checkpointer pool 失败", exc_info=True)

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

@@ -5,7 +5,7 @@ from __future__ import annotations
 import json
 from datetime import UTC, datetime
 
-from sqlalchemy import select
+from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from deerflow.persistence.ai_writing_sessions.model import AIWritingSessionRow
@@ -160,3 +160,38 @@ class AIWritingSessionRepository:
             await session.delete(row)
             await session.commit()
             return True
+
+    async def list_older_than(
+        self,
+        cutoff: datetime,
+        *,
+        only_finished: bool = False,
+        limit: int = 500,
+    ) -> list[str]:
+        """列出 updated_at 早于 cutoff 的会话 id（最多 limit 个）。
+
+        清理任务先 list 出 id 列表，再用这些 id 联动删 checkpointer
+        里对应的 thread，最后再删 MySQL 行 —— 保证两边一致。
+
+        Args:
+          cutoff: 截止时间，updated_at < cutoff 的会话被认为「过期」
+          only_finished: True 时只清 status in (done/error)，False 不看状态
+          limit: 单批最多返回多少条，用于分批处理避免大事务
+        """
+        stmt = select(AIWritingSessionRow.id).where(AIWritingSessionRow.updated_at < cutoff)
+        if only_finished:
+            stmt = stmt.where(AIWritingSessionRow.status.in_(("done", "error")))
+        stmt = stmt.order_by(AIWritingSessionRow.updated_at.asc()).limit(limit)
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            return [row[0] for row in result.all()]
+
+    async def delete_by_ids(self, session_ids: list[str]) -> int:
+        """按 id 列表批量删除。返回实际删除的行数。"""
+        if not session_ids:
+            return 0
+        stmt = delete(AIWritingSessionRow).where(AIWritingSessionRow.id.in_(session_ids))
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            await session.commit()
+            return int(result.rowcount or 0)

offline-backend-20260512/config.yaml

@@ -117,8 +117,22 @@ database:
   pool_size: 10
   echo_sql: false
 checkpointer:
+  # SQLite 文件路径用相对路径，由 resolve_sqlite_conn_str 解析成绝对路径。
+  # Docker 部署：把容器内 .deer-flow 目录挂卷出来即可持久化（撑住容器重启）。
+  # 这里同时被「主 agent」和「AI 写作 graph」共用 —— LangGraph 用
+  # (thread_id, checkpoint_ns) 做隔离，session_id 是 uuid，不会撞。
   type: sqlite
-  connection_string: D:\cursor-work\offline-backend-20260512\backend\.deer-flow\data\deerflow.db
+  connection_string: .deer-flow/data/checkpoints.db
+# AI 写作子系统的会话清理配置。
+# 业务表(ai_writing_sessions @ MySQL) + checkpointer(checkpoints.db @ SQLite) 联动清理。
+# 留意：清理走 cron 表达式调度，默认每天凌晨 3 点跑一次，按 updated_at 一刷拉清。
+ai_writing:
+  cleanup:
+    enabled: true
+    retention_days: 7              # 保留 7 天以内的会话，更早的整条删
+    cron: "0 3 * * *"              # 每天 03:00 本地时区
+    delete_only_finished: false    # false=不看状态一刷拉；true=只清 done/error
+    batch_size: 500                # 单批最多删多少行，避免长事务
 # 记忆子系统:跨会话持久化用户画像与 agent 私有记忆。
 # 三层覆盖优先级(高->低):
 #   users/{user_id}/memory_config.json > .deer-flow/memory_user_overrides.json > 此处