311 lines
11 KiB
Python
311 lines
11 KiB
Python
"""Regression tests for the strategy-selection built-in agent seed."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
from fastapi import HTTPException
|
|
|
|
from app.gateway.routers import _strategy_selection_seed as seed
|
|
from app.gateway.routers import agents as agents_router
|
|
from deerflow.config.paths import Paths
|
|
from deerflow.persistence.skills.memory import MemorySkillStore
|
|
from deerflow.skills.types import Skill, SkillCategory
|
|
|
|
|
|
@pytest.fixture()
|
|
def tmp_paths(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Paths:
|
|
paths = Paths(tmp_path)
|
|
monkeypatch.setattr(seed, "get_paths", lambda: paths)
|
|
return paths
|
|
|
|
|
|
def _asset(agent_id: str, filename: str) -> Path:
|
|
return seed._ASSETS_DIR / agent_id / filename
|
|
|
|
|
|
def test_seed_assets_define_all_nine_functional_agents() -> None:
|
|
agent_ids = tuple(seed.required_agent_ids())
|
|
|
|
assert len(agent_ids) == 9
|
|
assert len(set(agent_ids)) == 9
|
|
assert agent_ids[0] == seed.STRATEGY_INTENT_AGENT_ID
|
|
assert agent_ids[-2:] == (
|
|
seed.STRATEGY_JUDGE_AGENT_ID,
|
|
seed.STRATEGY_SOLUTION_AGENT_ID,
|
|
)
|
|
assert len(seed.STRATEGY_ANGLE_AGENT_IDS) == 6
|
|
|
|
for agent_id in agent_ids:
|
|
config_path = _asset(agent_id, "config.yaml")
|
|
soul_path = _asset(agent_id, "SOUL.md")
|
|
assert config_path.is_file(), agent_id
|
|
assert soul_path.is_file(), agent_id
|
|
config = yaml.safe_load(config_path.read_text(encoding="utf-8"))
|
|
soul = soul_path.read_text(encoding="utf-8")
|
|
assert config["id"] == agent_id
|
|
assert config["name"].strip()
|
|
assert config["description"].strip()
|
|
assert "作战战术" in soul
|
|
assert "定向政治说服" in soul
|
|
assert "降级" in soul
|
|
|
|
|
|
def test_seed_assets_use_role_appropriate_existing_skills_and_tools() -> None:
|
|
intent = yaml.safe_load(
|
|
_asset(seed.STRATEGY_INTENT_AGENT_ID, "config.yaml").read_text(encoding="utf-8")
|
|
)
|
|
assert intent["skills"] == ["intent-clarification-policy"]
|
|
assert intent["tool_groups"] == []
|
|
|
|
for agent_id in seed.STRATEGY_ANGLE_AGENT_IDS:
|
|
config = yaml.safe_load(_asset(agent_id, "config.yaml").read_text(encoding="utf-8"))
|
|
assert config["skills"] == ["policy-knowledge"]
|
|
assert config["tool_groups"] == ["web", "file:read"]
|
|
|
|
judge = yaml.safe_load(
|
|
_asset(seed.STRATEGY_JUDGE_AGENT_ID, "config.yaml").read_text(encoding="utf-8")
|
|
)
|
|
solution = yaml.safe_load(
|
|
_asset(seed.STRATEGY_SOLUTION_AGENT_ID, "config.yaml").read_text(encoding="utf-8")
|
|
)
|
|
assert judge["skills"] == []
|
|
assert judge["tool_groups"] == []
|
|
assert solution["skills"] == []
|
|
assert solution["tool_groups"] == ["file:write"]
|
|
|
|
|
|
def test_default_strategy_skills_are_enabled_in_the_system_catalog() -> None:
|
|
extensions_path = Path(__file__).resolve().parents[1] / "extensions_config.json"
|
|
extensions = yaml.safe_load(extensions_path.read_text(encoding="utf-8")) or {}
|
|
configured = extensions.get("skills") or {}
|
|
|
|
# A missing entry means enabled by default. An explicit false would make
|
|
# the seeded allowlist impossible to save under the server-side validator
|
|
# and silently remove the skill from strategy runs.
|
|
for skill_name in ("intent-clarification-policy", "policy-knowledge"):
|
|
assert (configured.get(skill_name) or {}).get("enabled", True) is True
|
|
|
|
|
|
def test_seed_assets_define_machine_readable_handoff_contracts() -> None:
|
|
intent_soul = _asset(seed.STRATEGY_INTENT_AGENT_ID, "SOUL.md").read_text(encoding="utf-8")
|
|
judge_soul = _asset(seed.STRATEGY_JUDGE_AGENT_ID, "SOUL.md").read_text(encoding="utf-8")
|
|
solution_soul = _asset(seed.STRATEGY_SOLUTION_AGENT_ID, "SOUL.md").read_text(encoding="utf-8")
|
|
|
|
assert "[INTENT_READY]" in intent_soul
|
|
for field in (
|
|
"name",
|
|
"detail",
|
|
"objective",
|
|
"prerequisites",
|
|
"constraints",
|
|
"deliverables",
|
|
"acceptanceCriteria",
|
|
"horizon",
|
|
"selectedAngleIds",
|
|
):
|
|
assert f'"{field}"' in intent_soul
|
|
assert "[JUDGEMENT_READY]" in judge_soul
|
|
for field in ("agentId", "score", "conclusion", "comment", "strengths", "risks"):
|
|
assert f'"{field}"' in judge_soul
|
|
assert '"recommendedAgentIds"' in judge_soul
|
|
assert "不得调用 `ask_clarification`" in judge_soul
|
|
assert "仍须一次性完成所有候选项的评判" in judge_soul
|
|
assert "/mnt/user-data/outputs/综合策略方案.md" in solution_soul
|
|
|
|
|
|
def test_seed_creates_once_and_never_overwrites_admin_changes(tmp_paths: Paths) -> None:
|
|
expected = set(seed.required_agent_ids())
|
|
assert set(seed.ensure_strategy_selection_agents()) == expected
|
|
|
|
for agent_id in expected:
|
|
agent_dir = tmp_paths.agent_dir(agent_id)
|
|
assert (agent_dir / "config.yaml").is_file()
|
|
assert (agent_dir / "SOUL.md").is_file()
|
|
|
|
customized_agent = seed.STRATEGY_SECURITY_AGENT_ID
|
|
customized_soul = tmp_paths.agent_dir(customized_agent) / "SOUL.md"
|
|
customized_soul.write_text("管理员现场修改", encoding="utf-8")
|
|
|
|
assert seed.ensure_strategy_selection_agents() == []
|
|
assert customized_soul.read_text(encoding="utf-8") == "管理员现场修改"
|
|
|
|
|
|
def test_seed_upgrades_only_an_exact_known_legacy_default(
|
|
tmp_paths: Paths,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
assert seed.ensure_strategy_selection_agents()
|
|
intent_soul = tmp_paths.agent_dir(seed.STRATEGY_INTENT_AGENT_ID) / "SOUL.md"
|
|
legacy_content = b"legacy bundled intent soul\r\n"
|
|
intent_soul.write_bytes(legacy_content)
|
|
monkeypatch.setitem(
|
|
seed._LEGACY_DEFAULT_SOUL_HASHES,
|
|
seed.STRATEGY_INTENT_AGENT_ID,
|
|
seed._normalized_text_hash(legacy_content),
|
|
)
|
|
|
|
assert seed.ensure_strategy_selection_agents() == []
|
|
assert (
|
|
intent_soul.read_bytes()
|
|
== _asset(
|
|
seed.STRATEGY_INTENT_AGENT_ID,
|
|
"SOUL.md",
|
|
).read_bytes()
|
|
)
|
|
|
|
intent_soul.write_text("管理员现场修改", encoding="utf-8")
|
|
assert seed.ensure_strategy_selection_agents() == []
|
|
assert intent_soul.read_text(encoding="utf-8") == "管理员现场修改"
|
|
|
|
|
|
class _SkillStorage:
|
|
def __init__(self, skills: list[Skill]) -> None:
|
|
self._skills = skills
|
|
|
|
def load_skills(self, *, enabled_only: bool) -> list[Skill]:
|
|
if enabled_only:
|
|
return [skill for skill in self._skills if skill.enabled]
|
|
return list(self._skills)
|
|
|
|
|
|
def _skill(name: str, *, category: SkillCategory, enabled: bool = True) -> Skill:
|
|
directory = Path("skills") / category.value / name
|
|
return Skill(
|
|
name=name,
|
|
description=name,
|
|
license=None,
|
|
skill_dir=directory,
|
|
skill_file=directory / "SKILL.md",
|
|
relative_path=Path(name),
|
|
category=category,
|
|
enabled=enabled,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_strategy_skill_validation_accepts_only_global_catalog_names(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
skill_store = MemorySkillStore()
|
|
await skill_store.create(
|
|
{
|
|
"name": "published-custom",
|
|
"owner_user_id": "publisher",
|
|
"published": True,
|
|
}
|
|
)
|
|
storage = _SkillStorage(
|
|
[
|
|
_skill("public-skill", category=SkillCategory.PUBLIC),
|
|
_skill("published-custom", category=SkillCategory.CUSTOM),
|
|
]
|
|
)
|
|
monkeypatch.setattr(agents_router, "get_or_new_skill_storage", lambda **_: storage)
|
|
monkeypatch.setattr(
|
|
agents_router,
|
|
"get_skill_usage_entries",
|
|
lambda: {},
|
|
)
|
|
|
|
result = await agents_router._validate_strategy_agent_skills(
|
|
[" public-skill ", "published-custom", "public-skill"],
|
|
app_config=object(), # type: ignore[arg-type]
|
|
skill_store=skill_store,
|
|
)
|
|
|
|
assert result == ["public-skill", "published-custom"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("requested", "detail_fragment"),
|
|
[
|
|
("missing", "Unknown strategy agent skills"),
|
|
("disabled", "Disabled or archived strategy agent skills"),
|
|
("stale", "Disabled or archived strategy agent skills"),
|
|
("private-custom", "must be globally available"),
|
|
("orphan-custom", "must be globally available"),
|
|
],
|
|
)
|
|
async def test_strategy_skill_validation_rejects_unselectable_names(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
requested: str,
|
|
detail_fragment: str,
|
|
) -> None:
|
|
skill_store = MemorySkillStore()
|
|
await skill_store.create(
|
|
{
|
|
"name": "private-custom",
|
|
"owner_user_id": "owner",
|
|
"published": False,
|
|
}
|
|
)
|
|
storage = _SkillStorage(
|
|
[
|
|
_skill("disabled", category=SkillCategory.PUBLIC, enabled=False),
|
|
_skill("stale", category=SkillCategory.PUBLIC),
|
|
_skill("private-custom", category=SkillCategory.CUSTOM),
|
|
_skill("orphan-custom", category=SkillCategory.CUSTOM),
|
|
]
|
|
)
|
|
monkeypatch.setattr(agents_router, "get_or_new_skill_storage", lambda **_: storage)
|
|
monkeypatch.setattr(
|
|
agents_router,
|
|
"get_skill_usage_entries",
|
|
lambda: {"stale": {"state": "stale"}},
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await agents_router._validate_strategy_agent_skills(
|
|
[requested],
|
|
app_config=object(), # type: ignore[arg-type]
|
|
skill_store=skill_store,
|
|
)
|
|
|
|
assert exc_info.value.status_code == 422
|
|
assert detail_fragment in str(exc_info.value.detail)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_strategy_skill_validation_allows_retained_offline_but_not_new_offline(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
skill_store = MemorySkillStore()
|
|
storage = _SkillStorage(
|
|
[
|
|
_skill("retained-offline", category=SkillCategory.PUBLIC, enabled=False),
|
|
_skill("new-offline", category=SkillCategory.PUBLIC, enabled=False),
|
|
]
|
|
)
|
|
monkeypatch.setattr(agents_router, "get_or_new_skill_storage", lambda **_: storage)
|
|
monkeypatch.setattr(agents_router, "get_skill_usage_entries", lambda: {})
|
|
|
|
assert await agents_router._validate_strategy_agent_skills(
|
|
["retained-offline"],
|
|
app_config=object(), # type: ignore[arg-type]
|
|
skill_store=skill_store,
|
|
existing_skills=["retained-offline"],
|
|
) == ["retained-offline"]
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await agents_router._validate_strategy_agent_skills(
|
|
["retained-offline", "new-offline"],
|
|
app_config=object(), # type: ignore[arg-type]
|
|
skill_store=skill_store,
|
|
existing_skills=["retained-offline"],
|
|
)
|
|
assert exc_info.value.status_code == 422
|
|
assert "new-offline" in str(exc_info.value.detail)
|
|
|
|
|
|
def test_fixed_strategy_role_agents_cannot_be_deleted() -> None:
|
|
for agent_id in seed.required_agent_ids():
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
agents_router._ensure_strategy_agent_deletable(agent_id)
|
|
assert exc_info.value.status_code == 409
|
|
|
|
agents_router._ensure_strategy_agent_deletable("ordinary-custom-agent")
|