feat(roundtable): 会商诊断日志 + 后台挂起卡死兜底(idle 超时)

- 新增会商全流程诊断日志：新表 roundtable_diagnostics + 仓储 + admin API
  (/api/roundtable-diagnostics) + 管理员诊断页(会商页右上角设置按钮入口)，
  覆盖第一/二/三步及后台执行器/进程内网关各处报错与关键事件，支持按
  范围/阶段/级别/事件/关键词筛选 + 分页。诊断写入全程 best-effort，绝不打断会商。
- 修复后台挂起「执行完第一个席位后永久卡死」：进程内网关每轮 agent run 改为
  idle(无活动)超时(默认 20min)——只要还在持续产出 token/工具事件就一直等，
  仅当连续无任何输出才判卡死并取消+报错+落 seat_turn_timeout 诊断；慢但活着的
  生成绝不打断。可经 ROUNDTABLE_SEAT_IDLE_TIMEOUT_SECONDS 调整。
- 一并提交进行中的会商 seat-skill 注入 / 业务链条编辑器相关改动。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

offline-backend-20260512/backend/CLAUDE.md

@@ -247,6 +247,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
 | **Public Shares** (`/api/public/shares`) | **Unauthenticated** (the `/api/public/` prefix is whitelisted in `auth_middleware._PUBLIC_PATH_PREFIXES`). `GET /{code}` - read-only conversation snapshot (title + serialized messages + artifacts) for a `view`-mode share; messages are read straight from the latest checkpoint. `GET /{code}/files/{path}` - serves a file referenced by that conversation (images, Markdown, text, PDF, ?) so the shared page's links are openable; path resolution is confined to the owner's thread directory. Active content (`text/html`, `application/xhtml+xml`, `image/svg+xml`) is forced to download (`_public_share_disposition`); everything else is served inline. Missing/revoked/non-`view` codes all return 404. |
 | **Roundtable Drafts** (`/api/roundtable-drafts`) | **Strictly per-user** persistence for the roundtable-planning page (replaces the old browser `localStorage`). `GET /` - list drafts (lightweight meta); `POST /` - create; `GET/PUT/DELETE /{id}` - full read / partial update / delete (all owner-scoped). A draft = one full session (Step 1 intent + Step 2 roundtable), step snapshots stored as JSON in `PortableLongText`. Recommendation history bound to a draft: `GET /{id}/recommendations` - list newest-first; `POST /{id}/recommendations` - append one run (objective/status/rationale/picks/candidates). taskId 深链会商记录 are **not** here — they live in the separate `/api/roundtable-task-drafts` store (see below) so personal history never mixes with task records. Backed by `deerflow.persistence.roundtable_drafts` (tables `roundtable_drafts`, `roundtable_recommend_history`); store wired in `deps.py` as `app.state.roundtable_draft_store`. Tests: `tests/test_roundtable_task_drafts.py`. |
 | **Roundtable Task Drafts** (`/api/roundtable-task-drafts`) | **Task-scoped, unpartitioned** store for taskId-deeplink roundtable 聊天记录 (无界嵌入抽屉). Keyed **only by `task_id`** (one task → many records), **never partitioned by user** — any authenticated caller may read/write/delete a task's records (`created_by` is audit-only). Physically separate from `roundtable_drafts` so the two never mix. `GET /by-task/{task_id}` - all drafts for a task, newest-first (history dropdown), enriched with the latest background-job status per draft via `RoundtableJobRepository.status_map_by_drafts`; `GET /by-task/{task_id}/latest` - newest draft (回显最新一条); `POST /` - create (`task_id` required); `GET/PUT/DELETE /{id}` - full read / partial update / delete (all user-agnostic). Backed by `deerflow.persistence.roundtable_task_drafts` (table `roundtable_task_drafts`, auto-created by `create_all`; migration `20260618_01`); store wired in `deps.py` as `app.state.roundtable_task_draft_store`. Existing `task_id`-bearing `roundtable_drafts` rows are migrated over by `scripts/migrate_roundtable_task_drafts.py`. Tests: `tests/test_roundtable_task_drafts.py`, `tests/test_roundtable_drafts_by_task.py`. |
+| **Roundtable Diagnostics** (`/api/roundtable-diagnostics`) | **Admin-only** read of 会商全流程诊断日志 (报错 + 关键事件). One row per event in `roundtable_diagnostics`: `{scope (background/foreground), stage (step1_intent/step2_init/step2_leader/step2_seat/step2_recommend/step2_orchestration/step3_summary/step3_report/step3_dashboard/step3_structure/upstream/job), level (info/warning/error), event, message, detail (JSON), job_id/draft_id/task_id/user_id/agent_id/agent_name/cycle, created_at}`. `GET /` - filter (scope/stage/level/event/job_id/draft_id/task_id/user_id/q/since/until) + paginate (newest-first); `GET /facets` - distinct values for filter dropdowns; `POST /admin/cleanup` - retention purge. Backed by `deerflow.persistence.roundtable_diagnostics` (table `roundtable_diagnostics`, auto-created by `create_all`; migration `20260619_01`); store wired in `deps.py` as `app.state.roundtable_diagnostic_store` + a process-level default store (`register_default_store`) so the harness-side engine/gateway can write across the app/harness boundary. **Writers** (all best-effort — diagnostics never break a run): background path uses `DiagnosticsRecorder` (bound to job ctx) threaded through `RoundtableJobExecutor` → `JobProgressWriter.diag(...)` (engine transitions/step3 failures/empty-delivery) + `InProcessRoundtableGateway` (per-turn timeout/report·summary·dashboard truncation+missing); foreground path uses `app/gateway/roundtable_diag.py::record_foreground_diag` in `multi_agent.py`/`intent.py`/`recommend.py` error frames. **Bug-fix shipped alongside**: `InProcessRoundtableGateway._run_agent_turn` guards each agent run with an **inactivity (idle) timeout** (`_await_run_with_idle_guard`, `ROUNDTABLE_SEAT_IDLE_TIMEOUT_SECONDS` default 1200s = 20min, tuned high for unstable intranet models) — **not** a total-duration cap, so a slow-but-alive run that keeps streaming tokens runs as long as it needs; only a run that emits **zero StreamBridge events for `idle_timeout` seconds straight** (a true freeze: 内网模型连接池挂死 / sqlite 锁在挂载卷上挂死) is cancelled + raises (顶层把 job 标 error，而非永久冻结) + records a `seat_turn_timeout` diagnostic. Activity is observed via `app.state.stream_bridge` (the worker publishes every chunk there). Optional absolute cap `ROUNDTABLE_SEAT_MAX_TOTAL_SECONDS` (default 0 = off). Frontend: admin page `pages/RoundtableDiagnosticsPage.tsx` (route `/page/strategy/admin/roundtable-diagnostics`), opened from the **会商页右上角设置按钮** (admin only — `handleSettingsClick` in `roundtable-planning/pages/RoundtablePlanningPage.tsx`); client `strategy-components/api/roundtable-diagnostics.ts`. Tests: `tests/test_roundtable_diagnostics.py`, `tests/test_roundtable_inprocess_timeout.py`. |
 | **Roundtable Draft Shares** (`/api/roundtable-draft-shares`) | Per-draft sharing for the roundtable-planning page, mirroring Thread Shares but the shared resource is a `roundtable_drafts` row. Two modes (`roundtable_draft_shares.mode`). **`import`**: `POST /` `{draft_id, mode}` - create (or reuse) a revocable code; `GET /{code}/preview` + `POST /{code}/import` - copy that draft (step1/2/3) into the caller's account under a deterministic per-(code,recipient) id (`imp_<sha1>` ? re-import is a no-op). **`view`**: a public, read-only result page ? see Public Roundtable Shares. Both: `GET /` - list my codes; `DELETE /{code}` - revoke. Backed by `deerflow.persistence.roundtable_draft_shares` (table `roundtable_draft_shares`, migration `20260606_01`); store wired in `deps.py` as `app.state.roundtable_draft_share_store`. Tests: `tests/test_roundtable_draft_shares.py`. |
 | **Public Roundtable Shares** (`/api/public/roundtable-shares`) | **Unauthenticated** (`/api/public/` whitelisted). `GET /{code}` - read-only snapshot for a `view`-mode draft share: returns the Step 3??????report HTML (`step3.html`, a self-contained document) + title/summary/`has_report`. The frontend renders it in a sandboxed `iframe` (no `allow-scripts`) at `#/roundtable-share/<code>`. Missing/revoked/non-`view` codes all return 404. |
 | **Scheduled Tasks** (`/api/scheduled-tasks`) | Per-user cron-like agent tasks. `GET/POST /` - list/create owned tasks; `PATCH/DELETE /{id}` - update/delete (owner); `POST /{id}/{pause,resume,trigger}`; `GET /{id}/runs`, `GET /runs/{rid}`, run-file preview/edit/download; `GET /runs/{rid}/messages` - agent messages for the run's underlying agent run (used to inspect a stuck/running run; authorized via `_load_run`); `GET /runs/{rid}/html` - generated HTML document for an `html_page` run; `DELETE /runs/{rid}` - owner-scoped, with an admin fallback so admins can delete any user's run; publish/subscribe under `/{id}/{publish,unpublish,subscribe}` and `GET /published`. **Admin (cross-user):** `GET /admin/all` - every task; `PATCH/DELETE /admin/{id}` - edit/delete any task; `POST /admin/{id}/trigger`; `GET /admin/{id}/runs`. Admin routes require `system_role == "admin"`. |

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

@@ -61,6 +61,7 @@ from app.gateway.routers import (
     flow_extract,
     recommended_questions,
     roundtable_chains,
+    roundtable_diagnostics,
     roundtable_draft_shares,
     roundtable_drafts,
     roundtable_jobs,
@@ -1370,6 +1371,7 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
 
     # Roundtable-planning background jobs at /api/roundtable-jobs
     app.include_router(roundtable_jobs.router)
+    app.include_router(roundtable_diagnostics.router)
 
     # Suggestions API is mounted at /api/threads/{thread_id}/suggestions
     app.include_router(suggestions.router)

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

@@ -295,6 +295,18 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
         # （Phase 2 后台编排图写进度用）。
         _register_roundtable_job_default_store(app.state.roundtable_job_store)
 
+        # 圆桌会商「诊断日志」store：会商全流程各处报错 / 关键事件落库，供管理员诊断页筛选。
+        # 同样登记进程级默认仓储——harness 侧编排引擎 / 进程内网关跨边界写诊断用。
+        from deerflow.persistence.roundtable_diagnostics import (
+            make_roundtable_diagnostic_store,
+        )
+        from deerflow.persistence.roundtable_diagnostics import (
+            register_default_store as _register_roundtable_diagnostic_default_store,
+        )
+
+        app.state.roundtable_diagnostic_store = make_roundtable_diagnostic_store(sf)
+        _register_roundtable_diagnostic_default_store(app.state.roundtable_diagnostic_store)
+
         # 圆桌后台作业执行器：把编排引擎作为后台任务跑。
         # 网关由环境变量 ``ROUNDTABLE_JOB_GATEWAY`` 选择（见 resolve_roundtable_gateway_mode）：
         #   - 缺省 / "inprocess"（Phase 3b）：进程内驱动真模型（leader/seat/report）——**默认**，
@@ -327,6 +339,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
                     seat_thinking_enabled=p.seat_thinking_enabled,
                     seat_reasoning_effort=p.seat_reasoning_effort,
                     seat_subagent_enabled=p.seat_subagent_enabled,
+                    # 诊断日志上下文：网关每轮 leader/seat/report/summary/dashboard 报错均带这些归属。
+                    job_id=p.job_id,
+                    draft_id=p.draft_id,
+                    task_id=p.task_id,
                 )
                 logger.info("[roundtable-jobs] gateway = inprocess (real models, default)")
 

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

@@ -0,0 +1,63 @@
+"""前台（实时会商）诊断日志助手。
+
+后台挂起路径用 ``DiagnosticsRecorder``（绑定作业上下文）；前台 multi_agent 路由用本助手
+从 ``request`` 取 store + 当前用户，best-effort 落一条 ``scope="foreground"`` 的诊断事件。
+任何异常静默吞掉——诊断日志绝不能反过来打断会商。
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from fastapi import Request
+
+from deerflow.persistence.roundtable_diagnostics import record_diagnostic
+from deerflow.runtime.user_context import get_effective_user_id
+
+logger = logging.getLogger(__name__)
+
+
+def _resolve_user_id(request: Request) -> str | None:
+    try:
+        user = getattr(request.state, "user", None)
+        if user is not None:
+            return str(user.id)
+        return get_effective_user_id()
+    except Exception:  # noqa: BLE001
+        return None
+
+
+async def record_foreground_diag(
+    request: Request,
+    *,
+    stage: str,
+    level: str,
+    event: str,
+    message: str | None = None,
+    detail: Any = None,
+    thread_id: str | None = None,
+    draft_id: str | None = None,
+    task_id: str | None = None,
+    agent_id: str | None = None,
+    agent_name: str | None = None,
+) -> None:
+    """从 request 落一条前台诊断事件（best-effort）。"""
+    try:
+        store = getattr(request.app.state, "roundtable_diagnostic_store", None)
+        await record_diagnostic(
+            store,
+            scope="foreground",
+            stage=stage,
+            level=level,
+            event=event,
+            message=message,
+            detail=detail,
+            draft_id=draft_id,
+            task_id=task_id,
+            user_id=_resolve_user_id(request),
+            agent_id=agent_id,
+            agent_name=agent_name,
+        )
+    except Exception:  # noqa: BLE001 — 诊断日志绝不打断会商
+        logger.debug("foreground roundtable diag failed", exc_info=True)

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

@@ -25,6 +25,7 @@ from typing import Any
 from langgraph.checkpoint.base import empty_checkpoint
 
 from app.gateway.roundtable_run_policy import roundtable_run_policy
+from app.gateway.roundtable_seat_skills import append_seat_skill_directive
 from app.gateway.routers._roundtable_seed import (
     COORDINATOR_AGENT_ID,
     DASHBOARD_AGENT_ID,
@@ -41,10 +42,31 @@ from deerflow.agents.roundtable_orchestrator.types import (
     SummaryResult,
 )
 from deerflow.config.paths import get_paths
+from deerflow.persistence.roundtable_diagnostics import DiagnosticsRecorder, get_default_store
 from deerflow.runtime.serialization import serialize_channel_values
 
 logger = logging.getLogger(__name__)
 
+# 单轮 agent run（leader/seat/report/summary/dashboard）的**无活动(idle)超时**（秒）。
+#
+# 背景：后台编排是**串行 await 每一个 run**，且历史上**没有任何超时保护**——只要某一轮
+# ``agent.astream`` 卡死（内网模型端点连接被打满 / sqlite 锁在挂载卷上挂死等环境相关原因），
+# 整个后台作业就永久冻结（用户侧表现为「执行完第一个席位后一直不动」）。
+#
+# ⚠️ 关键：这里**不是总时长超时**，而是 **idle 超时**——只要这轮 run 还在持续产出
+# （流式 token / 工具调用 / 进度事件，全部经 StreamBridge 外发），就一直等下去，哪怕这轮
+# 真的要跑 20 分钟、半小时也绝不打断（慢但活着的生成不该被切）。只有当 run **连续
+# ``_SEAT_IDLE_TIMEOUT_S`` 秒完全没有任何输出**（真卡死）时才取消它、抛错 → 顶层把 job 标
+# error（而非永久卡死）+ 落一条 seat_turn_timeout 诊断。默认 1200s（20 分钟）——内网模型
+# 很不稳定时单次「首 token / 工具间隔」可能拖很久，给足容忍度，宁可多等也别误杀慢生成；
+# idle 只兜真正「连续 20 分钟啥都不发生」的死锁。
+_SEAT_IDLE_TIMEOUT_S = int(os.getenv("ROUNDTABLE_SEAT_IDLE_TIMEOUT_SECONDS", "1200") or "1200")
+# 可选的绝对总时长上限（秒）：默认 0 = **关闭**（只靠 idle 超时，慢生成不受限）。
+# 仅当你确实想给单轮设一个天花板时再设它（如某些场景要硬兜底）。
+_SEAT_MAX_TOTAL_S = int(os.getenv("ROUNDTABLE_SEAT_MAX_TOTAL_SECONDS", "0") or "0")
+# idle 检测的心跳间隔：每隔这么久检查一次「距上次活动过了多久」。
+_IDLE_CHECK_INTERVAL_S = 15.0
+
 _ROUNDTABLE_THREAD_METADATA: dict[str, Any] = {"thread_type": "roundtable", "system": True}
 _REPORT_FILE = "方案总览.html"
 _SUMMARY_FILE = "方案总结报告.md"
@@ -183,6 +205,108 @@ def _tool_calls_since_last_human(messages: list[dict[str, Any]]) -> list[dict[st
     return out
 
 
+async def _await_run_with_idle_guard(
+    app,
+    record,
+    *,
+    idle_timeout: float,
+    diag: DiagnosticsRecorder | None,
+    stage: str,
+    agent_name: str,
+    cycle: int | None,
+    max_total: float = 0.0,
+) -> None:
+    """等待一轮 run 完成，但用 **无活动(idle) 超时**而非总时长超时。
+
+    只要 run 仍在持续产出（StreamBridge 上有新事件：流式 token / 工具调用 / 进度），就一直
+    等下去——慢但活着的生成绝不打断。只有当 run **连续 ``idle_timeout`` 秒完全没有任何事件**
+    （真卡死）时才取消它并抛 ``RuntimeError``。``max_total`` > 0 时附加一个绝对总时长上限。
+
+    活动信号取自 StreamBridge（worker 每产一块都 ``bridge.publish``）。拿不到 bridge / run_id
+    时退化为「总时长超时（若配置）或直接等待」。
+    """
+    task = getattr(record, "task", None)
+    if task is None:
+        return
+    run_id = getattr(record, "run_id", None)
+    bridge = getattr(app.state, "stream_bridge", None)
+    loop = asyncio.get_running_loop()
+    start = loop.time()
+
+    # 没有 bridge 可观察活动 → 退化：有 max_total 用之，否则直接等待（保持旧的「不卡死则一直等」）。
+    if bridge is None or run_id is None:
+        if max_total and max_total > 0:
+            await asyncio.wait_for(asyncio.shield(task), timeout=max_total)
+        else:
+            await task
+        return
+
+    from deerflow.runtime.stream_bridge.base import END_SENTINEL, HEARTBEAT_SENTINEL
+
+    last_activity = loop.time()
+    stuck = asyncio.Event()
+
+    async def _watch() -> None:
+        nonlocal last_activity
+        async for ev in bridge.subscribe(run_id, heartbeat_interval=_IDLE_CHECK_INTERVAL_S):
+            if ev is END_SENTINEL:
+                return  # run 已收尾，正常退出（不算卡死）
+            now = loop.time()
+            if ev is HEARTBEAT_SENTINEL:
+                if now - last_activity >= idle_timeout:
+                    stuck.set()
+                    return
+                if max_total and max_total > 0 and now - start >= max_total:
+                    stuck.set()
+                    return
+            else:
+                last_activity = now  # 任何真实事件 = 还活着
+
+    watch = asyncio.create_task(_watch())
+    try:
+        done, _pending = await asyncio.wait({task, watch}, return_when=asyncio.FIRST_COMPLETED)
+        if task in done:
+            return  # run 正常结束（run_agent 内部已吞掉异常并落终态）
+        # watch 先完成：要么判定卡死，要么因 END 正常退出。
+        if not stuck.is_set():
+            await task  # END 已出现，run 即将收尾，等它
+            return
+        # —— 判定卡死：取消该 run，落诊断，抛错 ——
+        idle_elapsed = loop.time() - last_activity
+        total = loop.time() - start
+        logger.error(
+            "roundtable %s turn '%s' stalled: no output for %.0fs (total %.0fs) on run %s",
+            stage, agent_name, idle_elapsed, total, run_id,
+        )
+        run_manager = getattr(app.state, "run_manager", None)
+        if run_manager is not None:
+            try:
+                await run_manager.cancel(run_id, action="interrupt")
+            except Exception:  # noqa: BLE001
+                logger.warning("cancel stalled run %s failed", run_id, exc_info=True)
+        if diag is not None:
+            await diag.record(
+                stage=stage, level="error", event="seat_turn_timeout",
+                message=(
+                    f"「{agent_name}」连续 {idle_elapsed:.0f}s 无任何输出、判定为卡死并已取消"
+                    f"（idle 上限 {idle_timeout:.0f}s；本轮已运行 {total:.0f}s）。常见原因：内网模型端点连接"
+                    "被打满 / sqlite checkpoint 在挂载卷上锁挂死。注意：慢但持续产出的生成不会被打断。"
+                ),
+                detail={"idle_seconds": round(idle_elapsed, 1), "total_seconds": round(total, 1), "idle_timeout": idle_timeout},
+                agent_name=agent_name, cycle=cycle,
+            )
+        raise RuntimeError(
+            f"roundtable seat turn '{agent_name}' stalled (no output for {idle_timeout:.0f}s)"
+        ) from None
+    finally:
+        if not watch.done():
+            watch.cancel()
+            try:
+                await watch
+            except BaseException:  # noqa: BLE001 — 取消 watch 的异常无关紧要
+                pass
+
+
 async def _run_agent_turn(
     app,
     *,
@@ -196,6 +320,9 @@ async def _run_agent_turn(
     thinking_enabled: bool,
     reasoning_effort: str,
     subagent_enabled: bool = False,
+    diag: DiagnosticsRecorder | None = None,
+    stage: str = "step2_seat",
+    cycle: int | None = None,
 ) -> list[dict[str, Any]]:
     """进程内驱动 agent_name 在 thread_id 上跑一轮（发 message），阻塞到完成，返回最终 messages。
 
@@ -239,9 +366,25 @@ async def _run_agent_turn(
     # 取该用户的后台额度后再开跑：start_run + await task 是最重的部分（建 agent、调模型、
     # 落 checkpoint）。额度满时本轮在此 await 排队，等同用户某个在跑轮次释放额度后再开。
     async with _user_background_semaphore(user_id):
-        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 已落盘
+        try:
+            record = await start_run(body, thread_id, request)  # type: ignore[arg-type]
+        except Exception as exc:  # noqa: BLE001 — start_run 报错（409 抢占 / DB 抖动等）也要落诊断
+            if diag is not None:
+                await diag.record(
+                    stage=stage, level="error", event="start_run_failed",
+                    message=f"启动 agent run 失败（{agent_name}）：{exc}",
+                    detail={"type": type(exc).__name__, "error": str(exc), "thread_id": thread_id},
+                    agent_name=agent_name, cycle=cycle,
+                )
+            raise
+        # idle 超时守护：慢但持续产出的生成一直等；只有连续 _SEAT_IDLE_TIMEOUT_S 秒**毫无输出**
+        # （真卡死）才取消并抛错。绝不因「这轮总耗时长」而打断（除非显式配了 _SEAT_MAX_TOTAL_S）。
+        await _await_run_with_idle_guard(
+            app, record,
+            idle_timeout=_SEAT_IDLE_TIMEOUT_S,
+            max_total=_SEAT_MAX_TOTAL_S,
+            diag=diag, stage=stage, agent_name=agent_name, cycle=cycle,
+        )
     return await _read_thread_messages(app, thread_id)
 
 
@@ -382,11 +525,23 @@ class InProcessRoundtableGateway:
         seat_thinking_enabled: bool | None = None,
         seat_reasoning_effort: str | None = None,
         seat_subagent_enabled: bool = False,
+        job_id: str | None = None,
+        draft_id: str | None = None,
+        task_id: str | None = None,
     ) -> None:
         self._app = app
         self._user_id = user_id
         self._agents = list(agents)
         self._model = model
+        # 诊断日志记录器（绑定本作业上下文）。用进程级默认 store（deps.py 启动时已登记）。
+        self._diag = DiagnosticsRecorder(
+            get_default_store(),
+            scope="background",
+            job_id=job_id,
+            draft_id=draft_id,
+            task_id=task_id,
+            user_id=user_id,
+        )
         # 席位执行模式（沿用 Step2 选择）：覆盖 seat policy 的 thinking/reasoning（None → policy 默认），
         # subagent 仅 ultra 档为真。只作用 special 席位；leader 派活不受影响。
         self._seat_thinking_enabled = seat_thinking_enabled
@@ -426,6 +581,8 @@ class InProcessRoundtableGateway:
             skill_stop_names=policy.skill_stop_names,
             thinking_enabled=policy.thinking_enabled,
             reasoning_effort=policy.reasoning_effort,
+            diag=self._diag,
+            stage="step2_leader",
         )
         ai = _latest_ai_message(messages)
         # tool_calls 取最后一条 AI 消息（派活就在这条）；正文取最后一条有文字的 AI 消息。
@@ -456,13 +613,17 @@ class InProcessRoundtableGateway:
 
     async def run_seat(self, *, thread_ids, agent_id, task, model) -> SeatTurn:
         policy = roundtable_run_policy("seat")
+        # 弱模型补偿：把该席位配置的技能（名称 + SKILL.md 路径 + 必须先 read_file 再按其
+        # 流程取信息的硬指令）显式拼进本轮任务，避免弱模型无视系统提示里的技能块（见
+        # roundtable_seat_skills 模块说明）。无配置技能时原样返回 task。
+        seat_task = append_seat_skill_directive(agent_id, task)
         # 席位模式覆盖 policy 的 thinking/reasoning（None → policy 默认），与前端 _special_run 一致。
         messages = await _run_agent_turn(
             self._app,
             user_id=self._user_id,
             thread_id=thread_ids[agent_id],
             agent_name=agent_id,
-            message=task,
+            message=seat_task,
             model=model or self._model,
             excluded_tools=policy.excluded_tools,
             skill_stop_names=policy.skill_stop_names,
@@ -477,6 +638,8 @@ class InProcessRoundtableGateway:
                 else self._seat_reasoning_effort
             ),
             subagent_enabled=self._seat_subagent_enabled,
+            diag=self._diag,
+            stage="step2_seat",
         )
         return SeatTurn(final_text=_latest_ai_text(messages), tool_calls=_tool_calls_since_last_human(messages))
 
@@ -503,19 +666,32 @@ class InProcessRoundtableGateway:
                 skill_stop_names=policy.skill_stop_names,
                 thinking_enabled=policy.thinking_enabled,
                 reasoning_effort=policy.reasoning_effort,
+                diag=self._diag,
+                stage="step3_report",
             )
             summary = _latest_ai_text(messages) or summary
             html = _read_report_html(self._app, report_thread_id, self._user_id)
             if _html_complete(html):
                 break
             if attempt >= _ARTIFACT_RETRY_LIMIT:
+                await self._diag.record(
+                    stage="step3_report", level="error", event="report_artifact_incomplete",
+                    message=f"结果绘制重试 {_ARTIFACT_RETRY_LIMIT} 次后 HTML 仍缺失/截断，回退为正文展示",
+                    detail={"thread_id": report_thread_id, "has_partial_html": bool(html and html.strip())},
+                    agent_name=REPORT_AGENT_ID,
+                )
                 break
+            kind = "truncated" if (html and html.strip()) else "missing"
             message = _REPORT_RETRY_TRUNCATED if (html and html.strip()) else _REPORT_RETRY_MISSING
             logger.warning(
                 "report html %s after attempt %d for thread %s; retrying with corrective prompt",
-                "truncated" if (html and html.strip()) else "missing",
-                attempt + 1,
-                report_thread_id,
+                kind, attempt + 1, report_thread_id,
+            )
+            await self._diag.record(
+                stage="step3_report", level="warning", event=f"report_{kind}",
+                message=f"结果绘制第 {attempt + 1} 次产物{('被截断' if kind == 'truncated' else '缺失')}，已追加纠偏提示重试",
+                detail={"thread_id": report_thread_id, "attempt": attempt + 1},
+                agent_name=REPORT_AGENT_ID,
             )
         # 截断的 HTML 也比纯文字强（浏览器容忍未闭合标签），仅在完全没有文件时才回退正文。
         final_html = html if (html and html.strip()) else summary
@@ -547,12 +723,20 @@ class InProcessRoundtableGateway:
                 skill_stop_names=policy.skill_stop_names,
                 thinking_enabled=policy.thinking_enabled,
                 reasoning_effort=policy.reasoning_effort,
+                diag=self._diag,
+                stage="step3_summary",
             )
             closing = _latest_ai_text(messages) or closing
             md = _read_summary_md(self._app, summary_thread_id, self._user_id)
             if md and md.strip():
                 break
             if attempt >= _ARTIFACT_RETRY_LIMIT:
+                await self._diag.record(
+                    stage="step3_summary", level="error", event="summary_artifact_missing",
+                    message=f"方案总结报告重试 {_ARTIFACT_RETRY_LIMIT} 次后 .md 仍缺失，回退为对话正文",
+                    detail={"thread_id": summary_thread_id},
+                    agent_name=SUMMARY_AGENT_ID,
+                )
                 break
             message = _SUMMARY_RETRY_MISSING
             logger.warning(
@@ -560,6 +744,12 @@ class InProcessRoundtableGateway:
                 attempt + 1,
                 summary_thread_id,
             )
+            await self._diag.record(
+                stage="step3_summary", level="warning", event="summary_missing",
+                message=f"方案总结报告第 {attempt + 1} 次未生成 .md，已追加纠偏提示重试",
+                detail={"thread_id": summary_thread_id, "attempt": attempt + 1},
+                agent_name=SUMMARY_AGENT_ID,
+            )
         final_md = md if (md and md.strip()) else closing
         return SummaryResult(md=final_md, closing=closing, model=model or self._model, thread_id=summary_thread_id)
 
@@ -589,11 +779,19 @@ class InProcessRoundtableGateway:
                 skill_stop_names=policy.skill_stop_names,
                 thinking_enabled=policy.thinking_enabled,
                 reasoning_effort=policy.reasoning_effort,
+                diag=self._diag,
+                stage="step3_dashboard",
             )
             body = _extract_report_json(_latest_ai_text(messages))
             if body:
                 return body
             if attempt >= _ARTIFACT_RETRY_LIMIT:
+                await self._diag.record(
+                    stage="step3_dashboard", level="error", event="dashboard_json_invalid",
+                    message=f"大屏数据重试 {_ARTIFACT_RETRY_LIMIT} 次后仍抽取不到合法 report-json，放弃大屏产物",
+                    detail={"thread_id": thread_id},
+                    agent_name=DASHBOARD_AGENT_ID,
+                )
                 break
             message = _DASHBOARD_RETRY
             logger.warning(
@@ -601,4 +799,10 @@ class InProcessRoundtableGateway:
                 attempt + 1,
                 thread_id,
             )
+            await self._diag.record(
+                stage="step3_dashboard", level="warning", event="dashboard_json_retry",
+                message=f"大屏数据第 {attempt + 1} 次抽取失败（缺 report-json / JSON 非法），已追加纠偏提示重试",
+                detail={"thread_id": thread_id, "attempt": attempt + 1},
+                agent_name=DASHBOARD_AGENT_ID,
+            )
         return None

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

@@ -26,6 +26,7 @@ from deerflow.agents.roundtable_orchestrator import (
     SeatRef,
     run_orchestration,
 )
+from deerflow.persistence.roundtable_diagnostics import DiagnosticsRecorder, get_default_store
 from deerflow.runtime.user_context import reset_current_user, set_current_user
 
 logger = logging.getLogger(__name__)
@@ -219,7 +220,20 @@ class RoundtableJobExecutor:
 
     async def _run(self, params: StartParams) -> None:
         token = set_current_user(_JobUser(params.user_id))
+        diag = DiagnosticsRecorder(
+            get_default_store(),
+            scope="background",
+            job_id=params.job_id,
+            draft_id=params.draft_id,
+            task_id=params.task_id,
+            user_id=params.user_id,
+        )
         try:
+            await diag.record(
+                stage="job", level="info", event="job_started",
+                message=f"后台挂起作业启动（mode={params.mode}{'，续跑' if params.is_resume else ''}）",
+                detail={"mode": params.mode, "is_resume": params.is_resume, "agents": [a.agent_id for a in params.agents]},
+            )
             gateway = self._gateway_factory(params)
             thread_ids = params.thread_ids or {}
             coordinator_name = params.coordinator_name
@@ -248,6 +262,7 @@ class RoundtableJobExecutor:
                 job_id=params.job_id,
                 agents=params.agents,
                 on_dialogue=_on_dialogue if (self._draft_store_for(params) and params.draft_id) else None,
+                recorder=diag,
             )
             result = await run_orchestration(
                 gateway=gateway,
@@ -283,8 +298,13 @@ class RoundtableJobExecutor:
             # 路由的 cancel 端点（executor.cancel + store.update_progress）权威写入。
             logger.info("roundtable job %s cancelled", params.job_id)
             raise
-        except Exception:  # noqa: BLE001 — 顶层兜底：run_orchestration 内部已尽量标 error
+        except Exception as exc:  # noqa: BLE001 — 顶层兜底：run_orchestration 内部已尽量标 error
             logger.exception("roundtable job %s crashed", params.job_id)
+            await diag.record(
+                stage="job", level="error", event="job_crashed",
+                message=f"后台挂起作业崩溃：{exc}",
+                detail={"type": type(exc).__name__, "error": str(exc)},
+            )
         finally:
             reset_current_user(token)
 
@@ -363,8 +383,16 @@ class RoundtableJobExecutor:
             update_kwargs["step3"] = effective_step3
         try:
             await draft_store.update_draft(params.draft_id, params.user_id, **update_kwargs)
-        except Exception:  # noqa: BLE001 — 写草稿失败不影响作业本身
+        except Exception as exc:  # noqa: BLE001 — 写草稿失败不影响作业本身
             logger.warning("write run to draft %s failed", params.draft_id, exc_info=True)
+            from deerflow.persistence.roundtable_diagnostics import record_diagnostic
+
+            await record_diagnostic(
+                scope="background", stage="job", level="error", event="draft_write_failed",
+                message=f"研讨结果写回草稿失败（status={status}）：{exc}",
+                detail={"type": type(exc).__name__, "error": str(exc)},
+                job_id=params.job_id, draft_id=params.draft_id, task_id=params.task_id, user_id=params.user_id,
+            )
 
     def cancel(self, job_id: str) -> bool:
         """请求取消一个在跑的作业。返回是否命中在跑任务。"""

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

@@ -0,0 +1,95 @@
+"""会商第二步席位「技能识别」强化（弱模型友好）。
+
+痛点：席位是一次自定义 agent run，其配置技能（agent ``config.yaml → skills``）只通过
+**系统提示**里的 ``<skill_system>`` 块注入（列出 name/description + SKILL.md 路径，靠模型
+自己 ``read_file`` 并遵循）。强模型会照做；**弱模型（如 deepseek-chat）往往无视系统提示里
+的技能块，从不 read_file SKILL.md**，于是配置的技能形同虚设——只凭模型自身知识作答。
+
+解法：在**派给席位的本轮任务消息（human turn）**里，显式把该席位配置的每个技能（名称 +
+描述 + SKILL.md 容器路径）列出来，并下达硬指令「先 read_file 对应 SKILL.md、按其说明用工具
+去取信息 / 走流程，再产出交付」。把指令放进 user turn（而非仅系统提示）对弱模型的牵引力
+强得多——这与本仓已有的弱模型补偿手法（``_SEAT_NO_ASK``、总控派活强化、AI 写作 SKILL.md
+注入，见 ``project-ai-writing-skill-agent``）一脉相承。
+
+两条席位执行路径共用本 helper：
+  - 前端手动：``multi_agent._special_run``（seat / seat_parallel 角色）；
+  - 后台挂起：``roundtable_inprocess_gateway.InProcessRoundtableGateway.run_seat``。
+
+无配置技能的席位 → 返回空串（完全无副作用）。可用 ``ROUNDTABLE_SEAT_SKILL_DIRECTIVE=0``
+全局关闭（默认开）。
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+
+logger = logging.getLogger(__name__)
+
+_TRUTHY = frozenset({"1", "true", "yes", "on", "y"})
+
+
+def _directive_enabled() -> bool:
+    """读 ``ROUNDTABLE_SEAT_SKILL_DIRECTIVE``（默认开）。空 / 未设 = 开。"""
+    return (os.getenv("ROUNDTABLE_SEAT_SKILL_DIRECTIVE") or "1").strip().lower() in _TRUTHY
+
+
+def build_seat_skill_directive(agent_id: str | None) -> str:
+    """为某席位 agent 构建「必须先读 SKILL.md 再用技能」的本轮任务追加指令。
+
+    返回 ``""`` 表示无需追加（开关关闭 / 无 agent_id / 该席位没有配置技能 / 解析失败），
+    调用方原样使用任务文本，零副作用。
+    """
+    if not _directive_enabled() or not agent_id:
+        return ""
+    try:
+        from deerflow.config.agents_config import load_agent_config
+
+        cfg = load_agent_config(agent_id)
+        configured = [name for name in (cfg.skills or []) if name] if cfg else []
+        if not configured:
+            return ""
+
+        from deerflow.skills.storage import get_or_new_skill_storage
+
+        try:
+            from deerflow.config import get_app_config
+
+            container_base_path = get_app_config().skills.container_path
+        except Exception:  # noqa: BLE001 — 配置读不到时退回默认挂载路径
+            container_base_path = "/mnt/skills"
+
+        by_name = {s.name: s for s in get_or_new_skill_storage().load_skills(enabled_only=True)}
+        # 保留席位 config 里的配置顺序，剔除已删除 / 未启用的技能名。
+        wanted = [by_name[name] for name in configured if name in by_name]
+        if not wanted:
+            return ""
+
+        lines: list[str] = [
+            "【你已配置以下专属技能，必须优先使用——这是硬性要求，不要只凭自身知识作答】",
+            "完成本轮交付前，请先判断哪些技能与任务相关，**用 read_file 工具读取对应技能的 SKILL.md 全文**，"
+            "严格按其中的说明（检索方式 / 接口 / 流程 / 输出格式）去**获取真实信息或执行处理流程**"
+            "（其中可能要求你调用 bash / write_file 等工具），再据此产出交付：",
+            "",
+        ]
+        for i, s in enumerate(wanted, 1):
+            desc = (s.description or "").strip().replace("\n", " ")
+            lines.append(f"{i}. 「{s.name}」{desc}".rstrip())
+            lines.append(f"   - 技能说明文件（先 read_file 它）：{s.get_container_file_path(container_base_path)}")
+        lines.append("")
+        lines.append(
+            "注意：只要某个技能与本轮任务相关，**务必先 read_file 它的 SKILL.md 并按其指引调用工具**，"
+            "不要跳过技能直接凭记忆回答；技能里写明的接口 / 命令要真正执行，不要假装调用。"
+        )
+        return "\n".join(lines)
+    except Exception:  # noqa: BLE001 — 强化失败绝不能拖垮席位本身的执行
+        logger.debug("build seat skill directive failed for agent %s", agent_id, exc_info=True)
+        return ""
+
+
+def append_seat_skill_directive(agent_id: str | None, task: str) -> str:
+    """把技能识别指令追加到席位本轮任务文本后（无指令时原样返回）。"""
+    directive = build_seat_skill_directive(agent_id)
+    if not directive:
+        return task
+    return f"{task}\n\n---\n\n{directive}"

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

@@ -61,6 +61,8 @@ import httpx
 from fastapi import APIRouter, Body, HTTPException, Request
 from fastapi.responses import StreamingResponse
 
+from app.gateway.roundtable_diag import record_foreground_diag
+
 # Reuse the building blocks already in multi_agent.py — keeps both routers
 # identical in connection handling, payload shape, and error mapping.
 from app.gateway.routers._roundtable_seed import ensure_roundtable_functional_agents
@@ -298,6 +300,11 @@ async def init_intent(request: Request, payload: dict = Body(...)) -> dict[str,
         # 在 !res.ok 分支里读 detail.hint / trace_tail 渲染诊断面板。
         diagnostic = _diagnose_exception(exc, context="intent_init")
         logger.error("intent init failed: %s", diagnostic, exc_info=exc)
+        await record_foreground_diag(
+            request, stage="step1_intent", level="error", event="intent_init_failed",
+            message=f"第一步意图初始化失败：{diagnostic.get('message') or exc}",
+            detail=diagnostic,
+        )
         raise HTTPException(status_code=500, detail=diagnostic) from exc
     return {
         "status": "success",
@@ -561,6 +568,11 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
                         "trace_tail": [],
                         "context": "intent_stream",
                     }
+                await record_foreground_diag(
+                    request, stage="step1_intent", level="error", event="intent_http_error",
+                    message=f"第一步意图生成失败：{error_detail.get('message')}",
+                    detail=error_detail,
+                )
                 err = {
                     "status": "error",
                     "content": error_detail["message"],
@@ -571,6 +583,11 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
             except Exception as exc:
                 diagnostic = _diagnose_exception(exc, context="intent_stream")
                 logger.error("intent stream failed: %s", diagnostic, exc_info=exc)
+                await record_foreground_diag(
+                    request, stage="step1_intent", level="error", event="intent_unexpected",
+                    message=f"第一步意图生成未预期异常：{diagnostic.get('message') or exc}",
+                    detail=diagnostic,
+                )
                 err = {
                     "status": "error",
                     "content": diagnostic["message"],

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

@@ -62,7 +62,9 @@ from fastapi import APIRouter, Body, HTTPException, Request
 from fastapi.responses import StreamingResponse
 from langgraph.checkpoint.base import empty_checkpoint
 
+from app.gateway.roundtable_diag import record_foreground_diag
 from app.gateway.roundtable_run_policy import merge_skill_stop_names, roundtable_run_policy
+from app.gateway.roundtable_seat_skills import append_seat_skill_directive
 from app.gateway.routers._roundtable_seed import (
     COORDINATOR_AGENT_ID,
     DASHBOARD_AGENT_ID,
@@ -80,6 +82,23 @@ logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/multi-agent", tags=["multi-agent"])
 
 
+def _stage_for_agent(agent_type: str, agent_name: str) -> str:
+    """把一次 /run/stream 的 (agent_type, agent_name) 映射成诊断日志的 stage。
+
+    Step3 的四个内置单例（report/summary/dashboard/structure）也走 special 路径，
+    据 agent_name 归到对应 step3_* 阶段；其余 leader→step2_leader、special→step2_seat。
+    """
+    if agent_name == REPORT_AGENT_ID:
+        return "step3_report"
+    if agent_name == SUMMARY_AGENT_ID:
+        return "step3_summary"
+    if agent_name == DASHBOARD_AGENT_ID:
+        return "step3_dashboard"
+    if agent_name == STRUCTURE_AGENT_ID:
+        return "step3_structure"
+    return "step2_leader" if agent_type == "leader" else "step2_seat"
+
+
 # 圆桌（intent / recommend / 席位 / 协调）创建的 thread 都是功能型内部线程,
 # 不是用户的普通对话。打上这个 metadata,前端「最近的对话」列表(recent-chat-list.tsx)
 # 据 `metadata.thread_type` 过滤掉,与 scheduler / bootstrap 系统会话同一套机制。
@@ -892,9 +911,19 @@ async def init_multi_agent(request: Request, payload: dict = Body(...)) -> Strea
             })
         except HTTPException as exc:
             logger.warning("multi-agent init HTTPException: %s %s", exc.status_code, exc.detail)
+            await record_foreground_diag(
+                request, stage="step2_init", level="error", event="init_http_error",
+                message=f"会商初始化失败 [{exc.status_code}]：{exc.detail}",
+                detail={"status_code": exc.status_code, "detail": str(exc.detail)},
+            )
             yield _frame({"event": "error", "detail": f"[{exc.status_code}] {exc.detail}"})
         except Exception as exc:
             logger.exception("multi-agent init unexpected failure")
+            await record_foreground_diag(
+                request, stage="step2_init", level="error", event="init_unexpected",
+                message=f"会商初始化未预期异常：{exc!r}",
+                detail={"type": type(exc).__name__, "error": str(exc)},
+            )
             yield _frame({"event": "error", "detail": f"unexpected: {exc!r}"})
 
     return StreamingResponse(
@@ -922,8 +951,16 @@ async def init_report_thread(request: Request) -> dict[str, str]:
       ``agent_name="roundtable-report"``、``d_agent_thread_id={REPORT_AGENT_ID: thread_id}``)
       跑这一轮,智能体把图表 HTML 写进该线程的 ``outputs/``,再由沙箱预览。
     """
-    ensure_roundtable_functional_agents()
-    thread_id = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
+    try:
+        ensure_roundtable_functional_agents()
+        thread_id = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
+    except Exception as exc:
+        await record_foreground_diag(
+            request, stage="step3_report", level="error", event="thread_init_failed",
+            message=f"创建结果绘制线程失败：{exc}", detail={"type": type(exc).__name__, "error": str(exc)},
+            agent_id=REPORT_AGENT_ID,
+        )
+        raise
     return {"agent_name": REPORT_AGENT_ID, "thread_id": thread_id}
 
 
@@ -942,8 +979,16 @@ async def init_summary_thread(request: Request) -> dict[str, str]:
       跑这一轮(后续问答/增量改**复用同一线程**,不重新 init),智能体把 Markdown 报告
       写进该线程的 ``outputs/方案总结报告.md``,再由沙箱预览。
     """
-    ensure_roundtable_functional_agents()
-    thread_id = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
+    try:
+        ensure_roundtable_functional_agents()
+        thread_id = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
+    except Exception as exc:
+        await record_foreground_diag(
+            request, stage="step3_summary", level="error", event="thread_init_failed",
+            message=f"创建总结报告线程失败：{exc}", detail={"type": type(exc).__name__, "error": str(exc)},
+            agent_id=SUMMARY_AGENT_ID,
+        )
+        raise
     return {"agent_name": SUMMARY_AGENT_ID, "thread_id": thread_id}
 
 
@@ -964,8 +1009,16 @@ async def init_dashboard_thread(request: Request) -> dict[str, str]:
       跑这一轮(后续问答/增量改**复用同一线程**,不重新 init),智能体在对话里输出
       ```report-json,前端用 ``extractReportData`` 抽取后按当前主题渲染大屏。
     """
-    ensure_roundtable_functional_agents()
-    thread_id = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
+    try:
+        ensure_roundtable_functional_agents()
+        thread_id = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
+    except Exception as exc:
+        await record_foreground_diag(
+            request, stage="step3_dashboard", level="error", event="thread_init_failed",
+            message=f"创建大屏数据线程失败：{exc}", detail={"type": type(exc).__name__, "error": str(exc)},
+            agent_id=DASHBOARD_AGENT_ID,
+        )
+        raise
     return {"agent_name": DASHBOARD_AGENT_ID, "thread_id": thread_id}
 
 
@@ -986,8 +1039,16 @@ async def init_structure_thread(request: Request) -> dict[str, str]:
       跑这一轮,把 taskId + 总结报告 + 结构化流程图交给它,由它挑选合适的入库技能(名称
       可能含「六步法」/自行判断适用性)读 SKILL.md 并按其契约调用完成入库。
     """
-    ensure_roundtable_functional_agents()
-    thread_id = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
+    try:
+        ensure_roundtable_functional_agents()
+        thread_id = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
+    except Exception as exc:
+        await record_foreground_diag(
+            request, stage="step3_structure", level="error", event="thread_init_failed",
+            message=f"创建数据入库线程失败：{exc}", detail={"type": type(exc).__name__, "error": str(exc)},
+            agent_id=STRUCTURE_AGENT_ID,
+        )
+        raise
     return {"agent_name": STRUCTURE_AGENT_ID, "thread_id": thread_id}
 
 
@@ -995,6 +1056,16 @@ async def init_structure_thread(request: Request) -> dict[str, str]:
 # 上游 LangGraph SSE 透传（intent / recommend / leader / special 共用）
 # ---------------------------------------------------------------------------
 
+# 回环 /runs/stream 在**响应头阶段**返回这些状态码时,视为「单 worker + 单 sqlite 锁被
+# 并发打满」造成的瞬时失败,退避重试(详见 _stream_upstream 内注释):
+#   409 同 thread 抢占竞态;429 限流;500/503/504 高并发下 start_run 里 DB/checkpoint 写抖动。
+# 不含 400/401/403/404/422 等确定性错误——那些重试也没用,立即上报。
+_UPSTREAM_RETRYABLE_STATUSES = frozenset({409, 429, 500, 502, 503, 504})
+# 重试次数(不含首次)。总尝试 = 1 + _UPSTREAM_MAX_RETRIES。
+_UPSTREAM_MAX_RETRIES = 2
+# 指数退避基数(秒):0.4 → 0.8 → …,足够让在跑的席位 run 让出 sqlite 锁,又不至于明显拖慢。
+_UPSTREAM_RETRY_BASE_DELAY = 0.4
+
 
 async def _stream_upstream(
     client: httpx.AsyncClient,
@@ -1017,36 +1088,74 @@ async def _stream_upstream(
 
     url = f"{base}/api/threads/{thread_id}/runs/stream"
     captured: list[str] = []
-    t_send = time.perf_counter()
-    async with client.stream("POST", url, json=body, headers=headers) as resp:
-        t_headers = time.perf_counter()
-        if resp.status_code != 200:
-            body_text = (await resp.aread()).decode("utf-8", "replace")[:1000]
-            raise HTTPException(
-                status_code=502,
-                detail=f"upstream {url} returned HTTP {resp.status_code}: {body_text}",
-            )
-        agent_for_log = (body.get("context") or {}).get("agent_name") or thread_id
-        logger.info(
-            "[upstream-timing] %s headers_in=%.3fs",
-            agent_for_log,
-            t_headers - t_send,
-        )
-        first_line = True
-        async for line in resp.aiter_lines():
-            if not line:
-                continue
-            if first_line:
-                first_line = False
-                logger.info(
-                    "[upstream-timing] %s first_line=%.3fs (LangGraph runtime startup + LLM first byte)",
-                    agent_for_log,
-                    time.perf_counter() - t_headers,
+    agent_for_log = (body.get("context") or {}).get("agent_name") or thread_id
+
+    # 头阶段非 200 的瞬时重试。
+    # 背景:网关是**单 worker** + checkpointer 是**单连接 aiosqlite**(全进程一把
+    # asyncio.Lock 串行 checkpoint 读写)。圆桌第三步进入时,前端把「方案总结/结果绘制/
+    # 大屏」三个重 run **同一刻并行**打出来,叠加第二步 on_disconnect=continue 仍在后台
+    # 跑的席位 run,瞬时把单锁/单 worker 打满 —— 此时回环子请求 /runs/stream 的 start_run
+    # 里那串 DB / checkpoint 写有概率抛错(或同 thread 抢占竞态),返回非 200,被本函数原样
+    # 包成 502 漏给前端(用户侧表现为"进第三步概率性 502")。
+    # 关键安全前提:这个非 200 发生在**响应头阶段**,本函数**尚未 yield 任何 SSE 行**给下游,
+    # 所以"换条连接重发"是幂等安全的(不会重复透传半截流)。对可恢复的瞬时状态码退避重试几次,
+    # 把绝大多数并发尖峰造成的偶发失败吃掉;非可恢复状态码(如 400/401/403/404)立即按原逻辑
+    # 报错,不浪费重试。一旦进入流式(已 yield 行)就绝不重试。
+    last_status = 0
+    last_body = ""
+    for attempt in range(_UPSTREAM_MAX_RETRIES + 1):
+        t_send = time.perf_counter()
+        async with client.stream("POST", url, json=body, headers=headers) as resp:
+            t_headers = time.perf_counter()
+            if resp.status_code != 200:
+                last_status = resp.status_code
+                last_body = (await resp.aread()).decode("utf-8", "replace")[:1000]
+                if resp.status_code in _UPSTREAM_RETRYABLE_STATUSES and attempt < _UPSTREAM_MAX_RETRIES:
+                    delay = _UPSTREAM_RETRY_BASE_DELAY * (2**attempt)
+                    logger.warning(
+                        "[upstream-retry] %s 回环 %s 返回 HTTP %s,%.2fs 后重试(第 %d/%d 次):%s",
+                        agent_for_log,
+                        url,
+                        resp.status_code,
+                        delay,
+                        attempt + 1,
+                        _UPSTREAM_MAX_RETRIES,
+                        last_body[:200],
+                    )
+                    await asyncio.sleep(delay)
+                    continue  # 重连:换条新连接重发同一 body(尚未 yield 任何行,幂等安全)
+                raise HTTPException(
+                    status_code=502,
+                    detail=f"upstream {url} returned HTTP {resp.status_code}: {last_body}",
                 )
-            yield line, None
-            if line.startswith("data:"):
-                captured.append(line)
-    yield None, captured
+            logger.info(
+                "[upstream-timing] %s headers_in=%.3fs%s",
+                agent_for_log,
+                t_headers - t_send,
+                f" (retried x{attempt})" if attempt else "",
+            )
+            first_line = True
+            async for line in resp.aiter_lines():
+                if not line:
+                    continue
+                if first_line:
+                    first_line = False
+                    logger.info(
+                        "[upstream-timing] %s first_line=%.3fs (LangGraph runtime startup + LLM first byte)",
+                        agent_for_log,
+                        time.perf_counter() - t_headers,
+                    )
+                yield line, None
+                if line.startswith("data:"):
+                    captured.append(line)
+        yield None, captured
+        return
+
+    # 理论上不可达(循环要么 return、要么 raise);留一条兜底防御。
+    raise HTTPException(
+        status_code=502,
+        detail=f"upstream {url} returned HTTP {last_status}: {last_body}",
+    )
 
 
 def _parse_last_messages(captured: list[str]) -> list[dict[str, Any]] | None:
@@ -1677,11 +1786,18 @@ async def _special_run(
     if files:
         _read_tools = {"read_file", "ls", "grep", "view_image"}
         seat_excluded = [t for t in seat_excluded if t not in _read_tools]
+    # 弱模型补偿：仅对真正的研讨席位（seat / seat_parallel；report/summary/dashboard/ingest
+    # 等 Step3 单例不在此列）把其配置技能（名称 + SKILL.md 路径 + 必须先 read_file 再按其
+    # 流程取信息的硬指令）显式拼进本轮任务，避免弱模型无视系统提示里的技能块（见
+    # roundtable_seat_skills 模块说明）。无配置技能时原样返回。
+    seat_message = new_message
+    if role in ("seat", "seat_parallel"):
+        seat_message = append_seat_skill_directive(agent_name, new_message)
     body = _run_payload(
         agent_name=agent_name,
         model_name=model_name,
         thread_id=thread_id,
-        new_message=new_message,
+        new_message=seat_message,
         excluded_tools=seat_excluded,
         skill_stop_names=merge_skill_stop_names(policy.skill_stop_names, skill_stop_names),
         # 席位执行模式覆盖 policy 的 thinking/reasoning（None → 回退 policy 默认）；
@@ -1899,6 +2015,13 @@ async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) ->
                         yield chunk if chunk.endswith("\n") else chunk + "\n"
             except HTTPException as exc:
                 logger.warning("multi-agent stream HTTPException: %s %s", exc.status_code, exc.detail)
+                await record_foreground_diag(
+                    request, stage=_stage_for_agent(agent_type, agent_name), level="error",
+                    event="run_http_error",
+                    message=f"「{agent_name}」运行失败 [{exc.status_code}]：{exc.detail}",
+                    detail={"status_code": exc.status_code, "detail": str(exc.detail), "agent_type": agent_type},
+                    agent_id=agent_name, agent_name=display_name or agent_name,
+                )
                 err_frame = {
                     "status": "error",
                     "content": f"[{exc.status_code}] {exc.detail}",
@@ -1907,6 +2030,13 @@ async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) ->
                 yield f"data: {json.dumps(err_frame, ensure_ascii=False)}\n\n"
             except Exception as exc:
                 logger.exception("multi-agent stream unexpected failure")
+                await record_foreground_diag(
+                    request, stage=_stage_for_agent(agent_type, agent_name), level="error",
+                    event="run_unexpected",
+                    message=f"「{agent_name}」运行未预期异常：{exc!r}",
+                    detail={"type": type(exc).__name__, "error": str(exc), "agent_type": agent_type},
+                    agent_id=agent_name, agent_name=display_name or agent_name,
+                )
                 err_frame = {
                     "status": "error",
                     "content": f"unexpected: {exc!r}",

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

@@ -47,6 +47,7 @@ import httpx
 from fastapi import APIRouter, Body, HTTPException, Request
 from fastapi.responses import StreamingResponse
 
+from app.gateway.roundtable_diag import record_foreground_diag
 from app.gateway.routers._roundtable_seed import ensure_roundtable_functional_agents
 from app.gateway.routers.multi_agent import (
     _ROUNDTABLE_THREAD_METADATA,
@@ -320,10 +321,20 @@ async def stream_recommend(request: Request, payload: dict = Body(...)) -> Strea
 
             except HTTPException as exc:
                 logger.warning("recommend stream HTTPException: %s %s", exc.status_code, exc.detail)
+                await record_foreground_diag(
+                    request, stage="step2_recommend", level="error", event="recommend_http_error",
+                    message=f"智能体推荐失败 [{exc.status_code}]：{exc.detail}",
+                    detail={"status_code": exc.status_code, "detail": str(exc.detail)},
+                )
                 err = {"status": "error", "content": f"[{exc.status_code}] {exc.detail}", "picks": None}
                 yield f"data: {json.dumps(err, ensure_ascii=False)}\n\n"
             except Exception as exc:
                 logger.exception("recommend stream unexpected failure")
+                await record_foreground_diag(
+                    request, stage="step2_recommend", level="error", event="recommend_unexpected",
+                    message=f"智能体推荐未预期异常：{exc!r}",
+                    detail={"type": type(exc).__name__, "error": str(exc)},
+                )
                 err = {"status": "error", "content": f"unexpected: {exc!r}", "picks": None}
                 yield f"data: {json.dumps(err, ensure_ascii=False)}\n\n"
 

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

@@ -43,6 +43,11 @@ class ChainSeatSchema(BaseModel):
     agent_id: str = Field(min_length=1, max_length=128)
     name: str = ""
     description: str = ""
+    # 可选「席位模型覆盖」：该席位作为子智能体执行时用的 LLM model name（须是 config
+    # models[] 里的一个 name）。None/"" = 不覆盖，运行时回退到全局所选模型。整条 seats
+    # 数组以 JSON 透传持久化（见 persistence/roundtable_chains），故新增此字段无需迁移；
+    # 仅需在此 schema 显式声明，否则 model_dump() 会把未知字段剥掉。
+    model: str | None = Field(default=None, max_length=128)
 
 
 class ChainResponse(BaseModel):

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

@@ -0,0 +1,173 @@
+"""圆桌会商诊断日志查询 API（管理员）。
+
+会商全流程（第一/二/三步 + 后台执行器 + 进程内网关）各处的报错与关键事件落在
+``roundtable_diagnostics`` 表，本路由暴露**管理员**筛选查询 + facets + 清理。
+
+Routes（prefix ``/api/roundtable-diagnostics``，全部 admin-gated）::
+    GET    ""            筛选 + 分页列出（scope/stage/level/event/job_id/draft_id/task_id/user_id/q/since/until）
+    GET    "/facets"     各筛选维度的 distinct 取值（下拉用）
+    POST   "/admin/cleanup"  按保留天数清理旧日志
+"""
+
+from __future__ import annotations
+
+import time
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from fastapi import APIRouter, HTTPException, Query, Request
+from pydantic import BaseModel, Field
+
+router = APIRouter(prefix="/api/roundtable-diagnostics", tags=["roundtable-diagnostics"])
+
+
+def _get_store(request: Request):
+    store = getattr(request.app.state, "roundtable_diagnostic_store", None)
+    if store is None:
+        raise HTTPException(status_code=503, detail="Roundtable diagnostic store not available")
+    return store
+
+
+async def _require_admin(request: Request) -> None:
+    """放行 admin / auth-disabled，其余 403（与 roundtable_jobs / task_buttons 同套语义）。"""
+    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="会商诊断日志仅限管理员查看")
+
+
+class DiagnosticItem(BaseModel):
+    id: str
+    scope: str
+    stage: str
+    level: str
+    event: str
+    message: str | None = None
+    detail: Any = None
+    jobId: str | None = None
+    draftId: str | None = None
+    taskId: str | None = None
+    userId: str | None = None
+    agentId: str | None = None
+    agentName: str | None = None
+    cycle: int | None = None
+    createdAt: str | None = None
+
+
+class DiagnosticListResponse(BaseModel):
+    items: list[DiagnosticItem]
+    total: int
+    limit: int
+    offset: int
+
+
+class FacetsResponse(BaseModel):
+    scope: list[str] = Field(default_factory=list)
+    stage: list[str] = Field(default_factory=list)
+    level: list[str] = Field(default_factory=list)
+    event: list[str] = Field(default_factory=list)
+
+
+class CleanupRequest(BaseModel):
+    retention_days: int = Field(default=30, ge=0)
+    batch_size: int = Field(default=1000, ge=1, le=10000)
+
+
+class CleanupResponse(BaseModel):
+    deleted: int
+    elapsed_seconds: float
+
+
+def _to_item(row: dict[str, Any]) -> DiagnosticItem:
+    return DiagnosticItem(
+        id=row["id"],
+        scope=row.get("scope") or "",
+        stage=row.get("stage") or "",
+        level=row.get("level") or "",
+        event=row.get("event") or "",
+        message=row.get("message"),
+        detail=row.get("detail"),
+        jobId=row.get("job_id"),
+        draftId=row.get("draft_id"),
+        taskId=row.get("task_id"),
+        userId=row.get("user_id"),
+        agentId=row.get("agent_id"),
+        agentName=row.get("agent_name"),
+        cycle=row.get("cycle"),
+        createdAt=row.get("created_at"),
+    )
+
+
+def _parse_dt(raw: str | None) -> datetime | None:
+    if not raw:
+        return None
+    try:
+        # 容忍前端传的 ``...Z`` 后缀（datetime.fromisoformat 在 3.11+ 接受，旧版不接受）。
+        return datetime.fromisoformat(raw.replace("Z", "+00:00"))
+    except Exception:
+        return None
+
+
+@router.get("", response_model=DiagnosticListResponse)
+async def list_diagnostics(
+    request: Request,
+    scope: str | None = None,
+    stage: str | None = None,
+    level: str | None = None,
+    event: str | None = None,
+    job_id: str | None = None,
+    draft_id: str | None = None,
+    task_id: str | None = None,
+    user_id: str | None = None,
+    q: str | None = None,
+    since: str | None = None,
+    until: str | None = None,
+    limit: int = Query(default=100, ge=1, le=500),
+    offset: int = Query(default=0, ge=0),
+) -> DiagnosticListResponse:
+    await _require_admin(request)
+    store = _get_store(request)
+    items, total = await store.list_logs(
+        scope=scope or None,
+        stage=stage or None,
+        level=level or None,
+        event=event or None,
+        job_id=job_id or None,
+        draft_id=draft_id or None,
+        task_id=task_id or None,
+        user_id=user_id or None,
+        q=q or None,
+        since=_parse_dt(since),
+        until=_parse_dt(until),
+        limit=limit,
+        offset=offset,
+    )
+    return DiagnosticListResponse(items=[_to_item(r) for r in items], total=total, limit=limit, offset=offset)
+
+
+@router.get("/facets", response_model=FacetsResponse)
+async def get_facets(request: Request) -> FacetsResponse:
+    await _require_admin(request)
+    store = _get_store(request)
+    facets = await store.facets()
+    return FacetsResponse(**facets)
+
+
+@router.post("/admin/cleanup", response_model=CleanupResponse)
+async def cleanup(request: Request, body: CleanupRequest) -> CleanupResponse:
+    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, 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))

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

@@ -101,6 +101,11 @@ async def run_orchestration(
     intent = (intent_text or seed_message or "").strip()
     try:
         await progress.begin()
+        await progress.diag(
+            stage="step2_orchestration", level="info", event="begin",
+            message=f"会商编排启动（mode={mode}，{len(agents)} 个席位）",
+            detail={"mode": mode, "agents": [a.agent_id for a in agents]},
+        )
         dag_plan = parse_dag_plan(plan) if mode == "dag" else None
         if mode == "dag" and dag_plan is not None:
             outcome = await _run_dag(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message, intent, dag_plan)
@@ -112,6 +117,11 @@ async def run_orchestration(
 
         if outcome.kind == "clarification":
             question = outcome.question or "总控需要补充信息，请补充后继续。"
+            await progress.diag(
+                stage="step2_leader", level="info", event="clarification",
+                message="总控发起澄清，会商暂停等待补充信息",
+                detail={"question": question[:500]},
+            )
             await progress.awaiting_input(question)
             return OrchestrationResult(
                 status="awaiting_input",
@@ -120,8 +130,17 @@ async def run_orchestration(
                 dialogues=progress.dialogues(),
             )
 
+        if outcome.kind == "max_cycles":
+            await progress.diag(
+                stage="step2_orchestration", level="warning", event="max_cycles",
+                message=f"达到最大研讨轮次（{MAX_CYCLES}），强制收口——可能未真正达成共识",
+            )
         # consensus 或 max_cycles：均收口。Step3（总结报告 + 结果绘制）按 enable_report 开关执行。
         await progress.consensus(100)
+        await progress.diag(
+            stage="step2_orchestration", level="info", event="consensus",
+            message="研讨收口，进入第三步" if enable_report else "研讨收口（未启用第三步报告）",
+        )
         step3 = None
         if enable_report:
             step3 = await _run_step3(
@@ -144,6 +163,11 @@ async def run_orchestration(
         )
     except Exception as exc:  # noqa: BLE001 — 顶层兜底：标记 error 并返回
         logger.exception("roundtable orchestration failed")
+        await progress.diag(
+            stage="step2_orchestration", level="error", event="orchestration_crash",
+            message=f"会商编排顶层异常：{exc}",
+            detail={"type": type(exc).__name__, "error": str(exc)},
+        )
         await progress.error(str(exc))
         return OrchestrationResult(status="error", error=str(exc), dialogues=progress.dialogues())
 
@@ -229,6 +253,12 @@ async def _run_recommend(
             seat_msg = _seat_message(intent_text, deliveries, name_by_id, task)
             seat = await gateway.run_seat(thread_ids=thread_ids, agent_id=agent_id, task=seat_msg, model=model)
             deliveries[agent_id] = seat.final_text
+            if not (seat.final_text or "").strip():
+                await progress.diag(
+                    stage="step2_seat", level="warning", event="seat_empty_delivery",
+                    message=f"席位「{name_by_id.get(agent_id, agent_id)}」本轮交付为空（可能模型未产出或被截断）",
+                    agent_id=agent_id, agent_name=name_by_id.get(agent_id, agent_id), cycle=cycle,
+                )
             await progress.add_seat_dialogue(
                 agent_id, name_by_id.get(agent_id, agent_id), seat.final_text, cycle,
                 tool_calls=seat.tool_calls, thread_id=thread_ids.get(agent_id),
@@ -298,6 +328,12 @@ async def _run_chain(
         seat_msg = _seat_message(intent_text, deliveries, name_by_id, task)
         seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=seat_msg, model=model)
         deliveries[seat.agent_id] = seat_turn.final_text
+        if not (seat_turn.final_text or "").strip():
+            await progress.diag(
+                stage="step2_seat", level="warning", event="seat_empty_delivery",
+                message=f"席位「{seat.name}」本轮交付为空（可能模型未产出或被截断）",
+                agent_id=seat.agent_id, agent_name=seat.name, cycle=i + 1,
+            )
         await progress.add_seat_dialogue(
             seat.agent_id, seat.name, seat_turn.final_text, i + 1,
             tool_calls=seat_turn.tool_calls, thread_id=thread_ids.get(seat.agent_id),
@@ -398,6 +434,12 @@ async def _run_dag(
             seat_msg = _seat_message(intent_text, deliveries, name_by_id, task)
             seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=seat_msg, model=model)
             deliveries[seat.agent_id] = seat_turn.final_text
+            if not (seat_turn.final_text or "").strip():
+                await progress.diag(
+                    stage="step2_seat", level="warning", event="seat_empty_delivery",
+                    message=f"席位「{seat.name}」本轮交付为空（可能模型未产出或被截断）",
+                    agent_id=seat.agent_id, agent_name=seat.name, cycle=i + 1,
+                )
             await progress.add_seat_dialogue(
                 seat.agent_id, seat.name, seat_turn.final_text, i + 1,
                 tool_calls=seat_turn.tool_calls, thread_id=thread_ids.get(seat.agent_id),
@@ -477,6 +519,11 @@ async def _run_step3(
                 }
         except Exception as exc:  # noqa: BLE001 — 总结失败不阻塞结果绘制
             logger.exception("roundtable step3 summary failed")
+            await progress.diag(
+                stage="step3_summary", level="error", event="summary_failed",
+                message=f"第三步「方案总结报告」生成失败：{exc}",
+                detail={"type": type(exc).__name__, "error": str(exc)},
+            )
             errors.append(exc)
 
     async def _report_task() -> None:
@@ -494,6 +541,11 @@ async def _run_step3(
             step3.update(report)
         except Exception as exc:  # noqa: BLE001 — 绘制失败时保留已写出的总结报告
             logger.exception("roundtable step3 report failed")
+            await progress.diag(
+                stage="step3_report", level="error", event="report_failed",
+                message=f"第三步「结果绘制」(方案总览.html) 生成失败：{exc}",
+                detail={"type": type(exc).__name__, "error": str(exc)},
+            )
             errors.append(exc)
 
     async def _dashboard_task() -> None:
@@ -510,6 +562,11 @@ async def _run_step3(
                 step3["dashboardJson"] = data_json
         except Exception as exc:  # noqa: BLE001 — 大屏失败不拖垮另两段
             logger.exception("roundtable step3 dashboard failed")
+            await progress.diag(
+                stage="step3_dashboard", level="error", event="dashboard_failed",
+                message=f"第三步「大屏数据」(report-json) 生成失败：{exc}",
+                detail={"type": type(exc).__name__, "error": str(exc)},
+            )
             errors.append(exc)
 
     await asyncio.gather(_summary_task(), _report_task(), _dashboard_task())

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

@@ -53,11 +53,15 @@ class JobProgressWriter:
         agents: list[SeatRef],
         event_sink: EventSink = None,
         on_dialogue: Any | None = None,
+        recorder: Any | None = None,
     ) -> None:
         self._store = store
         self._job_id = job_id
         self._agents = list(agents)
         self._event_sink = event_sink
+        # 诊断日志记录器（DiagnosticsRecorder，已绑定本作业上下文）。引擎经
+        # ``await progress.diag(...)`` 在每个报错 / 关键转折处落一条事件。None = 不记录。
+        self._recorder = recorder
         # 每轮 leader/seat 模型返回后回调（dialogues, consensus）—— 执行器据此把对话
         # 写进草稿（和手动跑同一存取路径）。
         self._on_dialogue = on_dialogue
@@ -105,6 +109,35 @@ class JobProgressWriter:
             except Exception:  # noqa: BLE001
                 logger.warning("roundtable job progress event sink failed", exc_info=True)
 
+    async def diag(
+        self,
+        *,
+        stage: str,
+        level: str,
+        event: str,
+        message: str | None = None,
+        detail: Any = None,
+        agent_id: str | None = None,
+        agent_name: str | None = None,
+        cycle: int | None = None,
+    ) -> None:
+        """落一条诊断日志（best-effort；无 recorder 时 no-op）。"""
+        if self._recorder is None:
+            return
+        try:
+            await self._recorder.record(
+                stage=stage,
+                level=level,
+                event=event,
+                message=message,
+                detail=detail,
+                agent_id=agent_id,
+                agent_name=agent_name,
+                cycle=cycle,
+            )
+        except Exception:  # noqa: BLE001 — 诊断日志绝不打断编排
+            logger.debug("roundtable diag record failed", exc_info=True)
+
     # ── 语义方法（引擎调用）─────────────────────────────────────────────────
 
     async def begin(self) -> None:

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260619_01_roundtable_diagnostics.py

@@ -0,0 +1,70 @@
+"""Create roundtable_diagnostics table.
+
+会商全流程诊断日志（第一/二/三步 + 后台执行器 + 进程内网关各处报错与关键事件），
+管理员诊断页的数据源。全局/共享，不按 user 分权。
+
+On a brand-new DB the table is created by ``create_all``; this revision patches
+existing DBs (alembic auto-upgrade is disabled at runtime — see
+``persistence/engine.py`` — so operators run this / the column_additions SQL).
+
+Revision ID: 20260619_01
+Revises: 20260618_01
+Create Date: 2026-06-19
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+from sqlalchemy.dialects.mysql import LONGTEXT
+
+
+revision = "20260619_01"
+down_revision = "20260618_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_diagnostics"):
+        return
+
+    long_text = LONGTEXT() if op.get_bind().dialect.name == "mysql" else sa.Text()
+
+    op.create_table(
+        "roundtable_diagnostics",
+        sa.Column("id", sa.String(64), primary_key=True),
+        sa.Column("scope", sa.String(16), nullable=False, server_default="background"),
+        sa.Column("stage", sa.String(32), nullable=False, server_default="job"),
+        sa.Column("level", sa.String(16), nullable=False, server_default="info"),
+        sa.Column("event", sa.String(64), nullable=False, server_default=""),
+        sa.Column("message", sa.Text(), nullable=True),
+        sa.Column("detail", long_text, nullable=True),
+        sa.Column("job_id", sa.String(64), nullable=True),
+        sa.Column("draft_id", sa.String(64), nullable=True),
+        sa.Column("task_id", sa.String(128), nullable=True),
+        sa.Column("user_id", sa.String(64), nullable=True),
+        sa.Column("agent_id", sa.String(128), nullable=True),
+        sa.Column("agent_name", sa.String(255), nullable=True),
+        sa.Column("cycle", sa.Integer(), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+    )
+    op.create_index("ix_roundtable_diagnostics_scope", "roundtable_diagnostics", ["scope"])
+    op.create_index("ix_roundtable_diagnostics_stage", "roundtable_diagnostics", ["stage"])
+    op.create_index("ix_roundtable_diagnostics_level", "roundtable_diagnostics", ["level"])
+    op.create_index("ix_roundtable_diagnostics_job_id", "roundtable_diagnostics", ["job_id"])
+    op.create_index("ix_roundtable_diagnostics_draft_id", "roundtable_diagnostics", ["draft_id"])
+    op.create_index("ix_roundtable_diagnostics_task_id", "roundtable_diagnostics", ["task_id"])
+    op.create_index("ix_roundtable_diagnostics_user_id", "roundtable_diagnostics", ["user_id"])
+    op.create_index("ix_roundtable_diagnostics_created_at", "roundtable_diagnostics", ["created_at"])
+
+
+def downgrade() -> None:
+    if _has_table("roundtable_diagnostics"):
+        op.drop_table("roundtable_diagnostics")

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

@@ -54,6 +54,7 @@ from deerflow.persistence.roundtable_drafts.model import (
     RoundtableDraftRow,
     RoundtableRecommendHistoryRow,
 )
+from deerflow.persistence.roundtable_diagnostics.model import RoundtableDiagnosticRow
 from deerflow.persistence.roundtable_jobs.model import RoundtableJobRow
 from deerflow.persistence.roundtable_task_drafts.model import RoundtableTaskDraftRow
 from deerflow.persistence.run.model import RunRow
@@ -108,6 +109,7 @@ __all__ = [
     "RoundtableChainRow",
     "RoundtableDraftRow",
     "RoundtableDraftShareRow",
+    "RoundtableDiagnosticRow",
     "RoundtableJobRow",
     "RoundtableRecommendHistoryRow",
     "RoundtableTaskDraftRow",

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

@@ -0,0 +1,112 @@
+"""圆桌会商诊断日志持久化。
+
+对外暴露：
+- ``make_roundtable_diagnostic_store(sf)`` —— 建仓储（deps.py 启动时调）。
+- ``register_default_store`` / ``get_default_store`` —— 进程级默认仓储（跨 app/harness 边界，
+  与 ``roundtable_jobs`` 同套做法：harness 侧编排引擎拿不到 FastAPI ``app.state``）。
+- ``record_diagnostic(...)`` —— **best-effort** 写一条事件（store 为 None / 出错均静默吞掉，
+  诊断日志本身绝不能反过来打断会商）。
+- ``DiagnosticsRecorder`` —— 绑定一次作业上下文（scope/job_id/draft_id/task_id/user_id）的
+  便捷记录器，引擎 / 网关 / 执行器逐处调用 ``await recorder.record(stage=..., level=..., ...)``。
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from deerflow.persistence.roundtable_diagnostics.model import RoundtableDiagnosticRow
+from deerflow.persistence.roundtable_diagnostics.sql import RoundtableDiagnosticRepository
+
+logger = logging.getLogger(__name__)
+
+
+def make_roundtable_diagnostic_store(session_factory) -> RoundtableDiagnosticRepository | None:
+    if session_factory is None:
+        return None
+    return RoundtableDiagnosticRepository(session_factory)
+
+
+# ── 进程级默认仓储（跨 app/harness 边界）──────────────────────────────────────────
+_default_store: RoundtableDiagnosticRepository | None = None
+
+
+def register_default_store(store: RoundtableDiagnosticRepository | None) -> None:
+    global _default_store
+    _default_store = store
+
+
+def get_default_store() -> RoundtableDiagnosticRepository | None:
+    return _default_store
+
+
+async def record_diagnostic(store: RoundtableDiagnosticRepository | None = None, **fields: Any) -> None:
+    """Best-effort 写一条诊断事件。``store`` 缺省用进程级默认仓储。任何异常静默吞掉。"""
+    repo = store or _default_store
+    if repo is None:
+        return
+    try:
+        await repo.record(**fields)
+    except Exception:  # noqa: BLE001 — 诊断日志写失败绝不打断会商
+        logger.debug("roundtable diagnostic record failed", exc_info=True)
+
+
+class DiagnosticsRecorder:
+    """绑定一次会商上下文的记录器。所有 ``record`` 均 best-effort。"""
+
+    def __init__(
+        self,
+        store: RoundtableDiagnosticRepository | None = None,
+        *,
+        scope: str = "background",
+        job_id: str | None = None,
+        draft_id: str | None = None,
+        task_id: str | None = None,
+        user_id: str | None = None,
+    ) -> None:
+        self._store = store
+        self._scope = scope
+        self._job_id = job_id
+        self._draft_id = draft_id
+        self._task_id = task_id
+        self._user_id = user_id
+
+    async def record(
+        self,
+        *,
+        stage: str,
+        level: str,
+        event: str,
+        message: str | None = None,
+        detail: Any = None,
+        agent_id: str | None = None,
+        agent_name: str | None = None,
+        cycle: int | None = None,
+    ) -> None:
+        await record_diagnostic(
+            self._store,
+            scope=self._scope,
+            stage=stage,
+            level=level,
+            event=event,
+            message=message,
+            detail=detail,
+            job_id=self._job_id,
+            draft_id=self._draft_id,
+            task_id=self._task_id,
+            user_id=self._user_id,
+            agent_id=agent_id,
+            agent_name=agent_name,
+            cycle=cycle,
+        )
+
+
+__all__ = [
+    "RoundtableDiagnosticRow",
+    "RoundtableDiagnosticRepository",
+    "DiagnosticsRecorder",
+    "make_roundtable_diagnostic_store",
+    "register_default_store",
+    "get_default_store",
+    "record_diagnostic",
+]

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

@@ -0,0 +1,52 @@
+"""圆桌会商「诊断日志」ORM 模型。
+
+记录会商全流程（第一步意图 / 第二步研讨 / 第三步报告 + 后台挂起执行器 / 进程内网关）
+各处可能发生的报错与关键事件，供管理员在前端诊断页筛选排查。
+
+设计要点：
+- **全局/共享**：不按 user 分权（``user_id`` 仅审计），管理员读全量；写入开放（best-effort）。
+- 一行 = 一条事件。``detail`` 用 ``PortableLongText`` 存结构化补充信息（堆栈尾、上下文）。
+- 按 ``created_at`` / ``stage`` / ``level`` / ``job_id`` / ``task_id`` 建索引，支撑筛选。
+"""
+
+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 RoundtableDiagnosticRow(Base):
+    __tablename__ = "roundtable_diagnostics"
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    # background（后台挂起）| foreground（前台实时会商）。
+    scope: Mapped[str] = mapped_column(String(16), nullable=False, default="background", index=True)
+    # 阶段：step1_intent / step2_init / step2_leader / step2_seat / step2_orchestration /
+    # step3_summary / step3_report / step3_dashboard / step3_structure / upstream / job。
+    stage: Mapped[str] = mapped_column(String(32), nullable=False, default="job", index=True)
+    # 级别：info | warning | error。
+    level: Mapped[str] = mapped_column(String(16), nullable=False, default="info", index=True)
+    # 短事件码（如 seat_turn_timeout / model_error / report_truncated / dispatch_invalid）。
+    event: Mapped[str] = mapped_column(String(64), nullable=False, default="")
+    # 人类可读的报错说明（中文）。
+    message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # 结构化补充：堆栈尾、上下文、原始 body 片段等。
+    detail: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+
+    # 关联上下文（均可空）。
+    job_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    draft_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    task_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    user_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    agent_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    agent_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    cycle: Mapped[int | None] = mapped_column(Integer, nullable=True)
+
+    created_at: Mapped[datetime] = mapped_column(
+        BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), index=True
+    )

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

@@ -0,0 +1,206 @@
+"""SQLAlchemy-backed 圆桌会商诊断日志 store。
+
+写入（``record``）best-effort、全局共享；读取（``list_logs`` / ``count`` / ``facets``）
+带多维筛选，供管理员诊断页用。``detail`` 以 JSON 序列化进 ``PortableLongText``。
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from typing import Any
+from uuid import uuid4
+
+from sqlalchemy import delete, func, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.roundtable_diagnostics.model import RoundtableDiagnosticRow
+
+
+def _dumps(value: Any) -> str | None:
+    if value is None:
+        return None
+    if isinstance(value, str):
+        return value
+    try:
+        return json.dumps(value, ensure_ascii=False, default=str)
+    except Exception:
+        return str(value)
+
+
+def _loads(value: Any) -> Any:
+    if not isinstance(value, str) or not value:
+        return value or None
+    try:
+        return json.loads(value)
+    except Exception:
+        return value  # 非 JSON 文本（如纯字符串 detail）原样返回
+
+
+class RoundtableDiagnosticRepository:
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    @staticmethod
+    def _to_dict(row: RoundtableDiagnosticRow) -> dict[str, Any]:
+        return {
+            "id": row.id,
+            "scope": row.scope,
+            "stage": row.stage,
+            "level": row.level,
+            "event": row.event,
+            "message": row.message,
+            "detail": _loads(row.detail),
+            "job_id": row.job_id,
+            "draft_id": row.draft_id,
+            "task_id": row.task_id,
+            "user_id": row.user_id,
+            "agent_id": row.agent_id,
+            "agent_name": row.agent_name,
+            "cycle": row.cycle,
+            "created_at": row.created_at.isoformat() if isinstance(row.created_at, datetime) else row.created_at,
+        }
+
+    async def record(
+        self,
+        *,
+        scope: str = "background",
+        stage: str = "job",
+        level: str = "info",
+        event: str = "",
+        message: str | None = None,
+        detail: Any = None,
+        job_id: str | None = None,
+        draft_id: str | None = None,
+        task_id: str | None = None,
+        user_id: str | None = None,
+        agent_id: str | None = None,
+        agent_name: str | None = None,
+        cycle: int | None = None,
+    ) -> dict[str, Any]:
+        """插入一条诊断事件，返回序列化 dict。"""
+        row = RoundtableDiagnosticRow(
+            id=uuid4().hex,
+            scope=(scope or "background")[:16],
+            stage=(stage or "job")[:32],
+            level=(level or "info")[:16],
+            event=(event or "")[:64],
+            message=message,
+            detail=_dumps(detail),
+            job_id=job_id,
+            draft_id=draft_id,
+            task_id=task_id,
+            user_id=user_id,
+            agent_id=(agent_id or None) if agent_id is None else str(agent_id)[:128],
+            agent_name=(agent_name or None) if agent_name is None else str(agent_name)[:255],
+            cycle=cycle,
+        )
+        async with self._sf() as session:
+            session.add(row)
+            await session.commit()
+            await session.refresh(row)
+            return self._to_dict(row)
+
+    def _apply_filters(
+        self,
+        stmt,
+        *,
+        scope: str | None,
+        stage: str | None,
+        level: str | None,
+        event: str | None,
+        job_id: str | None,
+        draft_id: str | None,
+        task_id: str | None,
+        user_id: str | None,
+        q: str | None,
+        since: datetime | None,
+        until: datetime | None,
+    ):
+        R = RoundtableDiagnosticRow
+        if scope:
+            stmt = stmt.where(R.scope == scope)
+        if stage:
+            stmt = stmt.where(R.stage == stage)
+        if level:
+            stmt = stmt.where(R.level == level)
+        if event:
+            stmt = stmt.where(R.event == event)
+        if job_id:
+            stmt = stmt.where(R.job_id == job_id)
+        if draft_id:
+            stmt = stmt.where(R.draft_id == draft_id)
+        if task_id:
+            stmt = stmt.where(R.task_id == task_id)
+        if user_id:
+            stmt = stmt.where(R.user_id == user_id)
+        if q:
+            like = f"%{q}%"
+            stmt = stmt.where(R.message.ilike(like))
+        if since is not None:
+            stmt = stmt.where(R.created_at >= since)
+        if until is not None:
+            stmt = stmt.where(R.created_at <= until)
+        return stmt
+
+    async def list_logs(
+        self,
+        *,
+        scope: str | None = None,
+        stage: str | None = None,
+        level: str | None = None,
+        event: str | None = None,
+        job_id: str | None = None,
+        draft_id: str | None = None,
+        task_id: str | None = None,
+        user_id: str | None = None,
+        q: str | None = None,
+        since: datetime | None = None,
+        until: datetime | None = None,
+        limit: int = 100,
+        offset: int = 0,
+    ) -> tuple[list[dict[str, Any]], int]:
+        """筛选 + 分页（newest-first）。返回 ``(items, total)``。"""
+        R = RoundtableDiagnosticRow
+        base = self._apply_filters(
+            select(R),
+            scope=scope, stage=stage, level=level, event=event,
+            job_id=job_id, draft_id=draft_id, task_id=task_id, user_id=user_id,
+            q=q, since=since, until=until,
+        )
+        count_stmt = self._apply_filters(
+            select(func.count()).select_from(R),
+            scope=scope, stage=stage, level=level, event=event,
+            job_id=job_id, draft_id=draft_id, task_id=task_id, user_id=user_id,
+            q=q, since=since, until=until,
+        )
+        page_stmt = base.order_by(R.created_at.desc(), R.id.desc()).limit(limit).offset(offset)
+        async with self._sf() as session:
+            total = int((await session.execute(count_stmt)).scalar() or 0)
+            rows = (await session.execute(page_stmt)).scalars().all()
+            return [self._to_dict(r) for r in rows], total
+
+    async def facets(self) -> dict[str, list[str]]:
+        """返回各筛选维度的 distinct 取值（供前端下拉），供筛选 UI 用。"""
+        R = RoundtableDiagnosticRow
+        out: dict[str, list[str]] = {}
+        async with self._sf() as session:
+            for key, col in (("scope", R.scope), ("stage", R.stage), ("level", R.level), ("event", R.event)):
+                rows = (await session.execute(select(col).distinct())).scalars().all()
+                out[key] = sorted([str(v) for v in rows if v])
+        return out
+
+    async def list_older_than(self, cutoff: datetime, *, limit: int = 1000) -> list[str]:
+        R = RoundtableDiagnosticRow
+        stmt = select(R.id).where(R.created_at < cutoff).limit(limit)
+        async with self._sf() as session:
+            return [str(v) for v in (await session.execute(stmt)).scalars().all()]
+
+    async def delete_by_ids(self, ids: list[str]) -> int:
+        if not ids:
+            return 0
+        R = RoundtableDiagnosticRow
+        async with self._sf() as session:
+            result = await session.execute(delete(R).where(R.id.in_(ids)))
+            await session.commit()
+            return int(result.rowcount or 0)