diff --git a/.gitignore b/.gitignore index 60e88df4..b5fa1a07 100644 --- a/.gitignore +++ b/.gitignore @@ -118,8 +118,10 @@ offline-backend-20260512/backend/tests/* # Independent enterprise-research persistence, HTTP and race regressions. !offline-backend-20260512/backend/tests/test_enterprise_research_tasks.py # Position-roundtable per-turn write/present cap (weak-model duplicate artifacts). -!offline-backend-20260512/backend/tests/test_position_artifact_cap.py -# Workflow Studio: schemas / definitions / engine / run-lifecycle regressions +!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 # 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. !offline-backend-20260512/backend/tests/test_workflow_schemas.py diff --git a/frontend-web/src/roundtable-planning/api/multi-agent.ts b/frontend-web/src/roundtable-planning/api/multi-agent.ts index 6eb5c0b4..d4509999 100644 --- a/frontend-web/src/roundtable-planning/api/multi-agent.ts +++ b/frontend-web/src/roundtable-planning/api/multi-agent.ts @@ -254,8 +254,7 @@ export interface StreamMultiAgentRequest { */ synthesisMode?: boolean; /** - * 仅 leader 用。置 true 抑制「把总控本轮输入广播给各席位 thread」的预广播。DAG「每 stage - * 派活轮」置真——否则总控的派活 prompt 会进席位历史,弱模型据此模仿派活。后端缺省 false。 + * 仅为兼容旧协议保留。总控 prompt 属于总控私有上下文,前端恒传 true,后端也会无条件隔离。 */ suppressPreBroadcast?: boolean; /** @@ -532,8 +531,8 @@ export async function streamMultiAgent( skill_stop_names: req.skillStopNames ?? [], // 仅 leader 用;综合轮让后端注入各席位完整交付正文。省略 → 后端按 false 处理。 synthesis_mode: req.synthesisMode ?? false, - // 仅 leader 用;DAG 派活轮抑制预广播,避免总控派活 prompt 污染席位 thread。省略 → false。 - suppress_pre_broadcast: req.suppressPreBroadcast ?? false, + // 总控 prompt 永不进入席位 thread;后端同样强制执行,避免旧调用方破坏角色隔离。 + suppress_pre_broadcast: req.suppressPreBroadcast ?? true, // 仅 special 用;DAG 并行批次禁写文件(seat_parallel policy)。省略 → 后端按 false 处理。 parallel_no_file: req.parallelNoFile ?? false, // 仅 special 用;前序席位完整交付,后端注入 context 供 read_peer_delivery 按需取全文。 diff --git a/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py b/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py index 6558f3e7..e30497db 100644 --- a/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py +++ b/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py @@ -31,8 +31,7 @@ from app.gateway.roundtable_model_fallback import ( reason_text, ) from app.gateway.roundtable_run_policy import roundtable_run_policy -from app.gateway.roundtable_seat_skills import append_seat_skill_directive -from app.gateway.services import RUN_SCOPE_BACKGROUND_CHILD +from app.gateway.roundtable_seat_skills import append_seat_role_boundary, append_seat_skill_directive from app.gateway.routers._roundtable_seed import ( COORDINATOR_AGENT_ID, DASHBOARD_AGENT_ID, @@ -40,6 +39,7 @@ from app.gateway.routers._roundtable_seed import ( SUMMARY_AGENT_ID, 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.report_json import extract_report_json from deerflow.agents.roundtable_orchestrator.types import ( @@ -806,6 +806,8 @@ class InProcessRoundtableGateway: seat_task = append_seat_skill_directive( agent_id, task, chain_enabled=bool(self._seat_skill_directives.get(agent_id, False)) ) + # 后台挂起路径与前台保持同一角色边界:席位只能执行具体任务,编排能力永远只属于总控。 + seat_task = append_seat_role_boundary(seat_task) # 每席位「推理深度覆盖」:业务链条里给该席位单独配了 seatMode → 用该档派生的 # thinking/reasoning/subagent **覆盖作业级**(self._seat_*);未配则跟随作业级。 # 作业级仍为 None 时落 policy 默认。与前端 useStep2Orchestration「覆盖优先、否则跟随 diff --git a/offline-backend-20260512/backend/app/gateway/roundtable_seat_skills.py b/offline-backend-20260512/backend/app/gateway/roundtable_seat_skills.py index f5cad7a6..ab86d8b6 100644 --- a/offline-backend-20260512/backend/app/gateway/roundtable_seat_skills.py +++ b/offline-backend-20260512/backend/app/gateway/roundtable_seat_skills.py @@ -37,6 +37,22 @@ logger = logging.getLogger(__name__) _TRUTHY = frozenset({"1", "true", "yes", "on", "y"}) _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: """``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): 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: return "" diff --git a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py index 07f25a14..7c718cb2 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py +++ b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py @@ -25,8 +25,8 @@ ------------------ 通过 ``_append_thread_message`` 向其它 thread 的 checkpoint ``messages`` 追加 **只读上下文**的 human 消息(非触发 run),使各席位 LangGraph 状态里能看到 -总控派活/用户干预/子 agent 完成摘要。前端编排循环仍靠显式多次 ``/run/stream`` -驱动各 agent 真正推理。 +子 agent 完成摘要。总控输入、派活 prompt 和派活说明只留在总控 thread, +绝不广播给席位;前端编排循环仍靠显式多次 ``/run/stream`` 驱动各 agent 真正推理。 Leader 终态帧(前端 ``isFinalStatusFrame``) ------------------------------------------- @@ -48,7 +48,6 @@ Leader 终态帧(前端 ``isFinalStatusFrame``) from __future__ import annotations import asyncio -import contextlib import json import logging 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_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.services import RUN_SCOPE_ORCHESTRATION_CHILD, internal_run_scope_headers -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.routers._roundtable_seed import ( ACTION_PLAN_AGENT_ID, COORDINATOR_AGENT_ID, @@ -78,6 +76,7 @@ from app.gateway.routers._roundtable_seed import ( SUMMARY_AGENT_ID, 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.extraction_spec import build_summary_directive from deerflow.config import get_app_config @@ -1554,16 +1553,15 @@ async def _leader_run( model_name: str, seat_meta: dict[str, dict[str, Any]] | None = None, synthesis_mode: bool = False, - suppress_pre_broadcast: bool = False, + suppress_pre_broadcast: bool = True, files: list[dict[str, Any]] | None = None, ) -> AsyncIterator[str | dict[str, Any]]: """总控(leader)单轮:透传上游 SSE + 末尾 yield 一个 dict 状态帧(由路由层 ``json.dumps``)。 时序要点: - 1. **预广播**(与 LLM 并行):把用户/综合消息写入其它席位 thread,skip 总控自身; + 1. 总控输入只进入总控 thread,绝不广播到席位 thread; 2. 跑总控 thread 的 stream,yield 每一行原始 SSE; - 3. 解析 messages → 澄清 / 空 dispatch(共识)/ 派活列表; - 4. 若有派活,**派活广播**后再 yield ``{status: dispatched, ...}``。 + 3. 解析 messages → 澄清 / 空 dispatch(共识)/ 派活列表并返回。 ``skill_stop_names`` 会与 ``agent_orchestration`` 合并,确保 middleware 在工具调用后停图。 @@ -1575,6 +1573,10 @@ async def _leader_run( ``synthesis_mode``:前端在「派活后让总控判断继续/综合」的轮次置真。为真时**额外**把各 已交付席位的**完整交付正文**注入本轮输入(``_build_full_deliverables_block``),供总控写 最终方案;平时(派活决策轮)只给摘要,让总控 thread 与每轮上下文保持精简、响应快。 + + ``suppress_pre_broadcast`` 仅为兼容旧客户端保留,服务端始终按抑制处理。角色隔离不能由 + 客户端布尔值决定:总控 prompt 可能含 ``agent_orchestration``,进入席位历史会诱导弱模型 + 模仿总控加载编排能力。 """ t0 = time.perf_counter() # 实读各席位 thread 的真实交付状态,构造「席位清单 + 客观交付状态(摘要)」块;综合轮再额外 @@ -1623,21 +1625,9 @@ async def _leader_run( part for part in (seat_status_block, full_deliverables_block, new_message) if part ) - # 与上游 LLM 并行,减少「总控已开始想但席位 thread 里还没有用户消息」的窗口。 - # - # ``suppress_pre_broadcast``:DAG「每 stage 派活轮」(方案 A)置真——此时 ``new_message`` 是 - # 「请你作为总控…通过 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;席位需要的上下文 + # 由总控下发的具体 task、peer_deliveries 和席位交付广播提供。 + _ = suppress_pre_broadcast thread_id = agent_threads[agent_name] # 角色参数取自单一数据源 roundtable_run_policy(leader:禁 web_search、停 # agent_orchestration 图、关 thinking + low reasoning——短决策、首字节快)。 @@ -1677,7 +1667,7 @@ async def _leader_run( body = _mk_body(current_model) t_payload_built = time.perf_counter() logger.info( - "[leader-timing] %s payload_built=%.3fs (broadcast running in background)", + "[leader-timing] %s payload_built=%.3fs", agent_name, 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)", agent_name, ) - if pre_broadcast_task is not None: - 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 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) leader_text = _flatten_content(dispatch_msg.get("content", "")) tool_calls = dispatch_msg.get("tool_calls") or [] @@ -1839,22 +1810,9 @@ async def _leader_run( # 非法 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 派活帧,让前端立刻启动席位 special run。总控派活说明不广播给 + # 任何席位,避免席位从历史中模仿总控角色;席位间承上依靠完成交付广播 / peer_deliveries。 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 @@ -1937,6 +1895,8 @@ async def _special_run( seat_message = new_message 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_role_boundary(seat_message) # 深链接业务规范(仅 summary 单例):把该业务的「业务链完整性核对」指令拼进本轮任务,强制总结报告 # 核对产出相对业务链完不完整。business_code 为空(普通会商)/ 非写实业务 → directive 为 "",原样不变。 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)。 # 兼容 no_file 别名;缺省 False → 普通 seat,旧前端不传也不破坏。 no_file: bool = bool(payload.get("parallel_no_file") or payload.get("no_file") or False) - # 抑制 leader 预广播(仅 leader 用):DAG「每 stage 派活轮」置真,避免把总控的派活 prompt - # 广播进席位 thread 让弱模型模仿派活(详见 _leader_run 的 suppress_pre_broadcast 注释)。 - # 缺省 False,recommend/chain 行为不变。 - suppress_pre_broadcast: bool = bool(payload.get("suppress_pre_broadcast") or False) + # 旧客户端兼容字段。服务端已强制隔离总控 thread,因此无论请求值为何都不会执行 leader→seat + # 广播;仍解析并透传只是保持调用签名稳定。 + suppress_pre_broadcast: bool = bool(payload.get("suppress_pre_broadcast", True)) # 前序席位完整交付(仅 special 用):前端 DAG stage>0 席位带上此前各阶段的全文,网关注入 # run context,供 read_peer_delivery 工具按需取全文(平时只看广播摘要,§12.9)。规整成 # [{name, content}] 且过滤空项;缺省 None。 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py index 53a338e4..8de25f82 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py @@ -40,7 +40,74 @@ logger = logging.getLogger(__name__) WRITING_SETUP_MAX_ROUNDS = 8 REPORT_STRUCTURE_SETUP_MAX_ROUNDS = 8 -_ES_QUERY_ROUTING_OPEN_TAG = '' +_ES_QUERY_ROUTING_OPEN_TAG = '' + +_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 = """ + +你不是多智能体圆桌总控,因此没有编排、派活或调度其它智能体的权限。 +不得加载、读取、检索或调用 agent_orchestration / agent-orchestration, +不得声称准备使用编排技能;请直接完成当前问答或你的本职任务。 + +""".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( @@ -535,8 +602,11 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig): if not is_writing_mode_report_path(writing_artifact_path): writing_artifact_path = None - agent_id = validate_agent_id(cfg.get("agent_id") or cfg.get("agent_name")) - requested_excluded_tools = cfg.get("excluded_tools") or None + agent_id = validate_agent_id(cfg.get("agent_id") or cfg.get("agent_name")) + requested_excluded_tools = _enforce_coordinator_only_tool_exclusions( + agent_id, + cfg.get("excluded_tools") or None, + ) 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 @@ -585,10 +655,14 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig): from deerflow.agents.deep_research.retrieval import merge_collector_skill_allowlist, runtime_extra_skills - runtime_available_skills = merge_collector_skill_allowlist( - agent_id, runtime_available_skills, runtime_extra_skills(cfg) - ) - prompt_available_skills = set(runtime_available_skills) if runtime_available_skills is not None else None + runtime_available_skills = merge_collector_skill_allowlist( + 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 config["metadata"].update( { @@ -669,13 +743,14 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig): # Special bootstrap agent with minimal prompt for initial custom agent creation flow return create_agent( model=create_chat_model(name=model_name, thinking_enabled=thinking_enabled, force_disable_thinking=thinking_force_disabled, app_config=resolved_app_config), - tools=get_available_tools( - model_name=model_name, - subagent_enabled=subagent_enabled, - is_scheduled_run=cfg.get("is_scheduled_run", False), - excluded_tools=requested_excluded_tools, - app_config=resolved_app_config, - ) + tools=get_available_tools( + model_name=model_name, + subagent_enabled=subagent_enabled, + is_scheduled_run=cfg.get("is_scheduled_run", False), + excluded_tools=requested_excluded_tools, + allow_agent_orchestration=agent_id == _ROUNDTABLE_COORDINATOR_AGENT_ID, + app_config=resolved_app_config, + ) + [setup_agent], middleware=_build_middlewares(config, model_name=model_name, app_config=resolved_app_config), system_prompt=apply_prompt_template( @@ -698,11 +773,12 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig): default_tools = get_available_tools( model_name=model_name, groups=agent_config.tool_groups if agent_config else None, - subagent_enabled=subagent_enabled, - is_scheduled_run=cfg.get("is_scheduled_run", False), - excluded_tools=requested_excluded_tools, - app_config=resolved_app_config, - ) + subagent_enabled=subagent_enabled, + is_scheduled_run=cfg.get("is_scheduled_run", False), + excluded_tools=requested_excluded_tools, + allow_agent_orchestration=agent_id == _ROUNDTABLE_COORDINATOR_AGENT_ID, + app_config=resolved_app_config, + ) if agent_id == FORCED_RESEARCH_AGENT_ID: default_tools = filter_forced_research_tools(default_tools) if is_notebook_mode: @@ -744,8 +820,8 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig): writing_mode=writing_mode, writing_artifact_path=writing_artifact_path, ) - else: - system_prompt = apply_prompt_template( + else: + system_prompt = apply_prompt_template( subagent_enabled=subagent_enabled, max_concurrent_subagents=max_concurrent_subagents, agent_id=agent_id, @@ -758,10 +834,12 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig): rag_mode_enabled=rag_mode_enabled, is_notebook_mode=is_notebook_mode, notebook_source_ids=notebook_source_ids, - is_scheduled_run=bool(cfg.get("is_scheduled_run", False)), - ) - - _log_es_query_routing_registration( + 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( system_prompt=system_prompt, app_config=resolved_app_config, agent_id=agent_id, diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/sandbox/tools.py b/offline-backend-20260512/backend/packages/harness/deerflow/sandbox/tools.py index 7d640a0b..838efdfc 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/sandbox/tools.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/sandbox/tools.py @@ -1474,7 +1474,7 @@ def grep_tool( @tool("read_file", parse_docstring=True) -def read_file_tool( +def read_file_tool( runtime: ToolRuntime[ContextT, ThreadState], description: str, path: str, @@ -1488,12 +1488,21 @@ def read_file_tool( path: The **absolute** path to the file to read. start_line: Optional starting line number (1-indexed, inclusive). Use with end_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: - sandbox = ensure_sandbox_initialized(runtime) - ensure_thread_directories_exist(runtime) - requested_path = path - if is_local_sandbox(runtime): + """ + 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) + ensure_thread_directories_exist(runtime) + if is_local_sandbox(runtime): thread_data = get_thread_data(runtime) validate_local_tool_path(path, thread_data, read_only=True) if _is_skills_path(path): diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/skill_tools.py b/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/skill_tools.py index 15a7024d..c4130891 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/skill_tools.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/skill_tools.py @@ -9,6 +9,21 @@ from langchain.tools import ToolRuntime, tool 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: """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 try: 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: return str(ctx_id) cfg = getattr(runtime, "config", None) 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: return str(cid) 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) if cfg is 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: logger.debug("Failed to load agent config for skill_list filter", exc_info=True) 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 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.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): skill_file = storage.get_skills_root_path() from pathlib import Path + from deerflow.skills.types import 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() scored: list[tuple[int, dict]] = [] for skill in skills: + if not can_access_coordinator_only_skill(agent_id, skill.name): + continue entry = usage.get(skill.name, {}) if allowed is not None: 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() result = [] for skill in skills: + if not can_access_coordinator_only_skill(agent_id, skill.name): + continue entry = usage.get(skill.name, {}) if allowed is not None: # 自定义 agent:严格按 config.yaml 的 skills 白名单过滤。 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/tools/tools.py b/offline-backend-20260512/backend/packages/harness/deerflow/tools/tools.py index a7575792..457de383 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/tools/tools.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/tools/tools.py @@ -16,11 +16,10 @@ BROWSER_ACT_TOOL_USE = "deerflow.tools.builtins.browser_act_tool:browser_act_too BROWSER_CONTEXT_TOOL_USES = {BROWSER_CONTEXT_TOOL_USE, BROWSER_ACT_TOOL_USE} BROWSER_CONTEXT_TOOL_NAMES = {"browser_fetch_page", "browser_act"} -BUILTIN_TOOLS = [ - present_file_tool, - ask_clarification_tool, - agent_orchestration_tool, -] +BUILTIN_TOOLS = [ + present_file_tool, + ask_clarification_tool, +] SUBAGENT_TOOLS = [ task_tool, @@ -39,24 +38,36 @@ def _is_host_bash_tool(tool: object) -> bool: return False -def _is_browser_context_tool(tool: object) -> bool: +def _is_browser_context_tool(tool: object) -> bool: """Return True if the tool config represents the browser extension bridge.""" use = getattr(tool, "use", None) name = getattr(tool, "name", None) - 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, include_mcp: bool = True, model_name: str | None = None, subagent_enabled: bool = False, is_scheduled_run: bool = False, - excluded_tools: list[str] | None = None, - *, - app_config: AppConfig | None = None, -) -> list[BaseTool]: + excluded_tools: list[str] | None = None, + *, + allow_agent_orchestration: bool = False, + app_config: AppConfig | None = None, +) -> list[BaseTool]: """Get all available tools from config. Note: MCP tools should be initialized at application startup using @@ -67,9 +78,11 @@ def get_available_tools( include_mcp: Whether to include tools from MCP servers (default: True). model_name: Optional model name to determine if vision tools should be included. subagent_enabled: Whether to include subagent tools (task, task_status). - excluded_tools: Optional list of tool names to exclude from the result - (applied AFTER all other resolution). Takes precedence over - ``app_config.excluded_tools``. + excluded_tools: Optional list of tool names to exclude from the result + (applied AFTER all other resolution). Takes precedence over + ``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: List of available tools. @@ -114,8 +127,10 @@ def get_available_tools( loaded_tools = [t for _, t in loaded_tools_raw] - # Conditionally add tools based on config - builtin_tools = BUILTIN_TOOLS.copy() + # Conditionally add tools based on config + builtin_tools = BUILTIN_TOOLS.copy() + if allow_agent_orchestration: + builtin_tools.append(agent_orchestration_tool) if not is_scheduled_run: builtin_tools.append(scheduled_task_tool) @@ -292,4 +307,8 @@ def get_available_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, + ) diff --git a/offline-backend-20260512/backend/tests/test_roundtable_role_boundary.py b/offline-backend-20260512/backend/tests/test_roundtable_role_boundary.py new file mode 100644 index 00000000..878451d7 --- /dev/null +++ b/offline-backend-20260512/backend/tests/test_roundtable_role_boundary.py @@ -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", "分析风险"]]