fix: 会商席位与总控角色隔离,编排能力只归总控

This commit is contained in:
董宏凯 committed 2026-08-30 20:02:09 +08:00
1 parent ff80cc110d
commit 3384f2e51c
10 files changed
+440 -77

No files matched your search

+2
View File
@@ -119,6 +119,8 @@ offline-backend-20260512/backend/tests/*
!offline-backend-20260512/backend/tests/test_enterprise_research_tasks.py !offline-backend-20260512/backend/tests/test_enterprise_research_tasks.py
# Position-roundtable per-turn write/present cap (weak-model duplicate artifacts). # Position-roundtable per-turn write/present cap (weak-model duplicate artifacts).
!offline-backend-20260512/backend/tests/test_position_artifact_cap.py !offline-backend-20260512/backend/tests/test_position_artifact_cap.py
# Roundtable coordinator/seat capability isolation regressions.
!offline-backend-20260512/backend/tests/test_roundtable_role_boundary.py
# Workflow Studio: schemas / definitions / engine / run-lifecycle regressions # Workflow Studio: schemas / definitions / engine / run-lifecycle regressions
# plus the concurrency remediation suite (atomic resume, draft CAS, publish # plus the concurrency remediation suite (atomic resume, draft CAS, publish
# counter, event seq) — required in CI per docs/WORKFLOW_STUDIO_BACKEND_REMEDIATION_ZH.md P1-06. # counter, event seq) — required in CI per docs/WORKFLOW_STUDIO_BACKEND_REMEDIATION_ZH.md P1-06.
@@ -254,8 +254,7 @@ export interface StreamMultiAgentRequest {
*/ */
synthesisMode?: boolean; synthesisMode?: boolean;
/** /**
* 仅 leader 用。置 true 抑制「把总控本轮输入广播给各席位 thread」的预广播。DAG「每 stage * 仅为兼容旧协议保留。总控 prompt 属于总控私有上下文,前端恒传 true,后端也会无条件隔离。
* 派活轮」置真——否则总控的派活 prompt 会进席位历史,弱模型据此模仿派活。后端缺省 false。
*/ */
suppressPreBroadcast?: boolean; suppressPreBroadcast?: boolean;
/** /**
@@ -532,8 +531,8 @@ export async function streamMultiAgent(
skill_stop_names: req.skillStopNames ?? [], skill_stop_names: req.skillStopNames ?? [],
// 仅 leader 用;综合轮让后端注入各席位完整交付正文。省略 → 后端按 false 处理。 // 仅 leader 用;综合轮让后端注入各席位完整交付正文。省略 → 后端按 false 处理。
synthesis_mode: req.synthesisMode ?? false, synthesis_mode: req.synthesisMode ?? false,
// 仅 leader 用;DAG 派活轮抑制预广播,避免总控派活 prompt 污染席位 thread。省略 → false。 // 总控 prompt 永不进入席位 thread;后端同样强制执行,避免旧调用方破坏角色隔离。
suppress_pre_broadcast: req.suppressPreBroadcast ?? false, suppress_pre_broadcast: req.suppressPreBroadcast ?? true,
// 仅 special 用;DAG 并行批次禁写文件(seat_parallel policy)。省略 → 后端按 false 处理。 // 仅 special 用;DAG 并行批次禁写文件(seat_parallel policy)。省略 → 后端按 false 处理。
parallel_no_file: req.parallelNoFile ?? false, parallel_no_file: req.parallelNoFile ?? false,
// 仅 special 用;前序席位完整交付,后端注入 context 供 read_peer_delivery 按需取全文。 // 仅 special 用;前序席位完整交付,后端注入 context 供 read_peer_delivery 按需取全文。
@@ -31,8 +31,7 @@ from app.gateway.roundtable_model_fallback import (
reason_text, reason_text,
) )
from app.gateway.roundtable_run_policy import roundtable_run_policy from app.gateway.roundtable_run_policy import roundtable_run_policy
from app.gateway.roundtable_seat_skills import append_seat_skill_directive from app.gateway.roundtable_seat_skills import append_seat_role_boundary, append_seat_skill_directive
from app.gateway.services import RUN_SCOPE_BACKGROUND_CHILD
from app.gateway.routers._roundtable_seed import ( from app.gateway.routers._roundtable_seed import (
COORDINATOR_AGENT_ID, COORDINATOR_AGENT_ID,
DASHBOARD_AGENT_ID, DASHBOARD_AGENT_ID,
@@ -40,6 +39,7 @@ from app.gateway.routers._roundtable_seed import (
SUMMARY_AGENT_ID, SUMMARY_AGENT_ID,
ensure_roundtable_functional_agents, ensure_roundtable_functional_agents,
) )
from app.gateway.services import RUN_SCOPE_BACKGROUND_CHILD
from deerflow.agents.roundtable_orchestrator.delivery import classify_seat_delivery, visible_seat_text from deerflow.agents.roundtable_orchestrator.delivery import classify_seat_delivery, visible_seat_text
from deerflow.agents.roundtable_orchestrator.report_json import extract_report_json from deerflow.agents.roundtable_orchestrator.report_json import extract_report_json
from deerflow.agents.roundtable_orchestrator.types import ( from deerflow.agents.roundtable_orchestrator.types import (
@@ -806,6 +806,8 @@ class InProcessRoundtableGateway:
seat_task = append_seat_skill_directive( seat_task = append_seat_skill_directive(
agent_id, task, chain_enabled=bool(self._seat_skill_directives.get(agent_id, False)) agent_id, task, chain_enabled=bool(self._seat_skill_directives.get(agent_id, False))
) )
# 后台挂起路径与前台保持同一角色边界:席位只能执行具体任务,编排能力永远只属于总控。
seat_task = append_seat_role_boundary(seat_task)
# 每席位「推理深度覆盖」:业务链条里给该席位单独配了 seatMode → 用该档派生的 # 每席位「推理深度覆盖」:业务链条里给该席位单独配了 seatMode → 用该档派生的
# thinking/reasoning/subagent **覆盖作业级**(self._seat_*);未配则跟随作业级。 # thinking/reasoning/subagent **覆盖作业级**(self._seat_*);未配则跟随作业级。
# 作业级仍为 None 时落 policy 默认。与前端 useStep2Orchestration「覆盖优先、否则跟随 # 作业级仍为 None 时落 policy 默认。与前端 useStep2Orchestration「覆盖优先、否则跟随
@@ -37,6 +37,22 @@ logger = logging.getLogger(__name__)
_TRUTHY = frozenset({"1", "true", "yes", "on", "y"}) _TRUTHY = frozenset({"1", "true", "yes", "on", "y"})
_FALSY = frozenset({"0", "false", "no", "off", "n"}) _FALSY = frozenset({"0", "false", "no", "off", "n"})
# 编排能力是圆桌总控的专属能力。两种拼写分别对应内置 tool 名和历史 custom skill
# 目录名;席位侧无论 agent 配置或业务链条开关如何,都不得把它们暴露进技能强化提示。
COORDINATOR_ONLY_SKILL_NAMES = frozenset({"agent_orchestration", "agent-orchestration"})
_SEAT_ROLE_BOUNDARY = (
"【圆桌席位角色边界(最高优先级)】你是负责完成具体任务的执行席位,不是总控智能体。"
"编排、派活和调度其它智能体的能力只允许总控智能体使用。"
"你不得加载、读取或调用 agent_orchestration / agent-orchestration,"
"不得向其它智能体派活,也不要声称准备使用编排技能;请直接完成上方交给你的具体任务并提交结果。"
)
def append_seat_role_boundary(task: str) -> str:
"""在席位本轮 human 消息末尾追加不可越权的角色边界。"""
return f"{task}\n\n---\n\n{_SEAT_ROLE_BOUNDARY}"
def _master_enabled() -> bool: def _master_enabled() -> bool:
"""``ROUNDTABLE_SEAT_SKILL_DIRECTIVE`` 总开关:默认放行(kill-switch)。 """``ROUNDTABLE_SEAT_SKILL_DIRECTIVE`` 总开关:默认放行(kill-switch)。
@@ -65,7 +81,11 @@ def build_seat_skill_directive(agent_id: str | None, *, chain_enabled: bool = Fa
if not (agent_enabled or chain_enabled): if not (agent_enabled or chain_enabled):
return "" return ""
configured = [name for name in (cfg.skills or []) if name] if cfg else [] configured = [
name
for name in (cfg.skills or [])
if name and name.strip().lower() not in COORDINATOR_ONLY_SKILL_NAMES
] if cfg else []
if not configured: if not configured:
return "" return ""
@@ -25,8 +25,8 @@
------------------ ------------------
通过 ``_append_thread_message`` 向其它 thread 的 checkpoint ``messages`` 追加 通过 ``_append_thread_message`` 向其它 thread 的 checkpoint ``messages`` 追加
**只读上下文**的 human 消息(非触发 run),使各席位 LangGraph 状态里能看到 **只读上下文**的 human 消息(非触发 run),使各席位 LangGraph 状态里能看到
总控派活/用户干预/子 agent 完成摘要。前端编排循环仍靠显式多次 ``/run/stream`` 子 agent 完成摘要。总控输入、派活 prompt 和派活说明只留在总控 thread,
驱动各 agent 真正推理。 绝不广播给席位;前端编排循环仍靠显式多次 ``/run/stream`` 驱动各 agent 真正推理。
Leader 终态帧(前端 ``isFinalStatusFrame``) Leader 终态帧(前端 ``isFinalStatusFrame``)
------------------------------------------- -------------------------------------------
@@ -48,7 +48,6 @@ Leader 终态帧(前端 ``isFinalStatusFrame``)
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import contextlib
import json import json
import logging import logging
import os import os
@@ -67,8 +66,7 @@ from langgraph.checkpoint.base import empty_checkpoint
from app.gateway.roundtable_diag import record_foreground_diag from app.gateway.roundtable_diag import record_foreground_diag
from app.gateway.roundtable_model_fallback import fallback_model_chain, is_llm_error_message, reason_text from app.gateway.roundtable_model_fallback import fallback_model_chain, is_llm_error_message, reason_text
from app.gateway.roundtable_run_policy import merge_skill_stop_names, roundtable_run_policy from app.gateway.roundtable_run_policy import merge_skill_stop_names, roundtable_run_policy
from app.gateway.services import RUN_SCOPE_ORCHESTRATION_CHILD, internal_run_scope_headers from app.gateway.roundtable_seat_skills import append_seat_role_boundary, append_seat_skill_directive
from app.gateway.roundtable_seat_skills import append_seat_skill_directive
from app.gateway.routers._roundtable_seed import ( from app.gateway.routers._roundtable_seed import (
ACTION_PLAN_AGENT_ID, ACTION_PLAN_AGENT_ID,
COORDINATOR_AGENT_ID, COORDINATOR_AGENT_ID,
@@ -78,6 +76,7 @@ from app.gateway.routers._roundtable_seed import (
SUMMARY_AGENT_ID, SUMMARY_AGENT_ID,
ensure_roundtable_functional_agents, ensure_roundtable_functional_agents,
) )
from app.gateway.services import RUN_SCOPE_ORCHESTRATION_CHILD, internal_run_scope_headers
from deerflow.agents.roundtable_orchestrator.delivery import classify_seat_delivery, visible_seat_text from deerflow.agents.roundtable_orchestrator.delivery import classify_seat_delivery, visible_seat_text
from deerflow.agents.roundtable_orchestrator.extraction_spec import build_summary_directive from deerflow.agents.roundtable_orchestrator.extraction_spec import build_summary_directive
from deerflow.config import get_app_config from deerflow.config import get_app_config
@@ -1554,16 +1553,15 @@ async def _leader_run(
model_name: str, model_name: str,
seat_meta: dict[str, dict[str, Any]] | None = None, seat_meta: dict[str, dict[str, Any]] | None = None,
synthesis_mode: bool = False, synthesis_mode: bool = False,
suppress_pre_broadcast: bool = False, suppress_pre_broadcast: bool = True,
files: list[dict[str, Any]] | None = None, files: list[dict[str, Any]] | None = None,
) -> AsyncIterator[str | dict[str, Any]]: ) -> AsyncIterator[str | dict[str, Any]]:
"""总控(leader)单轮:透传上游 SSE + 末尾 yield 一个 dict 状态帧(由路由层 ``json.dumps``)。 """总控(leader)单轮:透传上游 SSE + 末尾 yield 一个 dict 状态帧(由路由层 ``json.dumps``)。
时序要点: 时序要点:
1. **预广播**(与 LLM 并行):把用户/综合消息写入其它席位 thread,skip 总控自身; 1. 总控输入只进入总控 thread,绝不广播到席位 thread;
2. 跑总控 thread 的 stream,yield 每一行原始 SSE; 2. 跑总控 thread 的 stream,yield 每一行原始 SSE;
3. 解析 messages → 澄清 / 空 dispatch(共识)/ 派活列表; 3. 解析 messages → 澄清 / 空 dispatch(共识)/ 派活列表并返回。
4. 若有派活,**派活广播**后再 yield ``{status: dispatched, ...}``。
``skill_stop_names`` 会与 ``agent_orchestration`` 合并,确保 middleware 在工具调用后停图。 ``skill_stop_names`` 会与 ``agent_orchestration`` 合并,确保 middleware 在工具调用后停图。
@@ -1575,6 +1573,10 @@ async def _leader_run(
``synthesis_mode``:前端在「派活后让总控判断继续/综合」的轮次置真。为真时**额外**把各 ``synthesis_mode``:前端在「派活后让总控判断继续/综合」的轮次置真。为真时**额外**把各
已交付席位的**完整交付正文**注入本轮输入(``_build_full_deliverables_block``),供总控写 已交付席位的**完整交付正文**注入本轮输入(``_build_full_deliverables_block``),供总控写
最终方案;平时(派活决策轮)只给摘要,让总控 thread 与每轮上下文保持精简、响应快。 最终方案;平时(派活决策轮)只给摘要,让总控 thread 与每轮上下文保持精简、响应快。
``suppress_pre_broadcast`` 仅为兼容旧客户端保留,服务端始终按抑制处理。角色隔离不能由
客户端布尔值决定:总控 prompt 可能含 ``agent_orchestration``,进入席位历史会诱导弱模型
模仿总控加载编排能力。
""" """
t0 = time.perf_counter() t0 = time.perf_counter()
# 实读各席位 thread 的真实交付状态,构造「席位清单 + 客观交付状态(摘要)」块;综合轮再额外 # 实读各席位 thread 的真实交付状态,构造「席位清单 + 客观交付状态(摘要)」块;综合轮再额外
@@ -1623,21 +1625,9 @@ async def _leader_run(
part for part in (seat_status_block, full_deliverables_block, new_message) if part part for part in (seat_status_block, full_deliverables_block, new_message) if part
) )
# 与上游 LLM 并行,减少「总控已开始想但席位 thread 里还没有用户消息」的窗口。 # 兼容旧请求字段但不再信任它。总控输入与派活说明始终只留在总控 thread;席位需要的上下文
# # 由总控下发的具体 task、peer_deliveries 和席位交付广播提供。
# ``suppress_pre_broadcast``:DAG「每 stage 派活轮」(方案 A)置真——此时 ``new_message`` 是 _ = suppress_pre_broadcast
# 「请你作为总控…通过 agent_orchestration 派活给席位…」的**派活 prompt**,若广播进各席位
# thread,弱模型(deepseek)会从历史里捡到「派活」意图,**席位跟着模仿派活 / 绕道 bash 调
# 内部接口**(已观测 bug-1~4)。DAG 下席位的任务由前端 ``pickDagSeatTask`` 直接、干净地下发
# (含任务背景),此处不需要、也不应把总控的派活 prompt 播给席位。recommend/chain 不传该 flag,
# 行为不变。
pre_broadcast_task = (
None
if suppress_pre_broadcast
else asyncio.create_task(
_broadcast(client, base, headers, "总控智能体收到用户消息:" + new_message, agent_threads, [agent_name])
)
)
thread_id = agent_threads[agent_name] thread_id = agent_threads[agent_name]
# 角色参数取自单一数据源 roundtable_run_policy(leader:禁 web_search、停 # 角色参数取自单一数据源 roundtable_run_policy(leader:禁 web_search、停
# agent_orchestration 图、关 thinking + low reasoning——短决策、首字节快)。 # agent_orchestration 图、关 thinking + low reasoning——短决策、首字节快)。
@@ -1677,7 +1667,7 @@ async def _leader_run(
body = _mk_body(current_model) body = _mk_body(current_model)
t_payload_built = time.perf_counter() t_payload_built = time.perf_counter()
logger.info( logger.info(
"[leader-timing] %s payload_built=%.3fs (broadcast running in background)", "[leader-timing] %s payload_built=%.3fs",
agent_name, agent_name,
t_payload_built - t0, t_payload_built - t0,
) )
@@ -1733,27 +1723,8 @@ async def _leader_run(
"(expected on rollback preemption; otherwise check the upstream run's server-side error)", "(expected on rollback preemption; otherwise check the upstream run's server-side error)",
agent_name, agent_name,
) )
if pre_broadcast_task is not None:
with contextlib.suppress(Exception):
await pre_broadcast_task
return 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 and pre_broadcast_task is not None:
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) dispatch_msg = _find_dispatch_message(messages)
leader_text = _flatten_content(dispatch_msg.get("content", "")) leader_text = _flatten_content(dispatch_msg.get("content", ""))
tool_calls = dispatch_msg.get("tool_calls") or [] tool_calls = dispatch_msg.get("tool_calls") or []
@@ -1839,22 +1810,9 @@ async def _leader_run(
# 非法 id 已被丢弃;若 roster 校验后无合法派活(且重试已用尽),退化为空共识, # 非法 id 已被丢弃;若 roster 校验后无合法派活(且重试已用尽),退化为空共识,
# 而不是把假 id 漏给前端再报 400。 # 而不是把假 id 漏给前端再报 400。
dispatched: list[list[str]] = [[sub_name, task_text] for sub_name, task_text in valid_dispatch] dispatched: list[list[str]] = [[sub_name, task_text] for sub_name, task_text in valid_dispatch]
# 关键路径优先:**先 yield 派活帧**,让前端立刻启动席位 special run;派活广播(告知 # 关键路径优先:yield 派活帧,让前端立刻启动席位 special run。总控派活说明不广播给
# 其它席位"总控给 X 派了什么活")改成 detached 后台任务,不再 await 阻塞 —— 否则广播 # 任何席位,避免席位从历史中模仿总控角色;席位间承上依靠完成交付广播 / peer_deliveries。
# 读写各席位巨大 thread 状态会卡几十秒,白白拖慢席位启动(见 _spawn_detached_broadcast)。
yield {"status": dispatched, "content": leader_text, "agent_name": agent_name, "model_used": current_model} yield {"status": dispatched, "content": leader_text, "agent_name": agent_name, "model_used": current_model}
# DAG 派活轮(suppress_pre_broadcast)同样**不**把「总控给 X 派活」广播给兄弟席位 ——
# 否则弱模型从历史里看到派活语句又会模仿派活(席位任务由前端 pickDagSeatTask 干净下发,
# 跨席位承上靠各 special run 的"完成交付"后广播,不需要这条派活广播)。recommend/chain 照旧。
if not suppress_pre_broadcast:
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 return
@@ -1937,6 +1895,8 @@ async def _special_run(
seat_message = new_message seat_message = new_message
if role in ("seat", "seat_parallel"): if role in ("seat", "seat_parallel"):
seat_message = append_seat_skill_directive(agent_name, new_message, chain_enabled=seat_skill_directive) seat_message = append_seat_skill_directive(agent_name, new_message, chain_enabled=seat_skill_directive)
# 角色边界放在所有技能强化之后,确保本轮最后一条指令明确:席位永远不能加载/调用编排能力。
seat_message = append_seat_role_boundary(seat_message)
# 深链接业务规范(仅 summary 单例):把该业务的「业务链完整性核对」指令拼进本轮任务,强制总结报告 # 深链接业务规范(仅 summary 单例):把该业务的「业务链完整性核对」指令拼进本轮任务,强制总结报告
# 核对产出相对业务链完不完整。business_code 为空(普通会商)/ 非写实业务 → directive 为 "",原样不变。 # 核对产出相对业务链完不完整。business_code 为空(普通会商)/ 非写实业务 → directive 为 "",原样不变。
elif role == "summary" and business_code: elif role == "summary" and business_code:
@@ -2131,10 +2091,9 @@ async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) ->
# 置真,网关据此用 seat_parallel policy 额外禁 write_file/bash/str_replace(§4)。 # 置真,网关据此用 seat_parallel policy 额外禁 write_file/bash/str_replace(§4)。
# 兼容 no_file 别名;缺省 False → 普通 seat,旧前端不传也不破坏。 # 兼容 no_file 别名;缺省 False → 普通 seat,旧前端不传也不破坏。
no_file: bool = bool(payload.get("parallel_no_file") or payload.get("no_file") or False) no_file: bool = bool(payload.get("parallel_no_file") or payload.get("no_file") or False)
# 抑制 leader 预广播(仅 leader 用):DAG「每 stage 派活轮」置真,避免把总控的派活 prompt # 旧客户端兼容字段。服务端已强制隔离总控 thread,因此无论请求值为何都不会执行 leader→seat
# 广播进席位 thread 让弱模型模仿派活(详见 _leader_run 的 suppress_pre_broadcast 注释)。 # 广播;仍解析并透传只是保持调用签名稳定。
# 缺省 False,recommend/chain 行为不变。 suppress_pre_broadcast: bool = bool(payload.get("suppress_pre_broadcast", True))
suppress_pre_broadcast: bool = bool(payload.get("suppress_pre_broadcast") or False)
# 前序席位完整交付(仅 special 用):前端 DAG stage>0 席位带上此前各阶段的全文,网关注入 # 前序席位完整交付(仅 special 用):前端 DAG stage>0 席位带上此前各阶段的全文,网关注入
# run context,供 read_peer_delivery 工具按需取全文(平时只看广播摘要,§12.9)。规整成 # run context,供 read_peer_delivery 工具按需取全文(平时只看广播摘要,§12.9)。规整成
# [{name, content}] 且过滤空项;缺省 None。 # [{name, content}] 且过滤空项;缺省 None。
@@ -42,6 +42,73 @@ REPORT_STRUCTURE_SETUP_MAX_ROUNDS = 8
_ES_QUERY_ROUTING_OPEN_TAG = '<skill_routing skill="es_query">' _ES_QUERY_ROUTING_OPEN_TAG = '<skill_routing skill="es_query">'
_SKILLS_BLOCKED_BY_EXCLUDED_TOOL: dict[str, frozenset[str]] = {
# 内置 tool 与历史 custom skill 目录存在两种拼写;禁用编排工具时两者都必须从系统提示和
# runtime skill allowlist 中消失,防止误配的席位仍看到「加载编排技能」。
"agent_orchestration": frozenset({"agent_orchestration", "agent-orchestration"}),
}
_ROUNDTABLE_COORDINATOR_AGENT_ID = "roundtable-coordinator"
_NON_COORDINATOR_ORCHESTRATION_BOUNDARY = """
<orchestration_capability_boundary>
你不是多智能体圆桌总控,因此没有编排、派活或调度其它智能体的权限。
不得加载、读取、检索或调用 agent_orchestration / agent-orchestration,
不得声称准备使用编排技能;请直接完成当前问答或你的本职任务。
</orchestration_capability_boundary>
""".strip()
def _enforce_coordinator_only_tool_exclusions(
agent_id: str | None,
excluded_tools: list[str] | None,
) -> list[str] | None:
"""Make orchestration unavailable to every agent except the fixed roundtable coordinator."""
if agent_id == _ROUNDTABLE_COORDINATOR_AGENT_ID:
return excluded_tools
effective = list(excluded_tools or [])
if "agent_orchestration" not in effective:
effective.append("agent_orchestration")
return effective
def _append_orchestration_capability_boundary(system_prompt: str, agent_id: str | None) -> str:
"""Reinforce the physical tool/skill gate in every non-coordinator system prompt."""
if agent_id == _ROUNDTABLE_COORDINATOR_AGENT_ID:
return system_prompt
return f"{system_prompt}\n\n{_NON_COORDINATOR_ORCHESTRATION_BOUNDARY}"
def _filter_available_skills_for_excluded_tools(
available_skills: list[str] | None,
excluded_tools: list[str] | None,
) -> list[str] | None:
"""让工具级禁用同步收紧 skill allowlist,必要时把「全部」展开为安全列表。"""
if not excluded_tools:
return available_skills
excluded = {name.strip().lower() for name in excluded_tools if name}
blocked = {
skill_name
for tool_name, skill_names in _SKILLS_BLOCKED_BY_EXCLUDED_TOOL.items()
if tool_name in excluded
for skill_name in skill_names
}
if not blocked:
return available_skills
if available_skills is None:
# ``None`` 原本表示「全部启用技能可见」,无法表达「除了总控专属技能之外的全部」。
# 此时显式展开存储中的启用技能;扫描失败则 fail closed,避免角色隔离失效。
try:
from deerflow.skills.storage import get_or_new_skill_storage
available_skills = [
skill.name for skill in get_or_new_skill_storage().load_skills(enabled_only=True)
]
except Exception: # noqa: BLE001
logger.warning("Failed to expand skills while applying excluded-tool boundary", exc_info=True)
available_skills = []
return [name for name in available_skills if name.strip().lower() not in blocked]
def _log_es_query_routing_registration( def _log_es_query_routing_registration(
*, *,
@@ -536,7 +603,10 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
if not is_writing_mode_report_path(writing_artifact_path): if not is_writing_mode_report_path(writing_artifact_path):
writing_artifact_path = None writing_artifact_path = None
agent_id = validate_agent_id(cfg.get("agent_id") or cfg.get("agent_name")) agent_id = validate_agent_id(cfg.get("agent_id") or cfg.get("agent_name"))
requested_excluded_tools = cfg.get("excluded_tools") or None requested_excluded_tools = _enforce_coordinator_only_tool_exclusions(
agent_id,
cfg.get("excluded_tools") or None,
)
try: try:
agent_config = load_agent_config(agent_id) if not is_bootstrap and not is_writing_setup and not is_report_structure_setup else None agent_config = load_agent_config(agent_id) if not is_bootstrap and not is_writing_setup and not is_report_structure_setup else None
@@ -588,6 +658,10 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
runtime_available_skills = merge_collector_skill_allowlist( runtime_available_skills = merge_collector_skill_allowlist(
agent_id, runtime_available_skills, runtime_extra_skills(cfg) agent_id, runtime_available_skills, runtime_extra_skills(cfg)
) )
runtime_available_skills = _filter_available_skills_for_excluded_tools(
runtime_available_skills,
requested_excluded_tools,
)
prompt_available_skills = set(runtime_available_skills) if runtime_available_skills is not None else None prompt_available_skills = set(runtime_available_skills) if runtime_available_skills is not None else None
config["metadata"].update( config["metadata"].update(
@@ -674,6 +748,7 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
subagent_enabled=subagent_enabled, subagent_enabled=subagent_enabled,
is_scheduled_run=cfg.get("is_scheduled_run", False), is_scheduled_run=cfg.get("is_scheduled_run", False),
excluded_tools=requested_excluded_tools, excluded_tools=requested_excluded_tools,
allow_agent_orchestration=agent_id == _ROUNDTABLE_COORDINATOR_AGENT_ID,
app_config=resolved_app_config, app_config=resolved_app_config,
) )
+ [setup_agent], + [setup_agent],
@@ -701,6 +776,7 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
subagent_enabled=subagent_enabled, subagent_enabled=subagent_enabled,
is_scheduled_run=cfg.get("is_scheduled_run", False), is_scheduled_run=cfg.get("is_scheduled_run", False),
excluded_tools=requested_excluded_tools, excluded_tools=requested_excluded_tools,
allow_agent_orchestration=agent_id == _ROUNDTABLE_COORDINATOR_AGENT_ID,
app_config=resolved_app_config, app_config=resolved_app_config,
) )
if agent_id == FORCED_RESEARCH_AGENT_ID: if agent_id == FORCED_RESEARCH_AGENT_ID:
@@ -761,6 +837,8 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
is_scheduled_run=bool(cfg.get("is_scheduled_run", False)), is_scheduled_run=bool(cfg.get("is_scheduled_run", False)),
) )
system_prompt = _append_orchestration_capability_boundary(system_prompt, agent_id)
_log_es_query_routing_registration( _log_es_query_routing_registration(
system_prompt=system_prompt, system_prompt=system_prompt,
app_config=resolved_app_config, app_config=resolved_app_config,
@@ -1490,9 +1490,18 @@ def read_file_tool(
end_line: Optional ending line number (1-indexed, inclusive). Use with start_line to read a specific range. end_line: Optional ending line number (1-indexed, inclusive). Use with start_line to read a specific range.
""" """
try: try:
requested_path = path
if _is_skills_path(requested_path):
from deerflow.tools.builtins.skill_tools import (
_resolve_active_agent_id,
is_coordinator_only_skill_path,
)
if is_coordinator_only_skill_path(requested_path) and _resolve_active_agent_id(runtime) != "roundtable-coordinator":
# Match skill_view's non-leaking behavior: ordinary agents must not discover whether it exists.
return f"Error: File not found: {requested_path}"
sandbox = ensure_sandbox_initialized(runtime) sandbox = ensure_sandbox_initialized(runtime)
ensure_thread_directories_exist(runtime) ensure_thread_directories_exist(runtime)
requested_path = path
if is_local_sandbox(runtime): if is_local_sandbox(runtime):
thread_data = get_thread_data(runtime) thread_data = get_thread_data(runtime)
validate_local_tool_path(path, thread_data, read_only=True) validate_local_tool_path(path, thread_data, read_only=True)
@@ -9,6 +9,21 @@ from langchain.tools import ToolRuntime, tool
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_ROUNDTABLE_COORDINATOR_AGENT_ID = "roundtable-coordinator"
COORDINATOR_ONLY_SKILL_NAMES = frozenset({"agent_orchestration", "agent-orchestration"})
def can_access_coordinator_only_skill(agent_id: str | None, skill_name: str) -> bool:
"""Return whether this runtime identity may discover/read a reserved orchestration skill."""
normalized = (skill_name or "").strip().lower()
return normalized not in COORDINATOR_ONLY_SKILL_NAMES or agent_id == _ROUNDTABLE_COORDINATOR_AGENT_ID
def is_coordinator_only_skill_path(path: str) -> bool:
"""Recognize reserved skill directories in virtual/container paths."""
parts = {part.strip().lower() for part in (path or "").replace("\\", "/").split("/") if part}
return bool(parts & COORDINATOR_ONLY_SKILL_NAMES)
def _resolve_active_agent_id(runtime: ToolRuntime | None) -> str | None: def _resolve_active_agent_id(runtime: ToolRuntime | None) -> str | None:
"""Extract the agent_id under which the current tool call is running. """Extract the agent_id under which the current tool call is running.
@@ -22,12 +37,13 @@ def _resolve_active_agent_id(runtime: ToolRuntime | None) -> str | None:
return None return None
try: try:
if getattr(runtime, "context", None): if getattr(runtime, "context", None):
ctx_id = runtime.context.get("agent_id") ctx_id = runtime.context.get("agent_id") or runtime.context.get("agent_name")
if ctx_id: if ctx_id:
return str(ctx_id) return str(ctx_id)
cfg = getattr(runtime, "config", None) cfg = getattr(runtime, "config", None)
if isinstance(cfg, dict): if isinstance(cfg, dict):
cid = cfg.get("configurable", {}).get("agent_id") configurable = cfg.get("configurable", {})
cid = configurable.get("agent_id") or configurable.get("agent_name")
if cid: if cid:
return str(cid) return str(cid)
except Exception: except Exception:
@@ -61,7 +77,11 @@ def _load_agent_skill_whitelist(agent_id: str | None, extra_skills: list[str] |
cfg = load_agent_config(agent_id) cfg = load_agent_config(agent_id)
if cfg is None: if cfg is None:
return None return None
allowed = set(cfg.skills or []) allowed = {
name
for name in (cfg.skills or [])
if can_access_coordinator_only_skill(agent_id, name)
}
except Exception: except Exception:
logger.debug("Failed to load agent config for skill_list filter", exc_info=True) logger.debug("Failed to load agent config for skill_list filter", exc_info=True)
return None return None
@@ -137,6 +157,9 @@ def skill_view_impl(name: str, agent_id: str | None, extra_skills: list[str] | N
guessing skill names. Replies "not found" rather than "forbidden" to guessing skill names. Replies "not found" rather than "forbidden" to
avoid leaking which skills exist. avoid leaking which skills exist.
""" """
if not can_access_coordinator_only_skill(agent_id, name):
return f"Skill '{name}' not found."
from deerflow.skills.storage import get_or_new_skill_storage from deerflow.skills.storage import get_or_new_skill_storage
from deerflow.skills.usage import bump_view from deerflow.skills.usage import bump_view
@@ -151,6 +174,7 @@ def skill_view_impl(name: str, agent_id: str | None, extra_skills: list[str] | N
elif storage.public_skill_exists(name): elif storage.public_skill_exists(name):
skill_file = storage.get_skills_root_path() skill_file = storage.get_skills_root_path()
from pathlib import Path from pathlib import Path
from deerflow.skills.types import SKILL_MD_FILE from deerflow.skills.types import SKILL_MD_FILE
path = Path(skill_file) / "public" / name / SKILL_MD_FILE path = Path(skill_file) / "public" / name / SKILL_MD_FILE
@@ -241,6 +265,8 @@ def search_skills_impl(query: str, agent_id: str | None, top_n: int = 5, extra_s
base_path = _resolve_skills_container_base_path() base_path = _resolve_skills_container_base_path()
scored: list[tuple[int, dict]] = [] scored: list[tuple[int, dict]] = []
for skill in skills: for skill in skills:
if not can_access_coordinator_only_skill(agent_id, skill.name):
continue
entry = usage.get(skill.name, {}) entry = usage.get(skill.name, {})
if allowed is not None: if allowed is not None:
if skill.name not in allowed: if skill.name not in allowed:
@@ -331,6 +357,8 @@ def skill_list_impl(agent_id: str | None, extra_skills: list[str] | None = None)
usage = all_entries() usage = all_entries()
result = [] result = []
for skill in skills: for skill in skills:
if not can_access_coordinator_only_skill(agent_id, skill.name):
continue
entry = usage.get(skill.name, {}) entry = usage.get(skill.name, {})
if allowed is not None: if allowed is not None:
# 自定义 agent:严格按 config.yaml 的 skills 白名单过滤。 # 自定义 agent:严格按 config.yaml 的 skills 白名单过滤。
@@ -19,7 +19,6 @@ BROWSER_CONTEXT_TOOL_NAMES = {"browser_fetch_page", "browser_act"}
BUILTIN_TOOLS = [ BUILTIN_TOOLS = [
present_file_tool, present_file_tool,
ask_clarification_tool, ask_clarification_tool,
agent_orchestration_tool,
] ]
SUBAGENT_TOOLS = [ SUBAGENT_TOOLS = [
@@ -47,6 +46,17 @@ def _is_browser_context_tool(tool: object) -> bool:
return use in BROWSER_CONTEXT_TOOL_USES or name in BROWSER_CONTEXT_TOOL_NAMES return use in BROWSER_CONTEXT_TOOL_USES or name in BROWSER_CONTEXT_TOOL_NAMES
def _enforce_agent_orchestration_scope(
tools: list[BaseTool],
*,
allow_agent_orchestration: bool,
) -> list[BaseTool]:
"""Fail closed: only an explicitly authorized coordinator may receive orchestration."""
if allow_agent_orchestration:
return tools
return [tool for tool in tools if tool.name != "agent_orchestration"]
def get_available_tools( def get_available_tools(
groups: list[str] | None = None, groups: list[str] | None = None,
include_mcp: bool = True, include_mcp: bool = True,
@@ -55,6 +65,7 @@ def get_available_tools(
is_scheduled_run: bool = False, is_scheduled_run: bool = False,
excluded_tools: list[str] | None = None, excluded_tools: list[str] | None = None,
*, *,
allow_agent_orchestration: bool = False,
app_config: AppConfig | None = None, app_config: AppConfig | None = None,
) -> list[BaseTool]: ) -> list[BaseTool]:
"""Get all available tools from config. """Get all available tools from config.
@@ -70,6 +81,8 @@ def get_available_tools(
excluded_tools: Optional list of tool names to exclude from the result excluded_tools: Optional list of tool names to exclude from the result
(applied AFTER all other resolution). Takes precedence over (applied AFTER all other resolution). Takes precedence over
``app_config.excluded_tools``. ``app_config.excluded_tools``.
allow_agent_orchestration: Explicit coordinator-only capability gate. Defaults to
``False`` so ordinary agents and all indirect callers fail closed.
Returns: Returns:
List of available tools. List of available tools.
@@ -116,6 +129,8 @@ def get_available_tools(
# Conditionally add tools based on config # Conditionally add tools based on config
builtin_tools = BUILTIN_TOOLS.copy() builtin_tools = BUILTIN_TOOLS.copy()
if allow_agent_orchestration:
builtin_tools.append(agent_orchestration_tool)
if not is_scheduled_run: if not is_scheduled_run:
builtin_tools.append(scheduled_task_tool) builtin_tools.append(scheduled_task_tool)
@@ -292,4 +307,8 @@ def get_available_tools(
) )
unique_tools = filtered_tools unique_tools = filtered_tools
return unique_tools # Defense in depth: a config/MCP tool with the reserved name must not bypass the builtin gate.
return _enforce_agent_orchestration_scope(
unique_tools,
allow_agent_orchestration=allow_agent_orchestration,
)
@@ -0,0 +1,247 @@
"""Roundtable coordinator/seat capability isolation regressions."""
from __future__ import annotations
from types import SimpleNamespace
import pytest
import app.gateway.roundtable_seat_skills as seat_skills
import app.gateway.routers.multi_agent as multi_agent
from deerflow.agents.lead_agent.agent import (
_append_orchestration_capability_boundary,
_enforce_coordinator_only_tool_exclusions,
_filter_available_skills_for_excluded_tools,
)
from deerflow.tools.builtins.skill_tools import (
_resolve_active_agent_id,
can_access_coordinator_only_skill,
is_coordinator_only_skill_path,
skill_view_impl,
)
from deerflow.tools.tools import _enforce_agent_orchestration_scope
def _skill(name: str, description: str = "desc") -> SimpleNamespace:
return SimpleNamespace(
name=name,
description=description,
get_container_file_path=lambda base: f"{base}/custom/{name}/SKILL.md",
)
def test_seat_skill_directive_never_exposes_coordinator_only_skills(monkeypatch) -> None:
"""Even a misconfigured seat may see ordinary skills only, never orchestration."""
import deerflow.config as config_pkg
import deerflow.config.agents_config as agents_config
import deerflow.skills.storage as storage
monkeypatch.delenv("ROUNDTABLE_SEAT_SKILL_DIRECTIVE", raising=False)
monkeypatch.setattr(
agents_config,
"load_agent_config",
lambda agent_id: SimpleNamespace(
skills=["agent_orchestration", "knowledge-base-search", "agent-orchestration"],
seat_skill_directive=True,
),
)
monkeypatch.setattr(
storage,
"get_or_new_skill_storage",
lambda: SimpleNamespace(
load_skills=lambda enabled_only=True: [
_skill("agent_orchestration"),
_skill("knowledge-base-search", "知识库检索"),
_skill("agent-orchestration"),
]
),
)
monkeypatch.setattr(
config_pkg,
"get_app_config",
lambda: SimpleNamespace(skills=SimpleNamespace(container_path="/mnt/skills")),
)
directive = seat_skills.build_seat_skill_directive("seat-1")
assert "knowledge-base-search" in directive
assert "agent_orchestration" not in directive
assert "agent-orchestration" not in directive
def test_seat_role_boundary_is_the_last_instruction() -> None:
message = seat_skills.append_seat_role_boundary("完成风险分析")
assert message.startswith("完成风险分析\n\n---\n\n")
assert "不是总控智能体" in message
assert "不得加载、读取或调用 agent_orchestration" in message
assert message.endswith("请直接完成上方交给你的具体任务并提交结果。")
def test_excluded_orchestration_tool_also_hides_orchestration_skills(monkeypatch) -> None:
"""The system prompt/runtime allowlist must follow the tool-level seat policy."""
import deerflow.skills.storage as storage
skills = ["agent_orchestration", "knowledge-base-search", "agent-orchestration"]
assert _filter_available_skills_for_excluded_tools(skills, ["agent_orchestration"]) == [
"knowledge-base-search"
]
assert _filter_available_skills_for_excluded_tools(skills, ["web_search"]) == skills
monkeypatch.setattr(
storage,
"get_or_new_skill_storage",
lambda: SimpleNamespace(load_skills=lambda enabled_only=True: [_skill(name) for name in skills]),
)
assert _filter_available_skills_for_excluded_tools(None, ["agent_orchestration"]) == [
"knowledge-base-search"
]
def test_single_agent_tool_exclusions_fail_closed_but_coordinator_is_allowed() -> None:
assert _enforce_coordinator_only_tool_exclusions("ordinary-agent", None) == [
"agent_orchestration"
]
assert _enforce_coordinator_only_tool_exclusions(
"ordinary-agent", ["web_search", "agent_orchestration"]
) == ["web_search", "agent_orchestration"]
assert _enforce_coordinator_only_tool_exclusions("roundtable-coordinator", None) is None
def test_single_agent_system_prompt_forbids_orchestration_but_coordinator_prompt_does_not() -> None:
ordinary = _append_orchestration_capability_boundary("普通系统提示", "ordinary-agent")
coordinator = _append_orchestration_capability_boundary(
"总控系统提示", "roundtable-coordinator"
)
assert "没有编排、派活或调度其它智能体的权限" in ordinary
assert "不得加载、读取、检索或调用 agent_orchestration" in ordinary
assert coordinator == "总控系统提示"
def test_tool_loader_cannot_leak_orchestration_to_ordinary_callers() -> None:
tools = [SimpleNamespace(name="read_file"), SimpleNamespace(name="agent_orchestration")]
ordinary = _enforce_agent_orchestration_scope(tools, allow_agent_orchestration=False)
coordinator = _enforce_agent_orchestration_scope(tools, allow_agent_orchestration=True)
assert [tool.name for tool in ordinary] == ["read_file"]
assert [tool.name for tool in coordinator] == ["read_file", "agent_orchestration"]
def test_skill_discovery_and_direct_read_are_coordinator_only() -> None:
assert can_access_coordinator_only_skill("ordinary-agent", "agent-orchestration") is False
assert can_access_coordinator_only_skill("ordinary-agent", "agent_orchestration") is False
assert can_access_coordinator_only_skill("roundtable-coordinator", "agent-orchestration") is True
assert can_access_coordinator_only_skill(None, "knowledge-base-search") is True
assert skill_view_impl("agent-orchestration", "ordinary-agent") == (
"Skill 'agent-orchestration' not found."
)
def test_runtime_agent_name_and_reserved_skill_paths_are_recognized() -> None:
runtime = SimpleNamespace(context={"agent_name": "ordinary-agent"}, config={})
assert _resolve_active_agent_id(runtime) == "ordinary-agent"
assert is_coordinator_only_skill_path("/mnt/skills/custom/agent-orchestration/SKILL.md") is True
assert is_coordinator_only_skill_path(r"C:\skills\public\agent_orchestration\SKILL.md") is True
assert is_coordinator_only_skill_path("/mnt/skills/public/pdf/SKILL.md") is False
@pytest.mark.asyncio
async def test_foreground_seat_run_receives_role_boundary(monkeypatch) -> None:
"""The foreground special-run path must put the boundary in the actual human turn."""
seen: dict[str, str] = {}
async def _fake_stream(client, base, headers, thread_id, body):
seen["message"] = body["input"]["messages"][0]["content"][0]["text"]
yield None, ['data: {"messages":[{"type":"ai","content":"风险交付"}]}']
monkeypatch.setattr(multi_agent, "_stream_upstream", _fake_stream)
monkeypatch.setattr(multi_agent, "fallback_model_chain", lambda model: [model])
monkeypatch.setattr(multi_agent, "append_seat_skill_directive", lambda agent_id, task, **kwargs: task)
monkeypatch.setattr(multi_agent, "_spawn_detached_broadcast", lambda *args, **kwargs: None)
frames = [
frame
async for frame in multi_agent._special_run(
client=object(),
base="http://gateway",
headers={},
agent_threads={"seat-a": "seat-thread"},
agent_name="seat-a",
new_message="完成风险分析",
skill_stop_names=[],
model_name="test-model",
)
]
assert "不是总控智能体" in seen["message"]
assert "不得加载、读取或调用 agent_orchestration" in seen["message"]
assert frames[-1]["content"] == "风险交付"
@pytest.mark.asyncio
async def test_background_seat_run_receives_role_boundary(monkeypatch) -> None:
"""The in-process/background path must enforce the same seat boundary."""
import app.gateway.roundtable_inprocess_gateway as gateway_module
gateway = gateway_module.InProcessRoundtableGateway(
SimpleNamespace(state=SimpleNamespace()),
user_id="u1",
agents=[],
)
seen: dict[str, str] = {}
async def _fake_turn(**kwargs):
seen["message"] = kwargs["message"]
return ([{"type": "ai", "content": "风险交付"}], [], "test-model")
monkeypatch.setattr(gateway, "_run_turn_with_fallback", _fake_turn)
monkeypatch.setattr(
gateway_module,
"append_seat_skill_directive",
lambda agent_id, task, **kwargs: task,
)
result = await gateway.run_seat(
thread_ids={"seat-a": "seat-thread"},
agent_id="seat-a",
task="完成风险分析",
model="test-model",
)
assert "不是总控智能体" in seen["message"]
assert "不得加载、读取或调用 agent_orchestration" in seen["message"]
assert result.final_text == "风险交付"
@pytest.mark.asyncio
async def test_leader_never_broadcasts_orchestration_prompts_to_seats(monkeypatch) -> None:
"""Old clients cannot re-enable leader-to-seat prompt broadcasts with a false flag."""
captured = [
'data: {"messages":[{"type":"ai","content":"已派活",'
'"tool_calls":[{"name":"agent_orchestration","args":'
'{"agent_name":"seat-a","task":"分析风险"}}]}]}'
]
async def _fake_stream(*args, **kwargs):
yield None, captured
def _unexpected_broadcast(*args, **kwargs):
raise AssertionError("leader content must not be broadcast to seat threads")
monkeypatch.setattr(multi_agent, "_stream_upstream", _fake_stream)
monkeypatch.setattr(multi_agent, "fallback_model_chain", lambda model: [model])
monkeypatch.setattr(multi_agent, "_broadcast", _unexpected_broadcast)
monkeypatch.setattr(multi_agent, "_spawn_detached_broadcast", _unexpected_broadcast)
frames = [
frame
async for frame in multi_agent._leader_run(
client=object(),
base="http://gateway",
headers={},
agent_threads={"roundtable-coordinator": "leader-thread", "seat-a": "seat-thread"},
agent_name="roundtable-coordinator",
new_message="请通过 agent_orchestration 派活",
skill_stop_names=[],
model_name="test-model",
suppress_pre_broadcast=False,
)
]
assert frames[-1]["status"] == [["seat-a", "分析风险"]]