"""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")