From 009631b8eea8512f2398260ee4caaa1dcc468c6b Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 24 Aug 2026 13:48:11 +0800 Subject: [PATCH] refactor: unify capability shutdown convergence --- app/agent/runtime_loader.py | 13 ++--- app/modules/telegram/module.py | 10 +++- app/modules/telegram/telegram.py | 28 +++++++--- app/runtime/capabilities/runtime.py | 53 +++++++++++++----- app/runtime/extensions/host_module_adapter.py | 5 +- app/runtime/extensions/module_manager.py | 12 ++-- app/runtime/managed_resources.py | 12 ++-- app/startup/initializers/managed_resources.py | 8 +-- app/startup/initializers/modules.py | 36 +++++++++--- .../backend-architecture-next-stage.md | 19 ++++++- docs/rules/05-architecture.md | 5 ++ tests/test_agent_runtime_loader.py | 26 +++++++++ tests/test_capability_runtime.py | 23 ++++++++ tests/test_lifecycle_shutdown.py | 26 ++++++++- tests/test_managed_resources.py | 56 ++++++++++++++++++- tests/test_module_lifecycle.py | 48 +++++++++++++++- .../test_module_manager_capability_adapter.py | 19 ++++++- tests/test_telegram_typing_lifecycle.py | 1 + 18 files changed, 330 insertions(+), 70 deletions(-) diff --git a/app/agent/runtime_loader.py b/app/agent/runtime_loader.py index ec762efe2..4e0d66864 100644 --- a/app/agent/runtime_loader.py +++ b/app/agent/runtime_loader.py @@ -20,10 +20,7 @@ from app.agent.capabilities.adapter import ( build_agent_capability_registry, should_run_agent_service, ) -from app.runtime.capabilities.model import ( - CapabilityLifecycleState, - CapabilityMaterializationState, -) +from app.runtime.capabilities.model import CapabilityMaterializationState from app.runtime.capabilities.runtime import CapabilityRuntime @@ -146,10 +143,8 @@ async def close_materialized_terminal_sessions() -> None: async def begin_agent_shutdown() -> bool: - """不可逆关闭首用闸门，并返回 Agent service 是否真实收敛。""" + """不可逆关闭首用闸门，并返回全部 Agent 能力是否真实收敛。""" runtime = _ensure_runtime() - await runtime.shutdown_async(reason="application_shutdown") - return ( - runtime.snapshot(AGENT_SERVICE_CAPABILITY_ID).lifecycle - is CapabilityLifecycleState.STOPPED + return await runtime.shutdown_async( + reason="application_shutdown", ) diff --git a/app/modules/telegram/module.py b/app/modules/telegram/module.py index 48328875a..971266ee2 100644 --- a/app/modules/telegram/module.py +++ b/app/modules/telegram/module.py @@ -72,13 +72,17 @@ class TelegramModule(_MessageChannelModuleBase[Telegram]): """ return 0 - def stop(self) -> None: - """停止模块""" + def stop(self) -> bool: + """停止全部 Telegram 实例，并返回资源是否全部收敛。""" + converged = True for client in self.get_instances().values(): try: - client.stop() + if client.stop() is False: + converged = False except Exception as err: logger.error(f"停止Telegram模块实例失败：{err}") + converged = False + return converged def init_setting(self) -> Tuple[str, Union[str, bool]]: """ diff --git a/app/modules/telegram/telegram.py b/app/modules/telegram/telegram.py index a5bcc7e98..504e44d03 100644 --- a/app/modules/telegram/telegram.py +++ b/app/modules/telegram/telegram.py @@ -84,6 +84,7 @@ class Telegram: _typing_command_max_duration_seconds = 30 _typing_callback_max_duration_seconds = 60 _typing_join_timeout_seconds = 1 + _polling_join_timeout_seconds = 10 def __init__( self, @@ -1742,21 +1743,32 @@ class Telegram: # 清理菜单命令 self._bot.delete_my_commands() - def stop(self) -> None: + def stop(self) -> bool: """ - 停止Telegram消息接收服务 + 停止 Telegram 消息接收服务，并返回 polling/typing owner 是否收敛。 """ + converged = True with self._typing_lifecycle_lock: self._typing_accepting = False # 封口与 owner 快照处于同一临界区，停止后不会漏掉并发新增任务。 for chat_id in list(self._typing_tasks.keys()): - self._stop_typing_task(chat_id) - if not self._bot: - return + if not self._stop_typing_task(chat_id): + converged = False - self._bot.stop_bot() - if self._polling_thread: - self._polling_thread.join() + bot = self._bot + polling_thread = self._polling_thread + if bot: + bot.stop_bot() + if ( + polling_thread + and polling_thread.is_alive() + and polling_thread is not threading.current_thread() + ): + polling_thread.join(timeout=self._polling_join_timeout_seconds) + if polling_thread and polling_thread.is_alive(): + logger.error("Telegram polling 线程未在关闭预算内退出") + return False self._polling_thread = None self._bot = None logger.info("Telegram消息接收服务已停止") + return converged diff --git a/app/runtime/capabilities/runtime.py b/app/runtime/capabilities/runtime.py index cec42e120..43d8402ec 100644 --- a/app/runtime/capabilities/runtime.py +++ b/app/runtime/capabilities/runtime.py @@ -1126,7 +1126,8 @@ class CapabilityRuntime: """同步撤销并停止能力资源；物化实现保留供后续显式重启。""" self._stop_sync(capability_id, reason=reason, shutdown=False) - def _stop_sync(self, capability_id: str, *, reason: str, shutdown: bool) -> None: + def _stop_sync(self, capability_id: str, *, reason: str, shutdown: bool) -> bool: + """停止同步能力并返回资源 owner 是否已经真实收敛。""" state = self._state(capability_id) adapter = self._adapter(state, AdapterExecutionMode.SYNC) while True: @@ -1146,7 +1147,7 @@ class CapabilityRuntime: if stop_owner is None and pending is None: if state.lifecycle is not CapabilityLifecycleState.FAILED: state.lifecycle = CapabilityLifecycleState.STOPPED - return + return True state.generation += 1 generation = state.generation future: Future[Any] = Future() @@ -1166,8 +1167,10 @@ class CapabilityRuntime: except BaseException: if not shutdown: raise + if waiter_operation == "stop": + return False if waiter_operation == "stop": - return + return True continue break @@ -1207,6 +1210,7 @@ class CapabilityRuntime: reason=reason, started_at=started_at, ) + return True except BaseException as error: with state.lock: state.lifecycle = CapabilityLifecycleState.FAILED @@ -1229,12 +1233,20 @@ class CapabilityRuntime: ) if not shutdown: raise operation_error from error + return False async def stop_async(self, capability_id: str, *, reason: str) -> None: """异步撤销并停止能力资源。""" await self._stop_async(capability_id, reason=reason, shutdown=False) - async def _stop_async(self, capability_id: str, *, reason: str, shutdown: bool) -> None: + async def _stop_async( + self, + capability_id: str, + *, + reason: str, + shutdown: bool, + ) -> bool: + """停止异步能力并返回资源 owner 是否已经真实收敛。""" state = self._state(capability_id) adapter = self._adapter(state, AdapterExecutionMode.ASYNC) while True: @@ -1254,7 +1266,7 @@ class CapabilityRuntime: if stop_owner is None and pending is None: if state.lifecycle is not CapabilityLifecycleState.FAILED: state.lifecycle = CapabilityLifecycleState.STOPPED - return + return True state.generation += 1 generation = state.generation future: Future[Any] = Future() @@ -1274,8 +1286,10 @@ class CapabilityRuntime: except BaseException: if not shutdown: raise + if waiter_operation == "stop": + return False if waiter_operation == "stop": - return + return True continue break @@ -1321,6 +1335,7 @@ class CapabilityRuntime: reason=reason, started_at=started_at, ) + return True except BaseException as error: with state.lock: state.lifecycle = CapabilityLifecycleState.FAILED @@ -1343,9 +1358,10 @@ class CapabilityRuntime: ) if not shutdown: raise operation_error from error + return False - def shutdown(self, *, reason: str) -> None: - """不可逆关闭仅含同步 adapter 的 Runtime，并阻止并发首启重新发布。""" + def shutdown(self, *, reason: str) -> bool: + """不可逆关闭同步 Runtime，并返回全部 owner 是否收敛。""" async_kinds = { kind for kind, adapter in self._adapters.items() @@ -1357,21 +1373,32 @@ class CapabilityRuntime: ) with self._shutdown_lock: self._shutdown = True + converged = True for spec in self._registry.list_specs(): - self._stop_sync(spec.id, reason=reason, shutdown=True) + if not self._stop_sync(spec.id, reason=reason, shutdown=True): + converged = False + return converged - async def shutdown_async(self, *, reason: str) -> None: - """不可逆关闭混合同步/异步 adapter 的 Runtime。""" + async def shutdown_async(self, *, reason: str) -> bool: + """不可逆关闭混合 Runtime，并返回全部 owner 是否收敛。""" with self._shutdown_lock: self._shutdown = True + converged = True for spec in self._registry.list_specs(): adapter = self._adapters[spec.kind] if getattr(adapter, "execution_mode", None) is AdapterExecutionMode.ASYNC: - await self._stop_async(spec.id, reason=reason, shutdown=True) + stopped = await self._stop_async( + spec.id, + reason=reason, + shutdown=True, + ) else: - await asyncio.to_thread( + stopped = await asyncio.to_thread( self._stop_sync, spec.id, reason=reason, shutdown=True, ) + if not stopped: + converged = False + return converged diff --git a/app/runtime/extensions/host_module_adapter.py b/app/runtime/extensions/host_module_adapter.py index 483215329..feeb694bf 100644 --- a/app/runtime/extensions/host_module_adapter.py +++ b/app/runtime/extensions/host_module_adapter.py @@ -247,8 +247,9 @@ class HostModuleAdapter: @staticmethod def stop(spec: CapabilitySpec, instance: Any, generation: int) -> None: """停止实例拥有的资源；Runtime 会先撤销其运行态可见性。""" - del spec, generation - instance.stop() + del generation + if instance.stop() is False: + raise RuntimeError(f"Host Module {spec.id} 资源未完成收口") @staticmethod def cleanup( diff --git a/app/runtime/extensions/module_manager.py b/app/runtime/extensions/module_manager.py index 94308cb39..e4ca5da32 100644 --- a/app/runtime/extensions/module_manager.py +++ b/app/runtime/extensions/module_manager.py @@ -224,13 +224,17 @@ class ModuleManager(metaclass=Singleton): self._refresh_running_projection() logger.info("所有模块停止完成") - def shutdown(self) -> None: - """进程关闭时不可逆停止 Runtime，阻止并发能力重新发布。""" + def shutdown(self) -> bool: + """不可逆停止 Runtime，并返回所有模块 owner 是否收敛。""" logger.info("正在关闭模块运行时...") with self._lifecycle_lock: - self._runtime.shutdown(reason="application_shutdown") + converged = self._runtime.shutdown(reason="application_shutdown") self._refresh_running_projection() - logger.info("模块运行时关闭完成") + if converged: + logger.info("模块运行时关闭完成") + else: + logger.error("模块运行时关闭后仍有资源 owner 未收敛") + return converged def reload(self) -> None: """保留旧插件可观察的 stop、load、ModuleReload 同步顺序。""" diff --git a/app/runtime/managed_resources.py b/app/runtime/managed_resources.py index d6a21122b..2f3dcab93 100644 --- a/app/runtime/managed_resources.py +++ b/app/runtime/managed_resources.py @@ -48,8 +48,8 @@ class ManagedResourceRuntime(Protocol): async def stop_async(self, capability_id: str, *, reason: str) -> None: """通过异步 adapter 停止资源。""" - async def shutdown_async(self, *, reason: str) -> None: - """关闭混合同步和异步 adapter 的 Runtime。""" + async def shutdown_async(self, *, reason: str) -> bool: + """关闭混合 Runtime，并返回全部资源是否收敛。""" _runtime_lock = threading.RLock() @@ -174,9 +174,9 @@ async def stop_managed_resource_async(capability_id: str, *, reason: str) -> Non raise RuntimeError(f"未知 Managed Resource kind：{kind}") -async def shutdown_managed_resource_runtime(*, reason: str) -> None: - """关闭已配置 Runtime；未配置时直接返回，绝不因关闭而创建资源。""" +async def shutdown_managed_resource_runtime(*, reason: str) -> bool: + """关闭已配置 Runtime，未配置时按已收敛处理。""" runtime = _runtime(required=False) if runtime is None: - return - await runtime.shutdown_async(reason=reason) + return True + return await runtime.shutdown_async(reason=reason) diff --git a/app/startup/initializers/managed_resources.py b/app/startup/initializers/managed_resources.py index 4a394768a..e9ae7e09f 100644 --- a/app/startup/initializers/managed_resources.py +++ b/app/startup/initializers/managed_resources.py @@ -38,10 +38,10 @@ def init_managed_resources() -> CapabilityRuntime: return _managed_resource_runtime -async def stop_managed_resources() -> None: - """关闭已经初始化的资源 Runtime；未初始化时不执行发现或激活。""" +async def stop_managed_resources() -> bool: + """关闭已初始化的资源 Runtime，并返回 owner 是否收敛。""" with _runtime_lock: runtime = _managed_resource_runtime if runtime is None: - return - await runtime.shutdown_async(reason="application_shutdown") + return True + return await runtime.shutdown_async(reason="application_shutdown") diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py index 2977a37fe..615369894 100644 --- a/app/startup/initializers/modules.py +++ b/app/startup/initializers/modules.py @@ -601,23 +601,36 @@ async def settle_events() -> bool: return await event_manager.drain_async(seal=False) -async def stop_modules(): +async def stop_modules() -> bool: """ - 服务关闭 + 关闭模块服务，并返回全部资源 owner 是否收敛。 """ - async def run_step(name: str, callback: Callable[[], object]) -> bool: - """单个模块资源关闭失败时继续执行后续阶段""" + all_converged = True + + async def run_step( + name: str, + callback: Callable[[], object], + *, + record_failure: bool = True, + ) -> bool: + """执行单个关闭步骤，失败时继续收口并保留诚实结果。""" + nonlocal all_converged try: result = callback() if inspect.isawaitable(result): - await result - return True + result = await result + converged = result is not False + if not converged: + logger.error("关闭%s未收敛，继续执行后续资源收口", name) except asyncio.CancelledError: logger.warning("关闭%s时收到取消请求，继续执行资源收口", name) - return False + converged = False except Exception as err: logger.error(f"关闭{name}失败：{err}") - return True + converged = False + if not converged and record_failure: + all_converged = False + return converged await run_step("图片代理安全日志合并器", close_image_proxy_block_log_coalescer) await run_step("模块", lambda: ModuleManager().shutdown()) @@ -631,7 +644,9 @@ async def stop_modules(): await run_step("异步Redis缓存连接", lambda: AsyncRedisHelper().close()) # Web Agent 的取消 finally 可能还要写入最终展示快照，必须先完成任务收尾，再关闭写入准入。 web_agent_drained = await run_step( - "Web Agent后台任务", shutdown_web_agent_background_tasks + "Web Agent后台任务", + shutdown_web_agent_background_tasks, + record_failure=False, ) if not web_agent_drained: web_agent_drained = await run_step( @@ -648,15 +663,18 @@ async def stop_modules(): ) else: persistence_drained = False + all_converged = False logger.error("Web Agent任务未完成收尾，跳过持久化和数据库关闭以保护活动事务") if persistence_drained: await run_step("数据库任务", stop_database_worker) if _database_worker is None: await run_step("数据库连接", close_database) else: + all_converged = False logger.error("数据库任务未收敛，跳过数据库连接关闭以避免运行中事务使用已释放连接") await run_step("前端服务", stop_frontend) await run_step("临时文件", clear_temp) + return all_converged async def init_modules() -> HostRuntime: diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 4b46f3296..b1144d193 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -2,13 +2,13 @@ > 文档性质：当前架构复核、优秀 Python 后端实践对标、AI 可执行任务手册 > 适用仓库：`MoviePilot`，分支 `v3` -> 审计基线：`af9a141f`（2026-08-24） +> 审计基线：`2aa41ea8`（2026-08-24） > 审计范围：宿主后端；排除 `app/plugins/**` 运行时插件副本 > 规范优先级：`AGENTS.md` 与 `docs/rules/` 高于本文 > 相关文档：`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md` -> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径；阶段 10 已统一模块、插件与调度运行时的显式 getter 调用；阶段 11 已清除系统配置 getter 的 Oper 形别名；阶段 12 已完成工作流域的显式 Chain 数据端口迁移；阶段 13 已收口用户、交互与消息链的数据端口；阶段 14 已收口音乐订阅数据端口；阶段 15 已收口站点数据端口；阶段 16 已收口媒体服务器数据端口；阶段 17 已收口下载数据端口；阶段 18 已收口主订阅数据端口；阶段 19 已收口整理数据端口；阶段 20 已收口 Agent 数据端口；阶段 21 已收口监控历史端口；阶段 22 已统一服务配置应用边界；阶段 23 已补齐媒体服务器 API 遗留的类形配置读取路径；阶段 24 已清除 Scheduler 内部无 owner 的协程提交双轨；阶段 25 已补齐 TaskRegistry 跨线程 owner 并迁移整理 AI 接管；阶段 26 已统一 Agent 会话清理提交；阶段 27 已统一历史 AI 进度 owner；阶段 28 已托管旧插件订阅统计线程；阶段 29 已统一 Emby 系条目转换并清零重复代码白名单；阶段 30 已收口插件市场请求级子任务；阶段 31 已托管搜索 AI 推荐任务；阶段 32 已清除事件调度器绕过生命周期 owner 的投递回退；阶段 33 已统一宿主 Agent 运行时的获取路径；阶段 34 已统一 durable-required 事件与 Outbox topic 事实源；阶段 35 已统一 LLM provider 管理 API 的运行时解析路径；阶段 36 已统一 WebAgent 音频能力访问边界；阶段 37 已统一插件输入事件发布路径；阶段 38 已统一 WebAgent 通知事件监听与队列边界；阶段 39 已补齐搜索 SSE 断线时的上游任务清理；阶段 40 已补齐异步防抖取消的终态所有权；阶段 41 已统一优雅重启兜底线程的唯一所有权；阶段 42 已补齐 Telegram typing 的多实例隔离和终态 owner；阶段 43 已统一 Discord typing 的异步 owner 和 shutdown 收尾；阶段 44 已清除 WebAgent 测试临时事件循环提前关闭产生的 CI 红注解；阶段 45 已统一影视与字幕搜索的请求级逐页任务编排；阶段 46 已收口启动性能门禁的托管 runner 假失败与诊断输出；阶段 47 已补齐 Agent 渠道流式刷新任务的重入 owner；阶段 48 已统一工件上传 action 的 Node 24 主版本；阶段 49 已统一插件安装的同步/异步代际解析事实源；阶段 50 已统一插件市场 GitHub 请求降级策略；阶段 51 已统一插件索引请求与响应三态策略；阶段 52 已统一插件 Release 分页策略；阶段 53 已统一远端插件安装模式决策；阶段 54 已补齐同步安装成功后的临时回滚备份清理；阶段 55～56 已收口官方插件观察基线与报告保留策略；阶段 57 已统一进程级运行时 Facade 门禁并补齐 ModuleManager 边界；阶段 58 已消除 AgentTask 关闭回归的跨线程零时长等待竞态；阶段 59 已统一 Feishu 多实例长连接的 SDK 循环路由；阶段 60 已清除命令服务虚假的关停 owner 声明。 +> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径；阶段 10 已统一模块、插件与调度运行时的显式 getter 调用；阶段 11 已清除系统配置 getter 的 Oper 形别名；阶段 12 已完成工作流域的显式 Chain 数据端口迁移；阶段 13 已收口用户、交互与消息链的数据端口；阶段 14 已收口音乐订阅数据端口；阶段 15 已收口站点数据端口；阶段 16 已收口媒体服务器数据端口；阶段 17 已收口下载数据端口；阶段 18 已收口主订阅数据端口；阶段 19 已收口整理数据端口；阶段 20 已收口 Agent 数据端口；阶段 21 已收口监控历史端口；阶段 22 已统一服务配置应用边界；阶段 23 已补齐媒体服务器 API 遗留的类形配置读取路径；阶段 24 已清除 Scheduler 内部无 owner 的协程提交双轨；阶段 25 已补齐 TaskRegistry 跨线程 owner 并迁移整理 AI 接管；阶段 26 已统一 Agent 会话清理提交；阶段 27 已统一历史 AI 进度 owner；阶段 28 已托管旧插件订阅统计线程；阶段 29 已统一 Emby 系条目转换并清零重复代码白名单；阶段 30 已收口插件市场请求级子任务；阶段 31 已托管搜索 AI 推荐任务；阶段 32 已清除事件调度器绕过生命周期 owner 的投递回退；阶段 33 已统一宿主 Agent 运行时的获取路径；阶段 34 已统一 durable-required 事件与 Outbox topic 事实源；阶段 35 已统一 LLM provider 管理 API 的运行时解析路径；阶段 36 已统一 WebAgent 音频能力访问边界；阶段 37 已统一插件输入事件发布路径；阶段 38 已统一 WebAgent 通知事件监听与队列边界；阶段 39 已补齐搜索 SSE 断线时的上游任务清理；阶段 40 已补齐异步防抖取消的终态所有权；阶段 41 已统一优雅重启兜底线程的唯一所有权；阶段 42 已补齐 Telegram typing 的多实例隔离和终态 owner；阶段 43 已统一 Discord typing 的异步 owner 和 shutdown 收尾；阶段 44 已清除 WebAgent 测试临时事件循环提前关闭产生的 CI 红注解；阶段 45 已统一影视与字幕搜索的请求级逐页任务编排；阶段 46 已收口启动性能门禁的托管 runner 假失败与诊断输出；阶段 47 已补齐 Agent 渠道流式刷新任务的重入 owner；阶段 48 已统一工件上传 action 的 Node 24 主版本；阶段 49 已统一插件安装的同步/异步代际解析事实源；阶段 50 已统一插件市场 GitHub 请求降级策略；阶段 51 已统一插件索引请求与响应三态策略；阶段 52 已统一插件 Release 分页策略；阶段 53 已统一远端插件安装模式决策；阶段 54 已补齐同步安装成功后的临时回滚备份清理；阶段 55～56 已收口官方插件观察基线与报告保留策略；阶段 57 已统一进程级运行时 Facade 门禁并补齐 ModuleManager 边界；阶段 58 已消除 AgentTask 关闭回归的跨线程零时长等待竞态；阶段 59 已统一 Feishu 多实例长连接的 SDK 循环路由；阶段 60 已清除命令服务虚假的关停 owner 声明；阶段 61 已统一 Capability Runtime 同步/异步关闭的诚实收敛结果。 > 当前 canonical 状态：API/Application 公共复杂度基线已清零，组合根外 `SystemConfigOper()` 构造和 Model/Oper 隐式事务均为 0；命名 Chain/Agent 数据端口、TaskRegistry owner、Module Contract V2、typed Event、Outbox durable intent、请求关联和插件运行时 getter 已形成当前路径。插件仓适配、未知第三方 fallback 和其它 E1/E3 副作用仍按风险持续治理。 -> 最新阶段：阶段 60 已统一命令重建任务的唯一关停 owner。 +> 最新阶段：阶段 61 已统一 Capability Runtime 关闭收敛事实源。 ## 当前复核结论（2026-08-24） @@ -631,6 +631,19 @@ - `Command` 类身份、Application 命令门面、插件命令 Hook、热更新时序和 SDK/Compat 均未修改； V1/V2/V3 插件仍通过原有注册链路生效，且未修改插件仓。 +### 长期整改阶段 61：Capability Runtime 关闭收敛事实源统一（2026-08-24） + +- Capability Runtime 原本会在 stop 异常时保留 `pending_stop`，但同步 `shutdown()` 把结果丢弃； + `HostModuleAdapter` 同时忽略模块显式返回的 `False`，startup 内部 `run_step()` 又把异常视为成功。 + 这三层会让未释放资源仅留下日志，对外却报告整体关闭完成。 +- 当前以 Capability Runtime 的同步/异步 stop 作为单一事实源：Host Module `stop() is False` 会进入失败 + 路径并保留原 owner，Runtime/ModuleManager/startup 逐层返回未收敛；Agent 与 Managed Resource 关闭入口 + 也直接传播 Runtime 的整体布尔结果，不再用单个 service 快照或无返回包装器形成并行判断。其余模块、 + 数据库和临时资源仍继续尽力关闭，但失败不再被伪装成成功；后续重试成功后同一 owner 才会进入终态。 +- Telegram polling 是首个接入该合同的长连接：无界 `join()` 改为 10 秒预算，超时时保留 SDK 与线程句柄并 + 返回 `False`，不会因清空句柄而丢失重试能力。Telegram 配置、菜单、消息、typing 语义与类 identity 未变。 +- 本阶段收口宿主 Capability Runtime 生命周期结果；未修改插件仓、SDK/Compat 映射或 V1/V2/V3 插件 Hook。 + ### 总体判断 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**： diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 4694857e2..f1025f41e 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -190,6 +190,11 @@ mechanism remains in `app/adapters/system/resource.py`. 应用级启动顺序使用 `app/startup/lifecycle/components.py` 的组件描述声明依赖、 normal/safe-mode 范围、start/stop 顺序、超时预算和失败策略。新增进程级资源不得只在 `lifespan()` 中追加过程代码，必须先进入可导出的生命周期清单并补顺序快照测试。 +Host Module 的 `stop()` 可以显式返回 `False` 表示资源 owner 尚未收敛； +`HostModuleAdapter` 必须将它视为 stop 失败，Capability Runtime 保留原 owner 供后续重试， +ModuleManager 与 startup 组合根继续关闭其余资源但必须向上返回未收敛，不得把记录日志等同于成功。 +同步和异步 Capability Runtime 的 `shutdown` 必须使用同一布尔收敛合同；Agent、Managed Resource +等领域关闭入口必须直接传播 Runtime 的整体结果，不得以单个能力快照或无返回包装器覆盖失败。 API 中允许丢失或可重建的进程内任务必须登记到 `app/runtime/tasks.py`；登记器先于其他 运行资源启动，并在资源释放前停止接收、取消和有限等待。需要崩溃恢复的 E2/E3 副作用仍应 进入 Outbox 或持久任务表，不能把 TaskRegistry 当成 durable queue。 diff --git a/tests/test_agent_runtime_loader.py b/tests/test_agent_runtime_loader.py index cee62eead..c2bc6ed35 100644 --- a/tests/test_agent_runtime_loader.py +++ b/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() diff --git a/tests/test_capability_runtime.py b/tests/test_capability_runtime.py index 5958f15b7..3c0f3601e 100644 --- a/tests/test_capability_runtime.py +++ b/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 与同步入口遵守相同状态和发布边界。""" diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index ad34e79a3..007b1d9e9 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/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"] diff --git a/tests/test_managed_resources.py b/tests/test_managed_resources.py index 72bb9b547..80694fab8 100644 --- a/tests/test_managed_resources.py +++ b/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() diff --git a/tests/test_module_lifecycle.py b/tests/test_module_lifecycle.py index 7c78658c0..72610b30b 100644 --- a/tests/test_module_lifecycle.py +++ b/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() diff --git a/tests/test_module_manager_capability_adapter.py b/tests/test_module_manager_capability_adapter.py index e2c1eb679..367a939f1 100644 --- a/tests/test_module_manager_capability_adapter.py +++ b/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: diff --git a/tests/test_telegram_typing_lifecycle.py b/tests/test_telegram_typing_lifecycle.py index 07e718eb4..079cc35c4 100644 --- a/tests/test_telegram_typing_lifecycle.py +++ b/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 = {}