diff --git a/README.md b/README.md index 8f7906862..0c82fcdf3 100644 --- a/README.md +++ b/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: diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 1b996f014..9e7ccaa8f 100644 --- a/backend/AGENTS.md +++ b/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. diff --git a/backend/app/channels/telegram.py b/backend/app/channels/telegram.py index eca123f26..2bcff0670 100644 --- a/backend/app/channels/telegram.py +++ b/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} diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index 4058403d2..319557e07 100644 --- a/backend/tests/test_channels.py +++ b/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() diff --git a/config.example.yaml b/config.example.yaml index dafd8dd2b..4b732d55b 100644 --- a/config.example.yaml +++ b/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