fix(telegram): wait for retry_after before retrying a new message (#7990)

* fix(telegram): honor retry_after when retrying a new message

A 429 on sendMessage or sendRichMessage slept 2**attempt and dropped
the reply before Telegram's flood wait expired. Wait the longer of
that backoff and the server hint, and leave other channels unchanged.

Fixes #7989

* fix(telegram): bound new reply flood waits

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -1467,6 +1467,7 @@ Notes:
 - Slack stop waits up to five seconds for an in-flight Socket Mode connection and its cleanup. If cleanup is still pending, the previous instance stays tracked and a restart can be retried after it finishes. SDK cleanup failures report their original error and remain retryable.
 - In Telegram private chats, text, attachment and streaming replies (including "Working on it...") can still be delivered if their referenced user or bot message was deleted, such as after clearing chat history during a run. Group reply-target requirements remain unchanged to preserve forum routing.
 - In Telegram groups, replying to a delivered bot response or attachment continues its DeerFlow conversation. Reply mappings survive restarts and stay scoped to the user's connection when one is bound. New, unrelated group messages still start separate conversations; private-chat routing is unchanged.
+- New Telegram final-reply sends use the bot API's rate-limit wait up to 60 seconds per retry. Missing, invalid, or shorter hints use the usual backoff. The "Working on it..." acknowledgment is best effort.
 - Feishu/Lark now queues rapid follow-up messages per mapped DeerFlow `thread_id` instead of immediately surfacing the generic busy reply, and topic replies keep a per-message card with a compact source-message preview across queued/running/final patches.
 - Streaming IM channels treat backend `error` events like transport failures and follow the same reply and retry path. The channel logs the error type and message for diagnosis.
 

backend/app/channels/AGENTS.md

@@ -47,7 +47,7 @@ Shared attachment feedback uses public `attachment_delivery_text`, preserves rou
 6. Slack/Discord chat: `runs.wait()` → extract final response → publish outbound
 6b. GitHub chat (`ChannelRunPolicy.fire_and_forget=True`): `runs.create()` returns once the run is `pending`; the manager does not wait for the final state and does not publish an outbound; the agent posts its own reply via sandbox `gh`. `ConflictError` on a busy thread still trips the standard `THREAD_BUSY_MESSAGE` path (log-only on GitHub); with `buffer_followups_on_busy=True` (GitHub's default) the triggering message is captured into a per-thread buffer so a concurrent comment is not silently dropped (see "Follow-up buffering while busy").
 7. Feishu patches one card per source (`config.update_multi=true`), with topic previews and queued → running → final states. Opt-in cards show context/question; bounded tokens preserve request ID/source. Originator/option checks consume once; consumption fails closed across workers (`feishu_card_interaction.py`). Card replies keep bound aliases in fields/metadata; ordinary replies preserve `/new`. Accepted intake retires only that card's buttons/text hint before terminal PATCH; rejected intake keeps text fallback. Non-final chunks store each thread alias once per reply; the final rewrites all.
-8. Telegram previews: 1s private/3s group, 20/min, 4096 chars; drop rate-limited updates, split final overflow. Only private sends (`chat_id > 0`) use `allow_sending_without_reply`; degrade missing targets atomically. Preserve group/forum routing/retries and durable owner aliases. Stream mapping errors log one traceback/answer, then debug; delivery/retries continue.
+8. Telegram previews: 1s private/3s group, 20/min, 4096 chars. Drop 429s. Split final overflow. Cap new final-send flood waits at 60s/retry. Only private chats set `allow_sending_without_reply`. Degrade missing targets atomically. Keep group/forum routing/retries and durable owner aliases. Log one stream mapping traceback per answer, then DEBUG. Continue delivery/retries.
 9. DingTalk AI Cards (`card_template_id`): `runs.stream()` → create card with initial text → `PUT /v1.0/card/streaming` → finalize (`is_final=True`). Creation/streaming failure uses `sampleMarkdown`; tables preserve escaped pipes and empty edge cells. Card calls get the channel token, not the SDK's.
 10. For commands (`/new`, `/status`, `/models`, `/model`, `/memory`, `/goal`, `/agent`, `/help`): handle locally or query Gateway API. `/agent list` reads the effective owner's Custom Agents; `/agent use <name>` validates in the same owner bucket, creates a new thread, and persists `channel_agent_name` in its metadata. A Custom Agent writes canonical `metadata.agent_name` so thread-search routes Web continuation to `/workspace/agents/<name>/chats/<thread_id>`; `lead_agent` omits it and stays on the ordinary chat route. 
 `/model <name>` persists `channel_model_name` in the *current* thread's metadata (in place, no fresh thread); `/model` reports the effective model, `/model default` clears it (stale pins self-clear).

backend/app/channels/base.py

@@ -250,7 +250,12 @@ class Channel(ABC):
         log_prefix: str | None = None,
         operation_name: str = "send",
     ) -> T:
-        """Run an outbound send operation with the shared channel retry policy."""
+        """Run an outbound send operation with the shared channel retry policy.
+
+        The wait between attempts is ``_outbound_retry_delay``. The default
+        is ``2**attempt``; a provider that is told how long to wait overrides
+        that without changing the attempt bound or the other channels.
+        """
         prefix = log_prefix or f"[{self.name}]"
         last_exc: Exception | None = None
         for attempt in range(max_retries):
@@ -259,9 +264,9 @@ class Channel(ABC):
             except Exception as exc:
                 last_exc = exc
                 if attempt < max_retries - 1:
-                    delay = 2**attempt
+                    delay = self._outbound_retry_delay(attempt, exc)
                     logger.warning(
-                        "%s %s failed (attempt %d/%d), retrying in %ds: %s",
+                        "%s %s failed (attempt %d/%d), retrying in %ss: %s",
                         prefix,
                         operation_name,
                         attempt + 1,
@@ -276,6 +281,16 @@ class Channel(ABC):
             raise RuntimeError(f"{self.name} {operation_name} failed without an exception from any attempt")
         raise last_exc
 
+    def _outbound_retry_delay(self, attempt: int, exc: Exception) -> float:
+        """Seconds to wait before the next attempt.
+
+        ``exc`` is the failure just caught. The shared policy ignores it and
+        sleeps ``2**attempt`` (1s, then 2s, then 4s, ...). A provider that
+        receives a server wait hint overrides this and may return a longer
+        delay; it must not shorten the local backoff.
+        """
+        return 2**attempt
+
     def _log_future_error(self, fut: Any, name: str, msg_id: Any) -> None:
         """Callback for concurrent futures scheduled from channel worker threads."""
         try:

backend/app/channels/telegram.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import math
 import re
 import threading
 import time
@@ -34,6 +35,9 @@ TELEGRAM_MAX_INBOUND_FILE_BYTES = 20_000_000
 # Keep Telegram cleanup inside the Gateway's five-second shutdown-hook budget.
 TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS = 4.0
 TELEGRAM_BRIDGE_DRAIN_TIMEOUT_SECONDS = 1.0
+# Outbound listeners await sends inline. Bound each flood wait while still
+# allowing a full group rate-limit window.
+TELEGRAM_MAX_OUTBOUND_RETRY_DELAY_SECONDS = 60.0
 STREAM_EDIT_MIN_INTERVAL_SECONDS = 1.0
 # Groups (negative chat_id) are capped at 20 messages/minute by Telegram,
 # so stream edits there must pace well below the private-chat 1 msg/s guideline.
@@ -967,6 +971,24 @@ class TelegramChannel(Channel):
             "last_text": last_text,
         }
 
+    def _outbound_retry_delay(self, attempt: int, exc: Exception) -> float:
+        """Extend the shared backoff for new plain or Rich final-reply sends.
+
+        Only sends through ``_send_with_retry`` use this policy. The running
+        acknowledgment and intermediate stream updates are best effort.
+        Final edits have their own retry path.
+        """
+        delay = super()._outbound_retry_delay(attempt, exc)
+        if not self._is_retry_after(exc):
+            return delay
+        try:
+            hint = self._retry_after_seconds(exc)
+        except (TypeError, ValueError, OverflowError):
+            return delay
+        if not math.isfinite(hint):
+            return delay
+        return max(delay, min(hint, TELEGRAM_MAX_OUTBOUND_RETRY_DELAY_SECONDS))
+
     @staticmethod
     def _is_retry_after(exc: Exception) -> bool:
         return getattr(exc, "retry_after", None) is not None
@@ -1008,7 +1030,7 @@ class TelegramChannel(Channel):
         return chunks
 
     async def _send_running_reply(self, chat_id: str, reply_to_message_id: int) -> None:
-        """Send a 'Working on it...' reply and register it as the stream target."""
+        """Send a best-effort acknowledgment before dispatching inbound work."""
         if not self._application:
             return
         try:

backend/tests/test_telegram_send_retry_after.py

@@ -0,0 +1,351 @@
+"""Telegram's final-reply sends honor bounded Bot API flood waits."""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import timedelta
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+from app.channels.base import Channel
+from app.channels.message_bus import MessageBus, OutboundMessage
+
+
+def _run(coro):
+    loop = asyncio.new_event_loop()
+    try:
+        return loop.run_until_complete(coro)
+    finally:
+        loop.close()
+
+
+def _flood(retry_after):
+    exc = Exception("Flood control exceeded")
+    exc.retry_after = retry_after
+    return exc
+
+
+class _OtherChannel(Channel):
+    """Stand-in for a non-Telegram provider that shares ``_send_with_retry``."""
+
+    def __init__(self, bus: MessageBus) -> None:
+        super().__init__(name="other", bus=bus, config={})
+
+    async def start(self) -> None:
+        return None
+
+    async def stop(self) -> None:
+        return None
+
+    async def send(self, msg: OutboundMessage) -> None:
+        return None
+
+
+def _channel_with_bot():
+    from app.channels.telegram import TelegramChannel
+
+    bus = MessageBus()
+    channel = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+    bot = SimpleNamespace(sent=[], edited=[], rich=[], next_message_id=100)
+
+    async def send_message(**kwargs):
+        bot.sent.append(kwargs)
+        result = MagicMock()
+        result.message_id = bot.next_message_id
+        bot.next_message_id += 1
+        return result
+
+    async def edit_message_text(**kwargs):
+        bot.edited.append(kwargs)
+        result = MagicMock()
+        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
+    channel._application = SimpleNamespace(bot=bot)
+    return channel, bot
+
+
+def _patch_retry_sleep(monkeypatch):
+    """The new-message retry sleeps in the shared helper, not in telegram.py."""
+    sleeps: list[float] = []
+
+    async def fake_sleep(delay):
+        sleeps.append(delay)
+
+    monkeypatch.setattr("app.channels.base.asyncio.sleep", fake_sleep)
+    return sleeps
+
+
+def _final(text: str, *, chat_id: str = "12345") -> OutboundMessage:
+    return OutboundMessage(channel_name="telegram", chat_id=chat_id, thread_id="t1", text=text, is_final=True)
+
+
+class TestTelegramNewMessageRetryAfter:
+    def test_plain_new_message_waits_for_retry_after_seconds(self, monkeypatch):
+        async def go():
+            channel, bot = _channel_with_bot()
+            sleeps = _patch_retry_sleep(monkeypatch)
+            real_send = bot.send_message
+            calls = {"n": 0}
+
+            async def flaky(**kwargs):
+                calls["n"] += 1
+                if calls["n"] == 1:
+                    raise _flood(41)
+                return await real_send(**kwargs)
+
+            bot.send_message = flaky
+            await channel.send(_final("hello"))
+
+            assert sleeps == [41.0]
+            assert calls["n"] == 2
+            assert [message["text"] for message in bot.sent] == ["hello"]
+            assert channel._last_bot_message["12345"] == 100
+
+        _run(go())
+
+    def test_plain_new_message_waits_for_retry_after_timedelta(self, monkeypatch):
+        async def go():
+            channel, bot = _channel_with_bot()
+            sleeps = _patch_retry_sleep(monkeypatch)
+            real_send = bot.send_message
+            calls = {"n": 0}
+
+            async def flaky(**kwargs):
+                calls["n"] += 1
+                if calls["n"] == 1:
+                    raise _flood(timedelta(seconds=17))
+                return await real_send(**kwargs)
+
+            bot.send_message = flaky
+            await channel.send(_final("hello"))
+
+            assert sleeps == [17.0]
+            assert [message["text"] for message in bot.sent] == ["hello"]
+
+        _run(go())
+
+    def test_rich_new_message_waits_for_retry_after(self, monkeypatch):
+        async def go():
+            channel, bot = _channel_with_bot()
+            channel.config["rich_messages"] = True
+            sleeps = _patch_retry_sleep(monkeypatch)
+            real_rich = bot.do_api_request
+            calls = {"n": 0}
+
+            async def flaky(endpoint, api_kwargs):
+                calls["n"] += 1
+                if calls["n"] == 1:
+                    raise _flood(12)
+                return await real_rich(endpoint, api_kwargs)
+
+            bot.do_api_request = flaky
+            await channel.send(_final("**answer**"))
+
+            assert sleeps == [12.0]
+            assert calls["n"] == 2
+            assert bot.sent == []
+            assert [call[0] for call in bot.rich] == ["sendRichMessage"]
+            assert channel._last_bot_message["12345"] == 100
+
+        _run(go())
+
+    def test_overflow_without_preview_waits_before_the_next_chunk(self, monkeypatch):
+        async def go():
+            channel, bot = _channel_with_bot()
+            sleeps = _patch_retry_sleep(monkeypatch)
+            real_send = bot.send_message
+            overflow = "b" * 10
+            failed = {"left": 1}
+
+            async def flaky(**kwargs):
+                if kwargs.get("text") == overflow and failed["left"] > 0:
+                    failed["left"] -= 1
+                    raise _flood(9)
+                return await real_send(**kwargs)
+
+            bot.send_message = flaky
+            await channel.send(_final("a" * 4096 + overflow))
+
+            assert sleeps == [9.0]
+            assert [message["text"] for message in bot.sent] == ["a" * 4096, overflow]
+            assert channel._last_bot_message["12345"] == 101
+
+        _run(go())
+
+    def test_overflow_after_stream_preview_waits_before_the_next_chunk(self, monkeypatch):
+        async def go():
+            channel, bot = _channel_with_bot()
+            sleeps = _patch_retry_sleep(monkeypatch)
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: 1000.0)
+            await channel._send_running_reply("12345", 42)
+
+            real_send = bot.send_message
+            overflow = "b" * 10
+            failed = {"left": 1}
+
+            async def flaky(**kwargs):
+                if kwargs.get("text") == overflow and failed["left"] > 0:
+                    failed["left"] -= 1
+                    raise _flood(8)
+                return await real_send(**kwargs)
+
+            bot.send_message = flaky
+            await channel.send(
+                OutboundMessage(
+                    channel_name="telegram",
+                    chat_id="12345",
+                    thread_id="t1",
+                    text="a" * 4096 + overflow,
+                    is_final=True,
+                    thread_ts="42",
+                )
+            )
+
+            assert sleeps == [8.0]
+            assert bot.edited[0]["text"] == "a" * 4096
+            assert [message["text"] for message in bot.sent] == ["Working on it...", overflow]
+            assert channel._last_bot_message["12345"] == 101
+            assert "12345:42" not in channel._stream_messages
+
+        _run(go())
+
+    @pytest.mark.parametrize(("hint", "expected_wait"), [(30, 30.0), (timedelta(hours=2), 60.0)])
+    def test_repeated_flood_wait_still_stops_at_the_attempt_bound(self, monkeypatch, hint, expected_wait):
+        async def go():
+            channel, bot = _channel_with_bot()
+            sleeps = _patch_retry_sleep(monkeypatch)
+            calls = {"n": 0}
+
+            async def always_limited(**kwargs):
+                calls["n"] += 1
+                raise _flood(hint)
+
+            bot.send_message = always_limited
+            with pytest.raises(Exception, match="Flood control exceeded"):
+                await channel.send(_final("hello"), _max_retries=3)
+
+            assert calls["n"] == 3
+            assert sleeps == [expected_wait, expected_wait]
+            assert bot.sent == []
+
+        _run(go())
+
+    def test_each_attempt_uses_that_response_hint(self, monkeypatch):
+        async def go():
+            channel, bot = _channel_with_bot()
+            sleeps = _patch_retry_sleep(monkeypatch)
+            real_send = bot.send_message
+            hints = [5, 20]
+
+            async def flaky(**kwargs):
+                if hints:
+                    raise _flood(hints.pop(0))
+                return await real_send(**kwargs)
+
+            bot.send_message = flaky
+            await channel.send(_final("hello"))
+
+            assert sleeps == [5.0, 20.0]
+            assert [message["text"] for message in bot.sent] == ["hello"]
+
+        _run(go())
+
+    def test_cancellation_during_the_flood_wait_is_not_swallowed(self, monkeypatch):
+        async def go():
+            channel, bot = _channel_with_bot()
+            sleeps: list[float] = []
+
+            async def cancel_sleep(delay):
+                sleeps.append(delay)
+                raise asyncio.CancelledError()
+
+            monkeypatch.setattr("app.channels.base.asyncio.sleep", cancel_sleep)
+            calls = {"n": 0}
+
+            async def limited(**kwargs):
+                calls["n"] += 1
+                raise _flood(40)
+
+            bot.send_message = limited
+            with pytest.raises(asyncio.CancelledError):
+                await channel.send(_final("hello"))
+
+            assert calls["n"] == 1
+            assert sleeps == [40.0]
+            assert bot.sent == []
+
+        _run(go())
+
+    @pytest.mark.parametrize(
+        "hint",
+        [0.25, float("inf"), float("nan"), "invalid", {}, 10**400],
+        ids=["short", "infinite", "nan", "nonnumeric", "wrong-type", "overflow"],
+    )
+    def test_unusable_hint_keeps_the_local_backoff(self, monkeypatch, hint):
+        async def go():
+            channel, bot = _channel_with_bot()
+            sleeps = _patch_retry_sleep(monkeypatch)
+            real_send = bot.send_message
+            calls = {"n": 0}
+
+            async def flaky(**kwargs):
+                calls["n"] += 1
+                if calls["n"] == 1:
+                    raise _flood(hint)
+                return await real_send(**kwargs)
+
+            bot.send_message = flaky
+            await channel.send(_final("hello"))
+
+            assert sleeps == [1]
+            assert [message["text"] for message in bot.sent] == ["hello"]
+
+        _run(go())
+
+    def test_ordinary_network_error_keeps_exponential_backoff(self, monkeypatch):
+        async def go():
+            channel, bot = _channel_with_bot()
+            sleeps = _patch_retry_sleep(monkeypatch)
+            real_send = bot.send_message
+            left = {"n": 2}
+
+            async def flaky(**kwargs):
+                if left["n"] > 0:
+                    left["n"] -= 1
+                    raise ConnectionError("transient")
+                return await real_send(**kwargs)
+
+            bot.send_message = flaky
+            await channel.send(_final("hello"))
+
+            assert sleeps == [1, 2]
+            assert [message["text"] for message in bot.sent] == ["hello"]
+
+        _run(go())
+
+    def test_other_channels_keep_ignoring_retry_after(self, monkeypatch):
+        sleeps = _patch_retry_sleep(monkeypatch)
+        channel = _OtherChannel(MessageBus())
+        attempts = {"n": 0}
+
+        async def flaky():
+            attempts["n"] += 1
+            if attempts["n"] < 3:
+                raise _flood(50)
+            return "sent"
+
+        assert _run(channel._send_with_retry(flaky, max_retries=3)) == "sent"
+        assert attempts["n"] == 3
+        assert sleeps == [1, 2]