代码提交

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

@@ -46,6 +46,7 @@ from app.gateway.routers import (
     recommended_questions,
     roundtable_chains,
     roundtable_drafts,
+    roundtable_jobs,
     runs,
     scheduled_tasks,
     skills,
@@ -863,6 +864,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
     # Roundtable-planning business chains at /api/roundtable-chains
     app.include_router(roundtable_chains.router)
 
+    # Roundtable-planning background jobs at /api/roundtable-jobs
+    app.include_router(roundtable_jobs.router)
+
     # Suggestions API is mounted at /api/threads/{thread_id}/suggestions
     app.include_router(suggestions.router)
 

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

@@ -8,6 +8,7 @@ Initialization is handled directly in ``app.py`` via :class:`AsyncExitStack`.
 
 from __future__ import annotations
 
+import logging
 from collections.abc import AsyncGenerator, Callable
 from contextlib import AsyncExitStack, asynccontextmanager
 from typing import TYPE_CHECKING, TypeVar, cast
@@ -21,6 +22,8 @@ from deerflow.runtime import RunContext, RunManager, StreamBridge
 from deerflow.runtime.events.store.base import RunEventStore
 from deerflow.runtime.runs.store.base import RunStore
 
+logger = logging.getLogger(__name__)
+
 if TYPE_CHECKING:
     from app.gateway.auth.local_provider import LocalAuthProvider
     from app.gateway.auth.repositories.sqlite import SQLiteUserRepository
@@ -178,6 +181,42 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
         # （Phase 2 后台编排图写进度用）。
         _register_roundtable_job_default_store(app.state.roundtable_job_store)
 
+        # 圆桌后台作业执行器：把编排引擎作为后台任务跑。
+        # 网关由环境变量 ``ROUNDTABLE_JOB_GATEWAY`` 选择：
+        #   - "inprocess"（Phase 3b）：进程内驱动真模型（leader/seat/report）；
+        #   - 其它/缺省（Phase 3a）：模拟网关，产生真实进度但不调真模型，供前端联调。
+        # 默认 simulated —— 真网关需在能起完整后端栈 + 真模型的环境验证后再开。
+        if app.state.roundtable_job_store is not None:
+            import os
+
+            from app.gateway.roundtable_job_executor import RoundtableJobExecutor
+
+            gateway_mode = (os.getenv("ROUNDTABLE_JOB_GATEWAY") or "simulated").strip().lower()
+            if gateway_mode == "inprocess":
+                from app.gateway.roundtable_inprocess_gateway import InProcessRoundtableGateway
+
+                _app = app  # 闭包捕获 app，供网关访问 app.state（checkpointer/thread_store）
+                gateway_factory = lambda p: InProcessRoundtableGateway(  # noqa: E731
+                    _app, user_id=p.user_id, agents=p.agents, model=p.model
+                )
+                logger.info("[roundtable-jobs] gateway = inprocess (real models)")
+            else:
+                from deerflow.agents.roundtable_orchestrator import SimulatedRoundtableGateway
+
+                # delay=3s/步：让模拟后台任务有可观察的进行中时长（切页/重连仍在跑）。
+                gateway_factory = lambda p: SimulatedRoundtableGateway(  # noqa: E731
+                    p.agents, mode=p.mode, delay=3.0
+                )
+                logger.info("[roundtable-jobs] gateway = simulated")
+
+            app.state.roundtable_job_executor = RoundtableJobExecutor(
+                app.state.roundtable_job_store,
+                gateway_factory=gateway_factory,
+                draft_store=getattr(app.state, "roundtable_draft_store", None),
+            )
+        else:
+            app.state.roundtable_job_executor = None
+
         # 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/roundtable_inprocess_gateway.py

@@ -0,0 +1,272 @@
+"""圆桌真网关（Phase 3b）—— 进程内驱动真实 leader / seat / report agent run。
+
+实现 ``RoundtableGateway`` 协议（定义在 harness 的 roundtable_orchestrator.types），
+但**进程内**跑真模型：复用 ``services.start_run``（scheduled task 同款）驱动一轮 agent，
+跑完从 checkpointer 读最终 messages，用 ``multi_agent`` 的纯解析函数解析派活 / 澄清。
+不走 HTTP、不需要 Request（用 ``SimpleNamespace(app=app)`` 当虚拟 request，
+``set_current_user`` 由执行器设好）。
+
+本网关位于 app 层（依赖 app.state + multi_agent helpers），实现 harness 定义的协议——
+不违反 harness→app 边界（harness 永不 import app；app import harness 合法）。
+
+⚠️ 需在能起完整后端栈 + 真模型的环境验证（见 scripts/probe_roundtable_job.py）。
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import UTC, datetime
+from types import SimpleNamespace
+from typing import Any
+
+from langgraph.checkpoint.base import empty_checkpoint
+
+from deerflow.agents.roundtable_orchestrator.types import (
+    LeaderTurn,
+    ReportResult,
+    SeatRef,
+    SeatTurn,
+)
+from deerflow.config.paths import get_paths
+from deerflow.runtime.serialization import serialize_channel_values
+
+from app.gateway.routers._roundtable_seed import (
+    COORDINATOR_AGENT_ID,
+    REPORT_AGENT_ID,
+    ensure_roundtable_functional_agents,
+)
+
+logger = logging.getLogger(__name__)
+
+_ROUNDTABLE_THREAD_METADATA: dict[str, Any] = {"thread_type": "roundtable", "system": True}
+_REPORT_FILE = "方案总览.html"
+
+# 各角色的工具裁剪（镜像 multi_agent 的 _leader_run / _special_run）。
+_LEADER_EXCLUDED = ["web_search"]
+_LEADER_SKILL_STOP = ["agent_orchestration"]
+_SEAT_EXCLUDED = ["ask_clarification", "present_files", "view_image", "agent_orchestration"]
+_REPORT_EXCLUDED = [*_SEAT_EXCLUDED, "bash", "str_replace"]
+
+
+# ── 进程内底层操作 ───────────────────────────────────────────────────────────
+
+
+async def _create_thread(app) -> str:
+    """同进程建一个带圆桌 metadata 的 thread + 空 checkpoint（镜像 _create_thread_direct）。"""
+    checkpointer = app.state.checkpointer
+    thread_store = app.state.thread_store
+    thread_id = str(uuid.uuid4())
+    await thread_store.create(thread_id, assistant_id=None, metadata=dict(_ROUNDTABLE_THREAD_METADATA))
+    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
+    await checkpointer.aput(
+        config,
+        empty_checkpoint(),
+        {"step": -1, "source": "input", "writes": None, "parents": {}, "created_at": datetime.now(UTC).isoformat()},
+        {},
+    )
+    return thread_id
+
+
+async def _read_thread_messages(app, thread_id: str) -> list[dict[str, Any]]:
+    """读 thread 最新 checkpoint 的 messages（已 serialize 成 dict 列表）。"""
+    checkpointer = app.state.checkpointer
+    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
+    ct = await checkpointer.aget_tuple(config)
+    if ct is None:
+        return []
+    checkpoint = getattr(ct, "checkpoint", {}) or {}
+    values = serialize_channel_values(checkpoint.get("channel_values", {}) or {})
+    return values.get("messages") or []
+
+
+def _latest_ai_message(messages: list[dict[str, Any]]) -> dict[str, Any] | None:
+    for m in reversed(messages):
+        if not isinstance(m, dict):
+            continue
+        if (m.get("type") or m.get("role")) in {"ai", "assistant"}:
+            return m
+    return None
+
+
+def _message_text(message: dict[str, Any] | None) -> str:
+    if not message:
+        return ""
+    content = message.get("content")
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts = [p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text"]
+        return "\n".join(parts)
+    return ""
+
+
+async def _run_agent_turn(
+    app,
+    *,
+    thread_id: str,
+    agent_name: str,
+    message: str,
+    model: str | None,
+    excluded_tools: list[str],
+    skill_stop_names: list[str],
+    thinking_enabled: bool,
+    reasoning_effort: str,
+) -> list[dict[str, Any]]:
+    """进程内驱动 agent_name 在 thread_id 上跑一轮（发 message），阻塞到完成，返回最终 messages。
+
+    镜像 multi_agent ``_run_payload`` 的 body 形状：agent 通过 ``context.agent_name`` 选，
+    assistant_id 固定 lead_agent；关闭 memory recall / subagent / plan mode（临时研讨任务）。
+    """
+    from app.gateway.routers.thread_runs import RunCreateRequest
+    from app.gateway.services import start_run
+
+    body = RunCreateRequest(
+        assistant_id="lead_agent",
+        input={"messages": [{"type": "human", "content": [{"type": "text", "text": message}], "additional_kwargs": {}}]},
+        config={"recursion_limit": 1000},
+        context={
+            "agent_name": agent_name,
+            "model_name": model or "",
+            "mode": "pro",
+            "reasoning_effort": reasoning_effort,
+            "thinking_enabled": thinking_enabled,
+            "is_plan_mode": False,
+            "subagent_enabled": False,
+            "thread_id": thread_id,
+            "memory_recall_disabled": True,
+        },
+        excluded_tools=excluded_tools,
+        skill_stop_names=skill_stop_names,
+        on_disconnect="continue",
+        multitask_strategy="reject",
+    )
+    request = SimpleNamespace(app=app)
+    record = await start_run(body, thread_id, request)  # type: ignore[arg-type]
+    if getattr(record, "task", None) is not None:
+        await record.task  # 阻塞到这轮 agent 跑完，checkpoint 已落盘
+    return await _read_thread_messages(app, thread_id)
+
+
+def _read_report_html(app, thread_id: str, user_id: str) -> str | None:
+    """从 report thread 的 outputs/方案总览.html 读出报告 HTML。"""
+    try:
+        outputs_dir = get_paths().sandbox_outputs_dir(thread_id, user_id=user_id)
+        path = outputs_dir / _REPORT_FILE
+        if path.exists():
+            return path.read_text(encoding="utf-8")
+    except Exception:  # noqa: BLE001
+        logger.warning("read report html failed for thread %s", thread_id, exc_info=True)
+    return None
+
+
+def _build_seat_intro(agents: list[SeatRef]) -> str:
+    """镜像 init 注入 coordinator 的「可调度席位清单」。"""
+    lines = [f"- agent_name: {a.agent_id}（{a.name}）" for a in agents]
+    return (
+        "【圆桌系统初始化】本次研讨可调度的席位清单如下，请严格按 agent_name 通过 "
+        "agent_orchestration 技能派活，不要派给不在清单内的 agent：\n" + "\n".join(lines)
+    )
+
+
+# ── 网关实现 ─────────────────────────────────────────────────────────────────
+
+
+class InProcessRoundtableGateway:
+    """进程内真网关。executor 先调 ``init_threads`` 建好线程，再跑编排引擎。"""
+
+    def __init__(self, app, *, user_id: str, agents: list[SeatRef], model: str | None = None) -> None:
+        self._app = app
+        self._user_id = user_id
+        self._agents = list(agents)
+        self._model = model
+        self._seat_intro = ""
+        self._first_leader = True
+
+    async def init_threads(self) -> tuple[dict[str, str], str]:
+        """建 coordinator + 各席位 thread，返回 (thread_ids, coordinator_name)。
+
+        席位清单通过「首轮 leader 消息前缀」注入（coordinator thread 跨轮累积历史，
+        首轮带上即长期可见），无需单独写 checkpoint。
+        """
+        ensure_roundtable_functional_agents()
+        thread_ids: dict[str, str] = {}
+        for a in self._agents:
+            thread_ids[a.agent_id] = await _create_thread(self._app)
+        thread_ids[COORDINATOR_AGENT_ID] = await _create_thread(self._app)
+        self._seat_intro = _build_seat_intro(self._agents)
+        return thread_ids, COORDINATOR_AGENT_ID
+
+    async def run_leader(self, *, thread_ids, coordinator_name, message, model) -> LeaderTurn:
+        msg = message
+        if self._first_leader and self._seat_intro:
+            msg = self._seat_intro + "\n\n" + message
+            self._first_leader = False
+        messages = await _run_agent_turn(
+            self._app,
+            thread_id=thread_ids[coordinator_name],
+            agent_name=coordinator_name,
+            message=msg,
+            model=model or self._model,
+            excluded_tools=_LEADER_EXCLUDED,
+            skill_stop_names=_LEADER_SKILL_STOP,
+            thinking_enabled=False,
+            reasoning_effort="low",
+        )
+        ai = _latest_ai_message(messages)
+        final_text = _message_text(ai)
+        tool_calls = (ai.get("tool_calls") if ai else None) or []
+
+        # 澄清优先
+        clar = next((tc for tc in tool_calls if (tc.get("name") or "").lower() == "ask_clarification"), None)
+        if clar is not None:
+            cargs = clar.get("args") or {}
+            question = str(cargs.get("question") or "").strip() or "总控需要补充信息，请补充后继续。"
+            return LeaderTurn(final_text=final_text, clarification=question)
+
+        # 派活解析（复用 multi_agent 纯函数）
+        from app.gateway.routers.multi_agent import _extract_dispatch
+
+        dispatched: list[tuple[str, str]] = []
+        for tc in tool_calls:
+            try:
+                name, task = _extract_dispatch(tc)
+                if name and task:
+                    dispatched.append((name, task))
+            except Exception:  # noqa: BLE001
+                logger.warning("dispatch parse failed for tool_call", exc_info=True)
+        return LeaderTurn(final_text=final_text, dispatched=dispatched)
+
+    async def run_seat(self, *, thread_ids, agent_id, task, model) -> SeatTurn:
+        messages = await _run_agent_turn(
+            self._app,
+            thread_id=thread_ids[agent_id],
+            agent_name=agent_id,
+            message=task,
+            model=model or self._model,
+            excluded_tools=_SEAT_EXCLUDED,
+            skill_stop_names=[],
+            thinking_enabled=True,
+            reasoning_effort="medium",
+        )
+        return SeatTurn(final_text=_message_text(_latest_ai_message(messages)))
+
+    async def create_report_thread(self) -> str:
+        ensure_roundtable_functional_agents()
+        return await _create_thread(self._app)
+
+    async def run_report(self, *, report_thread_id, prompt, model) -> ReportResult:
+        messages = await _run_agent_turn(
+            self._app,
+            thread_id=report_thread_id,
+            agent_name=REPORT_AGENT_ID,
+            message=prompt,
+            model=model or self._model,
+            excluded_tools=_REPORT_EXCLUDED,
+            skill_stop_names=[],
+            thinking_enabled=True,
+            reasoning_effort="medium",
+        )
+        summary = _message_text(_latest_ai_message(messages))
+        html = _read_report_html(self._app, report_thread_id, self._user_id) or summary
+        return ReportResult(html=html, summary=summary, model=model or self._model)

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

@@ -0,0 +1,152 @@
+"""圆桌后台作业执行器（Phase 3a）。
+
+把 Phase 2 的编排引擎（``run_orchestration``）作为**后台 asyncio 任务**跑起来，
+survive 发起请求的生命周期。每个任务：
+
+  1. ``set_current_user``：用作业的 user_id 设进程内用户上下文，让引擎里的进度落库
+     （``JobProgressWriter`` → ``RoundtableJobRepository.update_progress(user_id=AUTO)``）
+     正确归属（与 scheduled task 后台执行同一套做法）；
+  2. 用 ``gateway_factory`` 构造一个 ``RoundtableGateway``（Phase 3a 默认模拟网关，
+     Phase 3b 换成进程内真网关），跑完整编排（循环 → 共识 → 报告）。
+
+任务句柄登记在 ``_tasks`` 里，供 ``cancel`` 用；完成后自动摘除。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from dataclasses import dataclass, field
+from typing import Callable
+
+from deerflow.agents.roundtable_orchestrator import (
+    JobProgressWriter,
+    SeatRef,
+    run_orchestration,
+)
+from deerflow.runtime.user_context import reset_current_user, set_current_user
+
+logger = logging.getLogger(__name__)
+
+
+class _JobUser:
+    """最小 CurrentUser：满足 user_context 的 ``.id`` 结构协议。"""
+
+    def __init__(self, user_id: str) -> None:
+        self.id = user_id
+
+
+@dataclass
+class StartParams:
+    job_id: str
+    user_id: str
+    agents: list[SeatRef]
+    seed_message: str
+    model: str | None = None
+    mode: str = "recommend"
+    intent_text: str = ""
+    thread_ids: dict[str, str] | None = None
+    coordinator_name: str = "roundtable-coordinator"
+    # 关联草稿：作业完成后把 step3 报告写回该草稿，使「查看报告」可用。
+    draft_id: str | None = None
+
+
+# 网关工厂：给定一次启动参数，返回一个 RoundtableGateway 实例。
+GatewayFactory = Callable[[StartParams], object]
+
+
+class RoundtableJobExecutor:
+    def __init__(self, store, *, gateway_factory: GatewayFactory, draft_store=None) -> None:
+        self._store = store
+        self._gateway_factory = gateway_factory
+        # 草稿仓储（可选）：作业完成后把 step3 写回草稿，供前端「查看报告」。
+        self._draft_store = draft_store
+        self._tasks: dict[str, asyncio.Task] = {}
+
+    def start_job(self, params: StartParams) -> bool:
+        """启动一个后台编排任务。同一 job 已在跑则忽略（幂等）。返回是否新启动。"""
+        if params.job_id in self._tasks and not self._tasks[params.job_id].done():
+            return False
+        task = asyncio.create_task(self._run(params))
+        self._tasks[params.job_id] = task
+        task.add_done_callback(lambda t, jid=params.job_id: self._tasks.pop(jid, None))
+        return True
+
+    async def _run(self, params: StartParams) -> None:
+        token = set_current_user(_JobUser(params.user_id))
+        try:
+            gateway = self._gateway_factory(params)
+            thread_ids = params.thread_ids or {}
+            coordinator_name = params.coordinator_name
+            # 真网关需要先进程内建好 leader/seat 线程；模拟网关无 init_threads。
+            # 仅首启建线程（resume 时 params.thread_ids 已带，复用、不重建）。
+            init = getattr(gateway, "init_threads", None)
+            if init is not None and not thread_ids:
+                thread_ids, coordinator_name = await init()
+                try:
+                    await self._store.update_progress(
+                        params.job_id, thread_ids=thread_ids, coordinator_name=coordinator_name
+                    )
+                except Exception:  # noqa: BLE001
+                    logger.warning("persist thread_ids for job %s failed", params.job_id, exc_info=True)
+
+            progress = JobProgressWriter(self._store, job_id=params.job_id, agents=params.agents)
+            result = await run_orchestration(
+                gateway=gateway,
+                progress=progress,
+                agents=params.agents,
+                thread_ids=thread_ids,
+                coordinator_name=coordinator_name,
+                model=params.model,
+                seed_message=params.seed_message,
+                mode=params.mode,
+                intent_text=params.intent_text,
+            )
+            # 完成且有报告：写回关联草稿的 step3 + 推进 furthest_step=3，
+            # 这样用户从历史「查看报告」加载草稿即可看到最终方案（沿用既有 hydrateStep3 路径）。
+            if (
+                result.status == "done"
+                and result.step3
+                and params.draft_id
+                and self._draft_store is not None
+            ):
+                try:
+                    await self._draft_store.update_draft(
+                        params.draft_id,
+                        params.user_id,
+                        step3=result.step3,
+                        furthest_step=3,
+                    )
+                except Exception:  # noqa: BLE001 — 写报告失败不影响作业本身已 done
+                    logger.warning("persist step3 to draft %s failed", params.draft_id, exc_info=True)
+        except asyncio.CancelledError:
+            # 干净停止即可，**不在被取消的任务里写 DB**：取消态下二次 await 会再抛
+            # CancelledError、并可能打断进行中的 sqlite 连接。``cancelled`` 终态由
+            # 路由的 cancel 端点（executor.cancel + store.update_progress）权威写入。
+            logger.info("roundtable job %s cancelled", params.job_id)
+            raise
+        except Exception:  # noqa: BLE001 — 顶层兜底：run_orchestration 内部已尽量标 error
+            logger.exception("roundtable job %s crashed", params.job_id)
+        finally:
+            reset_current_user(token)
+
+    def cancel(self, job_id: str) -> bool:
+        """请求取消一个在跑的作业。返回是否命中在跑任务。"""
+        task = self._tasks.get(job_id)
+        if task is None or task.done():
+            return False
+        task.cancel()
+        return True
+
+    def is_running(self, job_id: str) -> bool:
+        task = self._tasks.get(job_id)
+        return task is not None and not task.done()
+
+    async def await_job(self, job_id: str) -> None:
+        """等待某作业任务结束（测试用；生产不调用）。"""
+        task = self._tasks.get(job_id)
+        if task is not None:
+            try:
+                await task
+            except (Exception, asyncio.CancelledError):  # noqa: BLE001 — 取消是 BaseException
+                pass

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

@@ -0,0 +1,341 @@
+"""圆桌后台作业控制 API（Phase 3a）。
+
+把 Step 2「后台挂起」的任务交给后端编排引擎跑，并暴露进度订阅 / 续跑 / 取消。
+
+Routes（prefix ``/api/roundtable-jobs``）::
+    POST   ""                 启动一个后台作业（创建 job 行 + 拉起执行器任务）
+    GET    ""?draft_id=       查某草稿最近一个作业
+    GET    "/{job_id}"        作业快照
+    GET    "/{job_id}/stream" SSE 进度流（轮询 job 表，多 worker 安全）
+    POST   "/{job_id}/resume" 回答澄清后续跑
+    POST   "/{job_id}/cancel" 取消
+
+进度数据源是 ``roundtable_jobs`` 表（Phase 1），SSE 走 DB 轮询而非内存 pub/sub —— 这样
+刷新 / 换 worker / 重连都能继续收到进度，规避 ai_writing 踩过的「per-worker 内存 dict
+丢会话」坑。响应统一 camelCase（对齐前端 §2.5）。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import time
+from datetime import UTC, datetime, timedelta
+from typing import Any
+from uuid import uuid4
+
+from fastapi import APIRouter, HTTPException, Request
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel, Field
+
+from deerflow.agents.roundtable_orchestrator import SeatRef
+from deerflow.runtime.user_context import get_effective_user_id
+
+from app.gateway.roundtable_job_executor import StartParams
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/roundtable-jobs", tags=["roundtable-jobs"])
+
+_TERMINAL = {"done", "error", "cancelled"}
+# SSE 轮询间隔与最长存活（对齐 ai_writing：长连接最多挂 30 分钟）。
+_POLL_INTERVAL_S = 0.7
+_MAX_STREAM_TICKS = int(30 * 60 / _POLL_INTERVAL_S)
+
+# 续跑澄清时拼给 leader 的 prompt 前缀。
+_RESUME_PREFIX = "用户已回复你刚才的澄清问题，回答如下，请据此继续派活或给出结论：\n"
+
+
+# ── schemas ──────────────────────────────────────────────────────────────────
+
+
+class JobAgent(BaseModel):
+    agentId: str
+    name: str = ""
+
+
+class StartJobRequest(BaseModel):
+    draftId: str | None = Field(default=None, max_length=64)
+    intent: str = ""
+    agents: list[JobAgent] = Field(default_factory=list)
+    model: str | None = None
+    orchestrationMode: str = "recommend"
+    chain: dict[str, Any] | None = None
+    # 第一轮 leader prompt；缺省用 intent。
+    seedMessage: str | None = None
+    # 预建好的 thread 映射（Phase 3b 真网关用；模拟网关忽略）。
+    threadIds: dict[str, str] | None = None
+    coordinatorName: str = "roundtable-coordinator"
+
+
+class ResumeJobRequest(BaseModel):
+    answer: str = ""
+
+
+class CleanupRequest(BaseModel):
+    retention_days: int = Field(default=7, ge=0, description="保留多少天内的作业；0 = 不保留")
+    only_finished: bool = Field(default=True, description="仅清终态（done/error/cancelled），保护在跑作业")
+    batch_size: int = Field(default=500, ge=1, le=5000)
+
+
+class CleanupResponse(BaseModel):
+    deleted: int
+    elapsed_seconds: float
+    params: dict[str, Any]
+
+
+class RoundtableJobResponse(BaseModel):
+    id: str
+    draftId: str | None = None
+    status: str = "queued"
+    phase: str = "initializing"
+    cycle: int = 0
+    dispatchChain: list[dict[str, Any]] = Field(default_factory=list)
+    activeAgentId: str | None = None
+    consensusPercentage: int = 0
+    pendingClarification: str | None = None
+    orchestrationMode: str = "recommend"
+    chain: dict[str, Any] | None = None
+    error: str | None = None
+    createdAt: datetime | str | None = None
+    updatedAt: datetime | str | None = None
+
+
+# ── helpers ──────────────────────────────────────────────────────────────────
+
+
+def _current_user_id(request: Request) -> str:
+    user = getattr(request.state, "user", None)
+    if user is not None:
+        return str(user.id)
+    return get_effective_user_id()
+
+
+def _get_store(request: Request):
+    store = getattr(request.app.state, "roundtable_job_store", None)
+    if store is None:
+        raise HTTPException(status_code=503, detail="Roundtable job store not available")
+    return store
+
+
+def _get_executor(request: Request):
+    executor = getattr(request.app.state, "roundtable_job_executor", None)
+    if executor is None:
+        raise HTTPException(status_code=503, detail="Roundtable job executor not available")
+    return executor
+
+
+def _pending_nodes(agents: list[JobAgent]) -> list[dict[str, Any]]:
+    """启动即落一份「全 pending」的派活链路，使 GET / 首帧 SSE 立刻能渲染流程图。"""
+    return [
+        {"agentId": a.agentId, "name": a.name or a.agentId, "order": i + 1, "state": "pending"}
+        for i, a in enumerate(agents)
+    ]
+
+
+def _to_response(row: dict[str, Any]) -> RoundtableJobResponse:
+    return RoundtableJobResponse(
+        id=row["id"],
+        draftId=row.get("draft_id"),
+        status=row.get("status") or "queued",
+        phase=row.get("phase") or "initializing",
+        cycle=int(row.get("cycle") or 0),
+        dispatchChain=row.get("dispatch_chain") or [],
+        activeAgentId=row.get("active_agent_id"),
+        consensusPercentage=int(row.get("consensus_percentage") or 0),
+        pendingClarification=row.get("pending_clarification"),
+        orchestrationMode=row.get("orchestration_mode") or "recommend",
+        chain=row.get("chain"),
+        error=row.get("error"),
+        createdAt=row.get("created_at"),
+        updatedAt=row.get("updated_at"),
+    )
+
+
+def _seat_refs_from_nodes(nodes: list[dict[str, Any]] | None) -> list[SeatRef]:
+    out: list[SeatRef] = []
+    for n in nodes or []:
+        aid = str(n.get("agentId") or "")
+        if aid:
+            out.append(SeatRef(aid, str(n.get("name") or aid)))
+    return out
+
+
+# ── routes ───────────────────────────────────────────────────────────────────
+
+
+@router.post("", response_model=RoundtableJobResponse, status_code=201)
+async def start_job(request: Request, body: StartJobRequest) -> RoundtableJobResponse:
+    store = _get_store(request)
+    executor = _get_executor(request)
+    user_id = _current_user_id(request)
+
+    if not body.agents:
+        raise HTTPException(status_code=400, detail="agents is required")
+
+    agents = [SeatRef(a.agentId, a.name or a.agentId) for a in body.agents]
+    job_id = uuid4().hex
+    row = await store.create(
+        id=job_id,
+        draft_id=body.draftId,
+        user_id=user_id,
+        orchestration_mode=body.orchestrationMode,
+        coordinator_name=body.coordinatorName,
+        thread_ids=body.threadIds,
+        chain=body.chain,
+        dispatch_chain=_pending_nodes(body.agents),
+        status="queued",
+    )
+
+    executor.start_job(
+        StartParams(
+            job_id=job_id,
+            user_id=user_id,
+            agents=agents,
+            seed_message=(body.seedMessage or body.intent or "").strip(),
+            model=body.model,
+            mode=body.orchestrationMode,
+            intent_text=body.intent,
+            thread_ids=body.threadIds,
+            coordinator_name=body.coordinatorName,
+            draft_id=body.draftId,
+        )
+    )
+    return _to_response(row)
+
+
+@router.get("", response_model=RoundtableJobResponse | None)
+async def get_job_by_draft(request: Request, draft_id: str) -> RoundtableJobResponse | None:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+    row = await store.get_by_draft(draft_id, user_id=user_id)
+    return _to_response(row) if row is not None else None
+
+
+@router.get("/{job_id}", response_model=RoundtableJobResponse)
+async def get_job(request: Request, job_id: str) -> RoundtableJobResponse:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+    row = await store.get(job_id, user_id=user_id)
+    if row is None:
+        raise HTTPException(status_code=404, detail="Job not found")
+    return _to_response(row)
+
+
+@router.get("/{job_id}/stream")
+async def stream_job(request: Request, job_id: str) -> StreamingResponse:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+
+    def _frame(payload: dict[str, Any]) -> str:
+        return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
+
+    async def gen():
+        for _ in range(_MAX_STREAM_TICKS):
+            if await request.is_disconnected():
+                return
+            row = await store.get(job_id, user_id=user_id)
+            if row is None:
+                yield _frame({"event": "error", "detail": "Job not found"})
+                return
+            yield _frame({"event": "progress", **_to_response(row).model_dump(mode="json")})
+            if (row.get("status") or "") in _TERMINAL:
+                return
+            await asyncio.sleep(_POLL_INTERVAL_S)
+
+    return StreamingResponse(
+        gen(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"},
+    )
+
+
+@router.post("/{job_id}/resume", response_model=RoundtableJobResponse)
+async def resume_job(request: Request, job_id: str, body: ResumeJobRequest) -> RoundtableJobResponse:
+    store = _get_store(request)
+    executor = _get_executor(request)
+    user_id = _current_user_id(request)
+
+    row = await store.get(job_id, user_id=user_id)
+    if row is None:
+        raise HTTPException(status_code=404, detail="Job not found")
+    if (row.get("status") or "") != "awaiting_input":
+        raise HTTPException(status_code=409, detail="Job is not awaiting input")
+
+    answer = body.answer.strip()
+    if not answer:
+        raise HTTPException(status_code=400, detail="answer is required")
+
+    updated = await store.update_progress(
+        job_id, user_id=user_id, status="running", pending_clarification=None
+    )
+
+    executor.start_job(
+        StartParams(
+            job_id=job_id,
+            user_id=user_id,
+            agents=_seat_refs_from_nodes(row.get("dispatch_chain")),
+            seed_message=_RESUME_PREFIX + answer,
+            model=None,
+            mode=row.get("orchestration_mode") or "recommend",
+            intent_text=answer,
+            thread_ids=row.get("thread_ids"),
+            coordinator_name=row.get("coordinator_name") or "roundtable-coordinator",
+            draft_id=row.get("draft_id"),
+        )
+    )
+    return _to_response(updated or row)
+
+
+async def _require_admin(request: Request) -> None:
+    """放行 admin / auth-disabled，其余 403（与 ai_writing / scheduled_tasks 同套语义）。"""
+    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="圆桌作业清理管理接口仅限管理员")
+
+
+@router.post("/admin/cleanup", response_model=CleanupResponse)
+async def admin_cleanup(request: Request, body: CleanupRequest) -> CleanupResponse:
+    """清理过期作业行（活跃作业受 only_finished 保护，永不误删）。
+
+    分批 ``list_older_than`` → ``delete_by_ids``，避免大事务。``retention_days=0``
+    一刷清空所有终态作业（调试用）。
+    """
+    await _require_admin(request)
+    store = _get_store(request)
+    cutoff = datetime.now(UTC) - timedelta(days=body.retention_days)
+
+    t0 = time.perf_counter()
+    deleted = 0
+    while True:
+        ids = await store.list_older_than(cutoff, only_finished=body.only_finished, limit=body.batch_size)
+        if not ids:
+            break
+        deleted += await store.delete_by_ids(ids)
+        if len(ids) < body.batch_size:
+            break
+
+    return CleanupResponse(
+        deleted=deleted,
+        elapsed_seconds=round(time.perf_counter() - t0, 3),
+        params={"retention_days": body.retention_days, "only_finished": body.only_finished, "batch_size": body.batch_size},
+    )
+
+
+@router.post("/{job_id}/cancel", response_model=RoundtableJobResponse)
+async def cancel_job(request: Request, job_id: str) -> RoundtableJobResponse:
+    store = _get_store(request)
+    executor = _get_executor(request)
+    user_id = _current_user_id(request)
+
+    row = await store.get(job_id, user_id=user_id)
+    if row is None:
+        raise HTTPException(status_code=404, detail="Job not found")
+
+    executor.cancel(job_id)
+    updated = await store.update_progress(job_id, user_id=user_id, status="cancelled")
+    return _to_response(updated or row)

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

@@ -0,0 +1,48 @@
+"""圆桌总控编排引擎（Phase 2）。
+
+把前端 useStep2Orchestration 的编排循环移植到后端，作为可后台跑、可单测的纯引擎。
+Phase 3 会提供具体的 ``RoundtableGateway`` 实现（复用 multi_agent.py 内部函数）+
+执行器 / LangGraph 注册，把引擎接到真实的 LLM / 线程上。
+
+公开入口：
+  - ``run_orchestration`` —— 跑一次完整编排（循环 → 共识 → 报告绘制）。
+  - ``JobProgressWriter`` —— 进度落库 + 可选 SSE 外发。
+  - 协议 / 数据类（``RoundtableGateway`` / ``LeaderTurn`` / ``SeatTurn`` /
+    ``ReportResult`` / ``SeatRef``）。
+"""
+
+from __future__ import annotations
+
+from .engine import OrchestrationResult, build_report_prompt, run_orchestration
+from .progress import JobProgressWriter
+from .simulated_gateway import SimulatedRoundtableGateway
+from .types import (
+    CHAIN_CLOSEOUT_PROMPT,
+    CONTINUE_PROMPT,
+    MAX_CYCLES,
+    LeaderTurn,
+    ReportResult,
+    RoundtableGateway,
+    SeatRef,
+    SeatTurn,
+    build_chain_leader_task,
+    pick_chain_task,
+)
+
+__all__ = [
+    "run_orchestration",
+    "OrchestrationResult",
+    "build_report_prompt",
+    "JobProgressWriter",
+    "SimulatedRoundtableGateway",
+    "RoundtableGateway",
+    "LeaderTurn",
+    "SeatTurn",
+    "ReportResult",
+    "SeatRef",
+    "MAX_CYCLES",
+    "CONTINUE_PROMPT",
+    "CHAIN_CLOSEOUT_PROMPT",
+    "build_chain_leader_task",
+    "pick_chain_task",
+]

offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py

@@ -0,0 +1,299 @@
+"""圆桌总控编排引擎 —— 后台跑的编排循环（移植自前端 useStep2Orchestration.ts）。
+
+纯逻辑：通过 ``RoundtableGateway`` 协议驱动「单轮 leader / seat / report」，通过
+``JobProgressWriter`` 落库进度。两种模式：
+
+  - recommend（``runOrchestration``）：总控自由派活循环，leader → 顺序派 special →
+    leader …，直到 ``dispatched==[]``（共识）或达到 ``MAX_CYCLES``；
+  - chain（``runChainOrchestration``）：总控**按席位顺序**逐位派活，走完整链后由
+    leader 收口产出最终结论。
+
+共识达成（或 max cycles 强制收口）后进入 Step3 报告绘制，写出 ``方案总览.html``，
+返回 ``DraftStep3Snapshot`` 形状的 step3 快照。命中澄清则返回 ``awaiting_input``
+（thread 已 checkpoint，resume 时由调用方用用户答案再发起一轮即可续跑）。
+
+**不并行 special**（与前端同一不变量）：顺序 await 每个席位，避免 token 级抖动 +
+保证派活链路游标严格推进。
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+
+from .progress import JobProgressWriter
+from .types import (
+    CHAIN_CLOSEOUT_PROMPT,
+    CONTINUE_PROMPT,
+    MAX_CYCLES,
+    ReportResult,
+    RoundtableGateway,
+    SeatRef,
+    build_chain_leader_task,
+    pick_chain_task,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class OrchestrationResult:
+    """引擎一次运行的终态。"""
+
+    status: str  # "done" | "awaiting_input" | "error"
+    consensus: bool = False
+    step3: dict | None = None
+    pending_clarification: str | None = None
+    error: str | None = None
+    last_leader_content: str = ""
+
+
+@dataclass
+class _LoopOutcome:
+    """内部：编排循环（未进报告阶段）的结果。"""
+
+    kind: str  # "consensus" | "clarification" | "max_cycles"
+    question: str | None = None
+    last_leader_content: str = ""
+    # {agent_id: 最近一次交付正文}
+    deliveries: dict[str, str] = field(default_factory=dict)
+
+
+async def run_orchestration(
+    *,
+    gateway: RoundtableGateway,
+    progress: JobProgressWriter,
+    agents: list[SeatRef],
+    thread_ids: dict[str, str],
+    coordinator_name: str,
+    model: str | None,
+    seed_message: str,
+    mode: str = "recommend",
+    intent_text: str = "",
+) -> OrchestrationResult:
+    """运行一次完整编排（循环 → 共识 → 报告绘制）。
+
+    Args:
+      gateway: 单轮执行抽象出口。
+      progress: 进度写入器（落库 + 可选 SSE 外发）。
+      agents: 有序席位（recommend=选择顺序 / chain=链条顺序）。
+      thread_ids: {agent_id / coordinator_name: thread_id}，预先建好。
+      coordinator_name: 总控 agent_id（如 "roundtable-coordinator"）。
+      model: 选定模型（None/"" 走后端默认）。
+      seed_message: 第一轮 leader 的 prompt（启动=初始意图；resume=用户答案续轮）。
+      mode: "recommend" | "chain"。
+      intent_text: 任务意图，用于报告 prompt（缺省退回 seed_message）。
+    """
+    try:
+        await progress.begin()
+        if mode == "chain":
+            outcome = await _run_chain(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message)
+        else:
+            outcome = await _run_recommend(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message)
+
+        if outcome.kind == "clarification":
+            question = outcome.question or "总控需要补充信息，请补充后继续。"
+            await progress.awaiting_input(question)
+            return OrchestrationResult(
+                status="awaiting_input",
+                pending_clarification=question,
+                last_leader_content=outcome.last_leader_content,
+            )
+
+        # consensus 或 max_cycles：均收口 → 报告绘制
+        await progress.consensus(100)
+        step3 = await _run_report(
+            gateway,
+            progress,
+            agents=agents,
+            model=model,
+            intent_text=intent_text or seed_message,
+            consensus_text=outcome.last_leader_content,
+            deliveries=outcome.deliveries,
+        )
+        await progress.done()
+        return OrchestrationResult(
+            status="done",
+            consensus=True,
+            step3=step3,
+            last_leader_content=outcome.last_leader_content,
+        )
+    except Exception as exc:  # noqa: BLE001 — 顶层兜底：标记 error 并返回
+        logger.exception("roundtable orchestration failed")
+        await progress.error(str(exc))
+        return OrchestrationResult(status="error", error=str(exc))
+
+
+# ── recommend 模式 ───────────────────────────────────────────────────────────
+
+
+async def _run_recommend(
+    gateway: RoundtableGateway,
+    progress: JobProgressWriter,
+    agents: list[SeatRef],
+    thread_ids: dict[str, str],
+    coordinator_name: str,
+    model: str | None,
+    seed_message: str,
+) -> _LoopOutcome:
+    deliveries: dict[str, str] = {}
+    last_leader = ""
+    message = seed_message
+
+    for cycle in range(1, MAX_CYCLES + 1):
+        await progress.leader_thinking(cycle)
+        turn = await gateway.run_leader(
+            thread_ids=thread_ids,
+            coordinator_name=coordinator_name,
+            message=message,
+            model=model,
+        )
+        if turn.final_text:
+            last_leader = turn.final_text
+
+        if turn.is_clarification:
+            return _LoopOutcome("clarification", question=turn.clarification, last_leader_content=last_leader, deliveries=deliveries)
+        if turn.is_consensus:
+            return _LoopOutcome("consensus", last_leader_content=last_leader, deliveries=deliveries)
+
+        # **顺序**派活 —— 不要并行
+        await progress.dispatching(cycle)
+        for agent_id, task in turn.dispatched:
+            await progress.seat_active(agent_id)
+            seat = await gateway.run_seat(thread_ids=thread_ids, agent_id=agent_id, task=task, model=model)
+            deliveries[agent_id] = seat.final_text
+            await progress.seat_done(agent_id, cycle)
+
+        await progress.bump_consensus()
+        message = CONTINUE_PROMPT
+
+    # 达到最大轮次：强制收口（对齐前端 toast「已达到最大研讨轮次」）
+    logger.info("roundtable recommend loop hit MAX_CYCLES=%d, forcing consensus", MAX_CYCLES)
+    return _LoopOutcome("max_cycles", last_leader_content=last_leader, deliveries=deliveries)
+
+
+# ── chain 模式 ───────────────────────────────────────────────────────────────
+
+
+async def _run_chain(
+    gateway: RoundtableGateway,
+    progress: JobProgressWriter,
+    agents: list[SeatRef],
+    thread_ids: dict[str, str],
+    coordinator_name: str,
+    model: str | None,
+    seed_message: str,
+) -> _LoopOutcome:
+    deliveries: dict[str, str] = {}
+    last_leader = ""
+    total = len(agents)
+
+    for i, seat in enumerate(agents):
+        # (a) 总控按链条顺序只给当前席位派活
+        await progress.leader_thinking(i + 1)
+        leader_task = build_chain_leader_task(
+            seed_message=seed_message,
+            position=i,
+            total=total,
+            seat_agent_id=seat.agent_id,
+            seat_name=seat.name,
+            is_first=(i == 0),
+        )
+        turn = await gateway.run_leader(
+            thread_ids=thread_ids,
+            coordinator_name=coordinator_name,
+            message=leader_task,
+            model=model,
+        )
+        if turn.final_text:
+            last_leader = turn.final_text
+        if turn.is_clarification:
+            return _LoopOutcome("clarification", question=turn.clarification, last_leader_content=last_leader, deliveries=deliveries)
+
+        task = pick_chain_task(turn.dispatched, turn.final_text, seat.agent_id, seat.name)
+
+        # (b) 当前席位交付（顺序，不并行）
+        await progress.seat_active(seat.agent_id)
+        seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=task, model=model)
+        deliveries[seat.agent_id] = seat_turn.final_text
+        await progress.seat_done(seat.agent_id, i + 1)
+
+        if total > 0:
+            await progress.set_consensus_pct(round((i + 1) / total * 90))
+
+    # (c) 总控收口：综合全链交付，产出最终结论
+    if total > 0:
+        await progress.leader_thinking(total + 1)
+        closeout = await gateway.run_leader(
+            thread_ids=thread_ids,
+            coordinator_name=coordinator_name,
+            message=CHAIN_CLOSEOUT_PROMPT,
+            model=model,
+        )
+        if closeout.final_text:
+            last_leader = closeout.final_text
+
+    return _LoopOutcome("consensus", last_leader_content=last_leader, deliveries=deliveries)
+
+
+# ── 报告阶段 ─────────────────────────────────────────────────────────────────
+
+
+async def _run_report(
+    gateway: RoundtableGateway,
+    progress: JobProgressWriter,
+    *,
+    agents: list[SeatRef],
+    model: str | None,
+    intent_text: str,
+    consensus_text: str,
+    deliveries: dict[str, str],
+) -> dict:
+    await progress.report_drawing()
+    report_thread_id = await gateway.create_report_thread()
+    prompt = build_report_prompt(intent_text, agents, deliveries, consensus_text)
+    report: ReportResult = await gateway.run_report(report_thread_id=report_thread_id, prompt=prompt, model=model)
+    return {
+        "html": report.html,
+        "generatedAt": report.generated_at or datetime.now(UTC).isoformat(),
+        "model": report.model or (model or ""),
+        "summary": report.summary,
+    }
+
+
+def build_report_prompt(
+    intent_text: str,
+    agents: list[SeatRef],
+    deliveries: dict[str, str],
+    consensus_text: str,
+) -> str:
+    """拼装 Step3 报告绘制 prompt：意图 + 各席位交付 + 总控共识。
+
+    席位交付按 ``agents`` 顺序列出（缺交付的席位跳过）。报告智能体据此用 write_file
+    写出自包含的 ``outputs/方案总览.html``（具体绘制规则在 roundtable-report 的 SOUL）。
+    """
+    lines: list[str] = [
+        "【圆桌最终成果汇总 —— 请据此绘制方案总览 HTML 看板】",
+        "",
+        "## 任务意图",
+        (intent_text or "").strip() or "（未提供）",
+        "",
+        "## 各席位交付",
+    ]
+    name_by_id = {a.agent_id: a.name for a in agents}
+    has_delivery = False
+    for a in agents:
+        text = (deliveries.get(a.agent_id) or "").strip()
+        if not text:
+            continue
+        has_delivery = True
+        lines.append(f"### {name_by_id.get(a.agent_id, a.agent_id)}")
+        lines.append(text)
+        lines.append("")
+    if not has_delivery:
+        lines.append("（各席位暂无可用交付，请基于任务意图与总控结论绘制。）")
+        lines.append("")
+    lines.append("## 总控最终结论")
+    lines.append((consensus_text or "").strip() or "（未提供）")
+    return "\n".join(lines)

offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/progress.py

@@ -0,0 +1,142 @@
+"""编排进度写入器 —— 把引擎的状态流转持久化进 ``roundtable_jobs`` 表。
+
+引擎只调本类的语义方法（leader_thinking / seat_active / seat_done / consensus / …），
+由它维护「派活链路各节点状态」并落库（best-effort，失败不打断编排）。落库的
+``dispatch_chain`` 即 Phase 7 流程图弹窗 / Phase 6 历史下拉的数据源。
+
+进度同时可通过可选的 ``event_sink`` 回调外发（Phase 3 的 SSE 流用），引擎逻辑不感知。
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Awaitable, Callable
+
+from .types import SeatRef
+
+logger = logging.getLogger(__name__)
+
+# 节点状态机：pending → active → done。
+_PENDING = "pending"
+_ACTIVE = "active"
+_DONE = "done"
+
+EventSink = Callable[[dict[str, Any]], Awaitable[None]] | None
+
+
+class JobProgressWriter:
+    """跟踪派活链路节点状态并持久化进度。
+
+    Args:
+      store: ``RoundtableJobRepository``（或任何提供 ``update_progress`` 的对象）；
+             传 None 时所有持久化变 no-op（引擎仍可纯跑，便于干测）。
+      job_id: 目标作业 id。
+      agents: 有序席位列表（recommend=选择顺序 / chain=链条顺序）→ 决定节点 order。
+      event_sink: 可选异步回调，每次进度变更时透传一份归一化事件（Phase 3 SSE）。
+    """
+
+    def __init__(
+        self,
+        store: Any | None,
+        *,
+        job_id: str,
+        agents: list[SeatRef],
+        event_sink: EventSink = None,
+    ) -> None:
+        self._store = store
+        self._job_id = job_id
+        self._agents = list(agents)
+        self._event_sink = event_sink
+        self._state: dict[str, str] = {a.agent_id: _PENDING for a in agents}
+        self._spoke_cycle: dict[str, int] = {}
+        self._consensus = 0
+
+    # ── 节点快照 ──────────────────────────────────────────────────────────
+
+    def _nodes(self) -> list[dict[str, Any]]:
+        """构造 dispatch_chain 节点数组（§2.3）。
+
+        不含 avatarType —— 颜色由前端 ``getAvatarTypeFor(agentId, order)`` 派生，
+        避免把前端的 BUILTIN_AGENT_META 配色表重复移植到后端（易漂移）。
+        """
+        out: list[dict[str, Any]] = []
+        for i, a in enumerate(self._agents):
+            node: dict[str, Any] = {
+                "agentId": a.agent_id,
+                "name": a.name,
+                "order": i + 1,
+                "state": self._state.get(a.agent_id, _PENDING),
+            }
+            if a.agent_id in self._spoke_cycle:
+                node["lastSpokeCycle"] = self._spoke_cycle[a.agent_id]
+            out.append(node)
+        return out
+
+    async def _persist(self, **fields: Any) -> None:
+        """落库（始终带最新 dispatch_chain + consensus）+ 外发事件，均 best-effort。"""
+        fields.setdefault("dispatch_chain", self._nodes())
+        fields.setdefault("consensus_percentage", self._consensus)
+        if self._store is not None:
+            try:
+                await self._store.update_progress(self._job_id, **fields)
+            except Exception:  # noqa: BLE001 — 进度落库失败不能打断编排
+                logger.warning("roundtable job progress persist failed", exc_info=True)
+        if self._event_sink is not None:
+            try:
+                await self._event_sink({"job_id": self._job_id, **fields})
+            except Exception:  # noqa: BLE001
+                logger.warning("roundtable job progress event sink failed", exc_info=True)
+
+    # ── 语义方法（引擎调用）─────────────────────────────────────────────────
+
+    async def begin(self) -> None:
+        await self._persist(status="running", phase="initializing", cycle=0, active_agent_id=None)
+
+    async def leader_thinking(self, cycle: int) -> None:
+        await self._persist(phase="leader_thinking", cycle=cycle, active_agent_id=None)
+
+    async def dispatching(self, cycle: int) -> None:
+        await self._persist(phase="dispatching", cycle=cycle)
+
+    async def seat_active(self, agent_id: str) -> None:
+        if agent_id in self._state:
+            self._state[agent_id] = _ACTIVE
+        await self._persist(phase="dispatching", active_agent_id=agent_id)
+
+    async def seat_done(self, agent_id: str, cycle: int) -> None:
+        if agent_id in self._state:
+            self._state[agent_id] = _DONE
+        self._spoke_cycle[agent_id] = cycle
+        await self._persist(active_agent_id=None)
+
+    async def bump_consensus(self) -> None:
+        """recommend 每轮派活后 +8（封顶 95），对齐前端。"""
+        self._consensus = min(95, self._consensus + 8)
+        await self._persist()
+
+    async def set_consensus_pct(self, pct: int) -> None:
+        """chain 每席位后设进度（封顶 95），对齐前端 round((i+1)/total*90)。"""
+        self._consensus = min(95, int(pct))
+        await self._persist()
+
+    async def consensus(self, pct: int = 100) -> None:
+        self._consensus = int(pct)
+        await self._persist(phase="consensus", active_agent_id=None)
+
+    async def report_drawing(self) -> None:
+        await self._persist(phase="report_drawing", active_agent_id=None)
+
+    async def done(self) -> None:
+        self._consensus = 100
+        await self._persist(status="done", phase="report_done", active_agent_id=None)
+
+    async def awaiting_input(self, question: str) -> None:
+        await self._persist(
+            status="awaiting_input",
+            phase="leader_thinking",
+            active_agent_id=None,
+            pending_clarification=question,
+        )
+
+    async def error(self, message: str) -> None:
+        await self._persist(status="error", error=message)

offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/simulated_gateway.py

@@ -0,0 +1,64 @@
+"""模拟网关 —— 产生真实的编排进度但**不调真模型 / 真 LangGraph server**。
+
+用途：
+  - Phase 3a：让 ``/api/roundtable-jobs`` 控制面（路由 + 执行器 + SSE）端到端跑起来，
+    前端 Phase 4-8 可对着真实 API + 真实 job 表 + 真实 SSE 联调，无需真模型；
+  - 单测：``delay=0`` 时瞬时跑完，验证执行器 / 进度落库 / 路由生命周期。
+
+真网关 ``InProcessRoundtableGateway``（进程内驱动 leader/seat/report run，复用
+multi_agent 的纯解析函数 + ``set_current_user`` 进程内调 agent runtime）是 Phase 3b，
+需在 live stack 上验证。本模拟器与真网关实现同一个 ``RoundtableGateway`` 协议，可直接替换。
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from .types import LeaderTurn, ReportResult, SeatRef, SeatTurn
+
+
+class SimulatedRoundtableGateway:
+    """脚本化的假网关。
+
+    recommend：第 1 轮把所有席位派活一遍 → 第 2 轮共识；
+    chain：每个 per-seat leader 调用不显式派活（引擎据链条顺序强制跑席位）→ 收口。
+    """
+
+    def __init__(self, agents: list[SeatRef], *, delay: float = 0.6, mode: str = "recommend") -> None:
+        self._agents = list(agents)
+        self._delay = delay
+        self._mode = mode
+        self._leader_calls = 0
+
+    async def _tick(self) -> None:
+        if self._delay > 0:
+            await asyncio.sleep(self._delay)
+
+    async def run_leader(self, *, thread_ids, coordinator_name, message, model) -> LeaderTurn:
+        await self._tick()
+        self._leader_calls += 1
+        if self._mode == "chain":
+            # chain 模式由引擎按链条顺序强制跑席位，leader 只给说明、不派活。
+            return LeaderTurn(final_text="（模拟）总控：请当前席位按职责给出分析。", dispatched=[])
+        # recommend：首轮派活全部席位，次轮共识。
+        if self._leader_calls == 1:
+            return LeaderTurn(
+                final_text="（模拟）总控首轮派活：已为各席位分配本轮任务。",
+                dispatched=[(a.agent_id, f"请「{a.name}」就任务从你的专业职责给出分析与建议。") for a in self._agents],
+            )
+        return LeaderTurn(final_text="（模拟）各席位已充分研讨，结论收敛，达成共识。", dispatched=[])
+
+    async def run_seat(self, *, thread_ids, agent_id, task, model) -> SeatTurn:
+        await self._tick()
+        return SeatTurn(final_text=f"（模拟）{agent_id} 的交付：已基于任务背景完成本轮分析。")
+
+    async def create_report_thread(self) -> str:
+        return "sim-report-thread"
+
+    async def run_report(self, *, report_thread_id, prompt, model) -> ReportResult:
+        await self._tick()
+        html = (
+            "<!doctype html><html><head><meta charset='utf-8'><title>方案总览（模拟）</title></head>"
+            "<body><h1>方案总览（模拟）</h1><p>这是模拟网关生成的占位报告，用于前端联调。</p></body></html>"
+        )
+        return ReportResult(html=html, summary="（模拟）方案已生成", model=model)

offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py

@@ -0,0 +1,191 @@
+"""圆桌总控编排引擎 —— 共享类型、常量与纯函数。
+
+把前端 ``useStep2Orchestration.ts`` 的编排循环（``runOrchestration`` /
+``runChainOrchestration``）忠实移植到后端。本文件只放：
+  - ``RoundtableGateway`` 协议：引擎驱动「单轮 leader / seat / report」的抽象出口，
+    把对 FastAPI ``Request`` / ``multi_agent.py`` 内部函数的依赖隔离到 Phase 3 的
+    具体实现里（引擎本身纯逻辑、可用假网关单测）;
+  - ``LeaderTurn`` / ``SeatTurn`` / ``ReportResult`` / ``SeatRef`` 数据类;
+  - 常量（MAX_CYCLES、各轮 prompt）与 chain 模式的 prompt 构造纯函数。
+
+对应 ``frontend-web/docs/multi-agent-background-run-dev.md`` §Phase 2。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Protocol, runtime_checkable
+
+# 编排循环最大轮次，与前端 roundtable-constants.ts MAX_CYCLES 对齐。
+MAX_CYCLES = 8
+
+# recommend 模式每轮派活后、回到 leader 的「续轮」prompt（逐字对齐前端）。
+CONTINUE_PROMPT = (
+    "各子智能体已就上一轮派活完成交付,请你结合最新结论判断是否仍需继续派活;"
+    "若信息已足够,请直接输出最终方案并停止派活。"
+)
+
+# chain 模式全链走完后、总控收口的 prompt（逐字对齐前端）。
+CHAIN_CLOSEOUT_PROMPT = (
+    "各链条席位已按顺序完成发言,请你综合所有席位的交付,输出最终方案与结论。"
+    "不要再派活、不要再向任何席位分配任务。"
+)
+
+
+# ── 数据类 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass(frozen=True)
+class SeatRef:
+    """一个研讨席位（智能体）。chain 模式下 ``agents`` 的顺序即链条顺序。"""
+
+    agent_id: str
+    name: str
+
+
+@dataclass
+class LeaderTurn:
+    """总控一轮执行的归一化结果（由 gateway.run_leader 产出）。"""
+
+    final_text: str = ""
+    # [(agent_id, task), ...]；空 = 共识（不再派活）。
+    dispatched: list[tuple[str, str]] = field(default_factory=list)
+    # 命中澄清时的问题文本；非 None 即 clarification 分支。
+    clarification: str | None = None
+
+    @property
+    def is_clarification(self) -> bool:
+        return bool(self.clarification)
+
+    @property
+    def is_consensus(self) -> bool:
+        return not self.dispatched and not self.is_clarification
+
+
+@dataclass
+class SeatTurn:
+    """某席位一轮交付的归一化结果（由 gateway.run_seat 产出）。"""
+
+    final_text: str = ""
+
+
+@dataclass
+class ReportResult:
+    """Step3 报告绘制结果（由 gateway.run_report 产出）。"""
+
+    html: str = ""
+    summary: str | None = None
+    model: str | None = None
+    generated_at: str | None = None
+
+
+# ── 网关协议 ────────────────────────────────────────────────────────────────
+
+
+@runtime_checkable
+class RoundtableGateway(Protocol):
+    """引擎驱动「单轮执行」的抽象出口。
+
+    Phase 3 提供具体实现（``LoopbackRoundtableGateway``）：内部复用
+    ``multi_agent.py`` 的 ``_leader_run`` / ``_special_run`` / ``_create_thread_direct``
+    等（它们依赖 ``Request``，由 Phase 3 的启动上下文注入）。引擎只面向本协议，
+    因此可用假网关脱离真模型 / 真 LangGraph server 单测。
+
+    线程约定：席位 + 总控 thread 由调用方预先建好并通过 ``thread_ids`` 传入
+    （key = agent_id / coordinator_name，value = thread_id）；report thread 由引擎在
+    进入报告阶段时通过 ``create_report_thread`` 现建。
+    """
+
+    async def run_leader(
+        self,
+        *,
+        thread_ids: dict[str, str],
+        coordinator_name: str,
+        message: str,
+        model: str | None,
+    ) -> LeaderTurn: ...
+
+    async def run_seat(
+        self,
+        *,
+        thread_ids: dict[str, str],
+        agent_id: str,
+        task: str,
+        model: str | None,
+    ) -> SeatTurn: ...
+
+    async def create_report_thread(self) -> str: ...
+
+    async def run_report(
+        self,
+        *,
+        report_thread_id: str,
+        prompt: str,
+        model: str | None,
+    ) -> ReportResult: ...
+
+
+# ── chain 模式 prompt 构造（移植自前端 buildChainLeaderTask / pickChainTask）────
+
+
+def build_chain_leader_task(
+    *,
+    seed_message: str,
+    position: int,
+    total: int,
+    seat_agent_id: str,
+    seat_name: str,
+    is_first: bool,
+) -> str:
+    """构造 chain 模式下「让总控只给当前席位派活」的 leader prompt。"""
+    ordinal = f"{position + 1}/{total}"
+    if is_first:
+        return "\n".join(
+            [
+                "这是一次【按业务链条顺序】进行的多智能体圆桌研讨。任务背景如下：",
+                seed_message,
+                "",
+                f"本次研讨严格按预设链条顺序进行，共 {total} 个席位，现在是第 {ordinal} 个席位。",
+                f"请你作为总控，通过 agent_orchestration 技能**仅**向智能体「{seat_name}」"
+                f"（agent_name: {seat_agent_id}）派活，结合任务背景为它分配聚焦其职责的本轮任务。",
+                "不要派给其他任何智能体，也不要自己直接作答或给出最终结论。",
+            ]
+        )
+    return "\n".join(
+        [
+            f"链条研讨继续，现在轮到第 {ordinal} 个席位。",
+            f"请你作为总控，结合此前各席位已完成的交付，通过 agent_orchestration 技能**仅**向智能体"
+            f"「{seat_name}」（agent_name: {seat_agent_id}）派活，为它分配承接前序结论、聚焦其职责的本轮任务。",
+            "不要派给其他任何智能体，也不要自己直接作答或给出最终结论。",
+        ]
+    )
+
+
+def pick_chain_task(
+    dispatched: list[tuple[str, str]],
+    leader_content: str,
+    seat_agent_id: str,
+    seat_name: str,
+) -> str:
+    """从 leader 终态里取「派给当前链条席位」的任务文本，带纠偏（移植自前端 pickChainTask）。
+
+    1) 命中 agent_id 完全一致的派活 → 用它的 task;
+    2) 总控派了活但写偏 id → 仍用它给出的 task 文本（强制跑正确席位）;
+    3) 总控只写了说明没派活 → 用 leader 正文，再不行回退到模板任务。
+    """
+    wanted = seat_agent_id.lower()
+    for name, task in dispatched:
+        if name.lower() == wanted and (task or "").strip():
+            return task
+    for _name, task in dispatched:
+        if (task or "").strip():
+            return task
+    if (leader_content or "").strip():
+        return (
+            f"请你作为「{seat_name}」，基于以下总控说明与此前各席位的交付，"
+            f"从你的专业职责出发给出你的分析与交付：\n{leader_content.strip()}"
+        )
+    return (
+        f"请你作为「{seat_name}」，基于任务背景与此前各席位的交付，"
+        "从你的专业职责出发给出你的分析与交付。"
+    )

offline-backend-20260512/backend/scripts/probe_roundtable_job.py

@@ -0,0 +1,100 @@
+"""圆桌后台作业真网关 —— headless 验证脚本（Phase 3b）。
+
+用法（在能起完整后端栈 + 配好真模型的环境）：
+
+  1) 用真网关启动后端：
+       ROUNDTABLE_JOB_GATEWAY=inprocess make gateway
+     （或 docker 里给 gateway 容器加该环境变量）
+
+  2) 跑本脚本，它会：POST 启动一个作业 → 轮询 GET 打印进度 → 跑到 done 打印报告长度。
+       PYTHONPATH=. uv run python scripts/probe_roundtable_job.py
+
+环境变量：
+  GATEWAY_URL     默认 http://127.0.0.1:8001
+  AUTH_COOKIE     若后端开启鉴权，传 "session=xxx" 之类的 Cookie 头
+  PROBE_INTENT    自定义任务意图
+  PROBE_MODE      recommend（默认）| chain
+
+说明：鉴权关闭的 dev 模式无需 AUTH_COOKIE。真网关会进程内驱动 roundtable-coordinator /
+各席位 / roundtable-report 真模型，耗时取决于模型，可能数分钟。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import time
+
+import httpx
+
+GATEWAY_URL = os.getenv("GATEWAY_URL", "http://127.0.0.1:8001").rstrip("/")
+AUTH_COOKIE = os.getenv("AUTH_COOKIE", "")
+INTENT = os.getenv("PROBE_INTENT", "为一家中型制造企业设计一套数字化转型的整体方案")
+MODE = os.getenv("PROBE_MODE", "recommend")
+
+# 用 3 个内置圆桌席位（确保可见、已 seed）。
+AGENTS = [
+    {"agentId": "roundtable-intelligence", "name": "情报收集"},
+    {"agentId": "roundtable-solution-design", "name": "方案设计"},
+    {"agentId": "roundtable-summary", "name": "总结陈述"},
+]
+
+TERMINAL = {"done", "error", "cancelled"}
+
+
+def _headers() -> dict[str, str]:
+    h = {"Content-Type": "application/json"}
+    if AUTH_COOKIE:
+        h["Cookie"] = AUTH_COOKIE
+    return h
+
+
+async def main() -> None:
+    async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client:
+        print(f"[probe] POST {GATEWAY_URL}/api/roundtable-jobs  mode={MODE}")
+        resp = await client.post(
+            f"{GATEWAY_URL}/api/roundtable-jobs",
+            headers=_headers(),
+            json={"intent": INTENT, "agents": AGENTS, "orchestrationMode": MODE},
+        )
+        if resp.status_code != 201:
+            print(f"[probe] start failed: HTTP {resp.status_code}: {resp.text[:500]}")
+            return
+        job = resp.json()
+        job_id = job["id"]
+        print(f"[probe] started job {job_id}")
+
+        last = ""
+        t0 = time.perf_counter()
+        while True:
+            await asyncio.sleep(2.0)
+            r = await client.get(f"{GATEWAY_URL}/api/roundtable-jobs/{job_id}", headers=_headers())
+            if r.status_code != 200:
+                print(f"[probe] get failed: HTTP {r.status_code}: {r.text[:300]}")
+                return
+            j = r.json()
+            line = (
+                f"status={j['status']} phase={j['phase']} cycle={j['cycle']} "
+                f"active={j.get('activeAgentId')} consensus={j['consensusPercentage']}%"
+            )
+            if line != last:
+                print(f"[probe] +{time.perf_counter() - t0:5.0f}s  {line}")
+                last = line
+            if j["status"] in TERMINAL:
+                print(f"[probe] terminal: {j['status']}")
+                if j["status"] == "error":
+                    print(f"[probe] error detail: {j.get('error')}")
+                # 读草稿 step3 验证报告已落库
+                if j.get("draftId"):
+                    dr = await client.get(
+                        f"{GATEWAY_URL}/api/roundtable-drafts/{j['draftId']}", headers=_headers()
+                    )
+                    if dr.status_code == 200:
+                        step3 = dr.json().get("step3")
+                        html_len = len((step3 or {}).get("html") or "")
+                        print(f"[probe] draft.step3 html length = {html_len}")
+                return
+
+
+if __name__ == "__main__":
+    asyncio.run(main())