From 0d62822132bec78f72deb315ba6a6399e33e1025 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 29 Aug 2026 18:31:13 +0800 Subject: [PATCH] refactor(runtime): unify capability lifecycle state decisions --- app/runtime/capabilities/runtime.py | 1160 +++++++++-------- .../fixtures/architecture/mypy-baseline.json | 2 - .../fixtures/architecture/ruff-baseline.json | 6 - tests/test_capability_runtime.py | 176 ++- tests/test_managed_resources.py | 46 + .../test_module_manager_capability_adapter.py | 41 +- 6 files changed, 900 insertions(+), 531 deletions(-) diff --git a/app/runtime/capabilities/runtime.py b/app/runtime/capabilities/runtime.py index 43d8402ec..cb3009c7d 100644 --- a/app/runtime/capabilities/runtime.py +++ b/app/runtime/capabilities/runtime.py @@ -9,7 +9,7 @@ from collections import deque from concurrent.futures import Future from dataclasses import dataclass, field from types import MappingProxyType -from typing import Any, Callable, Mapping, Optional +from typing import Any, Callable, Mapping, Optional, Union from app.runtime.capabilities.errors import ( CapabilityAdapterContractError, @@ -48,6 +48,67 @@ class _CapabilityState: inflight_operation: Optional[str] = None +@dataclass(frozen=True, slots=True) +class _PublishedTransition: + """表示状态机已经拥有可直接返回的发布结果。""" + + result: Any + + +@dataclass(frozen=True, slots=True) +class _SettledTransition: + """表示停止状态无需 I/O 即已真实收敛。""" + + +@dataclass(frozen=True, slots=True) +class _TransitionWait: + """表示调用方必须等待另一位 owner 完成当前状态转换。""" + + future: Future[Any] + operation: Optional[str] + + +@dataclass(frozen=True, slots=True) +class _MaterializationClaim: + """记录一次物化状态转换的唯一 owner。""" + + generation: int + future: Future[Any] + + +@dataclass(frozen=True, slots=True) +class _ActivationClaim: + """记录启动状态转换以及需要先收敛的遗留资源。""" + + generation: int + future: Future[Any] + implementation: Any + pending_stop: Any + pending_cleanup: Any + pending_cleanup_error: Optional[BaseException] + + +@dataclass(frozen=True, slots=True) +class _ReloadClaim: + """记录重载状态转换及已撤销可见性的旧实例。""" + + generation: int + future: Future[Any] + implementation: Any + previous: Any + + +@dataclass(frozen=True, slots=True) +class _StopClaim: + """记录停止状态转换及 Runtime 仍持有的全部资源 owner。""" + + generation: int + future: Future[Any] + stop_owner: Any + pending_cleanup: Any + pending_cleanup_error: Optional[BaseException] + + class CapabilityRuntime: """协调能力实现物化和资源生命周期的通用运行时。""" @@ -193,6 +254,321 @@ class CapabilityRuntime: else: future.set_exception(error) + @staticmethod + def _claim_materialization( + state: _CapabilityState, + *, + retry: bool, + ) -> Union[_PublishedTransition, _TransitionWait, _MaterializationClaim]: + """声明一次物化转换，且不执行任何 adapter I/O。""" + with state.lock: + if state.materialization is CapabilityMaterializationState.RESOLVED: + return _PublishedTransition(state.implementation) + if state.materialization is CapabilityMaterializationState.FAILED and not retry: + raise CapabilityOperationError( + state.spec.id, + "materialize", + RuntimeError("能力处于 FAILED，必须显式 retry=True"), + ) + if state.inflight is not None: + return _TransitionWait(state.inflight, state.inflight_operation) + state.generation += 1 + future: Future[Any] = Future() + state.inflight = future + state.inflight_operation = "materialize" + state.materialization = CapabilityMaterializationState.RESOLVING + return _MaterializationClaim(state.generation, future) + + @staticmethod + def _claim_activation( + state: _CapabilityState, + *, + operation: str, + retry: bool, + ) -> Union[_PublishedTransition, _TransitionWait, _ActivationClaim]: + """声明一次启动转换并原子撤销遗留 owner 的共享状态。""" + with state.lock: + if operation == "activate" and state.instance is not None: + return _PublishedTransition(state.instance) + if state.pending_stop is not None and not retry: + raise CapabilityOperationError( + state.spec.id, + operation, + RuntimeError("能力仍持有未释放资源，必须先重试 stop"), + ) + if state.lifecycle is CapabilityLifecycleState.FAILED and not retry: + raise CapabilityOperationError( + state.spec.id, + operation, + RuntimeError("能力处于 FAILED，必须显式 retry=True"), + ) + if state.inflight is not None: + return _TransitionWait(state.inflight, state.inflight_operation) + state.generation += 1 + future: Future[Any] = Future() + state.inflight = future + state.inflight_operation = operation + pending_stop = state.pending_stop + state.lifecycle = ( + CapabilityLifecycleState.STOPPING + if pending_stop is not None + else CapabilityLifecycleState.STARTING + ) + if state.implementation is None: + state.materialization = CapabilityMaterializationState.RESOLVING + claim = _ActivationClaim( + generation=state.generation, + future=future, + implementation=state.implementation, + pending_stop=pending_stop, + pending_cleanup=state.pending_cleanup, + pending_cleanup_error=state.pending_cleanup_error, + ) + state.pending_stop = None + state.pending_cleanup = None + state.pending_cleanup_error = None + return claim + + @staticmethod + def _claim_reload( + state: _CapabilityState, + ) -> Union[_TransitionWait, _ReloadClaim]: + """声明一次重载转换并先撤销旧实例的外部可见性。""" + with state.lock: + if state.inflight is not None: + return _TransitionWait(state.inflight, state.inflight_operation) + if state.pending_stop is not None: + raise CapabilityOperationError( + state.spec.id, + "reload", + RuntimeError("能力仍持有未释放资源，必须先重试 stop"), + ) + if state.instance is None: + raise CapabilityOperationError( + state.spec.id, + "reload", + RuntimeError("只有 RUNNING 能力可以 reload"), + ) + state.generation += 1 + future: Future[Any] = Future() + state.inflight = future + state.inflight_operation = "reload" + claim = _ReloadClaim( + generation=state.generation, + future=future, + implementation=state.implementation, + previous=state.instance, + ) + state.instance = None + state.lifecycle = CapabilityLifecycleState.RELOADING + return claim + + @staticmethod + def _claim_stop( + state: _CapabilityState, + ) -> Union[_SettledTransition, _TransitionWait, _StopClaim]: + """声明一次停止转换并原子接管尚未释放的资源。""" + with state.lock: + if state.inflight is not None: + return _TransitionWait(state.inflight, state.inflight_operation) + stop_owner = ( + state.instance + if state.instance is not None + else state.pending_stop + ) + pending_cleanup = state.pending_cleanup + if stop_owner is None and pending_cleanup is None: + if state.lifecycle is not CapabilityLifecycleState.FAILED: + state.lifecycle = CapabilityLifecycleState.STOPPED + return _SettledTransition() + state.generation += 1 + future: Future[Any] = Future() + state.inflight = future + state.inflight_operation = "stop" + claim = _StopClaim( + generation=state.generation, + future=future, + stop_owner=stop_owner, + pending_cleanup=pending_cleanup, + pending_cleanup_error=state.pending_cleanup_error, + ) + state.instance = None + state.pending_stop = None + state.pending_cleanup = None + state.pending_cleanup_error = None + state.lifecycle = CapabilityLifecycleState.STOPPING + return claim + + @classmethod + def _publish_materialization( + cls, + state: _CapabilityState, + claim: _MaterializationClaim, + implementation: Any, + ) -> None: + """提交物化成功状态并唤醒全部等待者。""" + with state.lock: + state.implementation = implementation + state.materialization = CapabilityMaterializationState.RESOLVED + state.last_error = None + cls._finish_transition(state, claim.future, result=implementation) + + @classmethod + def _fail_materialization( + cls, + state: _CapabilityState, + claim: _MaterializationClaim, + error: BaseException, + ) -> BaseException: + """提交物化失败状态并返回稳定的公开异常。""" + with state.lock: + state.materialization = CapabilityMaterializationState.FAILED + state.last_error = error + operation_error = cls._wrap_error(state.spec.id, "materialize", error) + cls._finish_transition(state, claim.future, error=operation_error) + return operation_error + + @classmethod + def _publish_activation( + cls, + state: _CapabilityState, + claim: _ActivationClaim, + *, + implementation: Any, + candidate: Any, + ) -> None: + """仅在候选启动成功后提交实现与可见实例。""" + with state.lock: + state.implementation = implementation + state.materialization = CapabilityMaterializationState.RESOLVED + state.instance = candidate + state.lifecycle = CapabilityLifecycleState.RUNNING + state.last_error = None + cls._finish_transition(state, claim.future, result=candidate) + + @classmethod + def _fail_activation( + cls, + state: _CapabilityState, + claim: _ActivationClaim, + *, + operation: str, + implementation: Any, + pending_stop: Any, + error: BaseException, + lifecycle_error: BaseException, + ) -> BaseException: + """提交启动失败状态，并保留尚未释放的旧 owner。""" + with state.lock: + state.implementation = implementation + state.materialization = ( + CapabilityMaterializationState.RESOLVED + if implementation is not None + else CapabilityMaterializationState.FAILED + ) + state.instance = None + if pending_stop is not None: + state.pending_stop = pending_stop + state.lifecycle = cls._failure_lifecycle(lifecycle_error) + state.last_error = error + operation_error = cls._wrap_error(state.spec.id, operation, error) + cls._finish_transition(state, claim.future, error=operation_error) + return operation_error + + @classmethod + def _publish_reload( + cls, + state: _CapabilityState, + claim: _ReloadClaim, + candidate: Any, + ) -> None: + """提交重载后的新实例并恢复对外可见性。""" + with state.lock: + state.instance = candidate + state.lifecycle = CapabilityLifecycleState.RUNNING + state.last_error = None + cls._finish_transition(state, claim.future, result=candidate) + + @classmethod + def _fail_reload( + cls, + state: _CapabilityState, + claim: _ReloadClaim, + *, + previous_released: bool, + error: BaseException, + lifecycle_error: BaseException, + ) -> BaseException: + """提交重载失败状态，必要时保留未释放的旧实例。""" + with state.lock: + state.instance = None + if not previous_released: + state.pending_stop = claim.previous + state.lifecycle = cls._failure_lifecycle(lifecycle_error) + state.last_error = error + operation_error = cls._wrap_error(state.spec.id, "reload", error) + cls._finish_transition(state, claim.future, error=operation_error) + return operation_error + + @classmethod + def _publish_stop( + cls, + state: _CapabilityState, + claim: _StopClaim, + ) -> None: + """提交资源真实收敛后的停止状态。""" + with state.lock: + state.lifecycle = CapabilityLifecycleState.STOPPED + state.last_error = None + cls._finish_transition(state, claim.future, result=None) + + @classmethod + def _fail_stop( + cls, + state: _CapabilityState, + claim: _StopClaim, + *, + stop_owner: Any, + pending_cleanup: Any, + error: BaseException, + ) -> BaseException: + """提交停止失败状态并归还仍未释放的资源 owner。""" + with state.lock: + state.lifecycle = CapabilityLifecycleState.FAILED + state.last_error = error + if stop_owner is not None: + state.pending_stop = stop_owner + if pending_cleanup is not None: + state.pending_cleanup = pending_cleanup + state.pending_cleanup_error = error + operation_error = cls._wrap_error(state.spec.id, "stop", error) + cls._finish_transition(state, claim.future, error=operation_error) + return operation_error + + @staticmethod + def _publish_pending_stop_released(state: _CapabilityState) -> None: + """在遗留停止成功后把启动转换推进到 STARTING。""" + with state.lock: + state.lifecycle = CapabilityLifecycleState.STARTING + + @staticmethod + def _restore_pending_cleanup( + state: _CapabilityState, + candidate: Any, + error: Optional[BaseException], + ) -> None: + """清理重试失败时归还 Runtime 对候选资源的所有权。""" + with state.lock: + state.pending_cleanup = candidate + state.pending_cleanup_error = error + + @staticmethod + def _failure_lifecycle(error: BaseException) -> CapabilityLifecycleState: + """把关闭竞态与普通失败映射到统一生命周期结果。""" + if isinstance(error, CapabilityRuntimeClosedError): + return CapabilityLifecycleState.STOPPED + return CapabilityLifecycleState.FAILED + def _calibrate_consumer_materialization(self, state: _CapabilityState) -> None: """从 sys.modules 校准显式 consumer import，不触发任何新导入。""" module_name, symbol_name = state.spec.entrypoint.split(":", maxsplit=1) @@ -260,41 +636,21 @@ class CapabilityRuntime: while True: with self._shutdown_lock: self._ensure_open() - with state.lock: - if state.materialization is CapabilityMaterializationState.RESOLVED: - return state.implementation - if state.materialization is CapabilityMaterializationState.FAILED and not retry: - raise CapabilityOperationError( - state.spec.id, - "materialize", - RuntimeError("能力处于 FAILED，必须显式 retry=True"), - ) - if state.inflight is not None: - waiter = state.inflight - waiter_operation = state.inflight_operation - owner = None - else: - state.generation += 1 - generation = state.generation - future: Future[Any] = Future() - state.inflight = future - state.inflight_operation = "materialize" - state.materialization = CapabilityMaterializationState.RESOLVING - waiter = None - waiter_operation = None - owner = (generation, future) - if waiter is not None: - result = self._wait_sync(waiter) - if waiter_operation == "materialize": + decision = self._claim_materialization(state, retry=retry) + if isinstance(decision, _PublishedTransition): + return decision.result + if isinstance(decision, _TransitionWait): + result = self._wait_sync(decision.future) + if decision.operation == "materialize": return result continue + claim = decision break - generation, future = owner started_at = time.monotonic() self._emit( state, - generation=generation, + generation=claim.generation, operation="materialize", outcome="started", reason=reason, @@ -304,14 +660,10 @@ class CapabilityRuntime: implementation = self._sync_callback(adapter, "materialize", state.spec) implementation = self._canonical_implementation(state.spec, implementation) self._ensure_open() - with state.lock: - state.implementation = implementation - state.materialization = CapabilityMaterializationState.RESOLVED - state.last_error = None - self._finish_transition(state, future, result=implementation) + self._publish_materialization(state, claim, implementation) self._emit( state, - generation=generation, + generation=claim.generation, operation="materialize", outcome="succeeded", reason=reason, @@ -319,14 +671,10 @@ class CapabilityRuntime: ) return implementation except BaseException as error: - with state.lock: - state.materialization = CapabilityMaterializationState.FAILED - state.last_error = error - operation_error = self._wrap_error(state.spec.id, "materialize", error) - self._finish_transition(state, future, error=operation_error) + operation_error = self._fail_materialization(state, claim, error) self._emit( state, - generation=generation, + generation=claim.generation, operation="materialize", outcome="failed", reason=reason, @@ -349,41 +697,21 @@ class CapabilityRuntime: while True: with self._shutdown_lock: self._ensure_open() - with state.lock: - if state.materialization is CapabilityMaterializationState.RESOLVED: - return state.implementation - if state.materialization is CapabilityMaterializationState.FAILED and not retry: - raise CapabilityOperationError( - state.spec.id, - "materialize", - RuntimeError("能力处于 FAILED，必须显式 retry=True"), - ) - if state.inflight is not None: - waiter = state.inflight - waiter_operation = state.inflight_operation - owner = None - else: - state.generation += 1 - generation = state.generation - future: Future[Any] = Future() - state.inflight = future - state.inflight_operation = "materialize" - state.materialization = CapabilityMaterializationState.RESOLVING - waiter = None - waiter_operation = None - owner = (generation, future) - if waiter is not None: - result = await self._wait_async(waiter) - if waiter_operation == "materialize": + decision = self._claim_materialization(state, retry=retry) + if isinstance(decision, _PublishedTransition): + return decision.result + if isinstance(decision, _TransitionWait): + result = await self._wait_async(decision.future) + if decision.operation == "materialize": return result continue + claim = decision break - generation, future = owner started_at = time.monotonic() self._emit( state, - generation=generation, + generation=claim.generation, operation="materialize", outcome="started", reason=reason, @@ -393,14 +721,10 @@ class CapabilityRuntime: implementation = await self._async_callback(adapter, "materialize", state.spec) implementation = self._canonical_implementation(state.spec, implementation) self._ensure_open() - with state.lock: - state.implementation = implementation - state.materialization = CapabilityMaterializationState.RESOLVED - state.last_error = None - self._finish_transition(state, future, result=implementation) + self._publish_materialization(state, claim, implementation) self._emit( state, - generation=generation, + generation=claim.generation, operation="materialize", outcome="succeeded", reason=reason, @@ -408,14 +732,10 @@ class CapabilityRuntime: ) return implementation except BaseException as error: - with state.lock: - state.materialization = CapabilityMaterializationState.FAILED - state.last_error = error - operation_error = self._wrap_error(state.spec.id, "materialize", error) - self._finish_transition(state, future, error=operation_error) + operation_error = self._fail_materialization(state, claim, error) self._emit( state, - generation=generation, + generation=claim.generation, operation="materialize", outcome="failed", reason=reason, @@ -485,74 +805,27 @@ class CapabilityRuntime: while True: with self._shutdown_lock: self._ensure_open() - with state.lock: - if operation == "activate" and state.instance is not None: - return state.instance - if state.pending_stop is not None and not retry: - raise CapabilityOperationError( - state.spec.id, - operation, - RuntimeError("能力仍持有未释放资源，必须先重试 stop"), - ) - if state.lifecycle is CapabilityLifecycleState.FAILED and not retry: - raise CapabilityOperationError( - state.spec.id, - operation, - RuntimeError("能力处于 FAILED，必须显式 retry=True"), - ) - if state.inflight is not None: - waiter = state.inflight - waiter_operation = state.inflight_operation - owner = None - else: - state.generation += 1 - generation = state.generation - future: Future[Any] = Future() - state.inflight = future - state.inflight_operation = operation - pending_stop = state.pending_stop - state.lifecycle = ( - CapabilityLifecycleState.STOPPING - if pending_stop is not None - else CapabilityLifecycleState.STARTING - ) - if state.implementation is None: - state.materialization = CapabilityMaterializationState.RESOLVING - pending_cleanup = state.pending_cleanup - pending_error = state.pending_cleanup_error - state.pending_stop = None - state.pending_cleanup = None - state.pending_cleanup_error = None - implementation = state.implementation - waiter = None - waiter_operation = None - owner = ( - generation, - future, - implementation, - pending_stop, - pending_cleanup, - pending_error, - ) - if waiter is not None: - result = self._wait_sync(waiter) - if waiter_operation == operation: + decision = self._claim_activation( + state, + operation=operation, + retry=retry, + ) + if isinstance(decision, _PublishedTransition): + return decision.result + if isinstance(decision, _TransitionWait): + result = self._wait_sync(decision.future) + if decision.operation == operation: return result continue + claim = decision break - ( - generation, - future, - implementation, - pending_stop, - pending_cleanup, - pending_error, - ) = owner + implementation = claim.implementation + pending_stop = claim.pending_stop started_at = time.monotonic() self._emit( state, - generation=generation, + generation=claim.generation, operation=operation, outcome="started", reason=reason, @@ -566,25 +839,26 @@ class CapabilityRuntime: "stop", state.spec, pending_stop, - generation, + claim.generation, ) pending_stop = None - with state.lock: - state.lifecycle = CapabilityLifecycleState.STARTING - if pending_cleanup is not None: + self._publish_pending_stop_released(state) + if claim.pending_cleanup is not None: try: self._sync_callback( adapter, "cleanup", state.spec, - pending_cleanup, - generation, - pending_error or RuntimeError("pending cleanup"), + claim.pending_cleanup, + claim.generation, + claim.pending_cleanup_error or RuntimeError("pending cleanup"), ) except BaseException: - with state.lock: - state.pending_cleanup = pending_cleanup - state.pending_cleanup_error = pending_error + self._restore_pending_cleanup( + state, + claim.pending_cleanup, + claim.pending_cleanup_error, + ) raise if implementation is None: materialized = self._sync_callback(adapter, "materialize", state.spec) @@ -594,25 +868,30 @@ class CapabilityRuntime: "create", state.spec, implementation, - generation, + claim.generation, previous, ) if candidate is None: raise CapabilityAdapterContractError( f"adapter 没有返回 {state.spec.id} 的 candidate" ) - self._sync_callback(adapter, "start", state.spec, candidate, generation) + self._sync_callback( + adapter, + "start", + state.spec, + candidate, + claim.generation, + ) self._ensure_open() - with state.lock: - state.implementation = implementation - state.materialization = CapabilityMaterializationState.RESOLVED - state.instance = candidate - state.lifecycle = CapabilityLifecycleState.RUNNING - state.last_error = None - self._finish_transition(state, future, result=candidate) + self._publish_activation( + state, + claim, + implementation=implementation, + candidate=candidate, + ) self._emit( state, - generation=generation, + generation=claim.generation, operation=operation, outcome="succeeded", reason=reason, @@ -624,31 +903,22 @@ class CapabilityRuntime: state, adapter, candidate, - generation, + claim.generation, error, ) final_error = cleanup_error or error - with state.lock: - state.implementation = implementation - state.materialization = ( - CapabilityMaterializationState.RESOLVED - if implementation is not None - else CapabilityMaterializationState.FAILED - ) - state.instance = None - if pending_stop is not None: - state.pending_stop = pending_stop - state.lifecycle = ( - CapabilityLifecycleState.STOPPED - if isinstance(error, CapabilityRuntimeClosedError) - else CapabilityLifecycleState.FAILED - ) - state.last_error = final_error - operation_error = self._wrap_error(state.spec.id, operation, final_error) - self._finish_transition(state, future, error=operation_error) + operation_error = self._fail_activation( + state, + claim, + operation=operation, + implementation=implementation, + pending_stop=pending_stop, + error=final_error, + lifecycle_error=error, + ) self._emit( state, - generation=generation, + generation=claim.generation, operation=operation, outcome="failed", reason=reason, @@ -672,74 +942,27 @@ class CapabilityRuntime: while True: with self._shutdown_lock: self._ensure_open() - with state.lock: - if operation == "activate" and state.instance is not None: - return state.instance - if state.pending_stop is not None and not retry: - raise CapabilityOperationError( - state.spec.id, - operation, - RuntimeError("能力仍持有未释放资源，必须先重试 stop"), - ) - if state.lifecycle is CapabilityLifecycleState.FAILED and not retry: - raise CapabilityOperationError( - state.spec.id, - operation, - RuntimeError("能力处于 FAILED，必须显式 retry=True"), - ) - if state.inflight is not None: - waiter = state.inflight - waiter_operation = state.inflight_operation - owner = None - else: - state.generation += 1 - generation = state.generation - future: Future[Any] = Future() - state.inflight = future - state.inflight_operation = operation - pending_stop = state.pending_stop - state.lifecycle = ( - CapabilityLifecycleState.STOPPING - if pending_stop is not None - else CapabilityLifecycleState.STARTING - ) - if state.implementation is None: - state.materialization = CapabilityMaterializationState.RESOLVING - pending_cleanup = state.pending_cleanup - pending_error = state.pending_cleanup_error - state.pending_stop = None - state.pending_cleanup = None - state.pending_cleanup_error = None - implementation = state.implementation - waiter = None - waiter_operation = None - owner = ( - generation, - future, - implementation, - pending_stop, - pending_cleanup, - pending_error, - ) - if waiter is not None: - result = await self._wait_async(waiter) - if waiter_operation == operation: + decision = self._claim_activation( + state, + operation=operation, + retry=retry, + ) + if isinstance(decision, _PublishedTransition): + return decision.result + if isinstance(decision, _TransitionWait): + result = await self._wait_async(decision.future) + if decision.operation == operation: return result continue + claim = decision break - ( - generation, - future, - implementation, - pending_stop, - pending_cleanup, - pending_error, - ) = owner + implementation = claim.implementation + pending_stop = claim.pending_stop started_at = time.monotonic() self._emit( state, - generation=generation, + generation=claim.generation, operation=operation, outcome="started", reason=reason, @@ -753,25 +976,26 @@ class CapabilityRuntime: "stop", state.spec, pending_stop, - generation, + claim.generation, ) pending_stop = None - with state.lock: - state.lifecycle = CapabilityLifecycleState.STARTING - if pending_cleanup is not None: + self._publish_pending_stop_released(state) + if claim.pending_cleanup is not None: try: await self._async_callback( adapter, "cleanup", state.spec, - pending_cleanup, - generation, - pending_error or RuntimeError("pending cleanup"), + claim.pending_cleanup, + claim.generation, + claim.pending_cleanup_error or RuntimeError("pending cleanup"), ) except BaseException: - with state.lock: - state.pending_cleanup = pending_cleanup - state.pending_cleanup_error = pending_error + self._restore_pending_cleanup( + state, + claim.pending_cleanup, + claim.pending_cleanup_error, + ) raise if implementation is None: materialized = await self._async_callback(adapter, "materialize", state.spec) @@ -781,25 +1005,30 @@ class CapabilityRuntime: "create", state.spec, implementation, - generation, + claim.generation, previous, ) if candidate is None: raise CapabilityAdapterContractError( f"adapter 没有返回 {state.spec.id} 的 candidate" ) - await self._async_callback(adapter, "start", state.spec, candidate, generation) + await self._async_callback( + adapter, + "start", + state.spec, + candidate, + claim.generation, + ) self._ensure_open() - with state.lock: - state.implementation = implementation - state.materialization = CapabilityMaterializationState.RESOLVED - state.instance = candidate - state.lifecycle = CapabilityLifecycleState.RUNNING - state.last_error = None - self._finish_transition(state, future, result=candidate) + self._publish_activation( + state, + claim, + implementation=implementation, + candidate=candidate, + ) self._emit( state, - generation=generation, + generation=claim.generation, operation=operation, outcome="succeeded", reason=reason, @@ -811,31 +1040,22 @@ class CapabilityRuntime: state, adapter, candidate, - generation, + claim.generation, error, ) final_error = cleanup_error or error - with state.lock: - state.implementation = implementation - state.materialization = ( - CapabilityMaterializationState.RESOLVED - if implementation is not None - else CapabilityMaterializationState.FAILED - ) - state.instance = None - if pending_stop is not None: - state.pending_stop = pending_stop - state.lifecycle = ( - CapabilityLifecycleState.STOPPED - if isinstance(error, CapabilityRuntimeClosedError) - else CapabilityLifecycleState.FAILED - ) - state.last_error = final_error - operation_error = self._wrap_error(state.spec.id, operation, final_error) - self._finish_transition(state, future, error=operation_error) + operation_error = self._fail_activation( + state, + claim, + operation=operation, + implementation=implementation, + pending_stop=pending_stop, + error=final_error, + lifecycle_error=error, + ) self._emit( state, - generation=generation, + generation=claim.generation, operation=operation, outcome="failed", reason=reason, @@ -865,9 +1085,7 @@ class CapabilityRuntime: ) return None except BaseException as cleanup_error: - with state.lock: - state.pending_cleanup = candidate - state.pending_cleanup_error = cleanup_error + self._restore_pending_cleanup(state, candidate, cleanup_error) return RuntimeError(f"{error}; cleanup failed: {cleanup_error}") async def _cleanup_failed_async( @@ -891,9 +1109,7 @@ class CapabilityRuntime: ) return None except BaseException as cleanup_error: - with state.lock: - state.pending_cleanup = candidate - state.pending_cleanup_error = cleanup_error + self._restore_pending_cleanup(state, candidate, cleanup_error) return RuntimeError(f"{error}; cleanup failed: {cleanup_error}") def reload(self, capability_id: str, *, reason: str) -> Any: @@ -903,42 +1119,18 @@ class CapabilityRuntime: while True: with self._shutdown_lock: self._ensure_open() - with state.lock: - if state.inflight is not None: - waiter = state.inflight - waiter_operation = state.inflight_operation - else: - waiter = None - waiter_operation = None - if state.pending_stop is not None: - raise CapabilityOperationError( - state.spec.id, - "reload", - RuntimeError("能力仍持有未释放资源，必须先重试 stop"), - ) - if state.instance is None: - raise CapabilityOperationError( - state.spec.id, - "reload", - RuntimeError("只有 RUNNING 能力可以 reload"), - ) - previous = state.instance - state.generation += 1 - generation = state.generation - future: Future[Any] = Future() - state.inflight = future - state.inflight_operation = "reload" - state.instance = None - state.lifecycle = CapabilityLifecycleState.RELOADING - implementation = state.implementation - break - result = self._wait_sync(waiter) - if waiter_operation == "reload": - return result + decision = self._claim_reload(state) + if isinstance(decision, _TransitionWait): + result = self._wait_sync(decision.future) + if decision.operation == "reload": + return result + continue + claim = decision + break started_at = time.monotonic() self._emit( state, - generation=generation, + generation=claim.generation, operation="reload", outcome="started", reason=reason, @@ -947,30 +1139,38 @@ class CapabilityRuntime: candidate = None previous_released = False try: - self._sync_callback(adapter, "stop", state.spec, previous, generation) + self._sync_callback( + adapter, + "stop", + state.spec, + claim.previous, + claim.generation, + ) previous_released = True candidate = self._sync_callback( adapter, "create", state.spec, - implementation, - generation, - previous, + claim.implementation, + claim.generation, + claim.previous, ) if candidate is None: raise CapabilityAdapterContractError( f"adapter 没有返回 {state.spec.id} 的 candidate" ) - self._sync_callback(adapter, "start", state.spec, candidate, generation) + self._sync_callback( + adapter, + "start", + state.spec, + candidate, + claim.generation, + ) self._ensure_open() - with state.lock: - state.instance = candidate - state.lifecycle = CapabilityLifecycleState.RUNNING - state.last_error = None - self._finish_transition(state, future, result=candidate) + self._publish_reload(state, claim, candidate) self._emit( state, - generation=generation, + generation=claim.generation, operation="reload", outcome="succeeded", reason=reason, @@ -982,25 +1182,20 @@ class CapabilityRuntime: state, adapter, candidate, - generation, + claim.generation, error, ) final_error = cleanup_error or error - with state.lock: - state.instance = None - if not previous_released: - state.pending_stop = previous - state.lifecycle = ( - CapabilityLifecycleState.STOPPED - if isinstance(error, CapabilityRuntimeClosedError) - else CapabilityLifecycleState.FAILED - ) - state.last_error = final_error - operation_error = self._wrap_error(state.spec.id, "reload", final_error) - self._finish_transition(state, future, error=operation_error) + operation_error = self._fail_reload( + state, + claim, + previous_released=previous_released, + error=final_error, + lifecycle_error=error, + ) self._emit( state, - generation=generation, + generation=claim.generation, operation="reload", outcome="failed", reason=reason, @@ -1016,42 +1211,18 @@ class CapabilityRuntime: while True: with self._shutdown_lock: self._ensure_open() - with state.lock: - if state.inflight is not None: - waiter = state.inflight - waiter_operation = state.inflight_operation - else: - waiter = None - waiter_operation = None - if state.pending_stop is not None: - raise CapabilityOperationError( - state.spec.id, - "reload", - RuntimeError("能力仍持有未释放资源，必须先重试 stop"), - ) - if state.instance is None: - raise CapabilityOperationError( - state.spec.id, - "reload", - RuntimeError("只有 RUNNING 能力可以 reload"), - ) - previous = state.instance - state.generation += 1 - generation = state.generation - future: Future[Any] = Future() - state.inflight = future - state.inflight_operation = "reload" - state.instance = None - state.lifecycle = CapabilityLifecycleState.RELOADING - implementation = state.implementation - break - result = await self._wait_async(waiter) - if waiter_operation == "reload": - return result + decision = self._claim_reload(state) + if isinstance(decision, _TransitionWait): + result = await self._wait_async(decision.future) + if decision.operation == "reload": + return result + continue + claim = decision + break started_at = time.monotonic() self._emit( state, - generation=generation, + generation=claim.generation, operation="reload", outcome="started", reason=reason, @@ -1060,30 +1231,38 @@ class CapabilityRuntime: candidate = None previous_released = False try: - await self._async_callback(adapter, "stop", state.spec, previous, generation) + await self._async_callback( + adapter, + "stop", + state.spec, + claim.previous, + claim.generation, + ) previous_released = True candidate = await self._async_callback( adapter, "create", state.spec, - implementation, - generation, - previous, + claim.implementation, + claim.generation, + claim.previous, ) if candidate is None: raise CapabilityAdapterContractError( f"adapter 没有返回 {state.spec.id} 的 candidate" ) - await self._async_callback(adapter, "start", state.spec, candidate, generation) + await self._async_callback( + adapter, + "start", + state.spec, + candidate, + claim.generation, + ) self._ensure_open() - with state.lock: - state.instance = candidate - state.lifecycle = CapabilityLifecycleState.RUNNING - state.last_error = None - self._finish_transition(state, future, result=candidate) + self._publish_reload(state, claim, candidate) self._emit( state, - generation=generation, + generation=claim.generation, operation="reload", outcome="succeeded", reason=reason, @@ -1095,25 +1274,20 @@ class CapabilityRuntime: state, adapter, candidate, - generation, + claim.generation, error, ) final_error = cleanup_error or error - with state.lock: - state.instance = None - if not previous_released: - state.pending_stop = previous - state.lifecycle = ( - CapabilityLifecycleState.STOPPED - if isinstance(error, CapabilityRuntimeClosedError) - else CapabilityLifecycleState.FAILED - ) - state.last_error = final_error - operation_error = self._wrap_error(state.spec.id, "reload", final_error) - self._finish_transition(state, future, error=operation_error) + operation_error = self._fail_reload( + state, + claim, + previous_released=previous_released, + error=final_error, + lifecycle_error=error, + ) self._emit( state, - generation=generation, + generation=claim.generation, operation="reload", outcome="failed", reason=reason, @@ -1131,54 +1305,29 @@ class CapabilityRuntime: state = self._state(capability_id) adapter = self._adapter(state, AdapterExecutionMode.SYNC) while True: - with state.lock: - if state.inflight is not None: - waiter = state.inflight - waiter_operation = state.inflight_operation - owner = None - else: - stop_owner = ( - state.instance - if state.instance is not None - else state.pending_stop - ) - pending = state.pending_cleanup - pending_error = state.pending_cleanup_error - if stop_owner is None and pending is None: - if state.lifecycle is not CapabilityLifecycleState.FAILED: - state.lifecycle = CapabilityLifecycleState.STOPPED - return True - state.generation += 1 - generation = state.generation - future: Future[Any] = Future() - state.inflight = future - state.inflight_operation = "stop" - state.instance = None - state.pending_stop = None - state.pending_cleanup = None - state.pending_cleanup_error = None - state.lifecycle = CapabilityLifecycleState.STOPPING - waiter = None - waiter_operation = None - owner = (generation, future, stop_owner, pending, pending_error) - if waiter is not None: + decision = self._claim_stop(state) + if isinstance(decision, _SettledTransition): + return True + if isinstance(decision, _TransitionWait): try: - self._wait_sync(waiter) + self._wait_sync(decision.future) except BaseException: if not shutdown: raise - if waiter_operation == "stop": + if decision.operation == "stop": return False - if waiter_operation == "stop": + if decision.operation == "stop": return True continue + claim = decision break - generation, future, stop_owner, pending, pending_error = owner + stop_owner = claim.stop_owner + pending_cleanup = claim.pending_cleanup started_at = time.monotonic() self._emit( state, - generation=generation, + generation=claim.generation, operation="stop", outcome="started", reason=reason, @@ -1186,25 +1335,28 @@ class CapabilityRuntime: ) try: if stop_owner is not None: - self._sync_callback(adapter, "stop", state.spec, stop_owner, generation) + self._sync_callback( + adapter, + "stop", + state.spec, + stop_owner, + claim.generation, + ) stop_owner = None - if pending is not None: + if pending_cleanup is not None: self._sync_callback( adapter, "cleanup", state.spec, - pending, - generation, - pending_error or RuntimeError("pending cleanup"), + pending_cleanup, + claim.generation, + claim.pending_cleanup_error or RuntimeError("pending cleanup"), ) - pending = None - with state.lock: - state.lifecycle = CapabilityLifecycleState.STOPPED - state.last_error = None - self._finish_transition(state, future, result=None) + pending_cleanup = None + self._publish_stop(state, claim) self._emit( state, - generation=generation, + generation=claim.generation, operation="stop", outcome="succeeded", reason=reason, @@ -1212,19 +1364,16 @@ class CapabilityRuntime: ) return True except BaseException as error: - with state.lock: - state.lifecycle = CapabilityLifecycleState.FAILED - state.last_error = error - if stop_owner is not None: - state.pending_stop = stop_owner - if pending is not None: - state.pending_cleanup = pending - state.pending_cleanup_error = error - operation_error = self._wrap_error(state.spec.id, "stop", error) - self._finish_transition(state, future, error=operation_error) + operation_error = self._fail_stop( + state, + claim, + stop_owner=stop_owner, + pending_cleanup=pending_cleanup, + error=error, + ) self._emit( state, - generation=generation, + generation=claim.generation, operation="stop", outcome="failed", reason=reason, @@ -1250,54 +1399,29 @@ class CapabilityRuntime: state = self._state(capability_id) adapter = self._adapter(state, AdapterExecutionMode.ASYNC) while True: - with state.lock: - if state.inflight is not None: - waiter = state.inflight - waiter_operation = state.inflight_operation - owner = None - else: - stop_owner = ( - state.instance - if state.instance is not None - else state.pending_stop - ) - pending = state.pending_cleanup - pending_error = state.pending_cleanup_error - if stop_owner is None and pending is None: - if state.lifecycle is not CapabilityLifecycleState.FAILED: - state.lifecycle = CapabilityLifecycleState.STOPPED - return True - state.generation += 1 - generation = state.generation - future: Future[Any] = Future() - state.inflight = future - state.inflight_operation = "stop" - state.instance = None - state.pending_stop = None - state.pending_cleanup = None - state.pending_cleanup_error = None - state.lifecycle = CapabilityLifecycleState.STOPPING - waiter = None - waiter_operation = None - owner = (generation, future, stop_owner, pending, pending_error) - if waiter is not None: + decision = self._claim_stop(state) + if isinstance(decision, _SettledTransition): + return True + if isinstance(decision, _TransitionWait): try: - await self._wait_async(waiter) + await self._wait_async(decision.future) except BaseException: if not shutdown: raise - if waiter_operation == "stop": + if decision.operation == "stop": return False - if waiter_operation == "stop": + if decision.operation == "stop": return True continue + claim = decision break - generation, future, stop_owner, pending, pending_error = owner + stop_owner = claim.stop_owner + pending_cleanup = claim.pending_cleanup started_at = time.monotonic() self._emit( state, - generation=generation, + generation=claim.generation, operation="stop", outcome="started", reason=reason, @@ -1310,26 +1434,23 @@ class CapabilityRuntime: "stop", state.spec, stop_owner, - generation, + claim.generation, ) stop_owner = None - if pending is not None: + if pending_cleanup is not None: await self._async_callback( adapter, "cleanup", state.spec, - pending, - generation, - pending_error or RuntimeError("pending cleanup"), + pending_cleanup, + claim.generation, + claim.pending_cleanup_error or RuntimeError("pending cleanup"), ) - pending = None - with state.lock: - state.lifecycle = CapabilityLifecycleState.STOPPED - state.last_error = None - self._finish_transition(state, future, result=None) + pending_cleanup = None + self._publish_stop(state, claim) self._emit( state, - generation=generation, + generation=claim.generation, operation="stop", outcome="succeeded", reason=reason, @@ -1337,19 +1458,16 @@ class CapabilityRuntime: ) return True except BaseException as error: - with state.lock: - state.lifecycle = CapabilityLifecycleState.FAILED - state.last_error = error - if stop_owner is not None: - state.pending_stop = stop_owner - if pending is not None: - state.pending_cleanup = pending - state.pending_cleanup_error = error - operation_error = self._wrap_error(state.spec.id, "stop", error) - self._finish_transition(state, future, error=operation_error) + operation_error = self._fail_stop( + state, + claim, + stop_owner=stop_owner, + pending_cleanup=pending_cleanup, + error=error, + ) self._emit( state, - generation=generation, + generation=claim.generation, operation="stop", outcome="failed", reason=reason, diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index 58fb4c024..d718eab4a 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -2962,8 +2962,6 @@ "unused-coroutine": 3 }, "app/runtime/capabilities/runtime.py": { - "arg-type": 2, - "misc": 6, "no-untyped-def": 2 }, "app/runtime/compat/imports.py": { diff --git a/tests/fixtures/architecture/ruff-baseline.json b/tests/fixtures/architecture/ruff-baseline.json index c9ffc02aa..43f5e5e8e 100644 --- a/tests/fixtures/architecture/ruff-baseline.json +++ b/tests/fixtures/architecture/ruff-baseline.json @@ -899,9 +899,6 @@ "tests/test_capability_registry.py": { "I001": 1 }, - "tests/test_capability_runtime.py": { - "I001": 1 - }, "tests/test_cli_auto_update.py": { "I001": 1 }, @@ -1080,9 +1077,6 @@ "tests/test_metamusic.py": { "I001": 1 }, - "tests/test_module_manager_capability_adapter.py": { - "I001": 1 - }, "tests/test_module_quality.py": { "I001": 1 }, diff --git a/tests/test_capability_runtime.py b/tests/test_capability_runtime.py index 3c0f3601e..fa7429063 100644 --- a/tests/test_capability_runtime.py +++ b/tests/test_capability_runtime.py @@ -25,7 +25,6 @@ from app.runtime.capabilities.model import ( from app.runtime.capabilities.registry import CapabilityRegistry from app.runtime.capabilities.runtime import CapabilityRuntime - _MANIFEST = """ schema_version = 1 id = "sample.capability" @@ -102,6 +101,7 @@ class _SyncAdapter: self.fail_materialize = False self.fail_start = False self.fail_stop = False + self.fail_cleanup = False self.start_entered: Optional[threading.Event] = None self.start_release: Optional[threading.Event] = None self.stop_entered: Optional[threading.Event] = None @@ -140,6 +140,8 @@ class _SyncAdapter: def cleanup(self, spec, candidate: _Candidate, generation: int, error: BaseException) -> None: self.cleanup_calls += 1 + if self.fail_cleanup: + raise RuntimeError("cleanup failed") candidate.stopped = True @@ -156,6 +158,7 @@ class _AsyncAdapter: self.fail_materialize = False self.fail_start = False self.fail_stop = False + self.fail_cleanup = False self.start_entered = asyncio.Event() self.start_release = asyncio.Event() self.stop_entered: Optional[asyncio.Event] = None @@ -196,6 +199,8 @@ class _AsyncAdapter: async def cleanup(self, spec, candidate: _Candidate, generation: int, error: BaseException) -> None: self.cleanup_calls += 1 await asyncio.sleep(0) + if self.fail_cleanup: + raise RuntimeError("async cleanup failed") candidate.stopped = True @@ -983,3 +988,172 @@ def test_adapter_mode_requires_declared_enum_member(tmp_path: Path) -> None: with pytest.raises(CapabilityAdapterModeError): runtime.activate("sample.capability", reason="invalid_mode") + + +def _snapshot_signature(runtime: CapabilityRuntime) -> tuple: + """提取忽略具体错误文本的生命周期状态签名。""" + snapshot = runtime.snapshot("sample.capability") + return ( + snapshot.materialization, + snapshot.lifecycle, + snapshot.generation, + snapshot.visible, + snapshot.error is not None, + ) + + +def _observation_signature(runtime: CapabilityRuntime) -> tuple: + """提取可比较的转换观测签名，排除耗时和具体错误文本。""" + return tuple( + ( + item.generation, + item.operation, + item.outcome, + item.materialization, + item.lifecycle, + item.error is not None, + ) + for item in runtime.observations("sample.capability") + ) + + +def _sync_lifecycle_trace(tmp_path: Path) -> tuple[tuple, tuple]: + """执行同步双轴生命周期并返回状态和观测轨迹。""" + adapter = _SyncAdapter() + runtime = CapabilityRuntime(_registry(tmp_path), adapters={"sample": adapter}) + snapshots = [] + + runtime.materialize("sample.capability", reason="parity_materialize") + snapshots.append(_snapshot_signature(runtime)) + runtime.activate("sample.capability", reason="parity_activate") + snapshots.append(_snapshot_signature(runtime)) + runtime.reload("sample.capability", reason="parity_reload") + snapshots.append(_snapshot_signature(runtime)) + runtime.stop("sample.capability", reason="parity_stop") + snapshots.append(_snapshot_signature(runtime)) + + return tuple(snapshots), _observation_signature(runtime) + + +async def _async_lifecycle_trace(tmp_path: Path) -> tuple[tuple, tuple]: + """执行异步双轴生命周期并返回状态和观测轨迹。""" + adapter = _AsyncAdapter() + adapter.start_release.set() + runtime = CapabilityRuntime(_registry(tmp_path), adapters={"sample": adapter}) + snapshots = [] + + await runtime.materialize_async("sample.capability", reason="parity_materialize") + snapshots.append(_snapshot_signature(runtime)) + await runtime.activate_async("sample.capability", reason="parity_activate") + snapshots.append(_snapshot_signature(runtime)) + await runtime.reload_async("sample.capability", reason="parity_reload") + snapshots.append(_snapshot_signature(runtime)) + await runtime.stop_async("sample.capability", reason="parity_stop") + snapshots.append(_snapshot_signature(runtime)) + + return tuple(snapshots), _observation_signature(runtime) + + +def test_sync_and_async_lifecycle_publish_identical_state_traces( + tmp_path: Path, +) -> None: + """双 ABI 应共享完全一致的物化、启动、重载和停止状态决策。""" + sync_trace = _sync_lifecycle_trace(tmp_path / "sync") + sys.modules.pop("sample_implementation", None) + async_trace = asyncio.run(_async_lifecycle_trace(tmp_path / "async")) + + assert async_trace == sync_trace + + +def test_sync_and_async_start_failures_publish_identical_state( + tmp_path: Path, +) -> None: + """启动和候选清理的 I/O 不同，但失败后的状态提交必须一致。""" + sync_adapter = _SyncAdapter() + sync_adapter.fail_start = True + sync_runtime = CapabilityRuntime( + _registry(tmp_path / "sync"), + adapters={"sample": sync_adapter}, + ) + with pytest.raises(CapabilityOperationError, match="start failed"): + sync_runtime.activate("sample.capability", reason="parity_failure") + sys.modules.pop("sample_implementation", None) + + async_adapter = _AsyncAdapter() + async_adapter.fail_start = True + async_adapter.start_release.set() + async_runtime = CapabilityRuntime( + _registry(tmp_path / "async"), + adapters={"sample": async_adapter}, + ) + + async def fail_async() -> None: + """触发异步启动失败并验证公开异常合同。""" + with pytest.raises(CapabilityOperationError, match="async start failed"): + await async_runtime.activate_async( + "sample.capability", + reason="parity_failure", + ) + + asyncio.run(fail_async()) + + assert _snapshot_signature(async_runtime) == _snapshot_signature(sync_runtime) + assert _observation_signature(async_runtime) == _observation_signature(sync_runtime) + + +def test_sync_and_async_cleanup_retry_publish_identical_state( + tmp_path: Path, +) -> None: + """候选清理失败后的 owner 归还与显式重试在双 ABI 下必须一致。""" + sync_adapter = _SyncAdapter() + sync_adapter.fail_start = True + sync_adapter.fail_cleanup = True + sync_runtime = CapabilityRuntime( + _registry(tmp_path / "sync"), + adapters={"sample": sync_adapter}, + ) + with pytest.raises(CapabilityOperationError, match="cleanup failed"): + sync_runtime.activate("sample.capability", reason="cleanup_failure") + sync_adapter.fail_start = False + sync_adapter.fail_cleanup = False + sync_runtime.activate( + "sample.capability", + reason="cleanup_retry", + retry=True, + ) + sync_state = ( + _snapshot_signature(sync_runtime), + _observation_signature(sync_runtime), + ) + sys.modules.pop("sample_implementation", None) + + async_adapter = _AsyncAdapter() + async_adapter.fail_start = True + async_adapter.fail_cleanup = True + async_adapter.start_release.set() + async_runtime = CapabilityRuntime( + _registry(tmp_path / "async"), + adapters={"sample": async_adapter}, + ) + + async def exercise_async() -> None: + """触发异步清理失败后完成显式恢复。""" + with pytest.raises(CapabilityOperationError, match="async cleanup failed"): + await async_runtime.activate_async( + "sample.capability", + reason="cleanup_failure", + ) + async_adapter.fail_start = False + async_adapter.fail_cleanup = False + await async_runtime.activate_async( + "sample.capability", + reason="cleanup_retry", + retry=True, + ) + + asyncio.run(exercise_async()) + + assert ( + _snapshot_signature(async_runtime), + _observation_signature(async_runtime), + ) == sync_state diff --git a/tests/test_managed_resources.py b/tests/test_managed_resources.py index d976b2620..279f8c95b 100644 --- a/tests/test_managed_resources.py +++ b/tests/test_managed_resources.py @@ -200,6 +200,52 @@ def test_async_managed_resource_uses_async_adapter( assert AsyncResource.instances == [resource] +def test_async_managed_resource_recreates_owner_across_event_loops( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """停止后的下一次激活必须在新事件循环创建新资源，不能复用旧 owner。""" + module_name = "fixture_loop_bound_managed_resource" + module = ModuleType(module_name) + + class LoopBoundResource: + """记录资源实际启动和停止所在的事件循环。""" + + def __init__(self) -> None: + self.loop = None + + async def start(self) -> None: + self.loop = asyncio.get_running_loop() + + async def stop(self) -> None: + assert asyncio.get_running_loop() is self.loop + + module.LoopBoundResource = LoopBoundResource + monkeypatch.setitem(sys.modules, module_name, module) + _write_manifest( + tmp_path, + capability_id="fixture.loop_bound", + kind=MANAGED_RESOURCE_ASYNC_KIND, + entrypoint=f"{module_name}:LoopBoundResource", + ) + runtime = _runtime(tmp_path) + + async def exercise() -> LoopBoundResource: + """在当前事件循环完成一次资源启动与停止。""" + resource = await runtime.activate_async( + "fixture.loop_bound", + reason="loop_cycle", + ) + await runtime.stop_async("fixture.loop_bound", reason="loop_cycle") + return resource + + first = asyncio.run(exercise()) + second = asyncio.run(exercise()) + + assert first is not second + assert first.loop is not second.loop + + def test_shutdown_propagates_stop_failure_and_retains_owner_for_retry( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/test_module_manager_capability_adapter.py b/tests/test_module_manager_capability_adapter.py index c9f3d4493..1fb4610d2 100644 --- a/tests/test_module_manager_capability_adapter.py +++ b/tests/test_module_manager_capability_adapter.py @@ -25,7 +25,6 @@ from app.runtime.extensions.service_config import configure_service_config_reade from app.schemas.event import ConfigChangeEventData from app.schemas.types import EventType - _SAMPLE_MANIFEST = """ schema_version = 1 id = "SampleModule" @@ -430,6 +429,46 @@ def test_config_reconcile_reload_and_stop_preserve_manager_contract( assert restarted.events == ["create", "start", "stop"] +def test_module_manager_lifecycle_keeps_monotonic_transition_generations( + module_manager_harness, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Manager 的旧同步入口必须按共享状态机顺序提交每一代转换。""" + manager = module_manager_harness.manager + _enable_sample(module_manager_harness.config_values) + monkeypatch.setattr(eventmanager, "send_event", Mock()) + + manager.load_modules() + manager.reload() + manager.stop() + manager.load_modules() + module_manager_harness.config_values["Notifications"] = [] + manager.load_modules() + + observations = manager._runtime.observations("SampleModule") + assert [ + (item.generation, item.operation, item.outcome) + for item in observations + ] == [ + (1, "activate", "started"), + (1, "activate", "succeeded"), + (2, "stop", "started"), + (2, "stop", "succeeded"), + (3, "activate", "started"), + (3, "activate", "succeeded"), + (4, "stop", "started"), + (4, "stop", "succeeded"), + (5, "activate", "started"), + (5, "activate", "succeeded"), + (6, "stop", "started"), + (6, "stop", "succeeded"), + ] + snapshot = manager._runtime.snapshot("SampleModule") + assert snapshot.generation == 6 + assert snapshot.lifecycle is CapabilityLifecycleState.STOPPED + assert snapshot.visible is False + + def test_config_event_reloads_same_instance_and_tracks_selector_changes( module_manager_harness, ) -> None: