From 04898631cb72dbf84f5c066c3f87fc8c94df53a7 Mon Sep 17 00:00:00 2001 From: Gille <4317663+helix4u@users.noreply.github.com> Date: Fri, 10 Jul 2026 12:52:52 -0600 Subject: [PATCH] fix(telegram): recover final delivery after stream flood --- gateway/stream_consumer.py | 148 +++++++++++- plugins/platforms/telegram/adapter.py | 8 + tests/gateway/test_telegram_final_delivery.py | 216 ++++++++++++++++++ 3 files changed, 365 insertions(+), 7 deletions(-) create mode 100644 tests/gateway/test_telegram_final_delivery.py diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index a08e169f2f9..d345d8f27fd 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -982,6 +982,36 @@ class GatewayStreamConsumer: continuation = self._continuation_text(final_text) self._fallback_final_send = False if not continuation.strip(): + # Some platforms treat a successful streaming preview as durable + # delivery. Telegram clients can instead lose or retain only part + # of that preview after a failed final edit, so opt-in adapters + # commit the completed answer with a fresh final send. + if ( + final_text.strip() + and final_text == self._visible_prefix() + and getattr( + self.adapter, + "RESEND_FINAL_ON_EMPTY_STREAM_FALLBACK", + False, + ) is True + ): + delivery = await self._send_empty_fallback_final(final_text) + if delivery == "delivered": + return + self._already_sent = True + self._fallback_prefix = "" + self._fallback_preserve_partial_messages = False + if delivery == "ambiguous": + # A timeout may mean Telegram accepted the send but the + # client never received the response. Preserve duplicate + # suppression for that one uncertain outcome. + self._final_content_delivered = True + else: + # A confirmed failure leaves the gateway free to perform + # its normal final send. + self._final_response_sent = False + self._final_content_delivered = False + return # Nothing new to send — the visible partial already matches final text. # BUT: if final_text itself has meaningful content (e.g. a timeout # message after a long tool call), the prefix-based continuation @@ -1044,10 +1074,12 @@ class GatewayStreamConsumer: if result.success: break if attempt == 0 and self._is_flood_error(result): + delay = float(getattr(result, "retry_after", None) or 3.0) logger.debug( - "Flood control on fallback send, retrying in 3s" + "Flood control on fallback send, retrying in %.1fs", + delay, ) - await asyncio.sleep(3.0) + await asyncio.sleep(delay) else: break # non-flood error or second attempt failed @@ -1112,6 +1144,84 @@ class GatewayStreamConsumer: self._fallback_prefix = "" self._fallback_preserve_partial_messages = False + async def _send_empty_fallback_final(self, final_text: str) -> str: + """Commit a completed answer after Telegram finalization fails. + + Returns ``delivered`` on confirmed success, ``failed`` when the + gateway can safely retry, and ``ambiguous`` when a timeout may have + reached the platform already. + """ + stale_ids = set(self._preview_message_ids) + if self._message_id and self._message_id != "__no_edit__": + stale_ids.add(str(self._message_id)) + + result = None + for attempt in range(2): + try: + result = await self.adapter.send( + chat_id=self.chat_id, + content=final_text, + metadata=self._metadata_for_send(final=True), + ) + except Exception as exc: + logger.debug("Empty fallback final send failed: %s", exc) + return ( + "ambiguous" + if self._send_failure_may_have_delivered(exc) + else "failed" + ) + + if getattr(result, "success", False): + break + if attempt == 0 and self._is_flood_error(result): + delay = float(getattr(result, "retry_after", None) or 3.0) + logger.debug( + "Flood control on empty fallback final send; retrying in %.1fs", + delay, + ) + await asyncio.sleep(delay) + continue + return ( + "ambiguous" + if self._send_failure_may_have_delivered(result) + else "failed" + ) + + new_message_id = getattr(result, "message_id", None) + delete_fn = getattr(self.adapter, "delete_message", None) + if delete_fn is not None: + for stale_id in stale_ids: + if not stale_id or stale_id == new_message_id: + continue + try: + await delete_fn(self.chat_id, stale_id) + except Exception as exc: + logger.debug( + "Empty fallback preview cleanup failed (%s): %s", + stale_id, + exc, + ) + + self._preview_message_ids = set() + self._message_id = new_message_id or "__no_edit__" + self._already_sent = True + self._final_response_sent = True + self._final_content_delivered = True + self._last_sent_text = final_text + self._fallback_prefix = "" + self._fallback_preserve_partial_messages = False + self._notify_new_message() + return "delivered" + + @staticmethod + def _send_failure_may_have_delivered(result_or_exc: Any) -> bool: + """Return True for timeout failures where retrying may duplicate.""" + if getattr(result_or_exc, "retryable", None) is True: + return False + error = str(getattr(result_or_exc, "error", None) or result_or_exc).lower() + name = result_or_exc.__class__.__name__.lower() + return "timeout" in error or "timed out" in error or "timeout" in name + def _is_flood_error(self, result) -> bool: """Check if a SendResult failure is due to flood control / rate limiting.""" err = getattr(result, "error", "") or "" @@ -1246,11 +1356,12 @@ class GatewayStreamConsumer: if not prefix or not prefix.strip(): return try: - await self._edit_message( + result = await self._edit_message( message_id=self._message_id, content=prefix, ) - self._last_sent_text = prefix + if getattr(result, "success", False): + self._last_sent_text = prefix except Exception: pass # best-effort — don't let this block the fallback path @@ -1665,6 +1776,7 @@ class GatewayStreamConsumer: self._flood_strikes = 0 return True else: + immediate_final_fallback = False if ( finalize and is_turn_final @@ -1726,13 +1838,31 @@ class GatewayStreamConsumer: self._MAX_FLOOD_STRIKES, self._current_edit_interval, ) - if self._flood_strikes < self._MAX_FLOOD_STRIKES: + immediate_final_fallback = ( + finalize + and is_turn_final + and getattr( + self.adapter, + "FALLBACK_ON_FINAL_EDIT_FLOOD", + False, + ) is True + ) + if ( + self._flood_strikes < self._MAX_FLOOD_STRIKES + and not immediate_final_fallback + ): # Don't disable edits yet — just slow down. # Update _last_edit_time so the next edit # respects the new interval. self._last_edit_time = time.monotonic() return False + if immediate_final_fallback: + logger.debug( + "Turn-final edit hit flood control; " + "entering fallback immediately" + ) + # Non-flood error OR flood strikes exhausted: enter # fallback mode — send only the missing tail once the # final response is available. @@ -1745,8 +1875,12 @@ class GatewayStreamConsumer: self._edit_supported = False self._already_sent = True # Best-effort: strip the cursor from the last visible - # message so the user doesn't see a stuck ▉. - await self._try_strip_cursor() + # message so the user doesn't see a stuck ▉. A + # turn-final Telegram flood skips this cosmetic edit: + # another edit would consume the same flood budget and + # delay the fallback send that carries the answer. + if not immediate_final_fallback: + await self._try_strip_cursor() return False else: # Editing not supported — skip intermediate updates. diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 5d4c095e3fb..bd4308feb12 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -457,6 +457,14 @@ class TelegramAdapter(BasePlatformAdapter): # edit and the final edit, skipping the plain-text → MarkdownV2 conversion. # Fixes #25710. REQUIRES_EDIT_FINALIZE: bool = True + # Retrying a turn-final edit consumes more of the same Telegram flood + # budget while the completed answer remains undelivered. Move directly to + # the final fallback path instead. + FALLBACK_ON_FINAL_EDIT_FLOOD: bool = True + # A failed final edit can leave Telegram clients with only a partial or + # non-durable preview. Commit empty-tail fallbacks as a fresh final message + # instead of trusting the preview as completed delivery. + RESEND_FINAL_ON_EMPTY_STREAM_FALLBACK: bool = True # Adaptive text-batch ingress: short messages need a tighter delay so the # first token reaches the agent fast. Numbers tuned for "feels instant": diff --git a/tests/gateway/test_telegram_final_delivery.py b/tests/gateway/test_telegram_final_delivery.py new file mode 100644 index 00000000000..e3a7bbb9871 --- /dev/null +++ b/tests/gateway/test_telegram_final_delivery.py @@ -0,0 +1,216 @@ +"""Regression coverage for Telegram final delivery after streamed edit failure.""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.platforms.base import SendResult +from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig + + +def _adapter() -> MagicMock: + adapter = MagicMock() + adapter.REQUIRES_EDIT_FINALIZE = True + adapter.FALLBACK_ON_FINAL_EDIT_FLOOD = True + adapter.RESEND_FINAL_ON_EMPTY_STREAM_FALLBACK = True + adapter.MAX_MESSAGE_LENGTH = 4096 + adapter.edit_message = AsyncMock() + adapter.send = AsyncMock() + adapter.delete_message = AsyncMock(return_value=True) + return adapter + + +@pytest.mark.asyncio +async def test_turn_final_flood_immediately_delivers_missing_tail(): + """A short visible preview must not suppress the completed answer.""" + adapter = _adapter() + adapter.edit_message.return_value = SendResult( + success=False, + error="Flood control exceeded. Retry in 180 seconds", + retry_after=180.0, + ) + adapter.send.return_value = SendResult(success=True, message_id="tail-1") + + consumer = GatewayStreamConsumer( + adapter, + "chat-1", + StreamConsumerConfig(cursor=" ▉"), + metadata={"thread_id": "77"}, + ) + consumer._message_id = "preview-1" + consumer._last_sent_text = ":(" + consumer._already_sent = True + + ok = await consumer._send_or_edit( + ":( The completed answer follows here.", + finalize=True, + is_turn_final=True, + ) + + assert ok is False + assert consumer._flood_strikes == 1 + assert consumer._fallback_final_send is True + assert consumer.final_content_delivered is False + assert adapter.edit_message.await_count == 1 + + await consumer._send_fallback_final(":( The completed answer follows here.") + + adapter.send.assert_awaited_once() + assert adapter.send.await_args.kwargs["content"] == "The completed answer follows here." + assert adapter.send.await_args.kwargs["metadata"] == { + "thread_id": "77", + "notify": True, + } + adapter.delete_message.assert_not_awaited() + assert consumer.final_response_sent is True + assert consumer.final_content_delivered is True + + +@pytest.mark.asyncio +async def test_non_opt_in_adapter_keeps_adaptive_final_edit_retry(): + """Immediate final fallback remains scoped to opted-in adapters.""" + adapter = _adapter() + adapter.FALLBACK_ON_FINAL_EDIT_FLOOD = False + adapter.edit_message.return_value = SendResult( + success=False, + error="Flood control exceeded. Retry in 30 seconds", + retry_after=30.0, + ) + + consumer = GatewayStreamConsumer(adapter, "chat-1") + consumer._message_id = "preview-1" + consumer._last_sent_text = "partial" + consumer._already_sent = True + + ok = await consumer._send_or_edit( + "partial plus final", + finalize=True, + is_turn_final=True, + ) + + assert ok is False + assert consumer._flood_strikes == 1 + assert consumer._fallback_final_send is False + + +@pytest.mark.asyncio +async def test_turn_final_flood_commits_empty_tail_as_fresh_message(): + """Telegram gets a durable final even when the internal tail is empty.""" + adapter = _adapter() + adapter.edit_message.return_value = SendResult( + success=False, + error="Flood control exceeded. Retry in 30 seconds", + retry_after=30.0, + ) + adapter.send.return_value = SendResult(success=True, message_id="final-1") + + consumer = GatewayStreamConsumer( + adapter, + "chat-1", + StreamConsumerConfig(cursor=" ▉"), + ) + final_text = "The complete answer" + consumer._message_id = "preview-1" + consumer._preview_message_ids = {"preview-1"} + consumer._last_sent_text = f"{final_text} ▉" + consumer._already_sent = True + + ok = await consumer._send_or_edit( + final_text, + finalize=True, + is_turn_final=True, + ) + + assert ok is False + assert consumer._fallback_final_send is True + assert consumer.final_content_delivered is True + assert adapter.edit_message.await_count == 1 + + await consumer._send_fallback_final(final_text) + + adapter.send.assert_awaited_once() + assert adapter.send.await_args.kwargs["content"] == final_text + assert adapter.send.await_args.kwargs["metadata"] == {"notify": True} + adapter.delete_message.assert_awaited_once_with("chat-1", "preview-1") + assert consumer.message_id == "final-1" + assert consumer.final_response_sent is True + assert consumer.final_content_delivered is True + + +@pytest.mark.asyncio +async def test_empty_tail_commit_honors_retry_after(monkeypatch): + adapter = _adapter() + adapter.send.side_effect = [ + SendResult( + success=False, + error="Flood control exceeded", + retry_after=7.0, + ), + SendResult(success=True, message_id="final-1"), + ] + sleep = AsyncMock() + monkeypatch.setattr("gateway.stream_consumer.asyncio.sleep", sleep) + + consumer = GatewayStreamConsumer(adapter, "chat-1") + consumer._message_id = "preview-1" + consumer._last_sent_text = "Final answer" + consumer._fallback_final_send = True + + await consumer._send_fallback_final("Final answer") + + assert adapter.send.await_count == 2 + sleep.assert_awaited_once_with(7.0) + assert consumer.final_content_delivered is True + + +@pytest.mark.asyncio +async def test_ambiguous_empty_tail_timeout_preserves_duplicate_suppression(): + adapter = _adapter() + adapter.send.return_value = SimpleNamespace( + success=False, + error="Timed out", + retryable=False, + ) + + consumer = GatewayStreamConsumer(adapter, "chat-1") + consumer._message_id = "preview-1" + consumer._last_sent_text = "Final answer" + consumer._fallback_final_send = True + + await consumer._send_fallback_final("Final answer") + + adapter.delete_message.assert_not_awaited() + assert consumer.final_response_sent is False + assert consumer.final_content_delivered is True + + +@pytest.mark.asyncio +async def test_confirmed_empty_tail_send_failure_allows_gateway_retry(): + adapter = _adapter() + adapter.send.return_value = SendResult( + success=False, + error="network unavailable", + retryable=False, + ) + + consumer = GatewayStreamConsumer(adapter, "chat-1") + consumer._message_id = "preview-1" + consumer._last_sent_text = "Final answer" + consumer._fallback_final_send = True + consumer._final_content_delivered = True + + await consumer._send_fallback_final("Final answer") + + adapter.delete_message.assert_not_awaited() + assert consumer.final_response_sent is False + assert consumer.final_content_delivered is False + + +def test_timeout_exception_is_treated_as_ambiguous_delivery(): + class TimedOut(Exception): + pass + + assert GatewayStreamConsumer._send_failure_may_have_delivered( + TimedOut("request timed out") + ) is True