refactor: unify capability shutdown convergence

tests/test_agent_runtime_loader.py

@@ -91,6 +91,32 @@ async def test_shutdown_retains_nonconverged_agent_service_for_retry(
     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]:
     orchestrator = types.ModuleType("app.agent.orchestrator")
     orchestrator.agent_manager = manager if manager is not None else object()

tests/test_capability_runtime.py

@@ -791,6 +791,29 @@ async def test_stop_async_failure_retains_ownership_for_explicit_retry(tmp_path:
     assert runtime.snapshot("sample.capability").lifecycle is CapabilityLifecycleState.STOPPED
 
 
+@pytest.mark.asyncio
+async def test_shutdown_async_reports_unreleased_owner_until_retry(tmp_path: Path) -> None:
+    """异步 shutdown 不得把 stop 失败伪装成整体成功。"""
+    adapter = _AsyncAdapter()
+    runtime = CapabilityRuntime(_registry(tmp_path), adapters={"sample": adapter})
+    activation = asyncio.create_task(
+        runtime.activate_async("sample.capability", reason="initial")
+    )
+    await adapter.start_entered.wait()
+    adapter.start_release.set()
+    instance = await activation
+    adapter.fail_stop = True
+
+    assert await runtime.shutdown_async(reason="application_shutdown") is False
+    assert runtime.snapshot(
+        "sample.capability"
+    ).lifecycle is CapabilityLifecycleState.FAILED
+
+    adapter.fail_stop = False
+    assert await runtime.shutdown_async(reason="shutdown_retry") is True
+    assert adapter.stop_instances == [instance, instance]
+
+
 @pytest.mark.asyncio
 async def test_async_reload_uses_reloading_state_and_hides_candidate(tmp_path: Path) -> None:
     """异步 reload 与同步入口遵守相同状态和发布边界。"""

tests/test_lifecycle_shutdown.py

@@ -938,8 +938,21 @@ def test_stop_modules_continues_after_internal_owner_failure(monkeypatch):
     dependencies = _patch_module_shutdown_dependencies(monkeypatch)
     dependencies["module"].side_effect = RuntimeError("module failed")
 
-    asyncio.run(modules_initializer.stop_modules())
+    converged = asyncio.run(modules_initializer.stop_modules())
 
+    assert converged is False
+    for dependency in dependencies.values():
+        _assert_completed_once(dependency)
+
+
+def test_stop_modules_propagates_false_without_skipping_later_cleanup(monkeypatch):
+    """关闭回调显式返回 False 时不得被转换为整体成功。"""
+    dependencies = _patch_module_shutdown_dependencies(monkeypatch)
+    dependencies["module"].return_value = False
+
+    converged = asyncio.run(modules_initializer.stop_modules())
+
+    assert converged is False
     for dependency in dependencies.values():
         _assert_completed_once(dependency)
 
@@ -963,15 +976,22 @@ def test_stop_modules_drains_web_agent_tasks_before_persistence(monkeypatch):
         "get_configured_agent_chat_persistence",
         MagicMock(return_value=persistence),
     )
+
+    async def stop_database_worker() -> None:
+        """模拟生产 worker 关闭后释放组合根句柄。"""
+        order.append("database")
+        modules_initializer._database_worker = None
+
     monkeypatch.setattr(
         modules_initializer,
         "stop_database_worker",
-        AsyncMock(side_effect=lambda: order.append("database")),
+        AsyncMock(side_effect=stop_database_worker),
     )
     monkeypatch.setattr(modules_initializer, "_database_worker", object())
 
-    asyncio.run(modules_initializer.stop_modules())
+    converged = asyncio.run(modules_initializer.stop_modules())
 
+    assert converged is True
     assert order == ["web-agent", "persistence-admission", "persistence", "database"]
 
 

tests/test_managed_resources.py

@@ -143,7 +143,10 @@ def test_sync_managed_resource_is_single_flight(
         if observation.operation == "activate"
     ] == ["started", "succeeded"]
 
-    asyncio.run(shutdown_managed_resource_runtime(reason="test_shutdown"))
+    assert (
+        asyncio.run(shutdown_managed_resource_runtime(reason="test_shutdown"))
+        is True
+    )
 
     assert SyncResource.instances[0].stopped == 1
     with pytest.raises(CapabilityRuntimeClosedError):
@@ -198,6 +201,55 @@ def test_async_managed_resource_uses_async_adapter(
     assert AsyncResource.instances == [resource]
 
 
+def test_shutdown_propagates_stop_failure_and_retains_owner_for_retry(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """托管资源关闭失败必须向 startup 传播，并保留同一 owner 重试。"""
+    module_name = "fixture_retry_stop_managed_resource"
+    module = ModuleType(module_name)
+
+    class RetryStopResource:
+        """首次停止失败、第二次停止收敛的同步资源。"""
+
+        def __init__(self) -> None:
+            self.stop_calls = 0
+            self.fail_stop = True
+
+        def start(self) -> None:
+            """资源启动无需额外动作。"""
+
+        def stop(self) -> None:
+            """按测试开关模拟资源释放失败。"""
+            self.stop_calls += 1
+            if self.fail_stop:
+                raise RuntimeError("stop failed")
+
+    module.RetryStopResource = RetryStopResource
+    monkeypatch.setitem(sys.modules, module_name, module)
+    _write_manifest(
+        tmp_path,
+        capability_id="fixture.retry_stop",
+        kind=MANAGED_RESOURCE_SYNC_KIND,
+        entrypoint=f"{module_name}:RetryStopResource",
+    )
+    configure_managed_resource_runtime(_runtime(tmp_path))
+    resource = acquire_managed_resource("fixture.retry_stop", reason="test")
+
+    assert (
+        asyncio.run(shutdown_managed_resource_runtime(reason="test_shutdown"))
+        is False
+    )
+    assert resource.stop_calls == 1
+
+    resource.fail_stop = False
+    assert (
+        asyncio.run(shutdown_managed_resource_runtime(reason="shutdown_retry"))
+        is True
+    )
+    assert resource.stop_calls == 2
+
+
 def test_failed_start_is_cleaned_before_explicit_retry(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
@@ -334,6 +386,6 @@ def test_startup_shutdown_without_init_does_not_build_registry(monkeypatch) -> N
         build_registry,
     )
 
-    asyncio.run(managed_resources_initializer.stop_managed_resources())
+    assert asyncio.run(managed_resources_initializer.stop_managed_resources()) is True
 
     build_registry.assert_not_called()

tests/test_module_lifecycle.py

@@ -176,6 +176,7 @@ def test_telegram_stop_closes_sdk_and_waits_for_polling_thread():
     bot = Mock()
     client._bot = bot
     polling_thread = Mock()
+    polling_thread.is_alive.side_effect = [True, False]
     client._polling_thread = polling_thread
     client._typing_tasks = {}
     client._typing_stop_flags = {}
@@ -183,10 +184,51 @@ def test_telegram_stop_closes_sdk_and_waits_for_polling_thread():
     client._typing_lifecycle_lock = threading.RLock()
     client._typing_accepting = True
 
-    client.stop()
-    client.stop()
+    assert client.stop() is True
+    assert client.stop() is True
 
     bot.stop_bot.assert_called_once_with()
-    polling_thread.join.assert_called_once_with()
+    polling_thread.join.assert_called_once_with(
+        timeout=client._polling_join_timeout_seconds
+    )
     assert client._bot is None
     assert client._polling_thread is None
+
+
+def test_telegram_stop_keeps_polling_owner_when_thread_misses_deadline():
+    """polling 超过关闭预算时必须返回未收敛并保留原 owner。"""
+    client = Telegram.__new__(Telegram)
+    bot = Mock()
+    polling_thread = Mock()
+    polling_thread.is_alive.return_value = True
+    client._bot = bot
+    client._polling_thread = polling_thread
+    client._polling_join_timeout_seconds = 0.01
+    client._typing_tasks = {}
+    client._typing_stop_flags = {}
+    client._typing_lock = threading.RLock()
+    client._typing_lifecycle_lock = threading.RLock()
+    client._typing_accepting = True
+
+    assert client.stop() is False
+
+    polling_thread.join.assert_called_once_with(timeout=0.01)
+    assert client._bot is bot
+    assert client._polling_thread is polling_thread
+
+
+def test_telegram_module_reports_nonconverging_instance_after_stopping_peers():
+    """单实例未收敛时模块必须继续停止其余实例并返回 False。"""
+    module = TelegramModule()
+    blocked_client = Mock()
+    blocked_client.stop.return_value = False
+    healthy_client = Mock()
+    healthy_client.stop.return_value = True
+    module._instances = {
+        "blocked": blocked_client,
+        "healthy": healthy_client,
+    }
+
+    assert module.stop() is False
+    blocked_client.stop.assert_called_once_with()
+    healthy_client.stop.assert_called_once_with()

tests/test_module_manager_capability_adapter.py

@@ -16,7 +16,7 @@ import pytest
 from app.db.oper.systemconfig import SystemConfigOper
 from app.foundation.singleton import Singleton
 from app.runtime.capabilities.errors import CapabilityRuntimeClosedError
-from app.runtime.capabilities.model import SelectorSchema
+from app.runtime.capabilities.model import CapabilityLifecycleState, SelectorSchema
 from app.runtime.capabilities.registry import CapabilityRegistry
 from app.runtime.events import Event, EventHandlerBinding, eventmanager
 from app.runtime.extensions import module_manager as module_manager_extension
@@ -477,6 +477,23 @@ def test_shutdown_is_irreversible(module_manager_harness) -> None:
     assert type(running).instances == [running]
 
 
+def test_shutdown_reports_unreleased_module_owner(module_manager_harness) -> None:
+    """Host Module 返回 False 时 Runtime 必须保留 owner 并向组合根报告。"""
+    manager = module_manager_harness.manager
+    _enable_sample(module_manager_harness.config_values)
+    manager.load_modules()
+    running = manager.get_running_module("SampleModule")
+    running.stop = Mock(side_effect=[False, None])
+
+    assert manager.shutdown() is False
+    failed = manager._runtime.snapshot("SampleModule")
+    assert failed.lifecycle is CapabilityLifecycleState.FAILED
+    assert failed.visible is False
+
+    assert manager.shutdown() is True
+    assert running.stop.call_count == 2
+
+
 def test_all_real_host_modules_zero_arg_construct_without_starting_resources(
     tmp_path: Path,
 ) -> None:

tests/test_telegram_typing_lifecycle.py

@@ -62,6 +62,7 @@ def _telegram_client(bot=None) -> Telegram:
     """构造不连接外部服务且持有独立运行状态的 Telegram client。"""
     telegram = Telegram.__new__(Telegram)
     telegram._bot = bot or _FakeTelegramBot()
+    telegram._polling_thread = None
     telegram._telegram_token = "token"
     telegram._telegram_chat_id = "default-chat"
     telegram._user_chat_mapping = {}