feat: AI writing history, Word export, roundtable dispatch chain

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

@@ -168,6 +168,16 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
 
         app.state.roundtable_chain_store = make_roundtable_chain_store(sf)
 
+        from deerflow.persistence.roundtable_jobs import (
+            make_roundtable_job_store,
+            register_default_store as _register_roundtable_job_default_store,
+        )
+
+        app.state.roundtable_job_store = make_roundtable_job_store(sf)
+        # 让 LangGraph Server 路径下的圆桌编排图节点能跨 app/harness 边界拿到 repo
+        # （Phase 2 后台编排图写进度用）。
+        _register_roundtable_job_default_store(app.state.roundtable_job_store)
+
         # Run event store (has its own factory with config-driven backend selection)
         run_events_config = getattr(config, "run_events", None)
         app.state.run_event_store = make_run_event_store(run_events_config)

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

@@ -43,6 +43,9 @@ class DraftMetaResponse(BaseModel):
     furthest_step: int = 1
     created_at: datetime | str | None = None
     updated_at: datetime | str | None = None
+    # 最近一次后台作业的状态（Phase 6 历史下拉「进行中」loading 用）；无作业时为 None。
+    status: str | None = None
+    job_id: str | None = None
 
 
 class DraftListResponse(BaseModel):
@@ -137,6 +140,20 @@ async def list_drafts(request: Request) -> DraftListResponse:
     store = _get_store(request)
     user_id = _current_user_id(request)
     rows = await store.list_drafts(user_id)
+    # 富化：给每个草稿带上「最近一次后台作业」的状态 + jobId（Phase 6 历史下拉 loading 用）。
+    # job store 缺失（DB 未配置）或查询失败都不阻断列表，静默退化为无状态。
+    job_store = getattr(request.app.state, "roundtable_job_store", None)
+    if job_store is not None:
+        try:
+            status_map = await job_store.status_map_by_user(user_id)
+        except Exception:  # noqa: BLE001 — 富化失败不能拖垮列表
+            logger.warning("roundtable draft list: job status enrichment failed", exc_info=True)
+            status_map = {}
+        for r in rows:
+            info = status_map.get(r.get("id"))
+            if info:
+                r["status"] = info.get("status")
+                r["job_id"] = info.get("jobId")
     return DraftListResponse(drafts=[DraftMetaResponse(**r) for r in rows])
 
 

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260604_01_roundtable_jobs.py

@@ -0,0 +1,68 @@
+"""Create roundtable_jobs table.
+
+Backend background-orchestration jobs for the roundtable-planning Step 2「后台挂起」
+feature: one row per server-side run of the leader→dispatch→consensus→report
+pipeline. Big JSON columns (dispatch_chain / thread_ids / chain) are LONGTEXT on
+MySQL (can exceed the 64 KB TEXT cap) and plain TEXT elsewhere — matching
+``PortableLongText`` on ``RoundtableJobRow``. ``create_all`` adds the table for
+brand-new DBs; this revision creates it on DBs that predate it.
+
+Revision ID: 20260604_01
+Revises: 20260603_01
+Create Date: 2026-06-04
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+from sqlalchemy.dialects.mysql import LONGTEXT
+
+
+revision = "20260604_01"
+down_revision = "20260603_01"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    inspector = inspect(op.get_bind())
+    return table_name in inspector.get_table_names()
+
+
+def upgrade() -> None:
+    if _has_table("roundtable_jobs"):
+        return
+
+    long_text = LONGTEXT() if op.get_bind().dialect.name == "mysql" else sa.Text()
+
+    op.create_table(
+        "roundtable_jobs",
+        sa.Column("id", sa.String(64), primary_key=True),
+        sa.Column("draft_id", sa.String(64), nullable=True),
+        sa.Column("user_id", sa.String(64), nullable=True),
+        sa.Column("status", sa.String(32), nullable=False, server_default="queued"),
+        sa.Column("phase", sa.String(32), nullable=False, server_default="initializing"),
+        sa.Column("cycle", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("dispatch_chain", long_text, nullable=True),
+        sa.Column("active_agent_id", sa.String(128), nullable=True),
+        sa.Column("consensus_percentage", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("pending_clarification", sa.Text, nullable=True),
+        sa.Column("thread_ids", long_text, nullable=True),
+        sa.Column("coordinator_name", sa.String(128), nullable=True),
+        sa.Column("orchestration_mode", sa.String(16), nullable=False, server_default="recommend"),
+        sa.Column("chain", long_text, nullable=True),
+        sa.Column("error", sa.Text, nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index("ix_roundtable_jobs_draft_id", "roundtable_jobs", ["draft_id"])
+    op.create_index("ix_roundtable_jobs_user_id", "roundtable_jobs", ["user_id"])
+    op.create_index("ix_roundtable_jobs_status", "roundtable_jobs", ["status"])
+    op.create_index("ix_roundtable_jobs_updated_at", "roundtable_jobs", ["updated_at"])
+
+
+def downgrade() -> None:
+    if _has_table("roundtable_jobs"):
+        op.drop_table("roundtable_jobs")

offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py

@@ -27,6 +27,7 @@ from deerflow.persistence.roundtable_drafts.model import (
     RoundtableDraftRow,
     RoundtableRecommendHistoryRow,
 )
+from deerflow.persistence.roundtable_jobs.model import RoundtableJobRow
 from deerflow.persistence.run.model import RunRow
 from deerflow.persistence.scheduled_tasks.model import (
     ScheduledTaskDeliveryProfileRow,
@@ -53,6 +54,7 @@ __all__ = [
     "RecommendedQuestionRow",
     "RoundtableChainRow",
     "RoundtableDraftRow",
+    "RoundtableJobRow",
     "RoundtableRecommendHistoryRow",
     "RunEventRow",
     "RunRow",

offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_jobs/__init__.py

@@ -0,0 +1,47 @@
+"""Roundtable background-job persistence.
+
+Mirrors ``ai_writing_sessions``: exposes a store factory plus a process-level
+default-store singleton so the (harness-side) orchestration graph nodes can write
+progress without reaching into FastAPI ``request.app.state`` (harness → app imports
+are forbidden). app/gateway/deps.py registers the default store at startup.
+"""
+
+from __future__ import annotations
+
+from deerflow.persistence.roundtable_jobs.model import RoundtableJobRow
+from deerflow.persistence.roundtable_jobs.sql import RoundtableJobRepository
+
+
+def make_roundtable_job_store(session_factory) -> RoundtableJobRepository | None:
+    if session_factory is None:
+        return None
+    return RoundtableJobRepository(session_factory)
+
+
+# ── 进程级默认 repo 单例（跨 app/harness 边界）──────────────────────────────────
+#
+# Phase 2 的圆桌编排图注册成 LangGraph Server assistant 后，节点跑在 harness 进程里，
+# 拿不到 FastAPI 的 request.app.state；harness → app 又禁止 import。
+# 写入：app/gateway/deps.py 创建 session_factory 后 register_default_store(...)。
+# 读取：图节点里 get_default_store()，None 表示未配置 DB（跳过写入即可）。
+_default_store: RoundtableJobRepository | None = None
+
+
+def register_default_store(store: RoundtableJobRepository | None) -> None:
+    """登记进程级默认圆桌作业 store（幂等，可覆盖；传 None 显式清理，主要给测试）。"""
+    global _default_store
+    _default_store = store
+
+
+def get_default_store() -> RoundtableJobRepository | None:
+    """返回进程级默认圆桌作业 store；未配置时为 None。"""
+    return _default_store
+
+
+__all__ = [
+    "RoundtableJobRow",
+    "RoundtableJobRepository",
+    "make_roundtable_job_store",
+    "register_default_store",
+    "get_default_store",
+]

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

@@ -0,0 +1,67 @@
+"""ORM model for roundtable background-orchestration jobs.
+
+A "job" = one **backend** run of the roundtable orchestration pipeline kicked off
+from Step 2「后台挂起」: it drives the leader→dispatch→sub-agents loop to consensus
+and then Step 3 report drawing, entirely server-side, surviving page refresh/close.
+
+Bound 1:1 to a ``roundtable_drafts`` row (``draft_id``) so the history dropdown can
+show "this task is running" and the dispatch-chain modal can render live progress.
+
+Big JSON columns (``dispatch_chain`` / ``thread_ids`` / ``chain``) are stored as
+serialized strings in ``PortableLongText`` (MySQL ``LONGTEXT``) — same approach as
+``ai_writing_sessions`` / ``roundtable_drafts`` — to avoid MySQL's 64 KB ``TEXT`` cap.
+
+See ``frontend-web/docs/multi-agent-background-run-dev.md`` §2.5 for the field
+contract and §Phase 1 for the build steps.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import Integer, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+from deerflow.persistence.types import BeijingDateTime, PortableLongText
+
+
+class RoundtableJobRow(Base):
+    __tablename__ = "roundtable_jobs"
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    # The draft/session this job runs for. Indexed for "active job of draft" lookups.
+    draft_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    user_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    # Job status machine (§2.1): queued / running / awaiting_input / done / error / cancelled.
+    status: Mapped[str] = mapped_column(String(32), nullable=False, default="queued", index=True)
+    # Fine-grained phase (§2.2): initializing / leader_thinking / dispatching /
+    # consensus / report_drawing / report_done.
+    phase: Mapped[str] = mapped_column(String(32), nullable=False, default="initializing")
+    # Current orchestration cycle (0..MAX_CYCLES).
+    cycle: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    # Dispatch chain nodes (§2.3): JSON array of
+    # { agentId, name, avatarType, order, state, lastSpokeCycle? }.
+    dispatch_chain: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    # Currently-speaking seat agent_id (null = coordinator / none).
+    active_agent_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    consensus_percentage: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    # When status=awaiting_input: the coordinator's clarification question text.
+    pending_clarification: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # ThreadIdMap: JSON object { "<agent_id>": "<thread_id>", ... }.
+    thread_ids: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    coordinator_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    # "recommend"（总控自由派活）| "chain"（按业务链条顺序派活）.
+    orchestration_mode: Mapped[str] = mapped_column(String(16), nullable=False, default="recommend")
+    # chain 模式来源链条快照：JSON object { id, title }（recommend 模式为 null）.
+    chain: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    # Terminal error detail when status=error.
+    error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # 时间列必须用 BeijingDateTime（项目规范，与 ai_writing_sessions / roundtable_drafts 对齐）。
+    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
+    updated_at: Mapped[datetime] = mapped_column(
+        BeijingDateTime(),
+        nullable=False,
+        default=lambda: datetime.now(UTC),
+        onupdate=lambda: datetime.now(UTC),
+    )

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

@@ -0,0 +1,276 @@
+"""SQLAlchemy-backed roundtable background-job storage.
+
+JSON columns (``dispatch_chain`` / ``thread_ids`` / ``chain``) are stored as
+serialized strings in ``PortableLongText`` and round-tripped here with
+``json.dumps`` / ``json.loads`` — same approach as ``ai_writing_sessions`` /
+``roundtable_drafts``.
+
+User isolation mirrors ``ai_writing_sessions``: methods take ``user_id`` defaulting
+to ``AUTO`` (resolved from the run context) so backend graph nodes can write without
+threading a user id through, while gateway routes pass it explicitly.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.roundtable_jobs.model import RoundtableJobRow
+from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id
+
+_UNSET = object()
+
+# 终态：列表/清理判定用。
+TERMINAL_STATUSES = ("done", "error", "cancelled")
+
+
+def _loads(value: Any) -> Any:
+    if not isinstance(value, str):
+        return value
+    if not value:
+        return None
+    try:
+        return json.loads(value)
+    except Exception:
+        return None
+
+
+def _dumps(value: Any) -> str | None:
+    if value is None:
+        return None
+    return json.dumps(value, ensure_ascii=False)
+
+
+class RoundtableJobRepository:
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    @staticmethod
+    def _row_to_dict(row: RoundtableJobRow) -> dict[str, Any]:
+        d = row.to_dict()
+        for k in ("created_at", "updated_at"):
+            if isinstance(d.get(k), datetime):
+                d[k] = d[k].isoformat()
+        for k in ("dispatch_chain", "thread_ids", "chain"):
+            d[k] = _loads(d.get(k))
+        return d
+
+    # ── create ────────────────────────────────────────────────────────────
+
+    async def create(
+        self,
+        *,
+        id: str,
+        draft_id: str | None = None,
+        user_id: str | None | _AutoSentinel = AUTO,
+        orchestration_mode: str = "recommend",
+        coordinator_name: str | None = None,
+        thread_ids: dict[str, str] | None = None,
+        chain: dict[str, Any] | None = None,
+        dispatch_chain: list[dict[str, Any]] | None = None,
+        status: str = "queued",
+        phase: str = "initializing",
+    ) -> dict[str, Any]:
+        resolved_user_id = resolve_user_id(user_id, method_name="RoundtableJobRepository.create")
+        now = datetime.now(UTC)
+        row = RoundtableJobRow(
+            id=id,
+            draft_id=draft_id,
+            user_id=resolved_user_id,
+            status=status,
+            phase=phase,
+            cycle=0,
+            dispatch_chain=_dumps(dispatch_chain),
+            active_agent_id=None,
+            consensus_percentage=0,
+            pending_clarification=None,
+            thread_ids=_dumps(thread_ids),
+            coordinator_name=coordinator_name,
+            orchestration_mode=orchestration_mode,
+            chain=_dumps(chain),
+            error=None,
+            created_at=now,
+            updated_at=now,
+        )
+        async with self._sf() as session:
+            session.add(row)
+            await session.commit()
+            await session.refresh(row)
+            return self._row_to_dict(row)
+
+    # ── update progress ─────────────────────────────────────────────────────
+
+    async def update_progress(
+        self,
+        job_id: str,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+        status=_UNSET,
+        phase=_UNSET,
+        cycle=_UNSET,
+        dispatch_chain=_UNSET,
+        active_agent_id=_UNSET,
+        consensus_percentage=_UNSET,
+        pending_clarification=_UNSET,
+        thread_ids=_UNSET,
+        coordinator_name=_UNSET,
+        error=_UNSET,
+    ) -> dict[str, Any] | None:
+        resolved_user_id = resolve_user_id(user_id, method_name="RoundtableJobRepository.update_progress")
+        async with self._sf() as session:
+            row = await session.get(RoundtableJobRow, job_id)
+            if row is None:
+                return None
+            if resolved_user_id is not None and row.user_id != resolved_user_id:
+                return None
+            if status is not _UNSET:
+                row.status = status
+            if phase is not _UNSET:
+                row.phase = phase
+            if cycle is not _UNSET:
+                row.cycle = int(cycle)
+            if dispatch_chain is not _UNSET:
+                row.dispatch_chain = _dumps(dispatch_chain)
+            if active_agent_id is not _UNSET:
+                row.active_agent_id = active_agent_id
+            if consensus_percentage is not _UNSET:
+                row.consensus_percentage = int(consensus_percentage)
+            if pending_clarification is not _UNSET:
+                row.pending_clarification = pending_clarification
+            if thread_ids is not _UNSET:
+                row.thread_ids = _dumps(thread_ids)
+            if coordinator_name is not _UNSET:
+                row.coordinator_name = coordinator_name
+            if error is not _UNSET:
+                row.error = error
+            row.updated_at = datetime.now(UTC)
+            await session.commit()
+            await session.refresh(row)
+            return self._row_to_dict(row)
+
+    # ── read ────────────────────────────────────────────────────────────────
+
+    async def get(
+        self,
+        job_id: str,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> dict[str, Any] | None:
+        resolved_user_id = resolve_user_id(user_id, method_name="RoundtableJobRepository.get")
+        async with self._sf() as session:
+            row = await session.get(RoundtableJobRow, job_id)
+            if row is None:
+                return None
+            if resolved_user_id is not None and row.user_id != resolved_user_id:
+                return None
+            return self._row_to_dict(row)
+
+    async def get_by_draft(
+        self,
+        draft_id: str,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> dict[str, Any] | None:
+        """Return the most-recently-updated job for a draft (or None)."""
+        resolved_user_id = resolve_user_id(user_id, method_name="RoundtableJobRepository.get_by_draft")
+        stmt = select(RoundtableJobRow).where(RoundtableJobRow.draft_id == draft_id)
+        if resolved_user_id is not None:
+            stmt = stmt.where(RoundtableJobRow.user_id == resolved_user_id)
+        stmt = stmt.order_by(RoundtableJobRow.updated_at.desc()).limit(1)
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            row = result.scalars().first()
+            return self._row_to_dict(row) if row is not None else None
+
+    async def list_by_user(
+        self,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+        limit: int = 50,
+    ) -> list[dict[str, Any]]:
+        resolved_user_id = resolve_user_id(user_id, method_name="RoundtableJobRepository.list_by_user")
+        stmt = select(RoundtableJobRow)
+        if resolved_user_id is not None:
+            stmt = stmt.where(RoundtableJobRow.user_id == resolved_user_id)
+        stmt = stmt.order_by(RoundtableJobRow.updated_at.desc()).limit(limit)
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            return [self._row_to_dict(r) for r in result.scalars()]
+
+    async def status_map_by_user(
+        self,
+        user_id: str,
+        *,
+        limit: int = 200,
+    ) -> dict[str, dict[str, str]]:
+        """``{draft_id: {"jobId": ..., "status": ...}}`` for the latest job per draft.
+
+        Drives the history dropdown's per-draft status badge (Phase 6). Walks jobs
+        newest-first and keeps the first (latest) one seen for each draft_id.
+        """
+        stmt = (
+            select(RoundtableJobRow.id, RoundtableJobRow.draft_id, RoundtableJobRow.status)
+            .where(RoundtableJobRow.user_id == user_id)
+            .where(RoundtableJobRow.draft_id.is_not(None))
+            .order_by(RoundtableJobRow.updated_at.desc())
+            .limit(limit)
+        )
+        out: dict[str, dict[str, str]] = {}
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            for job_id, draft_id, status in result.all():
+                if draft_id is None or draft_id in out:
+                    continue
+                out[draft_id] = {"jobId": job_id, "status": status}
+        return out
+
+    # ── delete / cleanup ─────────────────────────────────────────────────────
+
+    async def delete(
+        self,
+        job_id: str,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> bool:
+        resolved_user_id = resolve_user_id(user_id, method_name="RoundtableJobRepository.delete")
+        async with self._sf() as session:
+            row = await session.get(RoundtableJobRow, job_id)
+            if row is None:
+                return False
+            if resolved_user_id is not None and row.user_id != resolved_user_id:
+                return False
+            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 个），供清理任务用。
+
+        only_finished=True 时只列终态（done/error/cancelled），避免清掉还在跑的作业。
+        """
+        stmt = select(RoundtableJobRow.id).where(RoundtableJobRow.updated_at < cutoff)
+        if only_finished:
+            stmt = stmt.where(RoundtableJobRow.status.in_(TERMINAL_STATUSES))
+        stmt = stmt.order_by(RoundtableJobRow.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, job_ids: list[str]) -> int:
+        if not job_ids:
+            return 0
+        stmt = delete(RoundtableJobRow).where(RoundtableJobRow.id.in_(job_ids))
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            await session.commit()
+            return int(result.rowcount or 0)