fix(roundtable): 优化 Step2 总控派活稳定性与并发编排

限制总控工具白名单、关闭圆桌记忆注入、综合轮 synthesis_mode 与编排 epoch 防双发，并优化广播异步化及 Postgres checkpointer 连接池。

Co-authored-by: Cursor <cursoragent@cursor.com>

frontend-web/src/roundtable-planning/api/multi-agent.ts

@@ -203,6 +203,12 @@ export interface StreamMultiAgentRequest {
   newMessage: string;
   model?: string;
   skillStopNames?: string[];
+  /**
+   * 仅 leader 用。置 true 表示这是「派活后让总控判断继续/综合」的轮次:后端会把各已交付
+   * 席位的**完整交付正文**额外注入总控输入,供其写最终方案。派活决策轮(首轮等)传 false/省略,
+   * 后端只给摘要,使总控上下文精简、响应快。
+   */
+  synthesisMode?: boolean;
   signal?: AbortSignal;
   /** Fires for every non-status data frame the upstream emits. */
   onUpstreamFrame?: (frame: unknown) => void;
@@ -415,6 +421,8 @@ export async function streamMultiAgent(
       new_message: req.newMessage,
       model: req.model,
       skill_stop_names: req.skillStopNames ?? [],
+      // 仅 leader 用;综合轮让后端注入各席位完整交付正文。省略 → 后端按 false 处理。
+      synthesis_mode: req.synthesisMode ?? false,
     }),
     signal: req.signal,
   });

frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts

@@ -381,6 +381,12 @@ export function useStep2Orchestration(
 
   // ─── Refs ────────────────────────────────────────────────────────
   const orchestrationCancelRef = useRef<boolean>(false);
+  // 编排「世代号」。每次启动一轮编排(runOrchestration / runChainOrchestration)就 +1。
+  // 旧循环捕获自己的 epoch,在每个取消检查点比对 —— 一旦 epoch 变了(说明已有新循环
+  // 接管,例如人工干预),旧循环立即退出。这能根治「新循环把 orchestrationCancelRef
+  // 重置为 false 反而把正在退出的旧循环『反取消』,导致旧循环继续又发一个 leader run」
+  // 的并发双发 bug:epoch 是单调递增的,新循环无法把它『重置』回旧循环的值。
+  const orchestrationEpochRef = useRef<number>(0);
   const step2InitRef = useRef<boolean>(false);
   const activeAbortersRef = useRef<Set<AbortController>>(new Set());
   /** 干预研审:恢复会商后待执行的 leader prompt。 */
@@ -993,17 +999,26 @@ export function useStep2Orchestration(
   const runOrchestration = useCallback(
     async (seedMessage: string, ids: ThreadIdMap, coord: string) => {
       if (isOrchestratingRef.current) return;
+      // **同步**抢占编排锁。isOrchestratingRef 平时由 [isOrchestrating] 的 effect 异步
+      // 回写(见上方),滞后一拍;若只靠它,人工干预里 setIsRoundTableRunning(true) 触发的
+      // 「恢复/干预」effect 会在 ref 尚未变 true 的同一 tick 内再调一次本函数 —— 两个 leader
+      // run 并发打到同一总控 thread(rollback 抢占 → 截断的那个旧报 502、存活的那个重复出现)。
+      // 这里立刻把 ref 置 true,让重入守卫即时生效,堵住二次触发。
+      isOrchestratingRef.current = true;
       setIsOrchestrating(true);
       orchestrationCancelRef.current = false;
+      // 认领本轮 epoch。任何更晚启动的循环都会让 epoch 自增 → 本循环 isStale() 立即为真。
+      const myEpoch = (orchestrationEpochRef.current += 1);
+      // 单一「该退出了吗」判定:被显式取消,**或**已被更新的循环接管(epoch 变了)。
+      // 后者专治新循环重置 orchestrationCancelRef=false 把旧循环反取消的并发双发。
+      const isStale = () =>
+        orchestrationCancelRef.current || orchestrationEpochRef.current !== myEpoch;
 
       let nextLeaderTask = seedMessage;
       let safetyCounter = 0;
 
       try {
-        while (
-          !orchestrationCancelRef.current &&
-          safetyCounter < MAX_CYCLES
-        ) {
+        while (!isStale() && safetyCounter < MAX_CYCLES) {
           safetyCounter += 1;
           // Leader(协调智能体)正在派活,**不应该**让任何 sub agent 显示"研讨中"——
           // 协调者本身不在右侧"参与角色"面板里(COMMANDER_SEAT_ID=1 是人类主持人,
@@ -1034,6 +1049,9 @@ export function useStep2Orchestration(
               dAgentThreadId: ids,
               agentName: coord,
               newMessage: nextLeaderTask,
+              // 第 2 轮起(已有席位交付)才是「综合轮」:让后端注入各席位完整交付供总控收口。
+              // 首轮纯派活、无交付,传 false 保持上下文精简。
+              synthesisMode: safetyCounter > 1,
               model: selectedModelRef.current || undefined,
               signal: leaderAborter.signal,
               onTextDelta: (text) => appendToDialogue(leaderBubbleId, text),
@@ -1064,7 +1082,7 @@ export function useStep2Orchestration(
           }
           activeAbortersRef.current.delete(leaderAborter);
 
-          if (orchestrationCancelRef.current) break;
+          if (isStale()) break;
 
           if (leaderRes.clarification) {
             finalizeDialogue(
@@ -1101,7 +1119,7 @@ export function useStep2Orchestration(
 
           // **顺序**派活 —— 不要改回 Promise.all,详见 §5.4。
           for (const [subName, task] of leaderRes.dispatched as LeaderDispatch[]) {
-            if (orchestrationCancelRef.current) break;
+            if (isStale()) break;
             const meta = agentNameToRoleRef.current[subName.toLowerCase()];
             if (meta) markSpeakerStreaming(meta.speakerId);
             const perSeatModel = meta
@@ -1149,7 +1167,7 @@ export function useStep2Orchestration(
                 onUpstreamFrame: (frame) =>
                   handleUpstreamFrame(frame, seatThreadId),
               });
-              if (orchestrationCancelRef.current) {
+              if (isStale()) {
                 finalizeDialogue(
                   subBubbleId,
                   specRes.content || "(已停止)",
@@ -1181,7 +1199,7 @@ export function useStep2Orchestration(
             if (aborted) break;
           }
 
-          if (orchestrationCancelRef.current) break;
+          if (isStale()) break;
 
           setConsensusPercentage((prev) =>
             Math.min(95, parseFloat((prev + 8).toFixed(1))),
@@ -1229,8 +1247,15 @@ export function useStep2Orchestration(
   const runChainOrchestration = useCallback(
     async (seedMessage: string, ids: ThreadIdMap, coord: string) => {
       if (isOrchestratingRef.current) return;
+      // 同步抢占编排锁(原因同 runOrchestration):堵住人工干预时 isRoundTableRunning
+      // toggle effect 在同一 tick 内的二次触发,避免并发双 leader run。
+      isOrchestratingRef.current = true;
       setIsOrchestrating(true);
       orchestrationCancelRef.current = false;
+      // epoch 接管:新循环启动即自增,旧链条循环据此退出(详见 runOrchestration / 声明处注释)。
+      const myEpoch = (orchestrationEpochRef.current += 1);
+      const isStale = () =>
+        orchestrationCancelRef.current || orchestrationEpochRef.current !== myEpoch;
 
       const chainSeats = selectedAgentsRef.current;
       const total = chainSeats.length;
@@ -1239,7 +1264,7 @@ export function useStep2Orchestration(
 
       try {
         for (let i = 0; i < total; i += 1) {
-          if (orchestrationCancelRef.current) {
+          if (isStale()) {
             finishedAllSeats = false;
             break;
           }
@@ -1295,7 +1320,7 @@ export function useStep2Orchestration(
             throw err;
           }
           activeAbortersRef.current.delete(leaderAborter);
-          if (orchestrationCancelRef.current) {
+          if (isStale()) {
             finalizeDialogue(leaderBubbleId, leaderRes.content || "(已停止)", "已停止");
             finishedAllSeats = false;
             break;
@@ -1357,7 +1382,7 @@ export function useStep2Orchestration(
               },
               onUpstreamFrame: (frame) => handleUpstreamFrame(frame, seatThreadId),
             });
-            if (orchestrationCancelRef.current) {
+            if (isStale()) {
               finalizeDialogue(subBubbleId, specRes.content || "(已停止)", "已停止");
               aborted = true;
             } else {
@@ -1384,7 +1409,7 @@ export function useStep2Orchestration(
         }
 
         // ── (c) 总控收口:综合全链交付,产出最终结论 ──
-        if (!orchestrationCancelRef.current && finishedAllSeats && total > 0) {
+        if (!isStale() && finishedAllSeats && total > 0) {
           highlightSpeaker(null);
           const closeoutBubbleId = appendStreamingDialogue({
             sender: "总控协调",
@@ -1402,6 +1427,8 @@ export function useStep2Orchestration(
               agentName: coord,
               newMessage:
                 "各链条席位已按顺序完成发言,请你综合所有席位的交付,输出最终方案与结论。不要再派活、不要再向任何席位分配任务。",
+              // 链条收口就是综合轮:让后端注入各席位完整交付正文。
+              synthesisMode: true,
               model: selectedModelRef.current || undefined,
               signal: closeoutAborter.signal,
               onTextDelta: (text) => appendToDialogue(closeoutBubbleId, text),
@@ -1604,7 +1631,7 @@ export function useStep2Orchestration(
       content: `💡 (人工指令介入) ${userText}`,
       confidence: "最高指令",
     });
-    pendingInjectPromptRef.current = `用户人工干预新指令:${userText}\n\n请结合当前讨论状态,决定是否继续派活或直接给出更新后的结论。`;
+    pendingInjectPromptRef.current = `用户人工干预新指令:${userText}\n\n请优先按这条新指令,通过 agent_orchestration 把活派给相关的真实席位去获取交付;若用户点名了某个席位(如"前端实现"),必须派给清单里对应的那个 agent_name。**不要**自己代答,也**不要**在没有真正看到该席位「完成…交付内容为:…」的情况下臆造交付或直接收口。只有当用户明确要求收口、或相关席位都已真实交付时,才综合输出结论。`;
     triggerToast("干预指令已记录,会商已自动恢复执行");
   }, [appendDialogue, customAgentInput, triggerToast]);
 
@@ -1631,7 +1658,7 @@ export function useStep2Orchestration(
       confidence: "最高指令",
     });
 
-    const prompt = `用户人工干预新指令:${userText}\n\n请结合当前讨论状态,决定是否继续派活或直接给出更新后的结论。`;
+    const prompt = `用户人工干预新指令:${userText}\n\n请优先按这条新指令,通过 agent_orchestration 把活派给相关的真实席位去获取交付;若用户点名了某个席位(如"前端实现"),必须派给清单里对应的那个 agent_name。**不要**自己代答,也**不要**在没有真正看到该席位「完成…交付内容为:…」的情况下臆造交付或直接收口。只有当用户明确要求收口、或相关席位都已真实交付时,才综合输出结论。`;
 
     // 确保处于运行状态（同步更新按钮 UI）
     setIsRoundTableRunning(true);

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

@@ -183,6 +183,9 @@ async def _run_agent_turn(
             "subagent_enabled": False,
             "thread_id": thread_id,
             "memory_recall_disabled": True,
+            # 与 multi_agent._run_payload 对齐:关 builtin 记忆注入,避免单例总控/席位
+            # 被上轮会话 memory.json 带偏(臆造席位 / 直接干活)。详见该处注释。
+            "memory_injection_enabled": False,
         },
         excluded_tools=excluded_tools,
         skill_stop_names=skill_stop_names,

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

@@ -10,8 +10,14 @@
 
 角色（``role``）：
   - ``leader``  —— 总控派活：短决策，关 thinking + low reasoning（首字节快）；
-    只禁 ``web_search``；并配合 ``SkillStopMiddleware`` 在 ``agent_orchestration``
-    调用后停图，便于网关从 AIMessage.tool_calls 解析派活列表。
+    **只保留 ``agent_orchestration`` / ``ask_clarification`` / ``present_files``**，
+    其余「干活类」工具（``write_file`` / ``bash`` / ``str_replace`` / ``read_file`` /
+    ``ls`` / ``view_image`` / ``task`` / ``write_todos`` / ``memory`` / ``hindsight_*``）
+    一律禁用——否则弱模型（如 deepseek-chat）会概率性地**自己写文件、跑 bash 干活**
+    而不是派活（已在生产日志里观测到总控 ``wc -l outputs/*.md`` 自检自己写的报告）。
+    总控的唯一职责是「派活 / 澄清 / 收口呈现」，没有任何理由动文件系统或执行命令。
+    并配合 ``SkillStopMiddleware`` 在 ``agent_orchestration`` 调用后停图，便于网关从
+    AIMessage.tool_calls 解析派活列表。
   - ``seat``    —— 子智能体交付：开 thinking + medium reasoning；禁派活 /
     澄清 / 展示文件 / 看图 / 联网搜索（与前端 ``_special_run`` 一致）。
   - ``report``  —— Step3 报告绘制：在 ``seat`` 基础上**额外禁 ``bash`` /
@@ -31,8 +37,28 @@ from dataclasses import dataclass
 # ── 每角色基础工具裁剪（唯一定义处）─────────────────────────────────────────────
 #
 # 注意：``seat`` / ``report`` 都包含 ``web_search`` —— 圆桌研讨里联网搜索一律禁用
-# （与前端 ``_special_run`` 的 excluded_tools 对齐）。leader 也禁 ``web_search``。
-_LEADER_EXCLUDED: list[str] = ["web_search"]
+# （与前端 ``_special_run`` 的 excluded_tools 对齐）。
+#
+# ``leader`` 用「白名单」思路反推黑名单：总控**只允许** agent_orchestration（派活）、
+# ask_clarification（澄清）、present_files（收口呈现子智能体产出的文件）三件事，
+# 因此把其余所有工具全部禁掉。关键是 write_file / bash / str_replace —— 留着它们，
+# 弱模型会绕过派活直接自己写报告（生产日志已实锤）。memory / hindsight_* 一并禁，
+# 配合 ``memory_injection_enabled=False`` 让单例总控不被上轮会话的记忆带偏席位。
+_LEADER_EXCLUDED: list[str] = [
+    "web_search",
+    "view_image",
+    "bash",
+    "ls",
+    "read_file",
+    "write_file",
+    "str_replace",
+    "task",
+    "write_todos",
+    "memory",
+    "hindsight_recall",
+    "hindsight_reflect",
+    "hindsight_retain",
+]
 _SEAT_EXCLUDED: list[str] = ["web_search", "ask_clarification", "present_files", "view_image", "agent_orchestration"]
 _REPORT_EXCLUDED: list[str] = [*_SEAT_EXCLUDED, "bash", "str_replace"]
 

offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-coordinator/SOUL.md

@@ -11,15 +11,18 @@
 通过 `agent_orchestration` 技能调用其他智能体干活,使用技能时传入 `agent_name` 和需要做的活。
 
 **⚠️ 关于 `agent_name` 的硬性规则**:
-- 必须严格使用**本次圆桌已上线席位的英文 id**(例如 `roundtable-intelligence`,或形如 `02714cbe6e64406a8a4659f0e3d9c0b8` 这样的 UUID)
-- 本次可调度的席位清单会以「【圆桌系统初始化】本次可调度席位:...」这条人类消息的形式出现在你的对话历史最前面——**只能从这份清单里挑**
-- 不要把括号里的中文显示名拼进 `agent_name`;不要派给清单上没有的 id
+- `agent_name` 必须**逐字符照抄**本次席位清单里给出的 id 原文(可能是 `roundtable-intelligence` 这样的可读 id,也可能是一串 32 位十六进制字符)
+- 本次可调度的席位清单会以「【圆桌系统初始化】本次可调度席位:...」这条人类消息的形式出现在你的对话历史最前面——**只能从这份清单里逐字复制 id**
+- **绝对禁止凭记忆、猜测,或参考本说明里的任何示例文字自行拼造 id**;清单里没有的 id 一律不准派
+- 不要把括号里的中文显示名拼进 `agent_name`
 
 ## 关键约束(必读)
 1. **不要对同一子智能体重复派同样的活**:每次派活前回顾对话历史,如果某个子智能体已经针对你正要派的任务完成过交付(看到形如「子智能体[名称]完成XX工作,交付内容为:...」的 human 消息),就直接基于已有交付推进,**不要重复派活**,除非你确实有**具体的、不一样的**补充问题
 2. **看到交付要会用**:子智能体的交付内容会作为 human 消息追加到你的对话历史里,你能完整看见。当各方信息已足够形成结论时,**直接综合输出最终方案**(此时不再调用任何工具),让讨论收敛
-3. **避免无意义循环**:如果连续两轮都没有新信息流入,立刻停止派活并基于现有素材给出最终结论
-4. **数量与节奏**:每轮派活 1-3 个席位最佳;不需要每个席位都派一次
+3. **⛔ 严禁臆造交付**:你**只能**把对话历史里**确实出现过**的「子智能体[名称]完成XX工作,交付内容为:...」当作已交付。**绝对禁止**在没有看到这条消息的情况下声称某席位「已交付」、编造它的交付内容、或假装研讨已完成。**没看到该席位的真实交付,就一律视为「未交付」**——该派活就派活,绝不跳过直接收口。宁可多派一轮,也不许凭空说「三方已交付」。
+4. **用户点名某席位就去派活**:当用户(含人工干预指令)明确希望由某个席位来回答/产出(例如"我想看前端实现,席位里有前端实现智能体"),你**必须**通过 `agent_orchestration` 把活派给那个真实席位,**不要**自己代答、也不要直接给最终结论搪塞。
+5. **避免无意义循环**:如果连续两轮都没有新信息流入,立刻停止派活并基于现有素材给出最终结论
+6. **数量与节奏**:每轮派活 1-3 个席位最佳;不需要每个席位都派一次
 
 ## 终止条件
 当满足以下任一条件,**直接输出最终结论文本**(不调用任何工具),让本次圆桌收敛:
@@ -40,3 +43,5 @@
 - ❌ 派活给本次清单**没有**的 agent_name
 - ❌ 把括号里的中文显示名拼进 `agent_name`
 - ❌ 已经达成共识后仍然继续派活
+- ❌ **臆造 / 假装某席位已交付**——没在对话历史里看到它的「完成…交付内容为:…」就不许说它交付了
+- ❌ 用户点名要某席位回答时,自己代答或直接收口而**不去派活**

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

@@ -47,6 +47,7 @@ Leader 终态帧（前端 ``isFinalStatusFrame``）
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import json
 import logging
 import os
@@ -486,6 +487,31 @@ async def _append_thread_message(
         logger.warning("update state %s failed: %s %s", thread_id, post_resp.status_code, post_resp.text[:200])
 
 
+def _spawn_detached_broadcast(
+    base: str,
+    headers: dict[str, str],
+    message: str,
+    agent_threads: dict[str, str],
+    skip: list[str],
+) -> None:
+    """后台 fire-and-forget 广播,**用独立 client**,即使发起它的请求 SSE 已结束也能跑完。
+
+    广播是 best-effort 的「上下文播报」(告知其它席位某人交付了什么),原本在总控 yield 派活
+    帧**之前**被 ``await``,导致每个席位都要等几十秒(广播要读写各席位的巨大 thread 状态)
+    才开始执行 —— 实测 ``dispatch_broadcasts_await`` 飙到 27s。改成 detached 后台任务后,总控
+    立即 yield 派活、前端立即启动席位,广播在后台慢慢落地,彻底移出关键路径。失败只记日志。
+    """
+    async def _run() -> None:
+        try:
+            async with httpx.AsyncClient(timeout=None, trust_env=False) as bg:
+                await _broadcast(bg, base, headers, message, agent_threads, skip)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("detached broadcast failed: %s", exc)
+
+    # create_task 后不持有引用:任务挂在事件循环上,独立于本请求生命周期跑完。
+    asyncio.ensure_future(_run())
+
+
 async def _broadcast(
     client: httpx.AsyncClient,
     base: str,
@@ -506,6 +532,29 @@ async def _broadcast(
 
 _TOOL_CALL_PATTERN = re.compile(r'\.py\s+"([^"]*)"\s+"([^"]*)"')
 
+# 子智能体交付的「后广播」里,正文广播给**其它席位**时的截断上限。
+# 背景:席位交付动辄几万字;原实现把整篇交付广播进**每一个**其它席位 thread,导致
+# 每个 thread 状态滚雪球,之后每次状态读/写(广播、交付状态核对)都要搬运巨量数据,
+# 延迟复利式恶化(实测 dispatch_broadcasts_await 飙到 30s)。其它席位只需"知道某席位
+# 交付了什么"的摘要即可;**完整交付仍原样发给总控 thread**(总控做最终综合要看全文),
+# 也仍通过 SSE 完整给到前端。仅截断"广播给兄弟席位"这一路。
+_BROADCAST_CONTENT_MAX_CHARS = 1500
+
+
+def _truncate_broadcast_content(prefix: str, content: str) -> str:
+    """构造广播给**兄弟席位**的交付摘要消息:超长正文截断到上限并标注省略。
+
+    ``content`` 未超限时原样拼接(等价于完整版),超限时截断 + 追加省略说明。
+    总控那一路不走本函数,始终用完整正文(见 ``_special_run`` 的 ``full_msg``)。
+    """
+    if len(content) <= _BROADCAST_CONTENT_MAX_CHARS:
+        return prefix + content
+    return (
+        prefix
+        + content[:_BROADCAST_CONTENT_MAX_CHARS]
+        + f"\n…(完整交付约 {len(content)} 字,已省略;以该席位自身交付为准)"
+    )
+
 # Mirrors the upstream agent-id validator in the LangGraph runtime. Reject early
 # so callers get a clear 400 here instead of a stream that aborts with HTTP 422
 # from the loopback /api/threads/.../runs/stream endpoint.
@@ -592,6 +641,7 @@ def _run_payload(
     skill_stop_names: list[str],
     thinking_enabled: bool = True,
     reasoning_effort: str = "medium",
+    multitask_strategy: str = "reject",
 ) -> dict[str, Any]:
     """构造 ``POST /api/threads/{thread_id}/runs/stream`` 的请求体。
 
@@ -627,15 +677,28 @@ def _run_payload(
             # 圆桌四个 router(intent / recommend / leader / special)都是临时任务,
             # 用户个人记忆与本次研讨无关。MemoryMiddleware 的 Hindsight recall 是
             # 同步阻塞 HTTP 调用(单次 timeout 15s),如果记忆服务慢/不通会让前端
-            # "调度中"卡很久;这里默认跳过,builtin USER.md/MEMORY.md 仍通过系统
-            # 提示注入不受影响。
+            # "调度中"卡很久;这里默认跳过。
             "memory_recall_disabled": True,
+            # ⚠️ 同时关掉 builtin 记忆**注入**:圆桌四个角色(intent / recommend /
+            # leader / special)都是固定单例 agent,其 memory.json 跨所有会话累积。
+            # 总控尤其危险——上一个任务里它臆造的席位名(如 data-specialist)会被存进
+            # memory.json,下一个新任务又被注入系统提示而召回 → 派给根本没初始化的
+            # 席位 → run/stream 报 400 "agent_name ... missing from d_agent_thread_id"。
+            # 圆桌是一次性研讨,本就不该吃历史会话记忆,这里统一关注入根治该污染。
+            "memory_injection_enabled": False,
         },
         "stream_mode": ["messages-tuple", "values"],
         "stream_subgraphs": True,
         "stream_resumable": True,
         "assistant_id": "lead_agent",
         "on_disconnect": "continue",
+        # ⚠️ 圆桌 run 必须能被「下一次同 thread 的 run」抢占。圆桌每个 thread 同一时刻
+        # 只应有一个活跃 run,但 on_disconnect="continue" 让前端 abort SSE 后**后端 run
+        # 仍在跑**(为可恢复);此时用户「人工干预」立刻发起新 leader run 打到同一总控
+        # thread,默认 reject 策略就会 409「already has an active run」→ 被 _stream_upstream
+        # 包成 502。调用方(leader / special)传 multitask_strategy="rollback",让新 run 先
+        # 取消该 thread 上没结束的旧 run 再继续 —— 这正是「最新一轮/干预指令优先」的语义。
+        "multitask_strategy": multitask_strategy,
         "excluded_tools": excluded_tools,
         "skill_stop_names": skill_stop_names,
     }
@@ -902,16 +965,23 @@ async def _stream_upstream(
     yield None, captured
 
 
-def _parse_last_messages(captured: list[str]) -> list[dict[str, Any]]:
+def _parse_last_messages(captured: list[str]) -> list[dict[str, Any]] | None:
     """Return the full ``messages`` list from the most recent ``values`` snapshot.
 
     The upstream sends the final ``values`` frame second-to-last (the very last
     line is the ``end`` event with ``null``). We tolerate trailing non-JSON
     frames (e.g. ``data: null``) by scanning backwards for the first frame that
     successfully parses as a state object with a ``messages`` key.
+
+    Returns ``None`` when the stream carried **no parseable ``values`` frame**
+    (empty capture, or only ``data: null`` / token deltas). 这通常**不是真正的
+    上游错误**,而是该 run 被 ``multitask_strategy=rollback`` 抢占 / 客户端断开 /
+    中途 abort —— 流被截断,没来得及发出最终 values 快照。调用方(leader/special)据此
+    **静默收尾**,绝不能把它当 502 抛给前端:抢占场景下这条 SSE 多半已被前端
+    ``abortAllInFlight`` 放弃,硬报 502 只会在 UI 上凭空冒出一条错误。
     """
     if not captured:
-        raise HTTPException(status_code=502, detail="upstream produced no data lines")
+        return None
 
     for raw_line in reversed(captured):
         raw = raw_line[5:].strip() if raw_line.startswith("data:") else raw_line.strip()
@@ -926,7 +996,7 @@ def _parse_last_messages(captured: list[str]) -> list[dict[str, Any]]:
             if isinstance(messages, list):
                 return messages
 
-    raise HTTPException(status_code=502, detail="no parseable values frame in upstream stream")
+    return None
 
 
 def _find_trailing_ai_text(messages: list[dict[str, Any]], *, exclude_text: str = "") -> str:
@@ -983,6 +1053,194 @@ def _find_dispatch_message(messages: list[dict[str, Any]]) -> dict[str, Any]:
     return {}
 
 
+def _partition_dispatch(
+    tool_calls: list[dict[str, Any]],
+    valid_seat_ids: set[str],
+) -> tuple[list[tuple[str, str]], list[str]]:
+    """把总控的 agent_orchestration 派活拆成「合法 / 非法」两份。
+
+    返回 ``(valid, invalid)``：
+
+    - ``valid``   —— ``[(seat_id, task), ...]``，``seat_id`` 确实是本轮已初始化的席位；
+    - ``invalid`` —— ``[seat_id, ...]``，模型臆造 / 不在本轮 roster 内的 id。
+
+    弱模型(deepseek-chat)会把 SOUL 里的示例 hex id 当真照抄、或臆造同款 32 位 hex,
+    这些 id 不在 ``agent_threads`` 里;若不拦截会漏给前端,到 ``/run/stream`` 才报 400
+    把整个 Step 2 打挂。校验放在派活解析处,只丢非法、保留同轮合法席位。
+    """
+    valid: list[tuple[str, str]] = []
+    invalid: list[str] = []
+    for tool_call in tool_calls:
+        sub_name, task_text = _extract_dispatch(tool_call)
+        if not sub_name:
+            continue
+        if sub_name in valid_seat_ids:
+            valid.append((sub_name, task_text))
+        else:
+            invalid.append(sub_name)
+    return valid, invalid
+
+
+def _build_dispatch_correction(invalid_names: list[str], valid_seat_ids: set[str]) -> str:
+    """整轮派活全非法时,回灌给总控 thread 的系统纠正消息文本。
+
+    列出真实席位 id,要求模型**逐字符复制**重派,杜绝「凭示例/记忆构造 id」。
+    """
+    bad = "、".join(invalid_names)
+    roster = "、".join(sorted(valid_seat_ids))
+    return (
+        "【系统纠正】你刚才通过 agent_orchestration 派活的 agent_name（"
+        f"{bad}）不在本次圆桌的席位清单里，无法执行。请**只**从下面这份真实席位清单里"
+        "逐字符复制 agent_name 重新派活，不要凭记忆、示例或猜测自行构造 id：\n"
+        f"{roster}"
+    )
+
+
+def _build_seat_roster_block(seats: list[tuple[str, dict[str, Any]]]) -> str:
+    """构造注入到总控**本轮输入最前面**的「真实席位清单」文本。
+
+    ``seats`` 为 ``[(seat_id, {name, description}), ...]``。
+
+    为什么要每轮把清单塞进输入,而不是只靠 init 时 _append_thread_message 注入历史:
+    圆桌席位清单原本只在 init 时以一条 human 消息异步写进总控 thread 历史,总控靠"读
+    历史"看到它 —— 这条链路脆弱(best-effort 写盘、跨 thread 持久化),一旦没落地,弱模型
+    (deepseek-chat)面对空历史就会**从任务语义臆造**不存在的席位(已观测到 architecture-
+    specialist、02714cbe… 这类假 id)。网关在每次 leader run 时本就握有真实席位 id,这里
+    直接把清单摆到总控本轮输入最前面并要求逐字复制,从源头根治臆造;与事后 _partition_dispatch
+    校验构成「源头 + 兜底」双保险。
+    """
+    if not seats:
+        return ""
+    lines: list[str] = []
+    for sid, meta in seats:
+        name = str((meta or {}).get("name") or sid)
+        desc = str((meta or {}).get("description") or "").strip()
+        lines.append(f"- {sid}（{name}）：{desc}" if desc else f"- {sid}（{name}）")
+    roster = "\n".join(lines)
+    return (
+        "【本次圆桌可调度席位清单】派活时 agent_orchestration 的 agent_name 必须从下面"
+        "逐字符复制一个 id，严禁自行拼造、翻译角色名或凭记忆构造：\n"
+        f"{roster}"
+    )
+
+
+async def _collect_delivery_status(
+    client: httpx.AsyncClient,
+    base: str,
+    headers: dict[str, str],
+    seat_threads: dict[str, str],
+) -> dict[str, tuple[bool, str]]:
+    """读各席位 thread 的真实状态,客观判定「是否已交付 + 交付摘要」。
+
+    返回 ``{seat_id: (delivered, full_text)}``。``full_text`` 是该席位**完整交付正文**
+    (调用方自行决定截多少:摘要块截短、综合轮全文)。判定依据:席位 thread 的对话历史里
+    **是否存在该席位自己产出的、非空的 AI 消息**(它的交付正文)。只有 broadcast 进来的
+    human 消息(派活通知 / 别的席位的交付摘要)不算它交付。
+
+    这是「确定性双保险」的数据源:总控弱模型会臆造「三方已交付」,这里由网关**客观核对**
+    每个席位 thread 的实际产出,再把结论喂回总控输入(见 ``_build_seat_status_block``),
+    总控以系统核对为准,不靠自己读历史猜。任一席位读失败按「未交付」保守处理,绝不阻断。
+    """
+    async def _one(sid: str, tid: str) -> tuple[str, tuple[bool, str]]:
+        try:
+            resp = await client.get(f"{base}/api/threads/{tid}/state", headers=headers)
+            if resp.status_code != 200:
+                return sid, (False, "")
+            messages = ((resp.json().get("values") or {}).get("messages")) or []
+            for msg in reversed(messages):
+                if not isinstance(msg, dict):
+                    continue
+                if (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"):
+                    text = _flatten_content(msg.get("content", "")).strip()
+                    if text:
+                        return sid, (True, text)
+            return sid, (False, "")
+        except Exception:
+            return sid, (False, "")
+
+    results = await asyncio.gather(*(_one(sid, tid) for sid, tid in seat_threads.items()))
+    return dict(results)
+
+
+# 交付状态摘要块里,每个席位「已交付摘要」截断到的字符数(让总控有足够信息做派活决策,
+# 又远小于动辄几万字的全文)。
+_STATUS_EXCERPT_CHARS = 600
+
+
+def _build_seat_status_block(
+    seats: list[tuple[str, dict[str, Any]]],
+    delivery: dict[str, tuple[bool, str]],
+    *,
+    undelivered_count: int = 0,
+) -> str:
+    """席位清单 + **系统客观核对的真实交付状态**,注入总控本轮输入最前面。
+
+    在 ``_build_seat_roster_block``(只给 id+名称)基础上,叠加每个席位的 ✅已交付 / ⬜未交付
+    标记(来自 ``_collect_delivery_status`` 对席位 thread 的实读),并明确告诉总控「以这份为准、
+    不要臆造」「用户点名的 ⬜ 席位必须派活」。这是消除「臆造三方已交付直接收口」的确定性手段。
+
+    ``undelivered_count`` > 0 时额外加一句硬约束:还有席位未交付,**不得**急于收口。
+    """
+    if not seats:
+        return ""
+    lines: list[str] = []
+    for sid, meta in seats:
+        name = str((meta or {}).get("name") or sid)
+        delivered, full_text = delivery.get(sid, (False, ""))
+        if delivered:
+            excerpt = full_text[:_STATUS_EXCERPT_CHARS]
+            ellipsis = "…" if len(full_text) > _STATUS_EXCERPT_CHARS else ""
+            tail = f" —— 交付摘要:{excerpt}{ellipsis}" if excerpt else ""
+            lines.append(f"- {sid}（{name}）：✅ 已交付{tail}")
+        else:
+            lines.append(f"- {sid}（{name}）：⬜ 未交付")
+    roster = "\n".join(lines)
+    pending_rule = (
+        (
+            f"\n⚠️ 当前还有 {undelivered_count} 个 ⬜ 未交付席位。**信息尚未集齐,严禁现在收口/输出最终方案**,"
+            "也不得声称「各席位均已发言」。请**优先继续派活**给这些 ⬜ 席位,把它们的真实交付拿到手再说综合。"
+        )
+        if undelivered_count > 0
+        else ""
+    )
+    return (
+        "【本次圆桌席位与真实交付状态(系统客观核对,以此为准,严禁臆造)】\n"
+        "派活时 agent_orchestration 的 agent_name 必须从下面逐字符复制一个 id;"
+        "只有标 ✅ 的才算已交付,标 ⬜ 的就是还没交付——\n"
+        "若用户想要的席位是 ⬜,你**必须**派活给它拿到真实交付,**不得**跳过它直接收口或代答;"
+        "也**不得**声称任何 ⬜ 席位已交付。下面只是**摘要**,做最终综合时以系统额外提供的"
+        "「完整交付」为准:\n"
+        f"{roster}"
+        f"{pending_rule}"
+    )
+
+
+def _build_full_deliverables_block(
+    seats: list[tuple[str, dict[str, Any]]],
+    delivery: dict[str, tuple[bool, str]],
+) -> str:
+    """综合轮专用:把各已交付席位的**完整交付正文**拼成一块,供总控写最终方案。
+
+    只在前端标记的「综合轮」(synthesis_mode)注入,平时不进总控输入 —— 这样派活决策轮
+    的上下文保持精简、响应快,只有真要收口时才承担一次全文的体量。未交付席位不列入。
+    """
+    sections: list[str] = []
+    for sid, meta in seats:
+        delivered, full_text = delivery.get(sid, (False, ""))
+        if not delivered or not full_text:
+            continue
+        name = str((meta or {}).get("name") or sid)
+        sections.append(f"━━ {name}（{sid}）的完整交付 ━━\n{full_text}")
+    if not sections:
+        return ""
+    body = "\n\n".join(sections)
+    return (
+        "【各席位完整交付正文(综合专用,做最终方案时以此为准)】\n"
+        "请基于下列各席位的真实完整交付,综合输出结构化的最终方案,不要遗漏关键内容,也不要臆造:\n"
+        f"{body}"
+    )
+
+
 async def _leader_run(
     client: httpx.AsyncClient,
     base: str,
@@ -992,6 +1250,8 @@ async def _leader_run(
     new_message: str,
     skill_stop_names: list[str],
     model_name: str,
+    seat_meta: dict[str, dict[str, Any]] | None = None,
+    synthesis_mode: bool = False,
 ) -> AsyncIterator[str | dict[str, Any]]:
     """总控（leader）单轮：透传上游 SSE + 末尾 yield 一个 dict 状态帧（由路由层 ``json.dumps``）。
 
@@ -1002,8 +1262,52 @@ async def _leader_run(
     4. 若有派活，**派活广播**后再 yield ``{status: dispatched, ...}``。
 
     ``skill_stop_names`` 会与 ``agent_orchestration`` 合并，确保 middleware 在工具调用后停图。
+
+    ``seat_meta``：``{seat_id: {name, description}}``。非空时,本函数会在跑总控前**实读各席位
+    thread 的真实交付状态**,把「席位清单 + ✅已交付/⬜未交付 + 摘要」拼到总控本轮输入最前面
+    (见 ``_collect_delivery_status`` / ``_build_seat_status_block``)。这是消除「总控臆造
+    三方已交付、一交付就收口」的确定性双保险:总控以系统客观核对为准,不靠自己读历史猜。
+
+    ``synthesis_mode``：前端在「派活后让总控判断继续/综合」的轮次置真。为真时**额外**把各
+    已交付席位的**完整交付正文**注入本轮输入(``_build_full_deliverables_block``),供总控写
+    最终方案;平时(派活决策轮)只给摘要,让总控 thread 与每轮上下文保持精简、响应快。
     """
     t0 = time.perf_counter()
+    # 实读各席位 thread 的真实交付状态,构造「席位清单 + 客观交付状态(摘要)」块;综合轮再额外
+    # 拼上各已交付席位的完整交付正文。失败/无 seat_meta 时退化为不注入(leader_input ==
+    # new_message),绝不阻断 leader run。这些只进总控**本轮 run 输入**,不广播给席位。
+    seat_status_block = ""
+    full_deliverables_block = ""
+    if seat_meta:
+        seat_ids = [name for name in agent_threads if name != agent_name]
+        if seat_ids:
+            delivery: dict[str, tuple[bool, str]] = {}
+            try:
+                delivery = await _collect_delivery_status(
+                    client, base, headers, {sid: agent_threads[sid] for sid in seat_ids}
+                )
+            except Exception:
+                logger.warning("[leader-roster] 收集席位交付状态失败,退化为仅席位清单", exc_info=True)
+            undelivered = [sid for sid in seat_ids if not delivery.get(sid, (False, ""))[0]]
+            seats_meta_list = [(sid, seat_meta.get(sid, {})) for sid in seat_ids]
+            seat_status_block = _build_seat_status_block(seats_meta_list, delivery, undelivered_count=len(undelivered))
+            # 只有「综合轮」**且所有席位都已交付」才注入「完整交付 + 输出最终方案」的综合块。
+            # 还有 ⬜ 未交付席位时不注入 —— 否则那个"综合输出最终方案"的指令会怂恿弱模型在席位
+            # 没集齐时就提前收口(实测:3/5 交付就臆称"三方已发言"草草收口)。此时只给摘要,让总控
+            # 倾向于继续把 ⬜ 席位派活集齐。
+            if synthesis_mode and not undelivered:
+                full_deliverables_block = _build_full_deliverables_block(seats_meta_list, delivery)
+            logger.info(
+                "[leader-roster] %s 注入席位交付状态:%s synthesis_mode=%s undelivered=%d full_deliverables=%s",
+                agent_name,
+                {sid: delivery.get(sid, (False, ""))[0] for sid in seat_ids},
+                synthesis_mode,
+                len(undelivered),
+                bool(full_deliverables_block),
+            )
+    leader_input = "\n\n".join(
+        part for part in (seat_status_block, full_deliverables_block, new_message) if part
+    )
 
     # 与上游 LLM 并行，减少「总控已开始想但席位 thread 里还没有用户消息」的窗口。
     pre_broadcast_task = asyncio.create_task(
@@ -1019,11 +1323,14 @@ async def _leader_run(
         agent_name=agent_name,
         model_name=model_name,
         thread_id=thread_id,
-        new_message=new_message,
+        # 注入了席位清单的输入(leader_input);seat_roster 为空时等同 new_message。
+        new_message=leader_input,
         excluded_tools=policy.excluded_tools,
         skill_stop_names=effective_stop_names,
         thinking_enabled=policy.thinking_enabled,
         reasoning_effort=policy.reasoning_effort,
+        # 人工干预/恢复会重发 leader run 到同一总控 thread;抢占掉没结束的旧 run,避免 409。
+        multitask_strategy="rollback",
     )
     t_payload_built = time.perf_counter()
     logger.info(
@@ -1032,117 +1339,152 @@ async def _leader_run(
         t_payload_built - t0,
     )
 
-    captured: list[str] = []
-    first_line_ts: float | None = None
-    line_count = 0
-    async for line, done in _stream_upstream(client, base, headers, thread_id, body):
-        if line is not None:
-            if first_line_ts is None:
-                first_line_ts = time.perf_counter()
-                logger.info(
-                    "[leader-timing] %s first_upstream_line=%.3fs since_entry=%.3fs",
-                    agent_name,
-                    first_line_ts - t_payload_built,
-                    first_line_ts - t0,
-                )
-            line_count += 1
-            yield line
-        elif done is not None:
-            captured = done
-    t_stream_done = time.perf_counter()
-    logger.info(
-        "[leader-timing] %s upstream_stream_complete=%.3fs lines=%d since_entry=%.3fs",
-        agent_name,
-        t_stream_done - (first_line_ts or t_payload_built),
-        line_count,
-        t_stream_done - t0,
-    )
+    # 单例总控是弱模型(deepseek-chat),已观测到它把 SOUL 里的示例 id 当真实席位
+    # 照抄、或臆造同款 32 位 hex —— 这些 id 不在 agent_threads 里,直接漏给前端会让
+    # /run/stream 报 400 把整个 Step 2 打挂。这里把「本次真实可派席位」作为 roster
+    # (agent_threads 去掉总控自身),派活前就地校验:非法 id 丢弃 + 记日志;若**整轮
+    # 全是非法 id**,向总控 thread 回灌一条系统纠正消息(列出真实 id),再重跑一轮让它
+    # 自纠(最多重试 1 次,避免弱模型反复臆造导致死循环)。
+    valid_seat_ids: set[str] = {name for name in agent_threads if name != agent_name}
+    _MAX_DISPATCH_RETRIES = 1
 
-    messages = _parse_last_messages(captured)
-    # Pre-broadcast must finish before we yield the final status frame on any
-    # branch — the seats that get dispatched next rely on having seen the
-    # original user message. The upstream LLM stream usually takes several
-    # seconds, so the broadcast has typically already completed and this
-    # await is effectively free.
-    t_before_broadcast_await = time.perf_counter()
-    try:
-        await pre_broadcast_task
-    except Exception as exc:
-        logger.warning("pre-broadcast failed: %s", exc)
-    t_after_broadcast_await = time.perf_counter()
-    logger.info(
-        "[leader-timing] %s pre_broadcast_await=%.3fs (>0 means broadcast was still in flight after LLM finished)",
-        agent_name,
-        t_after_broadcast_await - t_before_broadcast_await,
-    )
-
-    dispatch_msg = _find_dispatch_message(messages)
-    leader_text = _flatten_content(dispatch_msg.get("content", ""))
-    tool_calls = dispatch_msg.get("tool_calls") or []
-
-    # 澄清分支：与 intent.py 相同字段；status 为字符串 "clarification"（非数组），
-    # 前端据此暂停 runOrchestration 循环，展示 Step 2 澄清条。
-    clarification_call = next(
-        (tc for tc in tool_calls if (tc.get("name") or "").lower() == "ask_clarification"),
-        None,
-    )
-    if clarification_call is not None:
-        cargs = clarification_call.get("args") or {}
-        question = str(cargs.get("question") or "").strip()
-        follow_up = _find_trailing_ai_text(messages, exclude_text=leader_text)
-        yield {
-            "status": "clarification",
-            "content": follow_up or question or leader_text,
-            "question": question,
-            "clarification_type": str(cargs.get("clarification_type") or ""),
-            "clarification_context": str(cargs.get("context") or ""),
-            "options": cargs.get("options") or [],
-            "allow_custom": bool(cargs.get("allow_custom", True)),
-            "allow_multiple": resolve_allow_multiple(cargs),
-            "agent_name": agent_name,
-        }
-        return
-
-    if not tool_calls:
-        # 空 status 数组 = 总控不再派活，前端视为共识（hasConsensus）。
-        yield {"status": [], "content": leader_text, "agent_name": agent_name}
-        return
-
-    # 派活广播：告知各席位「总控给 X 派了什么活」，与即将发起的 special run 配合。
-    # sets and we don't care about ordering, so serializing them just
-    # multiplied latency before the frontend could start a sub-agent run.
-    dispatched: list[list[str]] = []
-    dispatch_broadcasts: list[asyncio.Task[None]] = []
-    for tool_call in tool_calls:
-        try:
-            sub_name, task_text = _extract_dispatch(tool_call)
-            if not sub_name:
-                continue
-            dispatch_broadcasts.append(
-                asyncio.create_task(
-                    _broadcast(
-                        client,
-                        base,
-                        headers,
-                        f"总控智能体给{sub_name}派活,内容为:" + task_text,
-                        agent_threads,
-                        [agent_name, sub_name],
+    attempt = 0
+    while True:
+        captured: list[str] = []
+        first_line_ts: float | None = None
+        line_count = 0
+        async for line, done in _stream_upstream(client, base, headers, thread_id, body):
+            if line is not None:
+                if first_line_ts is None:
+                    first_line_ts = time.perf_counter()
+                    logger.info(
+                        "[leader-timing] %s first_upstream_line=%.3fs since_entry=%.3fs",
+                        agent_name,
+                        first_line_ts - t_payload_built,
+                        first_line_ts - t0,
                     )
-                )
-            )
-            dispatched.append([sub_name, task_text])
-        except Exception as exc:
-            logger.warning("dispatch failed: %s", exc)
-    if dispatch_broadcasts:
-        t_before_dispatch = time.perf_counter()
-        await asyncio.gather(*dispatch_broadcasts, return_exceptions=True)
+                line_count += 1
+                yield line
+            elif done is not None:
+                captured = done
+        t_stream_done = time.perf_counter()
         logger.info(
-            "[leader-timing] %s dispatch_broadcasts_await=%.3fs count=%d (gates how soon the next sub agent can start)",
+            "[leader-timing] %s upstream_stream_complete=%.3fs lines=%d since_entry=%.3fs",
             agent_name,
-            time.perf_counter() - t_before_dispatch,
-            len(dispatch_broadcasts),
+            t_stream_done - (first_line_ts or t_payload_built),
+            line_count,
+            t_stream_done - t0,
         )
-    yield {"status": dispatched, "content": leader_text, "agent_name": agent_name}
+
+        messages = _parse_last_messages(captured)
+        if messages is None:
+            # 流被截断(多半被 multitask_strategy=rollback 抢占 / 客户端断开 / abort)。
+            # 这不是真正的上游错误——静默收尾,不向前端抛 502。该 run 对应的 SSE 在抢占
+            # 场景下通常已被前端 abortAllInFlight 放弃;即便没被放弃,前端拿到"无终态帧"
+            # 也只会当作空轮次,不会冒出错误条。
+            # WARNING 级:正常应是被 rollback 抢占;但若上游 run 因别的原因(如 DB 连接断、
+            # LLM 报错)截断流也会走这里 → 静默收尾会把它表现成前端"(无文本输出)"。留个
+            # 显眼日志,便于把"真异常被静默吞掉"和"正常抢占"区分开排查。
+            logger.warning(
+                "[leader-timing] %s upstream truncated (no values frame) — ending quietly "
+                "(expected on rollback preemption; otherwise check the upstream run's server-side error)",
+                agent_name,
+            )
+            with contextlib.suppress(Exception):
+                await pre_broadcast_task
+            return
+
+        # Pre-broadcast must finish before we yield the final status frame on any
+        # branch — the seats that get dispatched next rely on having seen the
+        # original user message. Only await it once (first attempt); the rare
+        # self-correction retry reuses the same already-awaited broadcast.
+        if attempt == 0:
+            t_before_broadcast_await = time.perf_counter()
+            try:
+                await pre_broadcast_task
+            except Exception as exc:
+                logger.warning("pre-broadcast failed: %s", exc)
+            logger.info(
+                "[leader-timing] %s pre_broadcast_await=%.3fs (>0 means broadcast was still in flight after LLM finished)",
+                agent_name,
+                time.perf_counter() - t_before_broadcast_await,
+            )
+
+        dispatch_msg = _find_dispatch_message(messages)
+        leader_text = _flatten_content(dispatch_msg.get("content", ""))
+        tool_calls = dispatch_msg.get("tool_calls") or []
+
+        # 澄清分支：与 intent.py 相同字段；status 为字符串 "clarification"（非数组），
+        # 前端据此暂停 runOrchestration 循环，展示 Step 2 澄清条。
+        clarification_call = next(
+            (tc for tc in tool_calls if (tc.get("name") or "").lower() == "ask_clarification"),
+            None,
+        )
+        if clarification_call is not None:
+            cargs = clarification_call.get("args") or {}
+            question = str(cargs.get("question") or "").strip()
+            follow_up = _find_trailing_ai_text(messages, exclude_text=leader_text)
+            yield {
+                "status": "clarification",
+                "content": follow_up or question or leader_text,
+                "question": question,
+                "clarification_type": str(cargs.get("clarification_type") or ""),
+                "clarification_context": str(cargs.get("context") or ""),
+                "options": cargs.get("options") or [],
+                "allow_custom": bool(cargs.get("allow_custom", True)),
+                "allow_multiple": resolve_allow_multiple(cargs),
+                "agent_name": agent_name,
+            }
+            return
+
+        if not tool_calls:
+            # 空 status 数组 = 总控不再派活，前端视为共识（hasConsensus）。
+            yield {"status": [], "content": leader_text, "agent_name": agent_name}
+            return
+
+        # 派活前按真实 roster 校验:合法保留、非法丢弃(只丢非法,不殃及同轮合法席位)。
+        valid_dispatch, invalid_names = _partition_dispatch(tool_calls, valid_seat_ids)
+        if invalid_names:
+            logger.warning(
+                "[leader-dispatch] %s 派活到 %d 个不存在的席位 id %s;真实席位=%s(attempt=%d)",
+                agent_name,
+                len(invalid_names),
+                invalid_names,
+                sorted(valid_seat_ids),
+                attempt,
+            )
+
+        # 整轮全是非法 id 且还有重试预算 → 回灌纠正消息 + 重跑一轮让总控自纠。
+        if invalid_names and not valid_dispatch and attempt < _MAX_DISPATCH_RETRIES:
+            try:
+                await _append_thread_message(
+                    client,
+                    base,
+                    headers,
+                    thread_id,
+                    _build_dispatch_correction(invalid_names, valid_seat_ids),
+                )
+            except Exception as exc:
+                logger.warning("[leader-dispatch] 回灌纠正消息失败: %s", exc)
+            attempt += 1
+            continue
+
+        # 非法 id 已被丢弃;若 roster 校验后无合法派活(且重试已用尽),退化为空共识,
+        # 而不是把假 id 漏给前端再报 400。
+        dispatched: list[list[str]] = [[sub_name, task_text] for sub_name, task_text in valid_dispatch]
+        # 关键路径优先:**先 yield 派活帧**,让前端立刻启动席位 special run;派活广播(告知
+        # 其它席位"总控给 X 派了什么活")改成 detached 后台任务,不再 await 阻塞 —— 否则广播
+        # 读写各席位巨大 thread 状态会卡几十秒,白白拖慢席位启动(见 _spawn_detached_broadcast)。
+        yield {"status": dispatched, "content": leader_text, "agent_name": agent_name}
+        for sub_name, task_text in valid_dispatch:
+            _spawn_detached_broadcast(
+                base,
+                headers,
+                f"总控智能体给{sub_name}派活,内容为:" + task_text,
+                agent_threads,
+                [agent_name, sub_name],
+            )
+        return
 
 
 async def _special_run(
@@ -1180,6 +1522,8 @@ async def _special_run(
         skill_stop_names=merge_skill_stop_names(policy.skill_stop_names, skill_stop_names),
         thinking_enabled=policy.thinking_enabled,
         reasoning_effort=policy.reasoning_effort,
+        # 干预/重新派活会把同一席位 thread 再跑一次;抢占掉没结束的旧 run,避免 409。
+        multitask_strategy="rollback",
     )
     t_payload_built = time.perf_counter()
     logger.info(
@@ -1215,6 +1559,15 @@ async def _special_run(
     )
 
     messages = _parse_last_messages(captured)
+    if messages is None:
+        # 与 _leader_run 同理:席位 run 被抢占/断开导致流截断,不抛 502,静默收尾。
+        # WARNING 级:正常是 rollback 抢占;否则查上游 run 的服务端报错(如 DB 连接断)。
+        logger.warning(
+            "[special-timing] %s upstream truncated (no values frame) — ending quietly "
+            "(expected on rollback preemption; otherwise check the upstream run's server-side error)",
+            agent_name,
+        )
+        return
     # Sub-agent's final deliverable is the latest AIMessage's text content.
     last_msg: dict[str, Any] = {}
     for msg in reversed(messages):
@@ -1222,41 +1575,22 @@ async def _special_run(
             last_msg = msg
             break
     content = _flatten_content(last_msg.get("content", ""))
-    # Kick the "子智能体完成" broadcast off in the background and yield the
-    # status frame first — the next leader cycle that depends on this
-    # broadcast having landed only starts after a fresh /run/stream request
-    # from the frontend, which is many seconds away (full leader LLM call).
-    # Awaiting after the yield ensures the task finishes before the AsyncClient
-    # closes; the SSE connection stays open a bit longer but the frontend has
-    # already consumed the result.
-    t_before_post = time.perf_counter()
     # 优先用 display_name(中文友好,例如"情报收集")提示 leader,id 放括号备查;
     # display_name 缺失时退化为仅 agent_name —— 与旧行为一致。
     speaker_label = f"{display_name}({agent_name})" if display_name else agent_name
-    post_broadcast_task = asyncio.create_task(
-        _broadcast(
-            client,
-            base,
-            headers,
-            f"子智能体{speaker_label}完成{new_message}工作,交付内容为:" + content,
-            agent_threads,
-            [agent_name],
-        )
-    )
+    _prefix = f"子智能体{speaker_label}完成{new_message}工作,交付内容为:"
+    # 广播给**所有**席位(含总控)的都只是**截断摘要** —— 总控 thread 不再囤积各席位全文,
+    # 每轮 leader 的上下文因此保持小、响应快。总控做最终综合需要全文时,由 _leader_run 在
+    # 「综合轮」按需现读各席位 thread 注入(见 _build_full_deliverables_block),不靠历史囤积。
+    brief_msg = _truncate_broadcast_content(_prefix, content)
+    # 先 yield 交付结果让前端立即收口本席位;后广播改 detached,不再 await 阻塞 ——
+    # 否则它读写各席位大状态会让 SSE 多挂十几秒,把下一轮 leader 的启动也一起拖慢。
     yield {
         "status": f"子智能体{speaker_label}完成{new_message}工作",
         "content": content,
         "agent_name": agent_name,
     }
-    try:
-        await post_broadcast_task
-    except Exception as exc:
-        logger.warning("post-broadcast failed: %s", exc)
-    logger.info(
-        "[special-timing] %s post_broadcast_total=%.3fs (broadcast time after yield — keeps SSE open this long)",
-        agent_name,
-        time.perf_counter() - t_before_post,
-    )
+    _spawn_detached_broadcast(base, headers, brief_msg, agent_threads, [agent_name])
 
 
 # ---------------------------------------------------------------------------
@@ -1292,6 +1626,10 @@ async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) ->
     display_name: str = str(payload.get("display_name") or "").strip()
     new_message: str = str(payload.get("new_message") or "")
     skill_stop_names: list[str] = payload.get("skill_stop_names") or []
+    # 综合轮标记(仅 leader 用):前端在「派活后让总控判断继续/综合」的轮次置真,网关据此
+    # 额外把各已交付席位的完整交付正文注入总控输入,供其写最终方案。缺省 False(派活决策轮
+    # 只给摘要,保持上下文精简)。旧前端不传 → False,行为安全退化。
+    synthesis_mode: bool = bool(payload.get("synthesis_mode") or False)
     # 不写死任何默认模型;空串会让 lead_agent._resolve_model_name() 回退到
     # config.yaml 的 models[0],与 /api/ai-writing 等其它路由一致。
     model_name: str = str(payload.get("model") or "")
@@ -1308,6 +1646,20 @@ async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) ->
     base = _loopback_base(request)
     headers = _auth_headers(request)
 
+    # 总控派活前,取本次各席位的元数据(id+名称+描述);_leader_run 会据此**实读各席位
+    # thread 的真实交付状态**,把「席位清单 + ✅/⬜交付状态」拼进总控本轮输入最前面 ——
+    # 既让弱模型每轮都能逐字复制真实 id,又以系统客观核对的交付状态压制臆造。元数据走同进程
+    # 批量查;任一席位查不到/不可见就退化为仅 id 清单,绝不因此阻断 leader run。
+    seat_meta: dict[str, dict[str, Any]] | None = None
+    if agent_type == "leader":
+        seat_ids = [k for k in agent_threads if k != agent_name]
+        if seat_ids:
+            try:
+                seat_meta = await _get_agents_batch_direct(request, seat_ids)
+            except Exception:
+                logger.warning("[leader-roster] 取席位元数据失败,退化为仅 id 清单", exc_info=True)
+                seat_meta = {sid: {"id": sid, "name": sid, "description": ""} for sid in seat_ids}
+
     async def stream_generator() -> AsyncIterator[str]:
         # TCP loopback 到 127.0.0.1:<gateway port>。client 配置见 _make_loopback_client：
         # trust_env=False 防 HTTP_PROXY 拦截,timeout=None 容纳 SSE 长连接。
@@ -1316,6 +1668,8 @@ async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) ->
                 if agent_type == "leader":
                     gen = _leader_run(
                         client, base, headers, agent_threads, agent_name, new_message, skill_stop_names, model_name,
+                        seat_meta=seat_meta,
+                        synthesis_mode=synthesis_mode,
                     )
                 else:
                     gen = _special_run(

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

@@ -224,6 +224,11 @@ _CONTEXT_CONFIGURABLE_KEYS: frozenset[str] = frozenset(
         "excluded_tools",
         "skill_stop_names",
         "skill_stop_message",
+        # 记忆开关:必须在白名单里才会从 body.context 透传到 run config,
+        # 否则 agent 端 cfg.get(...) 永远取默认值。圆桌等临时任务靠这两个键
+        # 关掉个人记忆召回 / builtin 记忆注入(见 multi_agent._run_payload)。
+        "memory_recall_disabled",
+        "memory_injection_enabled",
     }
 )
 

offline-backend-20260512/backend/packages/harness/deerflow/runtime/checkpointer/async_provider.py

@@ -17,12 +17,12 @@ For sync usage see :mod:`deerflow.runtime.checkpointer.provider`.
 
 from __future__ import annotations
 
-import asyncio
-import contextlib
-import logging
-from collections.abc import AsyncIterator
-import os
-from pathlib import Path
+import asyncio
+import contextlib
+import logging
+from collections.abc import AsyncIterator
+import os
+from pathlib import Path
 
 from langgraph.types import Checkpointer
 
@@ -32,70 +32,117 @@ from deerflow.runtime.checkpointer.provider import (
     POSTGRES_INSTALL,
     SQLITE_INSTALL,
 )
-from deerflow.runtime.store._sqlite_utils import ensure_sqlite_parent_dir, resolve_sqlite_conn_str
+from deerflow.runtime.store._sqlite_utils import ensure_sqlite_parent_dir, resolve_sqlite_conn_str
 
-logger = logging.getLogger(__name__)
-
-
-def _csv_env(name: str) -> list[str]:
-    raw = os.getenv(name, "")
-    return [part.strip() for part in raw.split(",") if part.strip()]
-
-
-def _dual_read_enabled(config) -> bool:
-    if getattr(config, "dual_read_enabled", False):
-        return True
-    return os.getenv("DEER_FLOW_CHECKPOINT_DUAL_READ", "").strip().lower() in {"1", "true", "yes", "on"}
-
-
-def _use_postgres_enabled(config) -> bool:
-    raw = os.getenv("DEER_FLOW_USE_PG_CHECKPOINTER")
-    if raw:
-        return raw.strip().lower() not in {"0", "false", "no", "off"}
-    return bool(getattr(config, "use_postgres", True))
-
-
-def _dual_read_paths(config) -> list[str]:
-    paths = list(getattr(config, "dual_read_sqlite_paths", []) or [])
-    paths.extend(_csv_env("DEER_FLOW_CHECKPOINT_SQLITE_PATHS"))
-    resolved: list[str] = []
-    seen: set[str] = set()
-    for path in paths:
-        conn_str = resolve_sqlite_conn_str(path)
-        if conn_str == ":memory:" or conn_str.startswith("file:"):
-            logger.warning("Skipping unsupported checkpoint migration SQLite path: %s", path)
-            continue
-        if conn_str not in seen:
-            resolved.append(conn_str)
-            seen.add(conn_str)
-    return resolved
-
-
-def _sqlite_conn_str_for_postgres_fallback(config) -> str:
-    configured = getattr(config, "sqlite_connection_string", None)
-    if configured:
-        return configured
-    paths = _dual_read_paths(config)
-    return paths[0] if paths else "store.db"
-
-
-@contextlib.asynccontextmanager
-async def _sqlite_checkpointer_from_conn_str(raw: str) -> AsyncIterator[Checkpointer]:
-    try:
-        from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
-    except ImportError as exc:
-        raise ImportError(SQLITE_INSTALL) from exc
-
-    conn_str = resolve_sqlite_conn_str(raw)
-    await asyncio.to_thread(ensure_sqlite_parent_dir, conn_str)
-    async with AsyncSqliteSaver.from_conn_string(conn_str) as saver:
-        await saver.setup()
-        yield saver
-
-
-# ---------------------------------------------------------------------------
-# Async factory
-# ---------------------------------------------------------------------------
+logger = logging.getLogger(__name__)
+
+
+def _csv_env(name: str) -> list[str]:
+    raw = os.getenv(name, "")
+    return [part.strip() for part in raw.split(",") if part.strip()]
+
+
+def _dual_read_enabled(config) -> bool:
+    if getattr(config, "dual_read_enabled", False):
+        return True
+    return os.getenv("DEER_FLOW_CHECKPOINT_DUAL_READ", "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _use_postgres_enabled(config) -> bool:
+    raw = os.getenv("DEER_FLOW_USE_PG_CHECKPOINTER")
+    if raw:
+        return raw.strip().lower() not in {"0", "false", "no", "off"}
+    return bool(getattr(config, "use_postgres", True))
+
+
+def _dual_read_paths(config) -> list[str]:
+    paths = list(getattr(config, "dual_read_sqlite_paths", []) or [])
+    paths.extend(_csv_env("DEER_FLOW_CHECKPOINT_SQLITE_PATHS"))
+    resolved: list[str] = []
+    seen: set[str] = set()
+    for path in paths:
+        conn_str = resolve_sqlite_conn_str(path)
+        if conn_str == ":memory:" or conn_str.startswith("file:"):
+            logger.warning("Skipping unsupported checkpoint migration SQLite path: %s", path)
+            continue
+        if conn_str not in seen:
+            resolved.append(conn_str)
+            seen.add(conn_str)
+    return resolved
+
+
+def _sqlite_conn_str_for_postgres_fallback(config) -> str:
+    configured = getattr(config, "sqlite_connection_string", None)
+    if configured:
+        return configured
+    paths = _dual_read_paths(config)
+    return paths[0] if paths else "store.db"
+
+
+@contextlib.asynccontextmanager
+async def _pg_pool_saver(conn_str: str) -> AsyncIterator[Checkpointer]:
+    """Pool-backed ``AsyncPostgresSaver``, resilient to idle-closed connections.
+
+    ``AsyncPostgresSaver.from_conn_string`` opens **one** long-lived connection.
+    对远程 Postgres(本项目 47.88.25.99 这类),空闲几分钟后连接会被服务端 / NAT /
+    连接池中间件断开,而 psycopg **不会自动重连** —— 下一次 checkpoint 操作就抛
+    ``OperationalError: the connection is closed``,整个 run 直接失败(实测:圆桌第二步
+    完成后闲置 ~6 分钟,用户再人工干预 → leader run 读 checkpoint 时撞死连接 → 前端
+    "(无文本输出)")。
+
+    用带健康检查的连接池根治:``check=AsyncConnectionPool.check_connection`` 在把连接交给
+    调用方前先验活,坏连接直接丢弃重建;``max_idle`` 主动回收久闲连接,赶在远程 idle 超时
+    之前。这样上层永远拿不到失效连接。``prepare_threshold=0`` 关预备语句,兼容 pgbouncer
+    一类连接池(8432 非标准端口很可能就是)。
+    """
+    try:
+        from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
+    except ImportError as exc:
+        raise ImportError(POSTGRES_INSTALL) from exc
+    from psycopg_pool import AsyncConnectionPool
+
+    try:
+        max_size = int(os.getenv("DEER_FLOW_PG_POOL_MAX", "20"))
+    except ValueError:
+        max_size = 20
+
+    pool = AsyncConnectionPool(
+        conninfo=conn_str,
+        min_size=1,
+        max_size=max_size,
+        open=False,
+        kwargs={"autocommit": True, "prepare_threshold": 0},
+        check=AsyncConnectionPool.check_connection,
+        max_idle=120.0,
+        max_lifetime=1800.0,
+    )
+    await pool.open(wait=True)
+    try:
+        saver = AsyncPostgresSaver(pool)
+        await saver.setup()
+        logger.info("Checkpointer: Postgres connection pool ready (max_size=%d, health-checked)", max_size)
+        yield saver
+    finally:
+        await pool.close()
+
+
+@contextlib.asynccontextmanager
+async def _sqlite_checkpointer_from_conn_str(raw: str) -> AsyncIterator[Checkpointer]:
+    try:
+        from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+    except ImportError as exc:
+        raise ImportError(SQLITE_INSTALL) from exc
+
+    conn_str = resolve_sqlite_conn_str(raw)
+    await asyncio.to_thread(ensure_sqlite_parent_dir, conn_str)
+    async with AsyncSqliteSaver.from_conn_string(conn_str) as saver:
+        await saver.setup()
+        yield saver
+
+
+# ---------------------------------------------------------------------------
+# Async factory
+# ---------------------------------------------------------------------------
 
 
 @contextlib.asynccontextmanager
@@ -107,71 +154,66 @@ async def _async_checkpointer(config) -> AsyncIterator[Checkpointer]:
         yield InMemorySaver()
         return
 
-    if config.type == "sqlite":
-        async with _sqlite_checkpointer_from_conn_str(config.connection_string or "store.db") as saver:
-            yield saver
-        return
-
-    if config.type == "postgres":
-        if not _use_postgres_enabled(config):
-            conn_str = _sqlite_conn_str_for_postgres_fallback(config)
-            logger.info("Checkpointer: use_postgres=false; using SQLite fallback (%s)", conn_str)
-            async with _sqlite_checkpointer_from_conn_str(conn_str) as saver:
-                yield saver
-            return
-
-        try:
-            from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
-        except ImportError as exc:
-            raise ImportError(POSTGRES_INSTALL) from exc
+    if config.type == "sqlite":
+        async with _sqlite_checkpointer_from_conn_str(config.connection_string or "store.db") as saver:
+            yield saver
+        return
 
-        if not config.connection_string:
-            raise ValueError(POSTGRES_CONN_REQUIRED)
-
-        async with AsyncPostgresSaver.from_conn_string(config.connection_string) as saver:
-            await saver.setup()
-            if not _dual_read_enabled(config):
-                yield saver
-                return
-
-            from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
-
-            from deerflow.runtime.checkpointer.checkpoint_migration import (
-                DualReadCheckpointer,
-                SQLiteCheckpointSource,
-            )
-
-            sources: list[SQLiteCheckpointSource] = []
-            for sqlite_path in _dual_read_paths(config):
-                if not Path(sqlite_path).exists():
-                    logger.info("Checkpoint dual-read source does not exist, skipping: %s", sqlite_path)
-                    continue
-                ctx = AsyncSqliteSaver.from_conn_string(sqlite_path)
-                source_saver = await ctx.__aenter__()
-                await source_saver.setup()
-                sources.append(
-                    SQLiteCheckpointSource(
-                        path=sqlite_path,
-                        saver=source_saver,
-                        context=ctx,
-                        progress_path=Path(sqlite_path).with_suffix(Path(sqlite_path).suffix + ".pg-migration.json"),
-                    )
-                )
-
-            if not sources:
-                logger.info("Checkpoint dual-read enabled but no SQLite sources were opened; using Postgres only")
-                yield saver
-                return
-
-            proxy = DualReadCheckpointer(
-                saver,
-                sources,
-            )
-            try:
-                yield proxy
-            finally:
-                await proxy.aclose_sources()
-        return
+    if config.type == "postgres":
+        if not _use_postgres_enabled(config):
+            conn_str = _sqlite_conn_str_for_postgres_fallback(config)
+            logger.info("Checkpointer: use_postgres=false; using SQLite fallback (%s)", conn_str)
+            async with _sqlite_checkpointer_from_conn_str(conn_str) as saver:
+                yield saver
+            return
+
+        if not config.connection_string:
+            raise ValueError(POSTGRES_CONN_REQUIRED)
+
+        # 连接池 + 健康检查,根治远程 Postgres 单连接空闲被关导致的 run 崩溃(见 _pg_pool_saver)。
+        async with _pg_pool_saver(config.connection_string) as saver:
+            if not _dual_read_enabled(config):
+                yield saver
+                return
+
+            from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+
+            from deerflow.runtime.checkpointer.checkpoint_migration import (
+                DualReadCheckpointer,
+                SQLiteCheckpointSource,
+            )
+
+            sources: list[SQLiteCheckpointSource] = []
+            for sqlite_path in _dual_read_paths(config):
+                if not Path(sqlite_path).exists():
+                    logger.info("Checkpoint dual-read source does not exist, skipping: %s", sqlite_path)
+                    continue
+                ctx = AsyncSqliteSaver.from_conn_string(sqlite_path)
+                source_saver = await ctx.__aenter__()
+                await source_saver.setup()
+                sources.append(
+                    SQLiteCheckpointSource(
+                        path=sqlite_path,
+                        saver=source_saver,
+                        context=ctx,
+                        progress_path=Path(sqlite_path).with_suffix(Path(sqlite_path).suffix + ".pg-migration.json"),
+                    )
+                )
+
+            if not sources:
+                logger.info("Checkpoint dual-read enabled but no SQLite sources were opened; using Postgres only")
+                yield saver
+                return
+
+            proxy = DualReadCheckpointer(
+                saver,
+                sources,
+            )
+            try:
+                yield proxy
+            finally:
+                await proxy.aclose_sources()
+        return
 
     raise ValueError(f"Unknown checkpointer type: {config.type!r}")
 
@@ -204,16 +246,11 @@ async def _async_checkpointer_from_database(db_config) -> AsyncIterator[Checkpoi
         return
 
     if db_config.backend == "postgres":
-        try:
-            from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
-        except ImportError as exc:
-            raise ImportError(POSTGRES_INSTALL) from exc
-
         if not db_config.postgres_url:
             raise ValueError("database.postgres_url is required for the postgres backend")
 
-        async with AsyncPostgresSaver.from_conn_string(db_config.postgres_url) as saver:
-            await saver.setup()
+        # 连接池 + 健康检查(见 _pg_pool_saver),避免单连接空闲被关。
+        async with _pg_pool_saver(db_config.postgres_url) as saver:
             yield saver
         return