fix(telegram): recover final delivery after stream flood

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.

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":

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