"""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 app.gateway.roundtable_run_policy import roundtable_run_policy from deerflow.agents.lead_agent.agent import ( _append_orchestration_capability_boundary, _enforce_coordinator_only_tool_exclusions, _filter_available_skills_for_excluded_tools, _knowledge_space_strict_exclusions, _runtime_tool_is_enabled, ) 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 assert _enforce_coordinator_only_tool_exclusions("task-analysis-coordinator", None) is None def test_tool_driving_middleware_respects_runtime_tool_policy() -> None: """A text-only coordinator must never receive a synthetic write_file call.""" assert _runtime_tool_is_enabled({}, "write_file") is True assert _runtime_tool_is_enabled({"allowed_tools": ["write_file"]}, "write_file") is True assert _runtime_tool_is_enabled({"allowed_tools": ["present_files"]}, "write_file") is False assert _runtime_tool_is_enabled({"excluded_tools": ["write_file"]}, "write_file") is False assert _runtime_tool_is_enabled( {"allowed_tools": ["write_file"], "excluded_tools": ["write_file"]}, "write_file", ) is False assert _runtime_tool_is_enabled({"disable_tools": True}, "write_file") is False def test_coordinator_guard_does_not_shrink_any_roundtable_runtime_budget() -> None: """Tool isolation must not shorten the working runtime of any role.""" for role in ( "leader", "seat", "seat_parallel", "report", "summary", "action_plan", "dashboard", "ingest", ): policy = roundtable_run_policy(role) assert policy.recursion_limit == 1000 if role != "leader": assert policy.allowed_tools is None seat = roundtable_run_policy("seat") for tool in ( "web_fetch", "browser_fetch_page", "browser_act", "tool_search", "page_report_template", "scheduled_task", "skill_manage", ): assert tool not in seat.excluded_tools def test_selected_knowledge_space_blocks_external_retrieval_surfaces() -> None: excluded = _knowledge_space_strict_exclusions(["custom_disabled"]) assert "custom_disabled" in excluded assert {"web_search", "browser_fetch_page", "tool_search", "task"}.issubset(excluded) assert "llmwiki_search" not in excluded assert {"search_skills", "skill_list", "skill_view"}.isdisjoint(excluded) assert _filter_available_skills_for_excluded_tools( ["data-analysis", "document-editing"], excluded, ) == ["data-analysis", "document-editing"] 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" ) task_analysis = _append_orchestration_capability_boundary( "总控系统提示", "task-analysis-coordinator" ) assert "没有编排、派活或调度其它智能体的权限" in ordinary assert "不得加载、读取、检索或调用 agent_orchestration" in ordinary assert coordinator == "总控系统提示" assert task_analysis == "总控系统提示" 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("task-analysis-coordinator", "agent-orchestration") is True assert can_access_coordinator_only_skill("task-analysis-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":"分析风险"}}]}]}' ] seen_inputs = [] async def _fake_stream(*args, **kwargs): seen_inputs.append(args[4]["input"]["messages"][0]["content"][0]["text"]) 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", "分析风险"]] assert "seat-a" in seen_inputs[0] assert "【本次圆桌" in seen_inputs[0] @pytest.mark.asyncio async def test_leader_final_closeout_is_tool_free_plain_text(monkeypatch) -> None: """Final synthesis must not loop through writer/presenter/orchestration tools.""" captured_bodies = [] async def _fake_delivery(*args, **kwargs): return {"seat-a": (True, "席位完整交付", 6, "席位完整交付")} async def _fake_stream(*args, **kwargs): captured_bodies.append(args[4]) yield None, ['data: {"messages":[{"type":"ai","content":"最终结论正文"}]}'] monkeypatch.setattr(multi_agent, "_collect_delivery_status", _fake_delivery) monkeypatch.setattr(multi_agent, "_stream_upstream", _fake_stream) monkeypatch.setattr(multi_agent, "fallback_model_chain", lambda model: [model]) 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="请输出最终方案", skill_stop_names=[], model_name="test-model", synthesis_mode=True, finalize_only=True, ) ] body = captured_bodies[0] assert body["context"]["allowed_tools"] == [] assert "write_file" in body["excluded_tools"] assert "present_files" in body["excluded_tools"] assert "最终收口·纯正文" in body["input"]["messages"][0]["content"][0]["text"] assert frames[-1]["status"] == [] @pytest.mark.asyncio async def test_leader_final_closeout_switches_only_after_completed_empty_response(monkeypatch) -> None: """最终收口完整结束仍无正文才换模；换模后仍以无工具正文完成。""" seen_models = [] async def _fake_stream(*args, **kwargs): body = args[4] model = body["context"]["model_name"] seen_models.append(model) content = "" if model == "model-a" else "换模后的最终结论" yield None, [f'data: {{"messages":[{{"type":"ai","content":"{content}"}}]}}'] monkeypatch.setattr(multi_agent, "_stream_upstream", _fake_stream) monkeypatch.setattr(multi_agent, "fallback_model_chain", lambda model: ["model-a", "model-b"]) frames = [ frame async for frame in multi_agent._leader_run( client=object(), base="http://gateway", headers={}, agent_threads={"roundtable-coordinator": "leader-thread"}, agent_name="roundtable-coordinator", new_message="请输出最终方案", skill_stop_names=[], model_name="model-a", synthesis_mode=True, finalize_only=True, ) ] assert seen_models == ["model-a", "model-b"] assert any(isinstance(frame, dict) and frame.get("status") == "model_switch" for frame in frames) assert frames[-1]["status"] == [] assert frames[-1]["content"] == "换模后的最终结论"