From 415335b21502076bcfe4c98be63a67953d35feac Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 24 Aug 2026 14:08:06 +0800 Subject: [PATCH] refactor: unify message channel shutdown convergence --- app/modules/__init__.py | 4 +- app/modules/_base/notification.py | 15 ++ app/modules/discord/__init__.py | 10 +- app/modules/discord/discord.py | 18 ++- app/modules/feishu/__init__.py | 10 +- app/modules/feishu/feishu.py | 21 ++- app/modules/qqbot/module.py | 10 +- app/modules/qqbot/qqbot.py | 25 +++- app/modules/slack/__init__.py | 10 +- app/modules/slack/slack.py | 5 +- app/modules/telegram/module.py | 10 +- app/modules/wechat/__init__.py | 11 +- app/modules/wechat/wechatbot.py | 30 +++- app/modules/wechatclawbot/__init__.py | 10 +- app/modules/wechatclawbot/wechatclawbot.py | 18 ++- .../backend-architecture-next-stage.md | 19 ++- docs/rules/05-architecture.md | 2 + tests/test_module_lifecycle.py | 137 +++++++++++++++++- 18 files changed, 276 insertions(+), 89 deletions(-) diff --git a/app/modules/__init__.py b/app/modules/__init__.py index 3cfb9f9aa..513ef2657 100644 --- a/app/modules/__init__.py +++ b/app/modules/__init__.py @@ -93,10 +93,10 @@ class _ModuleBase(ConfigReloadMixin, metaclass=ABCMeta): pass @abstractmethod - def stop(self) -> None: + def stop(self) -> Optional[bool]: """ 如果关闭时模块有服务需要停止，需要实现此方法 - :return: None，该方法可被多个模块同时处理 + :return: False 表示资源尚未收敛；None/True 表示本次停止完成 """ pass diff --git a/app/modules/_base/notification.py b/app/modules/_base/notification.py index 4f5025f47..491000f9c 100644 --- a/app/modules/_base/notification.py +++ b/app/modules/_base/notification.py @@ -73,6 +73,21 @@ class _MessageChannelModuleBase(_ModuleBase, _MessageBase[TService]): """ return bool(client.get_state()), "" + def _stop_service_instances(self) -> bool: + """停止全部渠道实例，并聚合客户端返回的资源收敛结果。""" + converged = True + for client in self.get_instances().values(): + stop = getattr(client, "stop", None) + if not callable(stop): + continue + try: + if stop() is False: + converged = False + except Exception as err: + logger.error("停止%s模块实例失败：%s", self.get_name(), err) + converged = False + return converged + def register_commands(self, commands: Dict[str, dict]) -> None: """ 注册命令，实现这个函数接收系统可用的命令菜单 diff --git a/app/modules/discord/__init__.py b/app/modules/discord/__init__.py index 9eba96a59..3c1f5576e 100644 --- a/app/modules/discord/__init__.py +++ b/app/modules/discord/__init__.py @@ -96,13 +96,9 @@ class DiscordModule(_MessageChannelModuleBase[Discord]): """ return 4 - def stop(self) -> None: - """停止模块""" - for client in self.get_instances().values(): - try: - client.stop() - except Exception as err: - logger.error(f"停止Discord模块实例失败：{err}") + def stop(self) -> bool: + """停止全部 Discord 实例，并返回资源是否全部收敛。""" + return self._stop_service_instances() def init_setting(self) -> Tuple[str, Union[str, bool]]: pass diff --git a/app/modules/discord/discord.py b/app/modules/discord/discord.py index 8a69aad93..47f4190c2 100644 --- a/app/modules/discord/discord.py +++ b/app/modules/discord/discord.py @@ -39,6 +39,8 @@ class Discord: """ _MAX_SLASH_COMMANDS = 100 + _client_close_timeout_seconds = 10 + _thread_join_timeout_seconds = 5 def __init__( self, @@ -247,10 +249,10 @@ class Discord: self._thread = threading.Thread(target=runner, daemon=True) self._thread.start() - def stop(self): - """停止 Discord 客户端，并在关闭事件循环前收口 typing owner。""" + def stop(self) -> bool: + """停止 Discord 客户端，并返回事件循环线程是否已经终止。""" if not self._client or not self._loop or not self._thread: - return + return True self._stop_requested.set() loop = self._loop thread = self._thread @@ -264,19 +266,23 @@ class Discord: try: asyncio.run_coroutine_threadsafe( self._client.close(), loop - ).result(timeout=10) + ).result(timeout=self._client_close_timeout_seconds) except Exception as err: logger.error(f"关闭 Discord Bot 失败：{err}") self._ready_event.clear() - thread.join(timeout=5) + if thread is not threading.current_thread(): + thread.join(timeout=self._thread_join_timeout_seconds) if thread.is_alive(): try: loop.call_soon_threadsafe(loop.stop) except Exception as err: logger.error(f"停止 Discord 事件循环失败：{err}") - thread.join(timeout=5) + if thread is not threading.current_thread(): + thread.join(timeout=self._thread_join_timeout_seconds) if thread.is_alive(): logger.error("Discord Bot 线程未在超时内停止") + return False + return True def get_state(self) -> bool: return self._ready_event.is_set() and self._client is not None diff --git a/app/modules/feishu/__init__.py b/app/modules/feishu/__init__.py index 3a391feb8..95cdaa038 100644 --- a/app/modules/feishu/__init__.py +++ b/app/modules/feishu/__init__.py @@ -48,13 +48,9 @@ class FeishuModule(_MessageChannelModuleBase[Feishu]): """ return False - def stop(self) -> None: - """停止模块""" - for client in self.get_instances().values(): - try: - client.stop() - except Exception as err: - logger.error(f"停止飞书模块实例失败：{err}") + def stop(self) -> bool: + """停止全部飞书实例，并返回资源是否全部收敛。""" + return self._stop_service_instances() def init_setting(self) -> Tuple[str, Union[str, bool]]: """通知模块通过系统通知配置控制实例化，这里不额外设置环境开关。""" diff --git a/app/modules/feishu/feishu.py b/app/modules/feishu/feishu.py index 563263a87..4046f1b2b 100644 --- a/app/modules/feishu/feishu.py +++ b/app/modules/feishu/feishu.py @@ -129,6 +129,8 @@ class Feishu: """飞书通知客户端，负责长连接收消息与主动发送通知。""" PROCESSING_REACTION_EMOJI = "GLANCE" + _ws_shutdown_timeout_seconds = 5 + _ws_join_timeout_seconds = 5 STREAM_CARD_TITLE_ELEMENT_ID = "mp_stream_title" STREAM_CARD_BODY_ELEMENT_ID = "mp_stream_body" IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".ico", ".tiff", ".heic"} @@ -678,12 +680,13 @@ class Feishu: """返回飞书客户端是否已就绪。""" return self._ready.is_set() and self._api_client is not None - def stop(self) -> None: - """停止飞书客户端并结束长连接线程。""" + def stop(self) -> bool: + """停止飞书客户端，并返回长连接线程是否已经终止。""" self._stop_event.set() self._ready.clear() ws_client = self._ws_client ws_loop = self._ws_loop + ws_thread = self._ws_thread if ws_client: try: ws_client._auto_reconnect = False @@ -692,11 +695,19 @@ class Feishu: self._shutdown_ws_client(), ws_loop, ) - shutdown_future.result(timeout=5) + shutdown_future.result(timeout=self._ws_shutdown_timeout_seconds) except Exception as err: logger.debug(f"停止飞书客户端失败：{err}") - if self._ws_thread and self._ws_thread.is_alive(): - self._ws_thread.join(timeout=5) + if ( + ws_thread + and ws_thread.is_alive() + and ws_thread is not threading.current_thread() + ): + ws_thread.join(timeout=self._ws_join_timeout_seconds) + if ws_thread and ws_thread.is_alive(): + logger.error("飞书长连接线程未在关闭预算内退出") + return False + return True def parse_message(self, body: Any) -> Optional[IncomingMessage]: """解析飞书转发到消息入口的 JSON 报文。""" diff --git a/app/modules/qqbot/module.py b/app/modules/qqbot/module.py index 899e18c43..2a81fd2a7 100644 --- a/app/modules/qqbot/module.py +++ b/app/modules/qqbot/module.py @@ -90,13 +90,9 @@ class QQBotModule(_MessageChannelModuleBase[QQBot]): """ return False - def stop(self) -> None: - """停止模块""" - for client in self.get_instances().values(): - try: - client.stop() - except Exception as err: - logger.error(f"停止QQ Bot模块实例失败：{err}") + def stop(self) -> bool: + """停止全部 QQ Bot 实例，并返回资源是否全部收敛。""" + return self._stop_service_instances() def init_setting(self) -> Tuple[str, Union[str, bool]]: pass diff --git a/app/modules/qqbot/qqbot.py b/app/modules/qqbot/qqbot.py index 72cda24db..dc201ce14 100644 --- a/app/modules/qqbot/qqbot.py +++ b/app/modules/qqbot/qqbot.py @@ -38,6 +38,8 @@ _MAX_IMAGE_SIZE: Tuple[int, int] = (512, 512) class QQBot: """QQ Bot 通知客户端""" + _gateway_join_timeout_seconds = 20 + def __init__( self, QQ_APP_ID: Optional[str] = None, @@ -161,8 +163,8 @@ class QQBot: except Exception as e: logger.error(f"QQ Bot Gateway 启动失败: {e}") - def stop(self) -> None: - """停止 Gateway 连接""" + def stop(self) -> bool: + """停止 Gateway 连接，并返回后台线程是否已经终止。""" if self._gateway_stop is not None: self._gateway_stop.set() try: @@ -170,12 +172,19 @@ class QQBot: self._gateway_ws_holder[0].close() except Exception as e: logger.debug(f"QQ Bot Gateway WebSocket close: {e}") - if self._gateway_thread is not None and self._gateway_thread.is_alive(): - self._gateway_thread.join(timeout=20) - if self._gateway_thread.is_alive(): - logger.warning( - "QQ Bot Gateway 线程在 stop 后仍未退出，可能存在重复收消息，请重启进程" - ) + gateway_thread = self._gateway_thread + if ( + gateway_thread is not None + and gateway_thread.is_alive() + and gateway_thread is not threading.current_thread() + ): + gateway_thread.join(timeout=self._gateway_join_timeout_seconds) + if gateway_thread is not None and gateway_thread.is_alive(): + logger.warning( + "QQ Bot Gateway 线程在 stop 后仍未退出，可能存在重复收消息，请重启进程" + ) + return False + return True def get_state(self) -> bool: """获取就绪状态""" diff --git a/app/modules/slack/__init__.py b/app/modules/slack/__init__.py index a0ce224c7..b1f729216 100644 --- a/app/modules/slack/__init__.py +++ b/app/modules/slack/__init__.py @@ -77,13 +77,9 @@ class SlackModule(_MessageChannelModuleBase[Slack]): """ return 3 - def stop(self) -> None: - """停止模块""" - for client in self.get_instances().values(): - try: - client.stop() - except Exception as err: - logger.error(f"停止Slack模块实例失败：{err}") + def stop(self) -> bool: + """停止全部 Slack 实例，并返回资源是否全部收敛。""" + return self._stop_service_instances() def init_setting(self) -> Tuple[str, Union[str, bool]]: pass diff --git a/app/modules/slack/slack.py b/app/modules/slack/slack.py index a8e280ba8..15b4e1db1 100644 --- a/app/modules/slack/slack.py +++ b/app/modules/slack/slack.py @@ -117,13 +117,16 @@ class Slack: timeout=timeout, ) - def stop(self): + def stop(self) -> bool: + """关闭 Socket Mode 服务，并返回资源是否成功释放。""" if self._service: try: self._service.close() logger.info("Slack消息接收服务已停止") except Exception as err: logger.error("Slack消息接收服务停止失败: %s" % str(err)) + return False + return True def get_state(self) -> bool: """ diff --git a/app/modules/telegram/module.py b/app/modules/telegram/module.py index 971266ee2..1fd0efc1b 100644 --- a/app/modules/telegram/module.py +++ b/app/modules/telegram/module.py @@ -74,15 +74,7 @@ class TelegramModule(_MessageChannelModuleBase[Telegram]): def stop(self) -> bool: """停止全部 Telegram 实例，并返回资源是否全部收敛。""" - converged = True - for client in self.get_instances().values(): - try: - if client.stop() is False: - converged = False - except Exception as err: - logger.error(f"停止Telegram模块实例失败：{err}") - converged = False - return converged + return self._stop_service_instances() def init_setting(self) -> Tuple[str, Union[str, bool]]: """ diff --git a/app/modules/wechat/__init__.py b/app/modules/wechat/__init__.py index 3e9ee7bfe..6d873c357 100644 --- a/app/modules/wechat/__init__.py +++ b/app/modules/wechat/__init__.py @@ -73,14 +73,9 @@ class WechatModule(_MessageChannelModuleBase[WeChat]): """ return 1 - def stop(self) -> None: - """停止模块""" - for client in self.get_instances().values(): - try: - if hasattr(client, "stop"): - client.stop() - except Exception as err: - logger.error(f"停止微信模块实例失败：{err}") + def stop(self) -> bool: + """停止全部微信实例，并返回长连接资源是否全部收敛。""" + return self._stop_service_instances() @staticmethod def _is_bot_mode(config: dict) -> bool: diff --git a/app/modules/wechat/wechatbot.py b/app/modules/wechat/wechatbot.py index f7236af79..e749772a7 100644 --- a/app/modules/wechat/wechatbot.py +++ b/app/modules/wechat/wechatbot.py @@ -38,6 +38,8 @@ class WeChatBot: _default_ws_url = "wss://openws.work.weixin.qq.com" _heartbeat_interval = 30 _ack_timeout = 10 + _gateway_join_timeout_seconds = 5 + _heartbeat_join_timeout_seconds = 2 def __init__(self, WECHAT_BOT_ID: Optional[str] = None, @@ -126,7 +128,8 @@ class WeChatBot: self._heartbeat_thread.start() logger.info(f"企业微信智能机器人长连接已启动：{self._config_name}") - def stop(self) -> None: + def stop(self) -> bool: + """停止网关与心跳线程，并返回两个 owner 是否均已终止。""" self._stop_event.set() self._authenticated.clear() if self._ws_app: @@ -134,10 +137,27 @@ class WeChatBot: self._ws_app.close() except Exception as err: logger.debug(f"关闭企业微信智能机器人连接失败：{err}") - if self._ws_thread and self._ws_thread.is_alive(): - self._ws_thread.join(timeout=5) - if self._heartbeat_thread and self._heartbeat_thread.is_alive(): - self._heartbeat_thread.join(timeout=2) + ws_thread = self._ws_thread + heartbeat_thread = self._heartbeat_thread + if ( + ws_thread + and ws_thread.is_alive() + and ws_thread is not threading.current_thread() + ): + ws_thread.join(timeout=self._gateway_join_timeout_seconds) + if ( + heartbeat_thread + and heartbeat_thread.is_alive() + and heartbeat_thread is not threading.current_thread() + ): + heartbeat_thread.join(timeout=self._heartbeat_join_timeout_seconds) + converged = not any( + thread and thread.is_alive() + for thread in (ws_thread, heartbeat_thread) + ) + if not converged: + logger.error("企业微信智能机器人线程未在关闭预算内退出") + return converged def get_state(self) -> bool: return self._ready and self._authenticated.is_set() diff --git a/app/modules/wechatclawbot/__init__.py b/app/modules/wechatclawbot/__init__.py index d28475320..dbc6560a8 100644 --- a/app/modules/wechatclawbot/__init__.py +++ b/app/modules/wechatclawbot/__init__.py @@ -69,13 +69,9 @@ class WechatClawBotModule(_MessageChannelModuleBase[WechatClawBot]): """ return False - def stop(self) -> None: - """停止模块""" - for client in self.get_instances().values(): - try: - client.stop() - except Exception as err: - logger.error(f"停止微信 ClawBot 模块实例失败：{err}") + def stop(self) -> bool: + """停止全部微信 ClawBot 实例，并返回资源是否全部收敛。""" + return self._stop_service_instances() def _test_connection(self, client) -> Tuple[bool, str]: """微信 ClawBot 的连接探测返回 (状态, 信息)。""" diff --git a/app/modules/wechatclawbot/wechatclawbot.py b/app/modules/wechatclawbot/wechatclawbot.py index 20fb587b3..531d07149 100644 --- a/app/modules/wechatclawbot/wechatclawbot.py +++ b/app/modules/wechatclawbot/wechatclawbot.py @@ -1441,6 +1441,7 @@ class WechatClawBot: _default_base_url = "https://ilinkai.weixin.qq.com" _qrcode_ttl_seconds = 240 _active_target_ttl_seconds = 24 * 60 * 60 + _poll_join_timeout_seconds = 5 @classmethod def _build_cache_key(cls, config_name: str) -> str: @@ -1706,12 +1707,21 @@ class WechatClawBot: """获取当前登录状态。""" return bool(self._state.get("bot_token")) - def stop(self) -> None: - """停止消息轮询。""" + def stop(self) -> bool: + """停止消息轮询，并保留超时线程 owner 供后续重试。""" self._stop_event.set() - if self._poll_thread and self._poll_thread.is_alive(): - self._poll_thread.join(timeout=5) + poll_thread = self._poll_thread + if ( + poll_thread + and poll_thread.is_alive() + and poll_thread is not threading.current_thread() + ): + poll_thread.join(timeout=self._poll_join_timeout_seconds) + if poll_thread and poll_thread.is_alive(): + logger.error("微信 ClawBot 消息轮询线程未在关闭预算内退出") + return False self._poll_thread = None + return True def _start_polling(self) -> None: """启动消息轮询线程。""" diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index b1144d193..95597e05e 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` -> 审计基线：`2aa41ea8`（2026-08-24） +> 审计基线：`009631b8`（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 声明；阶段 61 已统一 Capability Runtime 同步/异步关闭的诚实收敛结果。 +> 实施进度：阶段 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 同步/异步关闭的诚实收敛结果；阶段 62 已统一消息渠道长连接的多实例关闭收敛合同。 > 当前 canonical 状态：API/Application 公共复杂度基线已清零，组合根外 `SystemConfigOper()` 构造和 Model/Oper 隐式事务均为 0；命名 Chain/Agent 数据端口、TaskRegistry owner、Module Contract V2、typed Event、Outbox durable intent、请求关联和插件运行时 getter 已形成当前路径。插件仓适配、未知第三方 fallback 和其它 E1/E3 副作用仍按风险持续治理。 -> 最新阶段：阶段 61 已统一 Capability Runtime 关闭收敛事实源。 +> 最新阶段：阶段 62 已统一消息渠道长连接的关闭收敛合同。 ## 当前复核结论（2026-08-24） @@ -644,6 +644,19 @@ 返回 `False`，不会因清空句柄而丢失重试能力。Telegram 配置、菜单、消息、typing 语义与类 identity 未变。 - 本阶段收口宿主 Capability Runtime 生命周期结果；未修改插件仓、SDK/Compat 映射或 V1/V2/V3 插件 Hook。 +### 长期整改阶段 62：消息渠道长连接关闭收敛合同统一（2026-08-24） + +- 阶段 61 只让 Telegram polling 接入了 Host Module 的布尔收敛合同；QQBot、企业微信、 + WeChatClawBot、飞书和 Discord 的 Gateway/WebSocket/polling 线程仍在 join 超时后返回 `None`， + Slack 的 Socket Mode close 异常也只写日志。WeChatClawBot 还会无条件清空仍存活的轮询线程句柄，形成同一目标的两套实现。 +- 当前七个消息渠道模块统一复用 `_MessageChannelModuleBase._stop_service_instances()`：逐实例继续尽力停止， + 任一客户端异常或显式 `False` 都聚合为模块未收敛，再经 HostModuleAdapter 与 Capability Runtime 向 startup + 传播。各长连接客户端使用既有有界等待预算，只有线程真实终止才返回成功；超时 owner 和句柄继续保留供 + 后续 shutdown 重试。 +- 渠道名称、配置字段、优先级、消息解析、发送与命令 Hook、类 identity 均未改变；抽象 Module `stop()` + 只扩展为可选布尔结果，既有返回 `None` 的宿主模块和 V1/V2/V3 插件仍按成功兼容处理。未修改插件仓、 + SDK/Compat 映射或事件 payload。 + ### 总体判断 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**： diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index f1025f41e..5727653a0 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -195,6 +195,8 @@ Host Module 的 `stop()` 可以显式返回 `False` 表示资源 owner 尚未收 ModuleManager 与 startup 组合根继续关闭其余资源但必须向上返回未收敛，不得把记录日志等同于成功。 同步和异步 Capability Runtime 的 `shutdown` 必须使用同一布尔收敛合同；Agent、Managed Resource 等领域关闭入口必须直接传播 Runtime 的整体结果，不得以单个能力快照或无返回包装器覆盖失败。 +消息渠道模块必须通过 `_MessageChannelModuleBase._stop_service_instances()` 聚合多实例关闭结果； +长连接、轮询或 Socket 服务只有在真实终止后才能返回成功，超时 owner 不得清空句柄。 API 中允许丢失或可重建的进程内任务必须登记到 `app/runtime/tasks.py`；登记器先于其他 运行资源启动，并在资源释放前停止接收、取消和有限等待。需要崩溃恢复的 E2/E3 副作用仍应 进入 Outbox 或持久任务表，不能把 TaskRegistry 当成 durable queue。 diff --git a/tests/test_module_lifecycle.py b/tests/test_module_lifecycle.py index 72610b30b..0068397bc 100644 --- a/tests/test_module_lifecycle.py +++ b/tests/test_module_lifecycle.py @@ -5,18 +5,24 @@ import pytest from app.modules import _MessageBase from app.modules.discord import DiscordModule +from app.modules.discord.discord import Discord from app.modules.feishu import FeishuModule +from app.modules.feishu.feishu import Feishu from app.modules.filter import FilterModule from app.modules.plex import PlexModule from app.modules.qqbot.module import QQBotModule +from app.modules.qqbot.qqbot import QQBot from app.modules.slack import SlackModule +from app.modules.slack.slack import Slack from app.modules.telegram.module import TelegramModule from app.modules.telegram.telegram import Telegram from app.modules.themoviedb import TheMovieDbModule from app.modules.trimemedia.module import TrimeMediaModule from app.modules.ugreen.module import UgreenModule from app.modules.wechat import WechatModule +from app.modules.wechat.wechatbot import WeChatBot from app.modules.wechatclawbot import WechatClawBotModule +from app.modules.wechatclawbot.wechatclawbot import WechatClawBot def test_config_reload_stops_before_initializing_latest_generation(): @@ -217,9 +223,23 @@ def test_telegram_stop_keeps_polling_owner_when_thread_misses_deadline(): assert client._polling_thread is polling_thread -def test_telegram_module_reports_nonconverging_instance_after_stopping_peers(): - """单实例未收敛时模块必须继续停止其余实例并返回 False。""" - module = TelegramModule() +@pytest.mark.parametrize( + "module_type", + [ + DiscordModule, + FeishuModule, + QQBotModule, + SlackModule, + TelegramModule, + WechatModule, + WechatClawBotModule, + ], +) +def test_message_channel_module_reports_nonconverging_instance_after_peers( + module_type, +): + """渠道单实例未收敛时必须继续停止其余实例并返回 False。""" + module = module_type() blocked_client = Mock() blocked_client.stop.return_value = False healthy_client = Mock() @@ -232,3 +252,114 @@ def test_telegram_module_reports_nonconverging_instance_after_stopping_peers(): assert module.stop() is False blocked_client.stop.assert_called_once_with() healthy_client.stop.assert_called_once_with() + + +def test_qqbot_stop_retains_gateway_thread_until_retry() -> None: + """QQ Gateway 超时后不得把活线程伪装成已收敛。""" + client = QQBot.__new__(QQBot) + client._gateway_stop = threading.Event() + client._gateway_ws_holder = [] + gateway_thread = Mock() + gateway_thread.is_alive.return_value = True + client._gateway_thread = gateway_thread + client._gateway_join_timeout_seconds = 0.01 + + assert client.stop() is False + assert client._gateway_thread is gateway_thread + + gateway_thread.is_alive.return_value = False + assert client.stop() is True + + +def test_wechat_bot_stop_reports_each_live_thread_until_retry() -> None: + """企业微信网关或心跳任一存活时都必须返回未收敛。""" + client = WeChatBot.__new__(WeChatBot) + client._stop_event = threading.Event() + client._authenticated = threading.Event() + client._ws_app = None + ws_thread = Mock() + heartbeat_thread = Mock() + ws_thread.is_alive.return_value = True + heartbeat_thread.is_alive.return_value = True + client._ws_thread = ws_thread + client._heartbeat_thread = heartbeat_thread + client._gateway_join_timeout_seconds = 0.01 + client._heartbeat_join_timeout_seconds = 0.01 + + assert client.stop() is False + assert client._ws_thread is ws_thread + assert client._heartbeat_thread is heartbeat_thread + + ws_thread.is_alive.return_value = False + heartbeat_thread.is_alive.return_value = False + assert client.stop() is True + + +def test_wechat_clawbot_stop_keeps_poll_owner_until_retry() -> None: + """ClawBot 轮询超时后必须保留线程句柄供后续关闭重试。""" + client = WechatClawBot.__new__(WechatClawBot) + client._stop_event = threading.Event() + poll_thread = Mock() + poll_thread.is_alive.return_value = True + client._poll_thread = poll_thread + client._poll_join_timeout_seconds = 0.01 + + assert client.stop() is False + assert client._poll_thread is poll_thread + + poll_thread.is_alive.return_value = False + assert client.stop() is True + assert client._poll_thread is None + + +def test_feishu_stop_reports_live_ws_thread_until_retry() -> None: + """飞书 SDK 清理后线程仍存活时不得报告关闭完成。""" + client = Feishu.__new__(Feishu) + client._stop_event = threading.Event() + client._ready = threading.Event() + client._ws_client = None + client._ws_loop = None + ws_thread = Mock() + ws_thread.is_alive.return_value = True + client._ws_thread = ws_thread + client._ws_join_timeout_seconds = 0.01 + + assert client.stop() is False + assert client._ws_thread is ws_thread + + ws_thread.is_alive.return_value = False + assert client.stop() is True + + +def test_discord_stop_reports_live_event_loop_thread_until_retry() -> None: + """Discord 强制停止循环后线程仍存活时必须返回未收敛。""" + client = Discord.__new__(Discord) + client._client = Mock() + client._loop = Mock() + client._loop.is_running.return_value = False + event_loop_thread = Mock() + event_loop_thread.is_alive.return_value = True + client._thread = event_loop_thread + client._stop_requested = threading.Event() + client._ready_event = threading.Event() + client._thread_join_timeout_seconds = 0.01 + + assert client.stop() is False + assert client._thread is event_loop_thread + + event_loop_thread.is_alive.return_value = False + assert client.stop() is True + + +def test_slack_stop_propagates_socket_close_failure() -> None: + """Slack Socket Mode close 失败必须保留给 Runtime 重试。""" + client = Slack.__new__(Slack) + service = Mock() + service.close.side_effect = RuntimeError("close failed") + client._service = service + + assert client.stop() is False + assert client._service is service + + service.close.side_effect = None + assert client.stop() is True