394 lines
16 KiB
Python
394 lines
16 KiB
Python
"""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"] == "换模后的最终结论"
|