248 lines
10 KiB
Python
248 lines
10 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 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", "分析风险"]]
|