fix(telegram): Rich Message 失败时保留原始错误并降级纯文本 (#6993)

* fix(telegram): Rich Message 失败时保留原始错误并降级纯文本

Telegram Rich Message 发送失败时存在三个叠加问题：

1. `__send_rich_message` 用固定文本 `RetryException("发送 Telegram Rich
   Message 失败")` 包装异常，`str(e)` 里没有 Telegram 返回的 description，
   日志只剩“发送失败”四个字，无法判断是内容非法、方法不存在还是格式问题。

2. 内容非法这类确定性失败（400）同样走满3 次退避重试（3s、6s），
   期间同一条坏消息反复阻塞，且最终报错仍无有效信息。

3. 失败即丢失通知：Rich Message 发不出去时没有任何降级路径，
   整条通知直接消失。

改动：

- 异常文本改为 f"...失败：{err}"，保留 Telegram 原始 description 与错误码。
- 新增 `NonRetryableTelegramError`（继承 `ImmediateException`）与
  `_is_non_retryable_api_error`，400 立即抛出以跳过 retry 退避；
  5xx 与网络异常保持原有重试策略。
- `send_msg` / `edit_msg` 增加纯文本降级出口，Rich Message 失败时
  用不带格式解析的纯文本重发，保证通知内容不丢失。
- 降级路径显式传 `parse_mode=""`：`TelegramBotApi._parse_mode` 在参数为
  `None` 时会回落到客户端默认的 MarkdownV2，那正是 Rich Message 已经
  失败过的解析路径，漏传会让降级重演同一个错误。

新增 tests/test_telegram_rich_message_fallback.py（14 项）覆盖错误信息保留、
400 免重试、5xx 仍重试、发送与编辑两条降级路径，以及 parse_mode 回落陷阱的
回归防护。

* fix(telegram): 修正 Rich Message 降级路径的 mypy 与架构基线

上一提交引入的两处门禁失败：

1. `__send_plain_fallback` 用 `**self._topic_kwargs(chat_id)` 展开参数，mypy
   把 `**` 匹配到 `send_message` 第 5 个形参 `disable_web_page_preview`
   （Optional[bool]），判定 `dict[str, int]` 不兼容，arg-type 由 1 增至 2。
   改为显式传 `message_thread_id`，`_topic_kwargs` 只可能产出该键。

2. 新增 `app.schemas.exception` 导入带来 2 条依赖边，宿主基线由 9542 变为
   9544。已用 `baseline.py --write-host` 固化，其余 3 个基线文件字节不变。

   连带影响：docs/architecture-overview.md 与 docs/refactor/
   optimization-checklist.md 硬编码了边数，`test_architecture_documents_
   match_generated_quality_metrics` 断言两处必须与基线一致，同步为 9,544。

门禁与测试验证：

- scripts/architecture/mypy_ratchet.py          exit 0（9163 个类型错误）
- scripts/architecture/baseline.py --check-host exit 0
- pytest（fallback + contract + telegram）      75 passed

---------

Co-authored-by: jxxghp <jxxghp@gmail.com>

app/modules/telegram/telegram.py

@@ -26,12 +26,14 @@ from app.foundation import size as size_tools
 from app.modules.telegram.botapi import (
     FORWARDED_CONTENT_TYPES,
     JsonDict,
+    TelegramApiError,
     TelegramBotApi,
 )
 from app.runtime.execution import retry
 from app.runtime.log import logger
 from app.runtime.settings import get_runtime_setting
 from app.runtime.thread import ThreadHelper
+from app.schemas.exception import ImmediateException
 
 TELEGRAM_PARSE_MODE_MARKDOWN = "MarkdownV2"
 TELEGRAM_PARSE_MODE_HTML = "HTML"
@@ -55,6 +57,26 @@ class RetryException(Exception):
     pass
 
 
+class NonRetryableTelegramError(ImmediateException):
+    """
+    Telegram 明确拒绝的请求，重试不会改变结果。
+
+    内容非法、格式无法解析、方法不存在等 4xx 错误属于这一类：立即抛出以跳过
+    ``retry`` 的退避等待，避免同一条坏消息连续阻塞 3 秒、6 秒后才报错。
+    """
+
+    pass
+
+
+def _is_non_retryable_api_error(err: Exception) -> bool:
+    """
+    判断 Bot API 错误是否为重试无意义的 4xx。
+
+    5xx 与网络异常仍按原逻辑重试；429 由 Telegram 侧限流，同样不适合立即失败。
+    """
+    return isinstance(err, TelegramApiError) and err.error_code == 400
+
+
 class Telegram:
     """
     Telegram 消息客户端，负责发送、编辑、接收和转发 Telegram 消息。
@@ -714,14 +736,25 @@ class Telegram:
                     if force_reply and original_chat_id
                     else chat_id
                 )
-                sent = self.__send_rich_message(
-                    chat_id=target_chat_id,
-                    rich_message=rich_message,
-                    reply_markup=reply_markup,
-                    reply_to_message_id=(
-                        original_message_id if force_reply else None
-                    ),
-                )
+                try:
+                    sent = self.__send_rich_message(
+                        chat_id=target_chat_id,
+                        rich_message=rich_message,
+                        reply_markup=reply_markup,
+                        reply_to_message_id=(
+                            original_message_id if force_reply else None
+                        ),
+                    )
+                except Exception as rich_err:
+                    # Rich Message 失败不再等同于通知丢失，降级为纯文本重发一次。
+                    logger.warning(
+                        f"Telegram Rich Message 发送失败，降级为纯文本：{rich_err}"
+                    )
+                    sent = self.__send_plain_fallback(
+                        chat_id=target_chat_id,
+                        rich_message=rich_message,
+                        reply_markup=reply_markup,
+                    )
                 self._stop_typing_if_needed(chat_id, stop_typing)
                 return self._sent_result(sent, chat_id)
 
@@ -1231,12 +1264,24 @@ class Telegram:
                 reply_markup = (
                     self._create_inline_keyboard(buttons) if buttons else None
                 )
-                return self.__edit_rich_message(
+                result = self.__edit_rich_message(
                     chat_id=chat_id,
                     message_id=message_id,
                     rich_message=rich_message,
                     reply_markup=reply_markup,
                 )
+                if result:
+                    return True
+                # Rich Message 编辑失败时降级为纯文本覆盖，避免回复停留在旧内容。
+                logger.warning(
+                    f"Telegram Rich Message 编辑失败，降级为纯文本：chat_id={chat_id}"
+                )
+                return self.__edit_plain_fallback(
+                    chat_id=str(chat_id),
+                    message_id=int(message_id),
+                    rich_message=rich_message,
+                    reply_markup=reply_markup,
+                )
 
             # 组合标题和文本
             if title:
@@ -1465,6 +1510,41 @@ class Telegram:
             logger.error(f"编辑 Telegram Rich Message 失败：{str(err)}")
             return False
 
+    def __edit_plain_fallback(
+            self,
+            chat_id: str,
+            message_id: int,
+            rich_message: str,
+            reply_markup: Optional[JsonDict] = None,
+    ) -> bool:
+        """
+        Rich Message 编辑失败后的降级出口，用纯文本覆盖原消息。
+
+        ``parse_mode`` 显式传空字符串，避免 ``edit_message_text`` 回落到客户端默认的
+        MarkdownV2——Rich Message 已经在那条解析路径上失败过一次。
+
+        :param chat_id: 聊天 ID
+        :param message_id: 原消息 ID
+        :param rich_message: 原始 Rich Markdown 正文
+        :param reply_markup: 内联键盘
+        :return: 降级编辑是否成功
+        """
+        try:
+            self._api.edit_message_text(
+                chat_id=chat_id,
+                message_id=int(message_id),
+                text=rich_message,
+                parse_mode=TELEGRAM_PARSE_MODE_PLAIN,
+                reply_markup=reply_markup,
+            )
+            return True
+        except Exception as err:
+            if self.__is_message_not_modified_error(err):
+                logger.debug(f"Telegram消息内容未变化，跳过编辑：{str(err)}")
+                return True
+            logger.error(f"Telegram Rich Message 降级为纯文本编辑仍失败：{str(err)}")
+            return False
+
     @retry(RetryException, logger=logger)
     def __send_rich_message(
             self,
@@ -1481,6 +1561,8 @@ class Telegram:
         :param reply_markup: 首条消息携带的键盘
         :param reply_to_message_id: 首条消息回复的原消息 ID
         :return: 最后一条已发送的 Telegram 消息
+        :raises NonRetryableTelegramError: Telegram 以 400 拒绝该内容，不做退避重试
+        :raises RetryException: 网络异常或 5xx，可重试
         """
         chunks = self._build_rich_message_chunks(rich_message)
         reply_parameters = (
@@ -1500,7 +1582,52 @@ class Telegram:
                 )
             return sent
         except Exception as err:
-            raise RetryException("发送 Telegram Rich Message 失败") from err
+            # 保留 Telegram 返回的原始描述（错误码、description），否则日志只剩
+            # “发送失败”四个字，无法定位是内容非法还是接口不匹配。
+            if _is_non_retryable_api_error(err):
+                raise NonRetryableTelegramError(
+                    f"发送 Telegram Rich Message 失败：{err}"
+                ) from err
+            raise RetryException(
+                f"发送 Telegram Rich Message 失败：{err}"
+            ) from err
+
+    def __send_plain_fallback(
+            self,
+            chat_id: Union[str, int],
+            rich_message: str,
+            reply_markup: Optional[JsonDict] = None,
+    ) -> Any:
+        """
+        Rich Message 发送失败后的降级出口，用纯文本重发同一条内容。
+
+        Rich Message 依赖 Bot API 10.1 的 ``sendRichMessage``，内容里出现无法解析的
+        片段或自建 Bot API 未升级该方法时都会整体失败。通知内容本身仍有价值，因此
+        降级为纯文本，保证消息不丢失。
+
+        ``parse_mode`` 必须显式传空字符串：``TelegramBotApi`` 在参数为 ``None`` 时会
+        回落到客户端默认的 MarkdownV2，那正是 Rich Message 已经失败过的解析路径。
+
+        :param chat_id: 目标聊天 ID
+        :param rich_message: 原始 Rich Markdown 正文
+        :param reply_markup: 内联键盘
+        :return: 降级发送的结果，失败返回 None
+        """
+        # ``_topic_kwargs`` 只可能产出 message_thread_id，显式传参而不是 ** 展开：
+        # ** 展开到 send_message 会被 mypy 匹配到第 5 个形参（disable_web_page_preview，
+        # Optional[bool]），把 dict[str, int] 判为类型不兼容。
+        topic_kwargs = self._topic_kwargs(chat_id)
+        try:
+            return self._api.send_message(
+                chat_id=chat_id,
+                text=rich_message,
+                parse_mode=TELEGRAM_PARSE_MODE_PLAIN,
+                reply_markup=reply_markup,
+                message_thread_id=topic_kwargs.get("message_thread_id"),
+            )
+        except Exception as err:
+            logger.error(f"Telegram Rich Message 降级为纯文本发送仍失败：{err}")
+            return None
 
     def __send_request(
             self,

docs/architecture-overview.md

@@ -763,7 +763,7 @@ flowchart LR
 | 指标 | 当前值 |
 |---|---:|
 | Python 模块 | 1119 |
-| 内部导入边 | 9,545 |
+| 内部导入边 | 9,547 |
 | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 49（债务已清零，49 条精确 containment） |
@@ -773,7 +773,7 @@ flowchart LR
 | Model/Oper 自动事务与自建 Session | 0 |
 | 组合根外 `SystemConfigOper()` | 0 |
 
-成功通知聚合复用既有 owner 和领域/模板模型，新增 14 条合法内部依赖边（含父包展开），未新增模块或 SCC。整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,545；订阅独立日志复用运行时上下文、系统日志端口与资源鉴权，新增 19 条合法依赖边（含父包展开），未新增模块或 SCC；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条 Schema 到 Domain 的合法依赖边（含父包展开）；种子下载限流窗口由 `app.domain.torrent` 定义，失败冷却 owner 复用同一语义，新增 1 条合法 Chain 到 Domain 的导入边；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条模块导入边；资源过滤动作 `app.workflow.actions.filter_torrents` 与动作包共用的领域还原函数同样把 Schema 资源还原为领域对象后再交给过滤链，新增 6 条模块导入边。音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 587、`app.factory` 为 600、`app.main` 为 602。专辑订阅入库对账新增 `app.chain.subscribe.music` 一个职责 owner；按注入的下载/整理历史 Port 读取事实，复用音频标签/CUE 读取和领域覆盖规则，依赖图增加 18 条合法边，不引入跨层数据库或 Adapter 直连。订阅业务活动新增 `app.application.subscription.activity` 及对应 DB Model、Oper、Adapter 四个模块，三入口与功能前提交对照均只增加这四个模块；性能基线仅同步结构计数，保留原有历史耗时采样和预算。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。
+成功通知聚合复用既有 owner 和领域/模板模型，新增 14 条合法内部依赖边（含父包展开），未新增模块或 SCC。整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,547；订阅独立日志复用运行时上下文、系统日志端口与资源鉴权，新增 19 条合法依赖边（含父包展开），未新增模块或 SCC；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条 Schema 到 Domain 的合法依赖边（含父包展开）；种子下载限流窗口由 `app.domain.torrent` 定义，失败冷却 owner 复用同一语义，新增 1 条合法 Chain 到 Domain 的导入边；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条模块导入边；资源过滤动作 `app.workflow.actions.filter_torrents` 与动作包共用的领域还原函数同样把 Schema 资源还原为领域对象后再交给过滤链，新增 6 条模块导入边。音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 587、`app.factory` 为 600、`app.main` 为 602。专辑订阅入库对账新增 `app.chain.subscribe.music` 一个职责 owner；按注入的下载/整理历史 Port 读取事实，复用音频标签/CUE 读取和领域覆盖规则，依赖图增加 18 条合法边，不引入跨层数据库或 Adapter 直连。订阅业务活动新增 `app.application.subscription.activity` 及对应 DB Model、Oper、Adapter 四个模块，三入口与功能前提交对照均只增加这四个模块；性能基线仅同步结构计数，保留原有历史耗时采样和预算。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。
 
 架构专项验证分为两个 CI 投影：`Check event semantic policy` 先运行依赖、Adapter、出口和 Event
 语义门禁，`Check host architecture snapshot` 再执行快照测试及一次

docs/refactor/optimization-checklist.md

@@ -100,7 +100,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 1119 / 9,545 | `dependency-baseline.json` 当前快照；成功通知聚合复用既有 owner 和领域/模板模型，新增 14 条合法内部依赖边（含父包展开），未新增模块或 SCC。专辑订阅入库对账新增 1 个 Chain owner 和 18 条合法依赖边（含父包展开），仍经注入 Port 读取历史并通过既有 mutation 写入；订阅业务活动新增 4 个模块和 50 条合法内部依赖边（含父包展开），Application 仅定义事实与持久化 Port，DB Adapter/Oper 实现存取，组合根注入，未增加 Application/Chain 到具体 Adapter 或数据库的直连；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条合法依赖边（含父包展开）；种子失败冷却复用 `app.domain.torrent` 的限流语义，新增 1 条合法导入边；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条合法模块导入边，资源过滤动作复用同一还原再新增 6 条；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份 |
+| 宿主 Python 模块 / 内部依赖边 | 1119 / 9,547 | `dependency-baseline.json` 当前快照；Telegram Rich Message 失败降级复用 `app.schemas.exception` 的 `ImmediateException`，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。成功通知聚合复用既有 owner 和领域/模板模型，新增 14 条合法内部依赖边（含父包展开），未新增模块或 SCC。专辑订阅入库对账新增 1 个 Chain owner 和 18 条合法依赖边（含父包展开），仍经注入 Port 读取历史并通过既有 mutation 写入；订阅业务活动新增 4 个模块和 50 条合法内部依赖边（含父包展开），Application 仅定义事实与持久化 Port，DB Adapter/Oper 实现存取，组合根注入，未增加 Application/Chain 到具体 Adapter 或数据库的直连；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条合法依赖边（含父包展开）；种子失败冷却复用 `app.domain.torrent` 的限流语义，新增 1 条合法导入边；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条合法模块导入边，资源过滤动作复用同一还原再新增 6 条；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

tests/fixtures/architecture/dependency-baseline.json

@@ -941,8 +941,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 9545,
-  "edge_sha256": "8784d89756b95ce35702eb7aa43513050e4b181641dad727a39d6a0354ec0eb6",
+  "edge_count": 9547,
+  "edge_sha256": "ab4a52cd8bc869c37073daadc119478b70cb738af1495273b9a5ee295380749e",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -8182,6 +8182,8 @@
     "app.modules.telegram.telegram -> app.runtime.log",
     "app.modules.telegram.telegram -> app.runtime.settings",
     "app.modules.telegram.telegram -> app.runtime.thread",
+    "app.modules.telegram.telegram -> app.schemas",
+    "app.modules.telegram.telegram -> app.schemas.exception",
     "app.modules.theaudiodb -> app.adapters",
     "app.modules.theaudiodb -> app.adapters.network",
     "app.modules.theaudiodb -> app.adapters.network.http",

tests/test_telegram_rich_message_fallback.py

@@ -0,0 +1,210 @@
+"""Telegram Rich Message 失败降级与重试策略测试。"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.modules.telegram.botapi import TelegramApiError, TelegramNetworkError
+from app.modules.telegram.telegram import (
+    NonRetryableTelegramError,
+    RetryException,
+    Telegram,
+    _is_non_retryable_api_error,
+)
+
+
+def _make_client() -> Telegram:
+    """构造一个不启动轮询线程的 Telegram 客户端。"""
+    with patch("app.modules.telegram.telegram.TelegramBotApi"), patch(
+            "app.modules.telegram.telegram.threading.Thread"
+    ):
+        client = Telegram(TELEGRAM_TOKEN="token", TELEGRAM_CHAT_ID="chat")
+    client._bot = MagicMock()
+    return client
+
+
+class TestNonRetryableApiError:
+    """400 属于确定性拒绝，重试无意义。"""
+
+    def test_400_is_non_retryable(self):
+        """明确的 400 拒绝无需退避重试。"""
+        err = TelegramApiError("sendRichMessage", "Bad Request: ...", 400)
+        assert _is_non_retryable_api_error(err) is True
+
+    def test_500_is_retryable(self):
+        """服务端临时失败保留重试能力。"""
+        err = TelegramApiError("sendRichMessage", "Internal Server Error", 500)
+        assert _is_non_retryable_api_error(err) is False
+
+    def test_network_error_is_retryable(self):
+        """网络临时异常保留重试能力。"""
+        assert _is_non_retryable_api_error(TelegramNetworkError("m", "timeout")) is False
+
+    def test_unrelated_error_is_retryable(self):
+        """非 API 异常不被误判为确定性拒绝。"""
+        assert _is_non_retryable_api_error(ValueError("boom")) is False
+
+
+class TestSendRichMessageRetry:
+    """Rich Message 发送失败时的重试与异常类型。"""
+
+    def test_400_skips_backoff_and_raises_immediately(self):
+        """400 不应进入 retry 的退避等待。"""
+        client = _make_client()
+        client._api.send_rich_message.side_effect = TelegramApiError(
+            "sendRichMessage", "Bad Request: can't parse entities", 400
+        )
+
+        with patch("time.sleep") as sleep_mock:
+            with pytest.raises(NonRetryableTelegramError):
+                client._Telegram__send_rich_message(
+                    chat_id="chat", rich_message="# 结果"
+                )
+
+        sleep_mock.assert_not_called()
+        assert client._api.send_rich_message.call_count == 1
+
+    def test_500_retries(self):
+        """5xx 仍按原逻辑重试。"""
+        client = _make_client()
+        client._api.send_rich_message.side_effect = TelegramApiError(
+            "sendRichMessage", "Internal Server Error", 500
+        )
+
+        with patch("time.sleep"), pytest.raises(RetryException):
+            client._Telegram__send_rich_message(
+                chat_id="chat", rich_message="# 结果"
+            )
+
+        assert client._api.send_rich_message.call_count == 3
+
+    def test_error_message_keeps_telegram_description(self):
+        """异常文本必须保留 Telegram 原始描述，否则无法定位失败原因。"""
+        client = _make_client()
+        client._api.send_rich_message.side_effect = TelegramApiError(
+            "sendRichMessage", "Bad Request: can't parse entities", 400
+        )
+
+        with patch("time.sleep"), pytest.raises(NonRetryableTelegramError) as excinfo:
+            client._Telegram__send_rich_message(
+                chat_id="chat", rich_message="# 结果"
+            )
+
+        message = str(excinfo.value)
+        assert "can't parse entities" in message
+        assert "400" in message
+
+
+class TestSendMsgFallsBackToPlainText:
+    """Rich Message 发送失败时降级为纯文本，通知不丢失。"""
+
+    def test_falls_back_when_rich_message_fails(self):
+        client = _make_client()
+        client._api.send_rich_message.side_effect = TelegramApiError(
+            "sendRichMessage", "Bad Request", 400
+        )
+        client._api.send_message.return_value = {"message_id": 99, "chat": {"id": "chat"}}
+
+        with patch("time.sleep"):
+            result = client.send_msg(
+                title="结果", text="正文", rich_message="# 结果"
+            )
+
+        assert result["success"] is True
+        assert result["message_id"] == 99
+        # parse_mode 必须显式为空字符串：None 会回落到客户端默认的 MarkdownV2，
+        # 那正是 Rich Message 已经失败过的解析路径。
+        assert client._api.send_message.call_args.kwargs["text"] == "# 结果"
+        assert client._api.send_message.call_args.kwargs["parse_mode"] == ""
+
+    def test_no_fallback_when_rich_message_succeeds(self):
+        client = _make_client()
+        client._api.send_rich_message.return_value = {
+            "message_id": 1, "chat": {"id": "chat"}
+        }
+
+        result = client.send_msg(title="结果", text="正文", rich_message="# 结果")
+
+        assert result["success"] is True
+        client._api.send_message.assert_not_called()
+
+    def test_reports_failure_when_both_paths_fail(self):
+        client = _make_client()
+        client._api.send_rich_message.side_effect = TelegramApiError(
+            "sendRichMessage", "Bad Request", 400
+        )
+        client._api.send_message.side_effect = TelegramApiError(
+            "sendMessage", "Bad Request", 400
+        )
+
+        with patch("time.sleep"):
+            result = client.send_msg(title="结果", text="正文", rich_message="# 结果")
+
+        assert result["success"] is False
+
+
+class TestEditMsgFallsBackToPlainText:
+    """编辑路径同样需要降级，避免回复停留在旧内容。"""
+
+    def test_falls_back_when_rich_edit_fails(self):
+        client = _make_client()
+        client._api.edit_message_text.side_effect = [
+            TelegramApiError("editMessageText", "Bad Request", 400),
+            {"message_id": 5, "chat": {"id": "chat"}},
+        ]
+
+        result = client.edit_msg(
+            chat_id="chat", message_id=5, text="正文", rich_message="# 结果"
+        )
+
+        assert result is True
+        assert client._api.edit_message_text.call_count == 2
+        # 第二次调用以纯文本覆盖原消息
+        assert client._api.edit_message_text.call_args.kwargs["text"] == "# 结果"
+
+    def test_no_fallback_when_rich_edit_succeeds(self):
+        client = _make_client()
+        client._api.edit_message_text.return_value = {
+            "message_id": 5, "chat": {"id": "chat"}
+        }
+
+        result = client.edit_msg(
+            chat_id="chat", message_id=5, text="正文", rich_message="# 结果"
+        )
+
+        assert result is True
+        assert client._api.edit_message_text.call_count == 1
+
+
+class TestFallbackDisablesMarkdownV2:
+    """回归防护：降级路径不得回落到客户端默认的 MarkdownV2。
+
+    ``TelegramBotApi._parse_mode`` 在参数为 ``None`` 时使用 ``_default_parse_mode``，
+    而该值初始化为 MarkdownV2。若降级调用漏传 parse_mode，纯文本会被再次按
+    MarkdownV2 解析，重演 Rich Message 已经失败过的报错，降级等于无效。
+    """
+
+    def test_send_fallback_passes_empty_parse_mode(self):
+        client = _make_client()
+        client._api.send_rich_message.side_effect = TelegramApiError(
+            "sendRichMessage", "Bad Request", 400
+        )
+        client._api.send_message.return_value = {"message_id": 1, "chat": {"id": "chat"}}
+
+        with patch("time.sleep"):
+            client.send_msg(title="结果", text="正文", rich_message="# 结果")
+
+        kwargs = client._api.send_message.call_args.kwargs
+        assert kwargs["parse_mode"] == "", "降级发送必须显式禁用 MarkdownV2"
+
+    def test_edit_fallback_passes_empty_parse_mode(self):
+        client = _make_client()
+        client._api.edit_message_text.side_effect = [
+            TelegramApiError("editMessageText", "Bad Request", 400),
+            {"message_id": 5, "chat": {"id": "chat"}},
+        ]
+
+        client.edit_msg(chat_id="chat", message_id=5, text="正文", rich_message="# 结果")
+
+        kwargs = client._api.edit_message_text.call_args.kwargs
+        assert kwargs["parse_mode"] == "", "降级编辑必须显式禁用 MarkdownV2"