""" 验证 Agent 能力发现与首次加载相互隔离,并发首次使用只初始化一次。 """ from __future__ import annotations import asyncio import importlib import sys import threading import types from concurrent.futures import ThreadPoolExecutor import pytest from app.runtime.capabilities.errors import ( CapabilityOperationError, CapabilityRuntimeClosedError, ) from app.runtime.capabilities.model import ( CapabilityLifecycleState, CapabilityMaterializationState, ) @pytest.fixture def runtime_loader(monkeypatch): """ 为每个用例提供未构建、未关闭的 Agent Capability Runtime。 """ from app.agent import loader as module monkeypatch.setattr(module, "_agent_runtime", None) for implementation_module in ( "app.agent.manager", "app.agent.orchestrator", "app.agent.tools.factory", ): package_name, _, attribute = implementation_module.rpartition(".") package = importlib.import_module(package_name) # 重导入也会写入父包属性;与 sys.modules 一起恢复,避免后续补丁命中另一份模块。 # 先登记再移除属性,保证原本不存在的属性也能由 monkeypatch 完整撤销。 monkeypatch.setattr(package, attribute, vars(package).get(attribute), raising=False) monkeypatch.delattr(package, attribute) monkeypatch.delitem(sys.modules, implementation_module, raising=False) return module class _FakeManager: """ 记录异步 service 生命周期,并支持测试控制初始化时序。 """ def __init__(self) -> None: """ 初始化可观察的生命周期计数和并发控制状态。 """ self.initialize_calls = 0 self.close_calls = 0 self.fail_initialize = False self.initialize_entered: asyncio.Event | None = None self.initialize_release: asyncio.Event | None = None self.close_converged: bool | None = None async def initialize(self) -> None: """ 按测试设定等待屏障或抛错,模拟可控制的初始化时序。 """ self.initialize_calls += 1 if self.initialize_entered is not None: self.initialize_entered.set() if self.initialize_release is not None: await self.initialize_release.wait() if self.fail_initialize: raise RuntimeError("service initialization failed") async def close(self) -> bool | None: """ 记录关闭次数并返回当前测试指定的收敛结果。 """ self.close_calls += 1 return self.close_converged @pytest.mark.anyio async def test_shutdown_retains_nonconverged_agent_service_for_retry( runtime_loader, monkeypatch, ) -> None: """ Manager 返回未收敛时 Runtime 必须保留 owner,后续关闭可继续等待。 """ manager = _FakeManager() manager.close_converged = False modules = _fake_agent_modules(manager) monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", True, ) monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", lambda name: ( monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name] ), ) assert await runtime_loader.activate_agent_service() is manager assert await runtime_loader.begin_agent_shutdown() is False snapshot = runtime_loader._agent_runtime.snapshot("agent.service") assert snapshot.lifecycle is CapabilityLifecycleState.FAILED assert snapshot.visible is False manager.close_converged = True assert await runtime_loader.begin_agent_shutdown() is True snapshot = runtime_loader._agent_runtime.snapshot("agent.service") assert snapshot.lifecycle is CapabilityLifecycleState.STOPPED assert manager.close_calls == 2 @pytest.mark.anyio async def test_shutdown_propagates_runtime_wide_convergence( runtime_loader, monkeypatch, ) -> None: """ Agent 关闭不得用单个 service 快照覆盖 Runtime 的整体结果。 """ class RuntimeWithIndependentShutdownResult: """ 模拟其它 Agent 能力失败而 service 已停止的 Runtime。 """ async def shutdown_async(self, *, reason: str) -> bool: """ 记录关闭原因并返回 Runtime 级未收敛。 """ assert reason == "application_shutdown" return False @staticmethod def snapshot(_capability_id: str) -> types.SimpleNamespace: """ 提供旧实现读取的已停止 service 快照。 """ return types.SimpleNamespace(lifecycle=CapabilityLifecycleState.STOPPED) runtime = RuntimeWithIndependentShutdownResult() monkeypatch.setattr(runtime_loader, "_ensure_runtime", lambda: runtime) assert await runtime_loader.begin_agent_shutdown() is False def _fake_agent_modules(manager: object | None = None) -> dict[str, types.ModuleType]: """ 构造包含 canonical 符号的模块替身,隔离真实 Agent 服务初始化。 """ manager_module = types.ModuleType("app.agent.manager") manager_module.agent_manager = manager if manager is not None else object() orchestrator = types.ModuleType("app.agent.orchestrator") orchestrator.MoviePilotAgent = type("MoviePilotAgent", (), {}) tools = types.ModuleType("app.agent.tools.factory") tools.MoviePilotToolFactory = type("MoviePilotToolFactory", (), {}) return { manager_module.__name__: manager_module, orchestrator.__name__: orchestrator, tools.__name__: tools, } def test_registry_discovery_does_not_import_agent_implementation( runtime_loader, monkeypatch, ) -> None: """ 声明发现只能读取 TOML,不得导入 Agent、LLM 或工具实现。 """ imported = [] monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", lambda name: imported.append(name), ) runtime = runtime_loader._ensure_runtime() assert {spec.id for spec in runtime.list_specs()} == { "agent.manager", "agent.moviepilot_type", "agent.service", "agent.tool_factory", } assert imported == [] def test_agent_capability_entrypoints_resolve_canonical_symbols(runtime_loader) -> None: """ entrypoint 必须命中 canonical 模块的物理符号,Compat 惰性映射不满足强校验。 """ runtime = runtime_loader._ensure_runtime() for spec in runtime.list_specs(): module_name, symbol_name = spec.entrypoint.split(":", maxsplit=1) module = importlib.import_module(module_name) assert symbol_name in vars(module), spec.entrypoint implementation = vars(module)[symbol_name] assert runtime._canonical_implementation(spec, implementation) is implementation def test_concurrent_manager_first_use_is_single_flight( runtime_loader, monkeypatch, ) -> None: """ 并发首用必须只导入一次并向全部调用者发布同一 canonical 对象。 """ modules = _fake_agent_modules() import_calls = [] import_entered = threading.Event() import_release = threading.Event() def import_module(name: str): """ 挂起首次导入,验证并发请求只会加载一个服务 owner。 """ import_calls.append(name) import_entered.set() assert import_release.wait(timeout=5) monkeypatch.setitem(sys.modules, name, modules[name]) return modules[name] monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", import_module, ) with ThreadPoolExecutor(max_workers=8) as executor: futures = [executor.submit(runtime_loader.get_agent_manager) for _ in range(8)] assert import_entered.wait(timeout=5) import_release.set() results = [future.result(timeout=5) for future in futures] assert all( result is modules["app.agent.manager"].agent_manager for result in results ) assert ( runtime_loader.get_moviepilot_agent_type() is modules["app.agent.orchestrator"].MoviePilotAgent ) manager_snapshot = runtime_loader._agent_runtime.snapshot("agent.manager") assert manager_snapshot.materialization is CapabilityMaterializationState.RESOLVED assert manager_snapshot.lifecycle is CapabilityLifecycleState.DISCOVERED assert manager_snapshot.visible is False assert import_calls == ["app.agent.manager", "app.agent.orchestrator"] def test_tool_factory_has_independent_first_use_entrypoint( runtime_loader, monkeypatch, ) -> None: """ 工具工厂可独立首用,不需要先解析完整 Agent 编排模块。 """ modules = _fake_agent_modules() import_calls = [] def import_module(name: str): """ 登记当前能力的模块替身,避免触发真实服务的导入和初始化。 """ import_calls.append(name) monkeypatch.setitem(sys.modules, name, modules[name]) return modules[name] monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", import_module, ) assert runtime_loader.is_tool_factory_materialized() is False assert ( runtime_loader.get_tool_factory() is modules["app.agent.tools.factory"].MoviePilotToolFactory ) assert runtime_loader.is_tool_factory_materialized() is True assert import_calls == ["app.agent.tools.factory"] def test_materialization_query_does_not_construct_runtime( runtime_loader, monkeypatch, ) -> None: """ 只读物化查询在 Runtime 未构建时必须直接返回 False。 """ build_calls = [] monkeypatch.setattr( runtime_loader, "_build_agent_runtime", lambda: build_calls.append(True), ) assert runtime_loader.is_tool_factory_materialized() is False assert runtime_loader.get_running_agent_manager() is None assert build_calls == [] @pytest.mark.anyio async def test_service_first_and_entrypoint_first_share_canonical_identity( runtime_loader, monkeypatch, ) -> None: """ 资源轴和兼容物化轴无论谁先解析,都必须共享 canonical manager。 """ manager = _FakeManager() modules = _fake_agent_modules(manager) monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", True, ) def import_module(name: str): """ 登记当前能力的模块替身,避免触发真实服务的导入和初始化。 """ monkeypatch.setitem(sys.modules, name, modules[name]) return modules[name] monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", import_module, ) service_first = await runtime_loader.activate_agent_service() compat_after = runtime_loader.get_agent_manager() assert service_first is manager assert compat_after is manager assert runtime_loader.get_running_agent_manager() is manager assert manager.initialize_calls == 1 snapshot = runtime_loader._agent_runtime.snapshot("agent.service") assert snapshot.lifecycle is CapabilityLifecycleState.RUNNING assert snapshot.visible is True await runtime_loader.begin_agent_shutdown() assert manager.close_calls == 1 assert runtime_loader.get_running_agent_manager() is None with pytest.raises(CapabilityRuntimeClosedError): runtime_loader.get_agent_manager() with pytest.raises(CapabilityRuntimeClosedError): runtime_loader.get_moviepilot_agent_type() with pytest.raises(CapabilityRuntimeClosedError): runtime_loader.get_tool_factory() with pytest.raises(CapabilityRuntimeClosedError): await runtime_loader.activate_agent_service() @pytest.mark.anyio async def test_entrypoint_first_then_service_initializes_same_manager_once( runtime_loader, monkeypatch, ) -> None: """ 兼容 getter 先物化时不初始化,随后 service 只初始化同一对象一次。 """ manager = _FakeManager() modules = _fake_agent_modules(manager) monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", True, ) def import_module(name: str): """ 登记当前能力的模块替身,避免触发真实服务的导入和初始化。 """ monkeypatch.setitem(sys.modules, name, modules[name]) return modules[name] monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", import_module, ) assert runtime_loader.get_agent_manager() is manager assert manager.initialize_calls == 0 assert await runtime_loader.activate_agent_service() is manager assert await runtime_loader.activate_agent_service() is manager assert manager.initialize_calls == 1 @pytest.mark.anyio async def test_concurrent_service_first_use_initializes_once( runtime_loader, monkeypatch, ) -> None: """ 并发 service 首启只能 initialize 一次并发布同一实例。 """ manager = _FakeManager() manager.initialize_entered = asyncio.Event() manager.initialize_release = asyncio.Event() modules = _fake_agent_modules(manager) monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", True, ) monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", lambda name: ( monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name] ), ) tasks = [ asyncio.create_task(runtime_loader.activate_agent_service()) for _ in range(8) ] await manager.initialize_entered.wait() assert runtime_loader.get_agent_manager() is manager assert runtime_loader.get_running_agent_manager() is None manager.initialize_release.set() results = await asyncio.gather(*tasks) assert all(result is manager for result in results) assert manager.initialize_calls == 1 assert runtime_loader.get_running_agent_manager() is manager @pytest.mark.anyio async def test_concurrent_service_and_entrypoint_first_use_share_identity( runtime_loader, monkeypatch, ) -> None: """ 两个 spec 并发首解析时也只能引用同一个 canonical manager。 """ manager = _FakeManager() modules = _fake_agent_modules(manager) import_barrier = threading.Barrier(2) import_calls = [] monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", True, ) def import_module(name: str): """ 在共享屏障处等待,验证不同能力的导入不会互相串行阻塞。 """ import_calls.append(name) import_barrier.wait(timeout=5) monkeypatch.setitem(sys.modules, name, modules[name]) return modules[name] monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", import_module, ) service_task = asyncio.create_task(runtime_loader.activate_agent_service()) entrypoint_task = asyncio.create_task( asyncio.to_thread(runtime_loader.get_agent_manager) ) service, entrypoint = await asyncio.gather(service_task, entrypoint_task) assert service is manager assert entrypoint is manager assert runtime_loader.get_running_agent_manager() is manager assert manager.initialize_calls == 1 assert import_calls == ["app.agent.manager"] * 2 await runtime_loader.begin_agent_shutdown() @pytest.mark.anyio async def test_service_failure_is_not_published_and_requires_retry( runtime_loader, monkeypatch, ) -> None: """ 初始化失败必须清理候选、保持不可见,并要求显式 retry。 """ manager = _FakeManager() manager.fail_initialize = True modules = _fake_agent_modules(manager) monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", True, ) monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", lambda name: ( monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name] ), ) with pytest.raises(CapabilityOperationError, match="initialization failed"): await runtime_loader.activate_agent_service() assert runtime_loader.get_running_agent_manager() is None failed = runtime_loader._agent_runtime.snapshot("agent.service") assert failed.lifecycle is CapabilityLifecycleState.FAILED assert failed.visible is False assert manager.close_calls == 1 with pytest.raises(CapabilityOperationError, match="retry=True"): await runtime_loader.activate_agent_service() manager.fail_initialize = False assert await runtime_loader.activate_agent_service(retry=True) is manager assert manager.initialize_calls == 2 @pytest.mark.anyio async def test_shutdown_racing_service_first_use_fails_closed( runtime_loader, monkeypatch, ) -> None: """ 关闭与 service 首启竞争时不得发布对象,并须清理已初始化候选。 """ manager = _FakeManager() manager.initialize_entered = asyncio.Event() manager.initialize_release = asyncio.Event() modules = _fake_agent_modules(manager) monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", True, ) monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", lambda name: ( monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name] ), ) activation = asyncio.create_task(runtime_loader.activate_agent_service()) await manager.initialize_entered.wait() shutdown = asyncio.create_task(runtime_loader.begin_agent_shutdown()) await asyncio.sleep(0) manager.initialize_release.set() with pytest.raises(CapabilityRuntimeClosedError): await activation await shutdown assert runtime_loader.get_running_agent_manager() is None assert manager.initialize_calls == 1 assert manager.close_calls == 1 snapshot = runtime_loader._agent_runtime.snapshot("agent.service") assert snapshot.lifecycle is CapabilityLifecycleState.STOPPED assert snapshot.visible is False @pytest.mark.anyio async def test_disabled_service_reconcile_stays_unmaterialized( runtime_loader, monkeypatch, ) -> None: """ selector 为 false 时协调结果为空且不得导入 orchestrator。 """ imported = [] monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", False, ) monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", lambda name: imported.append(name), ) assert await runtime_loader.activate_agent_service() is None assert runtime_loader.get_running_agent_manager() is None snapshot = runtime_loader._agent_runtime.snapshot("agent.service") assert snapshot.materialization is CapabilityMaterializationState.UNRESOLVED assert snapshot.lifecycle is CapabilityLifecycleState.STOPPED assert imported == [] @pytest.mark.anyio async def test_empty_shutdown_does_not_import_agent_or_tools( runtime_loader, monkeypatch, ) -> None: """ 空载关闭只解析 data-only manifests,不导入编排器或工具实现。 """ imported = [] monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", lambda name: imported.append(name), ) await runtime_loader.begin_agent_shutdown() assert imported == [] assert runtime_loader.get_running_agent_manager() is None assert runtime_loader.is_tool_factory_materialized() is False @pytest.mark.anyio async def test_disable_reconcile_waits_for_concurrent_service_start( runtime_loader, monkeypatch, ) -> None: """ 关闭配置与首启竞争时必须等待启动并最终撤销实例。 """ manager = _FakeManager() manager.initialize_entered = asyncio.Event() manager.initialize_release = asyncio.Event() modules = _fake_agent_modules(manager) monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", True, ) monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", lambda name: ( monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name] ), ) activation = asyncio.create_task(runtime_loader.activate_agent_service()) await manager.initialize_entered.wait() monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", False, ) disable = asyncio.create_task( runtime_loader.reconcile_agent_service( reason="config_changed", changed_keys={"AI_AGENT_ENABLE"}, retry=True, ) ) await asyncio.sleep(0) manager.initialize_release.set() assert await activation is manager assert await disable is None assert runtime_loader.get_running_agent_manager() is None assert manager.initialize_calls == 1 assert manager.close_calls == 1 @pytest.mark.anyio async def test_config_reconcile_hot_switches_service_generations( runtime_loader, monkeypatch, ) -> None: """ watch 命中的配置切换应停止并重启同一 canonical service。 """ manager = _FakeManager() modules = _fake_agent_modules(manager) monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", lambda name: ( monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name] ), ) monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", False, ) assert await runtime_loader.activate_agent_service() is None assert ( await runtime_loader.reconcile_agent_service( reason="config_changed", changed_keys={"UNRELATED"}, retry=True, ) is None ) assert manager.initialize_calls == 0 monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", True, ) assert ( await runtime_loader.reconcile_agent_service( reason="config_changed", changed_keys={"AI_AGENT_ENABLE"}, retry=True, ) is manager ) assert manager.initialize_calls == 1 monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", False, ) assert ( await runtime_loader.reconcile_agent_service( reason="config_changed", changed_keys={"AI_AGENT_ENABLE"}, retry=True, ) is None ) assert manager.close_calls == 1 assert runtime_loader.get_running_agent_manager() is None monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", True, ) assert ( await runtime_loader.reconcile_agent_service( reason="config_changed", changed_keys={"AI_AGENT_ENABLE"}, retry=True, ) is manager ) assert manager.initialize_calls == 2 assert runtime_loader._agent_runtime.snapshot("agent.service").generation == 3 @pytest.mark.anyio async def test_config_reconcile_after_shutdown_cannot_restart_service( runtime_loader, monkeypatch, ) -> None: """ Runtime 关闭后即使 selector 再次为 true,配置协调也必须拒绝重启。 """ manager = _FakeManager() modules = _fake_agent_modules(manager) monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", True, ) monkeypatch.setattr( "app.agent.capabilities.adapter.importlib.import_module", lambda name: ( monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name] ), ) assert await runtime_loader.activate_agent_service() is manager await runtime_loader.begin_agent_shutdown() with pytest.raises(CapabilityRuntimeClosedError): await runtime_loader.reconcile_agent_service( reason="config_changed", changed_keys={"AI_AGENT_ENABLE"}, retry=True, ) assert manager.initialize_calls == 1 assert manager.close_calls == 1 assert runtime_loader.get_running_agent_manager() is None @pytest.mark.anyio async def test_real_agent_manager_can_restart_across_config_generations( runtime_loader, monkeypatch, ) -> None: """ 真实 AgentManager 在配置热切换后必须重新建立运行代际。 """ manager_module = importlib.import_module("app.agent.manager") memory_module = importlib.import_module("app.agent.memory") manager = manager_module.AgentManager() memory_events = [] async def close_memory() -> None: """ 记录记忆存储关闭顺序,核对生命周期退出的先后关系。 """ memory_events.append("close") monkeypatch.setattr( memory_module.memory_manager, "initialize", lambda: memory_events.append("initialize"), ) monkeypatch.setattr(memory_module.memory_manager, "close", close_memory) monkeypatch.setattr(manager_module, "agent_manager", manager) monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", False, ) assert await runtime_loader.activate_agent_service() is None monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", True, ) assert ( await runtime_loader.reconcile_agent_service( reason="config_changed", changed_keys={"AI_AGENT_ENABLE"}, retry=True, ) is manager ) assert manager._accepting_tasks is True monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", False, ) assert ( await runtime_loader.reconcile_agent_service( reason="config_changed", changed_keys={"AI_AGENT_ENABLE"}, retry=True, ) is None ) assert manager._accepting_tasks is False monkeypatch.setattr( "app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE", True, ) assert ( await runtime_loader.reconcile_agent_service( reason="config_changed", changed_keys={"AI_AGENT_ENABLE"}, retry=True, ) is manager ) assert manager._accepting_tasks is True assert memory_events == ["initialize", "close", "initialize"] await runtime_loader.begin_agent_shutdown() assert manager._accepting_tasks is False assert memory_events == ["initialize", "close", "initialize", "close"]