fix(telegram): render final replies as Rich Messages (#4387)

* fix: render Telegram replies as rich messages

* Исправление fallback Rich Messages в Telegram

README.md

@@ -440,6 +440,8 @@ channels:
   telegram:
     enabled: true
     bot_token: $TELEGRAM_BOT_TOKEN
+    # Optional: render final Markdown replies as Telegram Rich Messages.
+    rich_messages: false
     allowed_users: []               # empty = allow all
 
   wechat:

backend/AGENTS.md

@@ -739,7 +739,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
   A swallowed streaming failure publishes its final outbound before releasing the inbound dedupe key, so a provider redelivery can retry without overtaking the terminal reply.
 - `base.py` - Abstract `Channel` base class (start/stop/send lifecycle)
 - `service.py` - Manages lifecycle of all configured channels from `config.yaml`
-- `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, and edits the "Working on it..." stream target in place via `editMessageText`; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured, and overrides `receive_file` to download inbound images (`picture`/`richText`) and documents (`file`) by `downloadCode` into the thread uploads bucket, mirroring `feishu.py`)
+- `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, edits the "Working on it..." stream target in place via `editMessageText`, and can optionally send final Markdown replies as Rich Messages through `channels.telegram.rich_messages`; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured, and overrides `receive_file` to download inbound images (`picture`/`richText`) and documents (`file`) by `downloadCode` into the thread uploads bucket, mirroring `feishu.py`)
 - `github.py` - Webhook-driven GitHub channel. Inbound messages come from `POST /api/webhooks/github`; outbound is log-only because GitHub agents post explicitly with `gh` from their sandbox when they choose to comment or create a PR
 - `app/gateway/routers/channel_connections.py` - Browser-facing user connection and disconnect APIs
 - `deerflow.persistence.channel_connections` - SQL-backed user-owned connection, optional credential, connect state, and conversation store
@@ -771,7 +771,7 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_
 - `langgraph_url` - LangGraph-compatible Gateway API base URL (default: `http://localhost:8001/api`)
 - `gateway_url` - Gateway API URL for auxiliary commands (default: `http://localhost:8001`)
 - In Docker Compose, IM channels run inside the `gateway` container, so `localhost` points back to that container. Use `http://gateway:8001/api` for `langgraph_url` and `http://gateway:8001` for `gateway_url`, or set `DEER_FLOW_CHANNELS_LANGGRAPH_URL` / `DEER_FLOW_CHANNELS_GATEWAY_URL`.
-- Per-channel configs: `feishu` (app_id, app_secret), `slack` (bot_token, app_token), `telegram` (bot_token), `dingtalk` (client_id, client_secret, optional `card_template_id` for AI Card streaming), `github` (operator kill-switch `enabled`, plus `default_mention_login` for mention-required GitHub triggers)
+- Per-channel configs: `feishu` (app_id, app_secret), `slack` (bot_token, app_token), `telegram` (bot_token, optional `rich_messages` for final Markdown Rich Messages), `dingtalk` (client_id, client_secret, optional `card_template_id` for AI Card streaming), `github` (operator kill-switch `enabled`, plus `default_mention_login` for mention-required GitHub triggers)
 
 **User-owned channel connections** (`config.yaml` -> `channel_connections`):
 - Disabled by default. It is a user-binding layer on top of the existing `channels.*` runtime config, not a replacement for provider bot credentials.

backend/app/channels/telegram.py

@@ -24,6 +24,7 @@ from deerflow.uploads.manager import is_upload_staging_file, normalize_filename
 logger = logging.getLogger(__name__)
 
 TELEGRAM_MAX_MESSAGE_LENGTH = 4096
+TELEGRAM_MAX_RICH_MESSAGE_LENGTH = 32768
 # Telegram's hosted Bot API documents this as 20 MB (decimal bytes).
 TELEGRAM_MAX_INBOUND_FILE_BYTES = 20_000_000
 # Keep Telegram cleanup inside the Gateway's five-second shutdown-hook budget.
@@ -188,10 +189,23 @@ class TelegramChannel(Channel):
 
         state = self._stream_messages.pop(key, None)
         if state is not None:
+            if self._can_send_rich(msg.text) and await self._edit_rich_message(chat_id, state["message_id"], msg.text):
+                self._last_bot_message[msg.chat_id] = state["message_id"]
+                return
             await self._finalize_stream_message(chat_id, msg.chat_id, state, msg.text)
             return
 
-        await self._send_new_message(chat_id, msg.chat_id, msg.text, _max_retries=_max_retries)
+        if self._can_send_rich(msg.text):
+            try:
+                message_id = await self._send_new_rich_message(chat_id, msg.chat_id, msg.text, _max_retries=_max_retries)
+            except Exception as exc:
+                logger.warning("[Telegram] Rich Message send failed in chat=%s; falling back to plain text: %s", chat_id, exc)
+                message_id = None
+            if message_id is not None:
+                return
+
+        for chunk in self._split_message(msg.text):
+            await self._send_new_message(chat_id, msg.chat_id, chunk, _max_retries=_max_retries)
 
     async def _send_stream_update(self, chat_id: int, key: str, text: str, reply_to: int | None = None) -> None:
         """Edit the in-flight streamed message with accumulated text.
@@ -285,6 +299,51 @@ class TelegramChannel(Channel):
                 return False
         return False
 
+    def _can_send_rich(self, text: str) -> bool:
+        return bool(self.config.get("rich_messages")) and 0 < len(text) <= TELEGRAM_MAX_RICH_MESSAGE_LENGTH
+
+    async def _edit_rich_message(self, chat_id: int, message_id: int, text: str) -> bool:
+        """Replace a streamed preview with a persistent Telegram Rich Message."""
+        from telegram.error import BadRequest, EndPointNotFound
+
+        bot = self._application.bot
+        data = {"chat_id": chat_id, "message_id": message_id, "rich_message": {"markdown": text}}
+        for attempt in range(2):
+            try:
+                await bot.do_api_request("editMessageText", api_kwargs=data)
+                return True
+            except Exception as exc:
+                if self._is_retry_after(exc) and attempt == 0:
+                    await asyncio.sleep(self._retry_after_seconds(exc))
+                    continue
+                if isinstance(exc, (BadRequest, EndPointNotFound)):
+                    logger.warning("[Telegram] Rich Message rejected in chat=%s; falling back to plain text: %s", chat_id, exc)
+                    return False
+                logger.warning("[Telegram] final rich edit failed in chat=%s: %s", chat_id, exc)
+                return False
+        return False
+
+    async def _send_new_rich_message(self, chat_id: int, chat_key: str, text: str, *, _max_retries: int) -> int | None:
+        """Send raw agent Markdown through Bot API 10.1 Rich Messages."""
+        from telegram.error import BadRequest, EndPointNotFound
+
+        bot = self._application.bot
+
+        async def send_message() -> int | None:
+            try:
+                result = await bot.do_api_request(
+                    "sendRichMessage",
+                    api_kwargs={"chat_id": chat_id, "rich_message": {"markdown": text}},
+                )
+            except (BadRequest, EndPointNotFound) as exc:
+                logger.warning("[Telegram] Rich Message rejected in chat=%s; falling back to plain text: %s", chat_id, exc)
+                return None
+            message_id = int(result["message_id"])
+            self._last_bot_message[chat_key] = message_id
+            return message_id
+
+        return await self._send_with_retry(send_message, max_retries=_max_retries, log_prefix="[Telegram]")
+
     async def _send_new_message(self, chat_id: int, chat_key: str, text: str, *, _max_retries: int = 3) -> int | None:
         """Send a fresh message with retry/backoff. Returns the sent message_id."""
         kwargs: dict[str, Any] = {"chat_id": chat_id, "text": text}

backend/tests/test_channels.py

@@ -8881,6 +8881,7 @@ class TestTelegramStreaming:
         bot = SimpleNamespace()
         bot.sent = []
         bot.edited = []
+        bot.rich = []
         bot.next_message_id = 100
 
         async def send_message(**kwargs):
@@ -8896,8 +8897,15 @@ class TestTelegramStreaming:
             result.message_id = kwargs["message_id"]
             return result
 
+        async def do_api_request(endpoint, api_kwargs):
+            bot.rich.append((endpoint, api_kwargs))
+            result = {"message_id": bot.next_message_id}
+            bot.next_message_id += 1
+            return result
+
         bot.send_message = send_message
         bot.edit_message_text = edit_message_text
+        bot.do_api_request = do_api_request
         mock_app.bot = bot
         ch._application = mock_app
         return ch, bot
@@ -9194,6 +9202,104 @@ class TestTelegramStreaming:
 
         _run(go())
 
+    def test_final_uses_telegram_rich_markdown_when_enabled(self):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+            ch.config["rich_messages"] = True
+            markdown = "# Result\n\n**Bold** and `code`"
+
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text=markdown, is_final=True))
+
+            assert bot.rich == [
+                (
+                    "sendRichMessage",
+                    {"chat_id": 12345, "rich_message": {"markdown": markdown}},
+                )
+            ]
+            assert bot.sent == []
+
+        _run(go())
+
+    def test_final_replaces_plain_stream_with_rich_message(self, monkeypatch):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+            ch.config["rich_messages"] = True
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: 1000.0)
+
+            await ch._send_running_reply("12345", 42)
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="**partial", is_final=False, thread_ts="42"))
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="**final**", is_final=True, thread_ts="42"))
+
+            assert bot.rich == [
+                (
+                    "editMessageText",
+                    {
+                        "chat_id": 12345,
+                        "message_id": 100,
+                        "rich_message": {"markdown": "**final**"},
+                    },
+                )
+            ]
+            assert [message["text"] for message in bot.sent] == ["Working on it..."]
+
+        _run(go())
+
+    def test_rich_message_bad_request_falls_back_to_plain_text_once(self):
+        from telegram.error import BadRequest
+
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+            ch.config["rich_messages"] = True
+
+            async def reject_rich(endpoint, api_kwargs):
+                raise BadRequest("Can't parse rich message")
+
+            bot.do_api_request = reject_rich
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="**answer**", is_final=True))
+
+            assert [message["text"] for message in bot.sent] == ["**answer**"]
+
+        _run(go())
+
+    def test_rich_message_retryable_failure_falls_back_to_plain_text_once(self):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+            ch.config["rich_messages"] = True
+
+            async def fail_rich(endpoint, api_kwargs):
+                raise RuntimeError("network failed")
+
+            bot.do_api_request = fail_rich
+            await ch.send(
+                OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="**answer**", is_final=True),
+                _max_retries=1,
+            )
+
+            assert [message["text"] for message in bot.sent] == ["**answer**"]
+
+        _run(go())
+
+    def test_stream_rich_message_bad_request_falls_back_to_one_plain_text_edit(self, monkeypatch):
+        from telegram.error import BadRequest
+
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+            ch.config["rich_messages"] = True
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: 1000.0)
+
+            await ch._send_running_reply("12345", 42)
+
+            async def fail_rich(endpoint, api_kwargs):
+                raise BadRequest("Can't parse rich message")
+
+            bot.do_api_request = fail_rich
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="**answer**", is_final=True, thread_ts="42"))
+
+            assert [message["text"] for message in bot.sent] == ["Working on it..."]
+            assert bot.edited == [{"chat_id": 12345, "message_id": 100, "text": "**answer**"}]
+
+        _run(go())
+
     def test_final_edit_retries_once_after_rate_limit(self, monkeypatch):
         async def go():
             ch, bot = self._make_channel_with_bot()

config.example.yaml

@@ -2057,6 +2057,7 @@ run_ownership:
 #     enabled: false
 #     bot_token: $TELEGRAM_BOT_TOKEN
 #     allowed_users: []               # empty = allow all
+#     rich_messages: false             # Bot API 10.1 Rich Messages for final Markdown responses
 #
 #   wechat:
 #     enabled: false