"""Global strategy-agent role CRUD, permissions and visibility tests.""" from __future__ import annotations import asyncio from pathlib import Path from types import SimpleNamespace import pytest import yaml from fastapi import HTTPException from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.gateway.routers import _strategy_selection_seed as seed from app.gateway.routers import agents as agents_router from app.gateway.routers import parallel_agents, strategy_agents from deerflow.config import agents_config from deerflow.config.paths import Paths from deerflow.persistence.agents import MemoryAgentStore from deerflow.persistence.skills.memory import MemorySkillStore from deerflow.persistence.strategy_agent_roles import MemoryStrategyAgentRoleStore, StrategyAgentRoleRepository from deerflow.persistence.strategy_agent_roles.model import StrategyAgentRoleRow from deerflow.skills.types import Skill, SkillCategory 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 = SkillCategory.PUBLIC) -> 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=True, ) def test_generic_agent_namespace_guard_is_case_insensitive_and_canonical() -> None: assert agents_router._is_strategy_agent_id("STRATEGY-SELECTION-INTENT") assert strategy_agents._process_lock("STRATEGY-SELECTION-INTENT") is strategy_agents._process_lock("strategy-selection-intent") with pytest.raises(HTTPException) as exc_info: agents_router._require_canonical_record_id( "STRATEGY-SELECTION-INTENT", {"id": "strategy-selection-intent"}, ) assert exc_info.value.status_code == 409 @pytest.mark.asyncio async def test_generic_mutations_reject_strategy_namespace_before_storage( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( agents_router, "get_agents_api_config", lambda: SimpleNamespace(enabled=True), ) request = SimpleNamespace() with pytest.raises(HTTPException) as update_error: await agents_router.update_agent( seed.STRATEGY_INTENT_AGENT_ID.upper(), agents_router.AgentUpdateRequest(name="bypass"), request, ) assert update_error.value.status_code == 409 assert "/api/strategy-agents/" in str(update_error.value.detail) with pytest.raises(HTTPException) as delete_error: await agents_router.delete_agent( seed.STRATEGY_INTENT_AGENT_ID.upper(), request, ) assert delete_error.value.status_code == 409 assert "generic API" in str(delete_error.value.detail) @pytest.mark.asyncio async def test_strategy_mutation_rejects_case_insensitive_database_alias( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, agent_store, role_store = await _request(tmp_path, monkeypatch) canonical_id = seed.STRATEGY_INTENT_AGENT_ID await role_store.ensure_defaults(seed.default_strategy_agent_roles()) role = await role_store.get_role(canonical_id) record = await agent_store.get_any(canonical_id) assert role is not None and record is not None monkeypatch.setattr(strategy_agents, "_ensure_registry", lambda *_: _async_none()) monkeypatch.setattr(role_store, "get_role", lambda _: _async_value(role)) monkeypatch.setattr(agent_store, "get_any", lambda _: _async_value(record)) with pytest.raises(HTTPException) as exc_info: await strategy_agents._get_role_and_agent(request, canonical_id.upper()) assert exc_info.value.status_code == 409 async def _async_none() -> None: return None async def _async_value(value): return value async def _request( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, role: str = "admin", ) -> tuple[SimpleNamespace, MemoryAgentStore, MemoryStrategyAgentRoleStore]: paths = Paths(tmp_path) monkeypatch.setattr(seed, "get_paths", lambda: paths) monkeypatch.setattr(agents_router, "get_paths", lambda: paths) monkeypatch.setattr(strategy_agents, "get_paths", lambda: paths) monkeypatch.setattr(agents_config, "get_paths", lambda: paths) seed.ensure_strategy_selection_agents() agent_store = MemoryAgentStore() for agent_id in seed.required_agent_ids(): config = yaml.safe_load((paths.agent_dir(agent_id) / "config.yaml").read_text(encoding="utf-8")) await agent_store.ensure_builtin( { "id": agent_id, "name": config["name"], "description": config.get("description") or "", "published": True, } ) role_store = MemoryStrategyAgentRoleStore() skill_store = MemorySkillStore() storage = _SkillStorage( [ _skill("public-skill"), _skill("published-custom", SkillCategory.CUSTOM), _skill("private-custom", SkillCategory.CUSTOM), ] ) await skill_store.create( { "name": "published-custom", "owner_user_id": "publisher", "published": True, } ) await skill_store.create( { "name": "private-custom", "owner_user_id": "owner", "published": False, } ) monkeypatch.setattr(agents_router, "get_or_new_skill_storage", lambda **_: storage) monkeypatch.setattr(agents_router, "get_skill_usage_entries", lambda: {}) state = SimpleNamespace( agent_store=agent_store, strategy_agent_role_store=role_store, skill_store=skill_store, config=SimpleNamespace(models=[SimpleNamespace(name="model-a")]), ) request = SimpleNamespace( app=SimpleNamespace(state=state), state=SimpleNamespace( user=SimpleNamespace(id="actor", system_role=role), ), ) return request, agent_store, role_store @pytest.mark.asyncio async def test_catalog_has_fixed_categories_and_redacts_public_config( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, _, role_store = await _request(tmp_path, monkeypatch) public = await strategy_agents.list_strategy_agents(request) assert len(public.agents) == 9 assert [(item.category, item.fixed) for item in public.agents if item.fixed] == [ ("intent", True), ("judge", True), ("solution", True), ] assert all(item.model is None and item.soul is None and item.skills is None for item in public.agents) await role_store.update_role(seed.STRATEGY_SECURITY_AGENT_ID, enabled=False) active = await strategy_agents.list_strategy_agents(request) assert seed.STRATEGY_SECURITY_AGENT_ID not in {item.id for item in active.agents} admin = await strategy_agents.list_strategy_agents( request, include_disabled=True, include_config=True, ) disabled = next(item for item in admin.agents if item.id == seed.STRATEGY_SECURITY_AGENT_ID) assert disabled.enabled is False assert disabled.soul assert disabled.skills == ["policy-knowledge"] @pytest.mark.asyncio async def test_catalog_rejects_user_owned_reserved_default( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, agent_store, _ = await _request(tmp_path, monkeypatch) await agent_store.delete_builtin(seed.STRATEGY_INTENT_AGENT_ID) await agent_store.create( { "id": seed.STRATEGY_INTENT_AGENT_ID, "name": "collision", "user_id": "attacker", } ) with pytest.raises(HTTPException) as exc_info: await strategy_agents.list_strategy_agents(request) assert exc_info.value.status_code == 409 assert "owned by a user" in str(exc_info.value.detail) @pytest.mark.asyncio async def test_sensitive_catalog_and_writes_are_admin_only( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, _, _ = await _request(tmp_path, monkeypatch, role="user") # The active catalog is intentionally readable by the workbench. assert len((await strategy_agents.list_strategy_agents(request)).agents) == 9 with pytest.raises(HTTPException) as list_error: await strategy_agents.list_strategy_agents( request, include_disabled=True, include_config=True, ) assert list_error.value.status_code == 403 with pytest.raises(HTTPException) as create_error: await strategy_agents.create_strategy_angle( strategy_agents.StrategyAngleCreateRequest(name="非法新增"), request, ) assert create_error.value.status_code == 403 @pytest.mark.asyncio async def test_angle_crud_is_global_and_delete_is_soft( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, agent_store, role_store = await _request(tmp_path, monkeypatch) created = await strategy_agents.create_strategy_angle( strategy_agents.StrategyAngleCreateRequest( name="科技产业视角", description="评估关键技术和产业链韧性", soul="你是科技产业策略分析智能体。", model="model-a", skills=["public-skill", "published-custom"], ), request, ) assert created.category == "angle" assert created.fixed is False assert created.enabled is True assert created.id.startswith(seed.STRATEGY_ANGLE_AGENT_PREFIX) assert created.model == "model-a" assert created.skills == ["public-skill", "published-custom"] updated = await strategy_agents.update_strategy_agent( created.id, strategy_agents.StrategyAgentUpdateRequest( name="科技与产业链视角", enabled=False, sort_order=123, skills=["public-skill"], ), request, ) assert updated.name == "科技与产业链视角" assert updated.enabled is False assert updated.sort_order == 123 assert updated.skills == ["public-skill"] deleted = await strategy_agents.disable_strategy_angle(created.id, request) assert deleted.enabled is False # Soft delete protects saved projects and permits an administrator to # restore the angle without recreating its ID or files. assert await agent_store.get_any(created.id) is not None assert await role_store.get_role(created.id) is not None restored = await strategy_agents.update_strategy_agent( created.id, strategy_agents.StrategyAgentUpdateRequest(enabled=True), request, ) assert restored.enabled is True assert (tmp_path / "agents" / created.id / ".strategy-role.yaml").is_file() @pytest.mark.asyncio async def test_strategy_update_retains_existing_offline_skill_but_rejects_new_one( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, _, _ = await _request(tmp_path, monkeypatch) created = await strategy_agents.create_strategy_angle( strategy_agents.StrategyAngleCreateRequest( name="下线技能保留测试", skills=["public-skill"], ), request, ) storage = agents_router.get_or_new_skill_storage() public_skill = next(skill for skill in storage._skills if skill.name == "public-skill") public_skill.enabled = False retained = await strategy_agents.update_strategy_agent( created.id, strategy_agents.StrategyAgentUpdateRequest( description="只改说明并原样保留已下线技能", skills=["public-skill"], ), request, ) assert retained.skills == ["public-skill"] newly_offline = _skill("new-offline") newly_offline.enabled = False storage._skills.append(newly_offline) with pytest.raises(HTTPException) as exc_info: await strategy_agents.update_strategy_agent( created.id, strategy_agents.StrategyAgentUpdateRequest( skills=["public-skill", "new-offline"], ), request, ) assert exc_info.value.status_code == 422 assert "new-offline" in str(exc_info.value.detail) @pytest.mark.asyncio async def test_memory_mode_recovers_dynamic_angle_from_folder_sidecar( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, _, _ = await _request(tmp_path, monkeypatch) created = await strategy_agents.create_strategy_angle( strategy_agents.StrategyAngleCreateRequest( name="恢复测试角度", sort_order=177, ), request, ) await strategy_agents.disable_strategy_angle(created.id, request) request.app.state.agent_store = MemoryAgentStore() request.app.state.strategy_agent_role_store = MemoryStrategyAgentRoleStore() recovered = await strategy_agents.list_strategy_agents( request, include_disabled=True, include_config=True, ) item = next(entry for entry in recovered.agents if entry.id == created.id) assert item.name == "恢复测试角度" assert item.enabled is False assert item.sort_order == 177 assert (await request.app.state.agent_store.get_any(created.id))["user_id"] is None @pytest.mark.asyncio async def test_memory_mode_sidecar_keeps_default_angle_disabled_after_restart( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, _, _ = await _request(tmp_path, monkeypatch) await strategy_agents.update_strategy_agent( seed.STRATEGY_SECURITY_AGENT_ID, strategy_agents.StrategyAgentUpdateRequest(enabled=False, sort_order=188), request, ) request.app.state.agent_store = MemoryAgentStore() request.app.state.strategy_agent_role_store = MemoryStrategyAgentRoleStore() with pytest.raises(HTTPException) as run_error: await parallel_agents.run_parallel_agents_stream( request, { "task_name": "restart gate", "brief": "disabled default must stay disabled", "agent_ids": [seed.STRATEGY_SECURITY_AGENT_ID], }, ) assert run_error.value.status_code == 409 recovered = await strategy_agents.list_strategy_agents( request, include_disabled=True, include_config=True, ) item = next(entry for entry in recovered.agents if entry.id == seed.STRATEGY_SECURITY_AGENT_ID) assert item.enabled is False assert item.sort_order == 188 @pytest.mark.asyncio async def test_cancelled_create_compensates_every_persistence_layer( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, agent_store, role_store = await _request(tmp_path, monkeypatch) generated_id = f"{seed.STRATEGY_ANGLE_AGENT_PREFIX}cancelled" monkeypatch.setattr( strategy_agents, "uuid4", lambda: SimpleNamespace(hex="cancelled"), ) async def cancel_set_extras(*_args, **_kwargs): raise asyncio.CancelledError monkeypatch.setattr(agent_store, "set_extras", cancel_set_extras) with pytest.raises(asyncio.CancelledError): await strategy_agents.create_strategy_angle( strategy_agents.StrategyAngleCreateRequest(name="cancelled create"), request, ) assert await agent_store.get_any(generated_id) is None assert await role_store.get_role(generated_id) is None assert not (tmp_path / "agents" / generated_id).exists() @pytest.mark.asyncio async def test_cross_process_file_lock_excludes_second_handle( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: paths = Paths(tmp_path) monkeypatch.setattr(strategy_agents, "get_paths", lambda: paths) first = strategy_agents._open_cross_process_lock(seed.STRATEGY_INTENT_AGENT_ID) second = strategy_agents._open_cross_process_lock(seed.STRATEGY_INTENT_AGENT_ID) try: assert strategy_agents._try_cross_process_lock(first) is True assert strategy_agents._try_cross_process_lock(second) is False strategy_agents._release_cross_process_lock(first) first = None assert strategy_agents._try_cross_process_lock(second) is True finally: if first is not None: strategy_agents._release_cross_process_lock(first) if not second.closed: strategy_agents._release_cross_process_lock(second) @pytest.mark.asyncio async def test_update_failure_restores_files_cache_and_role( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, agent_store, role_store = await _request(tmp_path, monkeypatch) created = await strategy_agents.create_strategy_angle( strategy_agents.StrategyAngleCreateRequest( name="事务前名称", model="model-a", skills=["public-skill"], sort_order=166, ), request, ) agent_dir = tmp_path / "agents" / created.id before_files = {name: (agent_dir / name).read_text(encoding="utf-8") for name in ("config.yaml", "SOUL.md", ".strategy-role.yaml")} real_update = role_store.update_role failed = False async def fail_once(agent_id: str, **kwargs): nonlocal failed if kwargs.get("sort_order") == 188 and not failed: failed = True raise RuntimeError("injected role-store failure") return await real_update(agent_id, **kwargs) monkeypatch.setattr(role_store, "update_role", fail_once) with pytest.raises(RuntimeError, match="injected role-store failure"): await strategy_agents.update_strategy_agent( created.id, strategy_agents.StrategyAgentUpdateRequest( name="不应留下的名称", soul="不应留下的 SOUL", skills=[], sort_order=188, ), request, ) record = await agent_store.get_any(created.id) role = await role_store.get_role(created.id) assert record is not None and record["name"] == "事务前名称" assert role is not None and role["sort_order"] == 166 assert (await agent_store.get_extras_for([created.id]))[created.id]["skills"] == ["public-skill"] assert {name: (agent_dir / name).read_text(encoding="utf-8") for name in ("config.yaml", "SOUL.md", ".strategy-role.yaml")} == before_files @pytest.mark.asyncio async def test_fixed_roles_cannot_be_disabled_or_deleted( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, _, _ = await _request(tmp_path, monkeypatch) with pytest.raises(HTTPException) as update_error: await strategy_agents.update_strategy_agent( seed.STRATEGY_INTENT_AGENT_ID, strategy_agents.StrategyAgentUpdateRequest(enabled=False), request, ) assert update_error.value.status_code == 409 with pytest.raises(HTTPException) as delete_error: await strategy_agents.disable_strategy_angle( seed.STRATEGY_JUDGE_AGENT_ID, request, ) assert delete_error.value.status_code == 409 @pytest.mark.asyncio async def test_model_and_global_skill_visibility_are_validated( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, _, _ = await _request(tmp_path, monkeypatch) with pytest.raises(HTTPException) as model_error: await strategy_agents.create_strategy_angle( strategy_agents.StrategyAngleCreateRequest( name="未知模型", model="not-configured", ), request, ) assert model_error.value.status_code == 422 with pytest.raises(HTTPException) as skill_error: await strategy_agents.create_strategy_angle( strategy_agents.StrategyAngleCreateRequest( name="私有技能", skills=["private-custom"], ), request, ) assert skill_error.value.status_code == 422 assert "globally available" in str(skill_error.value.detail) defaulted = await strategy_agents.create_strategy_angle( strategy_agents.StrategyAngleCreateRequest( name="默认角色模板", description="关注跨领域风险", ), request, ) assert defaulted.soul assert "默认角色模板" in defaulted.soul assert "不替代其他角度" in defaulted.soul @pytest.mark.asyncio async def test_disabled_strategy_agent_cannot_be_started_via_parallel_api( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, _, role_store = await _request(tmp_path, monkeypatch) await role_store.ensure_defaults(seed.default_strategy_agent_roles()) await role_store.update_role(seed.STRATEGY_SECURITY_AGENT_ID, enabled=False) with pytest.raises(HTTPException) as exc_info: await parallel_agents.run_parallel_agents_stream( request, { "task_name": "测试", "brief": "验证停用闸门", "agent_ids": [seed.STRATEGY_SECURITY_AGENT_ID], }, ) assert exc_info.value.status_code == 409 @pytest.mark.asyncio async def test_parallel_api_rejects_user_owned_strategy_backing_agent( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: request, agent_store, role_store = await _request(tmp_path, monkeypatch) await role_store.ensure_defaults(seed.default_strategy_agent_roles()) await agent_store.delete_builtin(seed.STRATEGY_SECURITY_AGENT_ID) await agent_store.create( { "id": seed.STRATEGY_SECURITY_AGENT_ID, "name": "collision", "user_id": "attacker", } ) with pytest.raises(HTTPException) as exc_info: await parallel_agents.run_parallel_agents_stream( request, { "task_name": "测试", "brief": "验证归属闸门", "agent_ids": [seed.STRATEGY_SECURITY_AGENT_ID], }, ) assert exc_info.value.status_code == 409 assert "owned by a user" in str(exc_info.value.detail) @pytest.mark.asyncio async def test_memory_registry_repairs_corrupted_fixed_role_metadata() -> None: store = MemoryStrategyAgentRoleStore() await store.create_role( { "agent_id": seed.STRATEGY_INTENT_AGENT_ID, "category": "angle", "fixed": False, "enabled": False, "sort_order": 999, } ) await store.ensure_defaults(seed.default_strategy_agent_roles()) repaired = await store.get_role(seed.STRATEGY_INTENT_AGENT_ID) assert repaired is not None assert repaired["category"] == "intent" assert repaired["fixed"] is True assert repaired["enabled"] is True assert repaired["sort_order"] == 0 @pytest.mark.asyncio async def test_sql_registry_preserves_disabled_defaults_and_soft_records( tmp_path: Path, ) -> None: engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'strategy-agent-roles.db'}") async with engine.begin() as connection: await connection.run_sync(StrategyAgentRoleRow.__table__.create) store = StrategyAgentRoleRepository(async_sessionmaker(engine, expire_on_commit=False)) try: defaults = seed.default_strategy_agent_roles() assert await store.ensure_defaults(defaults) == 9 assert await store.ensure_defaults(defaults) == 0 await store.update_role(seed.STRATEGY_SECURITY_AGENT_ID, enabled=False) # Re-seeding must never undo an administrator's soft delete. assert await store.ensure_defaults(defaults) == 0 disabled = await store.get_role(seed.STRATEGY_SECURITY_AGENT_ID) assert disabled is not None assert disabled["enabled"] is False # Canonical fixed-role invariants are repaired, unlike angle lifecycle # state which intentionally remains administrator-controlled. async with store._sf() as session: row = await session.get( StrategyAgentRoleRow, seed.STRATEGY_INTENT_AGENT_ID, ) row.category = "angle" row.fixed = False row.enabled = False row.sort_order = 999 await session.commit() assert await store.ensure_defaults(defaults) == 0 repaired = await store.get_role(seed.STRATEGY_INTENT_AGENT_ID) assert repaired is not None assert repaired["category"] == "intent" assert repaired["fixed"] is True assert repaired["enabled"] is True assert repaired["sort_order"] == 0 dynamic = await store.create_role( { "agent_id": "strategy-selection-angle-sql-test", "category": "angle", "fixed": False, "enabled": True, "sort_order": 150, } ) assert dynamic["category"] == "angle" assert await store.update_role(dynamic["agent_id"], enabled=False) assert await store.delete_role(dynamic["agent_id"]) is True assert await store.get_role(dynamic["agent_id"]) is None finally: await engine.dispose()