diff --git a/app/adapters/system/resource.py b/app/adapters/system/resource.py index a9c2a7f1d..9c5018b20 100644 --- a/app/adapters/system/resource.py +++ b/app/adapters/system/resource.py @@ -28,6 +28,12 @@ def configure_resource_version_provider(provider: ResourceVersionProvider) -> No _resource_version_provider = provider +def reset_resource_version_provider() -> None: + """撤销当前 lifespan 的资源版本来源,恢复安全空版本。""" + global _resource_version_provider + _resource_version_provider = _unavailable_resource_versions + + class ResourceHelper: """ 检测和更新资源包 diff --git a/app/agent/tools/manager.py b/app/agent/tools/manager.py index 59867e542..3c08c3212 100644 --- a/app/agent/tools/manager.py +++ b/app/agent/tools/manager.py @@ -59,6 +59,14 @@ class MoviePilotToolsManager: self._catalog_materialized = False self._catalog_managed_by_factory = False + def _invalidate_catalog_locked(self) -> None: + """在持有工具锁时清除由旧数据上下文构造的目录快照。""" + self.tools = [] + self.catalog = None + self._plugin_agent_tools_revision = -1 + self._catalog_materialized = False + self._catalog_managed_by_factory = False + def set_data_context(self, data: AgentDataContext) -> None: """绑定组合根上下文,并使工厂拥有的旧工具快照原子失效。""" with self._tools_lock: @@ -67,11 +75,13 @@ class MoviePilotToolsManager: if self._catalog_materialized and not self._catalog_managed_by_factory: raise RuntimeError("调用方工具目录已物化,不能替换数据上下文") self._data = data - self.tools = [] - self.catalog = None - self._plugin_agent_tools_revision = -1 - self._catalog_materialized = False - self._catalog_managed_by_factory = False + self._invalidate_catalog_locked() + + def reset_data_context(self) -> None: + """撤销当前 lifespan 数据上下文,并清除引用旧仓储的工具快照。""" + with self._tools_lock: + self._data = None + self._invalidate_catalog_locked() @staticmethod def _summarize_error(error: Exception) -> str: diff --git a/app/application/database.py b/app/application/database.py index 70a0ddde1..03f9b638b 100644 --- a/app/application/database.py +++ b/app/application/database.py @@ -97,6 +97,11 @@ def configure_database_governance(governance: DatabaseGovernance) -> None: _DATABASE_GOVERNANCE.append(governance) +def reset_database_governance() -> None: + """撤销当前 lifespan 的数据库治理门面,恢复未装配状态。""" + _DATABASE_GOVERNANCE.clear() + + def get_database_governance() -> DatabaseGovernance: """返回启动阶段登记的数据库治理门面。""" if not _DATABASE_GOVERNANCE: diff --git a/app/runtime/event/binding.py b/app/runtime/event/binding.py index 29b492dbe..32498400e 100644 --- a/app/runtime/event/binding.py +++ b/app/runtime/event/binding.py @@ -44,6 +44,11 @@ class EventBindingResolver: with self._lock: self._resolvers()[name] = resolver + def unregister(self, name: str) -> None: + """移除命名实例解析器;名称不存在时保持幂等。""" + with self._lock: + self._resolvers().pop(name, None) + def unresolved_handlers(self) -> tuple[str, ...]: """返回本进程中未被显式 resolver 接管的类处理器。""" with self._lock: diff --git a/app/runtime/events.py b/app/runtime/events.py index 4bfe129c4..017fb5673 100644 --- a/app/runtime/events.py +++ b/app/runtime/events.py @@ -191,6 +191,10 @@ class EventManager(metaclass=Singleton): """ self.__binding_resolver.register(name, resolver) + def unregister_handler_instance_resolver(self, name: str) -> None: + """撤销命名实例解析器;重复撤销不影响其它 owner。""" + self.__binding_resolver.unregister(name) + def unresolved_handler_bindings(self) -> tuple[str, ...]: """返回未命中显式 resolver 的类处理器诊断清单。""" return self.__binding_resolver.unresolved_handlers() @@ -200,6 +204,10 @@ class EventManager(metaclass=Singleton): with self.__lock: self.__error_notifier = notifier + def reset_error_notifier(self) -> None: + """撤销当前 lifespan 的事件错误通知回调。""" + self.set_error_notifier(None) + def start(self): """ 开始广播事件处理线程 diff --git a/app/runtime/resources.py b/app/runtime/resources.py index 73c95aa5a..cfcf104c0 100644 --- a/app/runtime/resources.py +++ b/app/runtime/resources.py @@ -64,6 +64,16 @@ def configure_managed_resource_runtime(runtime: ManagedResourceRuntime) -> None: _managed_resource_runtime = runtime +def reset_managed_resource_runtime() -> None: + """释放已关闭 Runtime 的门面引用;未配置时保持幂等。""" + global _managed_resource_runtime + with _runtime_lock: + runtime = _managed_resource_runtime + if runtime is not None and not runtime.is_shutdown: + raise RuntimeError("Managed Resource Runtime 尚未关闭,不能释放引用") + _managed_resource_runtime = None + + @overload def _runtime(*, required: Literal[True]) -> ManagedResourceRuntime: ... diff --git a/app/scheduler/facade.py b/app/scheduler/facade.py index fae04e96b..3f0ac3add 100644 --- a/app/scheduler/facade.py +++ b/app/scheduler/facade.py @@ -104,6 +104,15 @@ class Scheduler( raise RuntimeError("Scheduler 已运行,不能替换业务能力") self._services = services + def reset_runtime_bindings(self) -> None: + """在调度器停止后撤销当前 lifespan 的仓储与业务能力引用。""" + with self._lock: + if self._lifecycle_state not in {"new", "stopped"}: + raise RuntimeError("Scheduler 仍在运行,不能撤销运行时绑定") + self._agent_tasks = None + self._services = None + self._agent_task_interruptions_reconciled = False + def _scheduler_services(self) -> SchedulerServices: """返回显式注入能力;缺少组合根装配时稳定失败。""" if self._services is None: diff --git a/app/startup/initializers/resources.py b/app/startup/initializers/resources.py index 7902ee42d..bcb39de3d 100644 --- a/app/startup/initializers/resources.py +++ b/app/startup/initializers/resources.py @@ -15,6 +15,7 @@ from app.runtime.resources import ( MANAGED_RESOURCE_ASYNC_KIND, MANAGED_RESOURCE_SYNC_KIND, configure_managed_resource_runtime, + reset_managed_resource_runtime, ) _runtime_lock = threading.RLock() @@ -44,3 +45,14 @@ async def stop_managed_resources() -> bool: if runtime is None: return True return await runtime.shutdown_async(reason="application_shutdown") + + +def reset_managed_resources() -> None: + """释放已关闭的具体 Runtime 及其门面引用,不模拟进程内完整重启。""" + global _managed_resource_runtime + with _runtime_lock: + runtime = _managed_resource_runtime + if runtime is not None and not runtime.is_shutdown: + raise RuntimeError("Managed Resource Runtime 尚未关闭,不能释放 owner") + reset_managed_resource_runtime() + _managed_resource_runtime = None diff --git a/tests/test_residual_provider_resets.py b/tests/test_residual_provider_resets.py new file mode 100644 index 000000000..4b58bac83 --- /dev/null +++ b/tests/test_residual_provider_resets.py @@ -0,0 +1,215 @@ +"""验证剩余启动 provider 可撤销、幂等且不会跨 lifespan 串用对象。""" + +from __future__ import annotations + +from collections.abc import Iterator +from types import SimpleNamespace +from typing import Any + +import pytest + +from app.adapters.system import resource as resource_module +from app.adapters.system.resource import ( + configure_resource_version_provider, + reset_resource_version_provider, +) +from app.agent.tools.manager import MoviePilotToolsManager +from app.application.database import ( + configure_database_governance, + get_database_governance, + reset_database_governance, +) +from app.runtime import resources as managed_resource_facade +from app.runtime.events import EventManager +from app.runtime.resources import ( + configure_managed_resource_runtime, + reset_managed_resource_runtime, +) +from app.scheduler.facade import Scheduler +from app.schemas.types import EventType +from app.startup.initializers import resources as managed_resource_initializer + +_MISSING = object() + + +@pytest.fixture +def database_governance_state() -> Iterator[None]: + """保存并恢复进程级数据库治理门面,避免污染其它测试。""" + try: + previous: Any = get_database_governance() + except RuntimeError: + previous = _MISSING + yield + reset_database_governance() + if previous is not _MISSING: + configure_database_governance(previous) + + +def test_database_governance_reset_contract(database_governance_state: None) -> None: + """数据库治理 reset 后拒绝旧对象,并允许下一代重新装配。""" + first = object() + second = object() + + configure_database_governance(first) # type: ignore[arg-type] + assert get_database_governance() is first + + reset_database_governance() + reset_database_governance() + with pytest.raises(RuntimeError, match="数据库治理服务尚未配置"): + get_database_governance() + + configure_database_governance(second) # type: ignore[arg-type] + assert get_database_governance() is second + assert get_database_governance() is not first + + +def test_resource_version_provider_reset_contract() -> None: + """资源版本 reset 恢复空版本,下一代不得调用上一代 provider。""" + previous = resource_module._resource_version_provider + calls: list[str] = [] + try: + configure_resource_version_provider( + lambda: calls.append("first") or ("1.0", "1.1") + ) + assert resource_module._resource_version_provider() == ("1.0", "1.1") + + reset_resource_version_provider() + reset_resource_version_provider() + assert resource_module._resource_version_provider() == ("0", "0") + + configure_resource_version_provider( + lambda: calls.append("second") or ("2.0", "2.1") + ) + assert resource_module._resource_version_provider() == ("2.0", "2.1") + assert calls == ["first", "second"] + finally: + configure_resource_version_provider(previous) + + +def test_tool_manager_data_context_reset_contract() -> None: + """工具管理器 reset 应清除旧上下文和由它物化的工具目录。""" + manager = MoviePilotToolsManager() + first = object() + second = object() + + manager.set_data_context(first) # type: ignore[arg-type] + manager.tools = [object()] + manager.catalog = object() # type: ignore[assignment] + manager._catalog_materialized = True + manager._catalog_managed_by_factory = True + + manager.reset_data_context() + manager.reset_data_context() + assert manager._data is None + assert manager.tools == [] + assert manager.catalog is None + assert manager._plugin_agent_tools_revision == -1 + assert manager._catalog_materialized is False + + manager.set_data_context(second) # type: ignore[arg-type] + assert manager._data is second + assert manager._data is not first + + +def test_scheduler_runtime_binding_reset_contract() -> None: + """停止态 Scheduler reset 后拒绝旧仓储和服务,并允许下一代装配。""" + scheduler = object.__new__(Scheduler) + Scheduler.__init__(scheduler) + first_tasks = object() + first_services = object() + second_tasks = object() + second_services = object() + + scheduler.configure_agent_tasks(first_tasks) # type: ignore[arg-type] + scheduler.configure_services(first_services) # type: ignore[arg-type] + assert scheduler._agent_task_repository() is first_tasks + assert scheduler._scheduler_services() is first_services + + scheduler.reset_runtime_bindings() + scheduler.reset_runtime_bindings() + with pytest.raises(RuntimeError, match="AgentTask 仓储尚未注入"): + scheduler._agent_task_repository() + with pytest.raises(RuntimeError, match="业务能力尚未注入"): + scheduler._scheduler_services() + + scheduler.configure_agent_tasks(second_tasks) # type: ignore[arg-type] + scheduler.configure_services(second_services) # type: ignore[arg-type] + assert scheduler._agent_task_repository() is second_tasks + assert scheduler._scheduler_services() is second_services + assert scheduler._agent_task_repository() is not first_tasks + + +def test_event_manager_host_binding_reset_contract() -> None: + """事件总线可分别撤销 resolver、错误通知和宿主 listener。""" + manager = object.__new__(EventManager) + EventManager.__init__(manager) + + def first_resolver(_owner: type) -> None: + """代表上一 lifespan 的实例解析器。""" + + def second_resolver(_owner: type) -> None: + """代表下一 lifespan 的实例解析器。""" + + def first_notifier(_title: str, _message: str) -> None: + """代表上一 lifespan 的错误通知器。""" + + def second_notifier(_title: str, _message: str) -> None: + """代表下一 lifespan 的错误通知器。""" + + def listener(_event: object) -> None: + """提供可按稳定标识注册和撤销的测试 listener。""" + + manager.register_handler_instance_resolver("host", first_resolver) + manager.set_error_notifier(first_notifier) + manager.add_event_listener(EventType.NoticeMessage, listener) + + manager.unregister_handler_instance_resolver("host") + manager.unregister_handler_instance_resolver("host") + manager.reset_error_notifier() + manager.reset_error_notifier() + manager.remove_event_listener(EventType.NoticeMessage, listener) + manager.remove_event_listener(EventType.NoticeMessage, listener) + assert manager._EventManager__handler_instance_resolvers == {} + assert manager._EventManager__error_notifier is None + assert not manager._EventManager__broadcast_subscribers.get(EventType.NoticeMessage) + + manager.register_handler_instance_resolver("host", second_resolver) + manager.set_error_notifier(second_notifier) + assert manager._EventManager__handler_instance_resolvers["host"] is second_resolver + assert manager._EventManager__error_notifier is second_notifier + assert second_resolver is not first_resolver + + +def test_managed_resource_owner_reset_releases_only_closed_runtime( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """托管资源 reset 只释放已关闭引用,不把释放动作伪装为完整重启。""" + first = SimpleNamespace(is_shutdown=True) + second = SimpleNamespace(is_shutdown=True) + monkeypatch.setattr(managed_resource_initializer, "_managed_resource_runtime", first) + monkeypatch.setattr(managed_resource_facade, "_managed_resource_runtime", first) + + managed_resource_initializer.reset_managed_resources() + managed_resource_initializer.reset_managed_resources() + assert managed_resource_initializer._managed_resource_runtime is None + assert managed_resource_facade._managed_resource_runtime is None + + configure_managed_resource_runtime(second) # type: ignore[arg-type] + assert managed_resource_facade._managed_resource_runtime is second + assert managed_resource_facade._managed_resource_runtime is not first + reset_managed_resource_runtime() + + +def test_managed_resource_owner_refuses_to_release_live_runtime( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """托管资源仍活动时 reset 必须保留 owner,避免隐藏未收口资源。""" + runtime = SimpleNamespace(is_shutdown=False) + monkeypatch.setattr(managed_resource_initializer, "_managed_resource_runtime", runtime) + monkeypatch.setattr(managed_resource_facade, "_managed_resource_runtime", runtime) + + with pytest.raises(RuntimeError, match="尚未关闭"): + managed_resource_initializer.reset_managed_resources() + + assert managed_resource_initializer._managed_resource_runtime is runtime + assert managed_resource_facade._managed_resource_runtime is runtime