fix: isolate telegram typing lifecycle

app/modules/telegram/telegram.py

@@ -77,18 +77,13 @@ class Telegram:
 
     _bot: TeleBot = None
     _callback_handlers: Dict[str, Callable] = {}  # 存储回调处理器
-    _user_chat_mapping: Dict[
-        str, str
-    ] = {}  # userid -> chat_id mapping for reply targeting
     _bot_username: Optional[str] = None  # Bot username for mention detection
-    _typing_tasks: Dict[str, threading.Thread] = {}  # chat_id -> typing任务
-    _typing_stop_flags: Dict[str, threading.Event] = {}  # chat_id -> 停止信号
-    _typing_lock = threading.RLock()
     _typing_interval_seconds = 5
     _typing_initial_delay_seconds = 1
     _typing_max_duration_seconds = 10 * 60
     _typing_command_max_duration_seconds = 30
     _typing_callback_max_duration_seconds = 60
+    _typing_join_timeout_seconds = 1
 
     def __init__(
             self,
@@ -103,6 +98,13 @@ class Telegram:
         self._telegram_token = TELEGRAM_TOKEN
         self._telegram_chat_id = TELEGRAM_CHAT_ID
         self._polling_thread = None
+        # 一个 Telegram 配置对应一个 SDK client，运行状态不能被其他配置共享。
+        self._user_chat_mapping: Dict[str, str] = {}
+        self._typing_tasks: Dict[str, threading.Thread] = {}
+        self._typing_stop_flags: Dict[str, threading.Event] = {}
+        self._typing_lock = threading.RLock()
+        self._typing_lifecycle_lock = threading.RLock()
+        self._typing_accepting = True
         if not TELEGRAM_TOKEN or not TELEGRAM_CHAT_ID:
             logger.error("Telegram配置不完整！")
             return
@@ -482,70 +484,101 @@ class Telegram:
             chat_id: Union[str, int],
             max_duration_seconds: Optional[float] = None,
             initial_delay_seconds: Optional[float] = None,
-    ) -> None:
+    ) -> bool:
         """
-        启动持续发送正在输入状态的任务
+        启动持续发送正在输入状态的任务。
+
+        :return: 是否取得该会话的唯一 typing owner
         """
         chat_id_str = str(chat_id)
-        # 如果已有任务在运行，先停止
-        self._stop_typing_task(chat_id_str)
+        with self._typing_lifecycle_lock:
+            if not self._typing_accepting:
+                logger.debug("Telegram client已停止，拒绝启动typing任务")
+                return False
+            # 如果已有任务在运行，先停止；阻塞的旧 SDK 请求不能被新 owner 覆盖。
+            if not self._stop_typing_task(chat_id_str):
+                logger.warning(
+                    "Telegram typing旧任务尚未结束，拒绝并行启动: chat_id=%s",
+                    chat_id_str,
+                )
+                return False
 
-        # 使用独立 Event 避免同一 chat 新旧 typing 线程互相误改停止标记。
-        stop_event = threading.Event()
-        max_duration = max_duration_seconds or self._typing_max_duration_seconds
-        initial_delay = (
-            self._typing_initial_delay_seconds
-            if initial_delay_seconds is None
-            else max(initial_delay_seconds, 0)
-        )
+            # 使用独立 Event 避免同一 chat 新旧 typing 线程互相误改停止标记。
+            stop_event = threading.Event()
+            max_duration = max_duration_seconds or self._typing_max_duration_seconds
+            initial_delay = (
+                self._typing_initial_delay_seconds
+                if initial_delay_seconds is None
+                else max(initial_delay_seconds, 0)
+            )
 
-        def typing_worker():
-            """延迟首发并定期发送 typing 状态的后台线程。"""
-            started_at = time.monotonic()
-            try:
-                # Telegram 没有撤销 typing 的接口，短响应先等待一小段时间，
-                # 避免回复已经发出后客户端仍残留几秒“正在输入”。
-                if initial_delay and stop_event.wait(initial_delay):
-                    return
-                while not stop_event.is_set():
-                    if time.monotonic() - started_at >= max_duration:
-                        logger.warning(
-                            "Telegram typing状态超过最大续期，自动停止: chat_id=%s",
-                            chat_id_str,
-                        )
-                        break
-                    try:
-                        if self._bot:
-                            self._bot.send_chat_action(chat_id, "typing")
-                    except Exception as e:
-                        logger.debug(f"发送typing状态失败: {e}")
-                    # Telegram 客户端约 5-6 秒后会隐藏 typing，需要周期性续发。
-                    stop_event.wait(self._typing_interval_seconds)
-            finally:
+            def typing_worker():
+                """延迟首发并定期发送 typing 状态的后台线程。"""
+                started_at = time.monotonic()
+                try:
+                    # Telegram 没有撤销 typing 的接口，短响应先等待一小段时间，
+                    # 避免回复已经发出后客户端仍残留几秒“正在输入”。
+                    if initial_delay and stop_event.wait(initial_delay):
+                        return
+                    while not stop_event.is_set():
+                        if time.monotonic() - started_at >= max_duration:
+                            logger.warning(
+                                "Telegram typing状态超过最大续期，自动停止: chat_id=%s",
+                                chat_id_str,
+                            )
+                            break
+                        try:
+                            if self._bot:
+                                self._bot.send_chat_action(chat_id, "typing")
+                        except Exception as e:
+                            logger.debug(f"发送typing状态失败: {e}")
+                        # Telegram 客户端约 5-6 秒后会隐藏 typing，需要周期性续发。
+                        stop_event.wait(self._typing_interval_seconds)
+                finally:
+                    with self._typing_lock:
+                        current = self._typing_tasks.get(chat_id_str)
+                        if current is threading.current_thread():
+                            self._typing_tasks.pop(chat_id_str, None)
+                            self._typing_stop_flags.pop(chat_id_str, None)
+
+            thread = threading.Thread(
+                target=typing_worker,
+                name=f"MoviePilot-TelegramTyping-{chat_id_str}"[:120],
+                daemon=True,
+            )
+            with self._typing_lock:
+                self._typing_stop_flags[chat_id_str] = stop_event
+                self._typing_tasks[chat_id_str] = thread
+                try:
+                    thread.start()
+                except BaseException:
+                    self._typing_stop_flags.pop(chat_id_str, None)
+                    self._typing_tasks.pop(chat_id_str, None)
+                    raise
+            return True
+
+    def _stop_typing_task(self, chat_id: Union[str, int]) -> bool:
+        """
+        停止正在输入状态的任务，并保留尚未结束的 owner。
+
+        :return: 任务是否已经进入终态
+        """
+        chat_id_str = str(chat_id)
+        with self._typing_lifecycle_lock:
+            with self._typing_lock:
+                stop_event = self._typing_stop_flags.get(chat_id_str)
+                task = self._typing_tasks.get(chat_id_str)
+            if stop_event:
+                stop_event.set()
+            if task and task.is_alive() and task is not threading.current_thread():
+                task.join(timeout=self._typing_join_timeout_seconds)
+            task_finished = task is None or not task.is_alive()
+            if task_finished:
                 with self._typing_lock:
-                    current = self._typing_tasks.get(chat_id_str)
-                    if current is threading.current_thread():
+                    if self._typing_tasks.get(chat_id_str) is task:
                         self._typing_tasks.pop(chat_id_str, None)
                         self._typing_stop_flags.pop(chat_id_str, None)
-
-        thread = threading.Thread(target=typing_worker, daemon=True)
-        with self._typing_lock:
-            self._typing_stop_flags[chat_id_str] = stop_event
-            self._typing_tasks[chat_id_str] = thread
-        thread.start()
-
-    def _stop_typing_task(self, chat_id: Union[str, int]) -> None:
-        """
-        停止正在输入状态的任务
-        """
-        chat_id_str = str(chat_id)
-        with self._typing_lock:
-            stop_event = self._typing_stop_flags.pop(chat_id_str, None)
-            task = self._typing_tasks.pop(chat_id_str, None)
-        if stop_event:
-            stop_event.set()
-        if task and task.is_alive() and task is not threading.current_thread():
-            task.join(timeout=1)
+            return task_finished
 
     def _stop_typing_if_needed(
             self, chat_id: Union[str, int], stop_typing: bool
@@ -574,8 +607,7 @@ class Telegram:
         target_chat_id = target_chat_id or (str(userid) if userid else None)
         if not target_chat_id:
             return False
-        self._start_typing_task(target_chat_id)
-        return True
+        return self._start_typing_task(target_chat_id)
 
     def stop_typing(
             self,
@@ -1714,9 +1746,11 @@ class Telegram:
         """
         停止Telegram消息接收服务
         """
-        # 停止所有typing任务
-        for chat_id in list(self._typing_tasks.keys()):
-            self._stop_typing_task(chat_id)
+        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
 

docs/refactor/backend-architecture-next-stage.md

@@ -6,7 +6,7 @@
 > 审计范围：宿主后端；排除 `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 已统一优雅重启兜底线程的唯一所有权。
+> 实施进度：阶段 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。
 
 ## 当前复核结论（2026-08-24）
 
@@ -438,6 +438,18 @@
 - 该 daemon monitor 故意跨越正常 shutdown drain，以便进程卡死时仍能请求 Docker 重启，因此不纳入
   `TaskRegistry` 或普通线程池等待。`SystemHelper` 公开方法、SDK/Compat 映射和 V1/V2/V3 插件 ABI 均未修改。
 
+### 长期整改阶段 42：Telegram typing 多实例与终态所有权收口（2026-08-24）
+
+- `TelegramModule` 支持多个通知配置实例，但用户到 chat 的映射、typing 线程、停止信号和锁原先都在
+  `Telegram` 类级共享；两个配置遇到相同 chat ID 时会相互停止或覆盖 owner。生产 client 现在各自持有
+  完整运行状态，同一配置内则由 lifecycle 锁串行替换，保留一个 chat 一个 typing owner。
+- 停止等待超过预算时不再提前删除仍存活的线程句柄，新请求会拒绝覆盖阻塞中的旧 owner；线程启动失败会
+  回滚 owner 和停止信号，client 停止时先封住新增任务再取得完整快照。SDK 请求恢复后，线程仍由自己的
+  `finally` 在真实终态释放登记。
+- `Telegram` 类路径、构造参数、`start_typing()`/`stop_typing()` 布尔合同、消息格式、模块方法及配置字段
+  均保持不变；私有类级可变状态已清除，正常 V1/V2/V3 模块实例不再共享运行状态。本阶段未修改插件仓、
+  SDK 或 Compat 映射。
+
 ### 总体判断
 
 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**：

tests/test_module_lifecycle.py

@@ -8,13 +8,13 @@ from app.modules.discord import DiscordModule
 from app.modules.feishu import FeishuModule
 from app.modules.filter import FilterModule
 from app.modules.plex import PlexModule
-from app.modules.qqbot import QQBotModule
+from app.modules.qqbot.module import QQBotModule
 from app.modules.slack import SlackModule
-from app.modules.telegram import TelegramModule
+from app.modules.telegram.module import TelegramModule
 from app.modules.telegram.telegram import Telegram
 from app.modules.themoviedb import TheMovieDbModule
-from app.modules.trimemedia import TrimeMediaModule
-from app.modules.ugreen import UgreenModule
+from app.modules.trimemedia.module import TrimeMediaModule
+from app.modules.ugreen.module import UgreenModule
 from app.modules.wechat import WechatModule
 from app.modules.wechatclawbot import WechatClawBotModule
 
@@ -177,6 +177,11 @@ def test_telegram_stop_closes_sdk_and_waits_for_polling_thread():
     client._bot = bot
     polling_thread = Mock()
     client._polling_thread = polling_thread
+    client._typing_tasks = {}
+    client._typing_stop_flags = {}
+    client._typing_lock = threading.RLock()
+    client._typing_lifecycle_lock = threading.RLock()
+    client._typing_accepting = True
 
     client.stop()
     client.stop()