refactor(startup): complete provider reset contracts

This commit is contained in:
jxxghp
2026-08-29 18:32:03 +08:00
parent 0d62822132
commit 95606801af
9 changed files with 285 additions and 5 deletions

View File

@@ -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:
"""
检测和更新资源包

View File

@@ -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:

View File

@@ -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:

View File

@@ -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:

View File

@@ -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):
"""
开始广播事件处理线程

View File

@@ -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: ...

View File

@@ -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:

View File

@@ -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

View File

@@ -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