fix(telegram): dedup saturated mid-stream overflow previews to stop flood-control edit storms (#58563)

Post-#48648, oversized mid-stream edits truncate to a 4096-char preview
instead of splitting. But when rich messages raise the consumer's overflow
budget to 32k, the consumer keeps accumulating past 4096 and keeps issuing
progressive edits every edit_interval — each one truncating to the SAME
preview text. Telegram counts every one of those no-op requests against the
flood budget: a long streamed reply fires ~1 identical edit per 0.8s for
the rest of the stream, trips flood control (200s+ penalties), and the
final delivery hangs behind inline flood sleeps. Users see the bot stuck
'streaming' and the chat unresponsive.

Fix at the chokepoint: track the last truncated preview per
(chat_id, message_id) and skip the API call when the new truncation is
identical. Previews still update when the visible prefix actually changes
(e.g. chunk-count marker 1/2 → 1/3). State clears on finalize and when
content shrinks back under the cap, so dedup can never mask a real edit.

Live repro: 19,956-char streamed reply, transport=edit, rich available —
4x flood-control hits within ~700ms, 250s penalties, hung final delivery.
E2E harness on the same stream: 14 edit calls on main vs 7 with the fix
(the delta is pure no-op duplicates; scales with stream length).

plugins/platforms/telegram/adapter.py

@@ -639,6 +639,14 @@ class TelegramAdapter(BasePlatformAdapter):
         # Tracks status bubbles owned by this adapter so subsequent calls with the
         # same key edit the same message instead of appending new ones (#30045).
         self._status_message_ids: Dict[tuple, str] = {}
+        # Last truncated mid-stream preview delivered per (chat_id, message_id).
+        # Once an oversized streaming edit saturates at the 4096 preview cap,
+        # every subsequent progressive edit truncates to the SAME text; sending
+        # it again is a no-op that still burns Telegram's flood budget (~1
+        # edit/0.8s × the rest of the stream ⇒ flood control with 200s+
+        # penalties, hanging final delivery). Dedup here so a saturated preview
+        # goes quiet until finalize. Bounded: entries are dropped on finalize.
+        self._last_overflow_preview: Dict[tuple, str] = {}
         # Background task that runs post-connect housekeeping (command-menu
         # registration + DM-topic setup) off the connect path so a slow Bot
         # API call (e.g. a set_my_commands stall for certain tokens) cannot
@@ -3855,12 +3863,32 @@ class TelegramAdapter(BasePlatformAdapter):
         # the next token chunk the full accumulated text is re-edited into the
         # continuation, triggering another split → infinite duplication loop
         # (#48648).  The full content is delivered when finalize=True.
+        _preview_key = (str(chat_id), str(message_id))
+        _saturated_preview = False
+        if finalize:
+            # Any saturation state for this message is finished with — the
+            # final edit always delivers real (full) content.
+            self._last_overflow_preview.pop(_preview_key, None)
         if utf16_len(content) > self.MAX_MESSAGE_LENGTH:
             if finalize:
                 return await self._edit_overflow_split(
                     chat_id, message_id, content, finalize=finalize, metadata=metadata,
                 )
             content = self._truncate_stream_overflow_preview(content)
+            _saturated_preview = True
+            # Saturated-preview dedup: past the cap, every progressive edit
+            # truncates to the same text. Re-sending it is a visual no-op that
+            # still burns flood budget (Telegram counts the request and answers
+            # "message is not modified"). ~1 edit/0.8s for the rest of a long
+            # stream trips flood control (200s+ penalties) and hangs the final
+            # delivery. Skip silently until finalize.
+            if self._last_overflow_preview.get(_preview_key) == content:
+                return SendResult(success=True, message_id=message_id)
+        elif not finalize:
+            # Content shrank back under the cap (segment break / new message
+            # id) — clear stale saturation state so dedup can't mask a real
+            # edit later.
+            self._last_overflow_preview.pop(_preview_key, None)
 
         try:
             if not finalize:
@@ -3869,6 +3897,8 @@ class TelegramAdapter(BasePlatformAdapter):
                     message_id=int(message_id),
                     text=content,
                 )
+                if _saturated_preview:
+                    self._last_overflow_preview[_preview_key] = content
                 return SendResult(success=True, message_id=message_id)
 
             formatted = self.format_message(content)
@@ -3916,11 +3946,15 @@ class TelegramAdapter(BasePlatformAdapter):
                     )
                 # Mid-stream: truncate and retry instead of splitting (#48648).
                 truncated = self._truncate_stream_overflow_preview(content)
+                if self._last_overflow_preview.get(_preview_key) == truncated:
+                    # Saturated-preview dedup (see pre-flight path above).
+                    return SendResult(success=True, message_id=message_id)
                 await self._bot.edit_message_text(
                     chat_id=normalize_telegram_chat_id(chat_id),
                     message_id=int(message_id),
                     text=truncated,
                 )
+                self._last_overflow_preview[_preview_key] = truncated
                 return SendResult(success=True, message_id=message_id)
             # Flood control / RetryAfter — short waits are retried inline,
             # long waits return a failure immediately so streaming can fall back

tests/gateway/test_telegram_format.py

@@ -1002,6 +1002,89 @@ class TestEditMessageStreamingSafety:
         edited_text = adapter._bot.edit_message_text.call_args.kwargs["text"]
         assert len(edited_text) <= adapter.MAX_MESSAGE_LENGTH
 
+    @pytest.mark.asyncio
+    async def test_saturated_preview_dedups_repeat_oversized_edits(self):
+        """Once a mid-stream preview saturates at the truncation cap, further
+        oversized edits truncate to the SAME text — re-sending them is a
+        visual no-op that burns Telegram's flood budget (~1 edit/0.8s for the
+        rest of a long stream ⇒ flood control with 200s+ penalties, hanging
+        final delivery). The adapter must skip identical saturated previews
+        without an API call."""
+        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
+        adapter._bot = MagicMock()
+        adapter._bot.edit_message_text = AsyncMock()
+        adapter._bot.send_message = AsyncMock()
+
+        # First oversized edit: delivers the truncated preview (1 API call).
+        r1 = await adapter.edit_message("123", "456", "x" * 6000, finalize=False)
+        assert r1.success is True
+        assert adapter._bot.edit_message_text.await_count == 1
+
+        # Stream keeps growing within the same chunk count: previews truncate
+        # identically — no API calls. (7000 and 8000 chars both truncate to
+        # the same "…(1/2)" preview; 9000 crosses into "(1/3)" — a real change
+        # that SHOULD be delivered, at most one edit per ~4096 chars of growth
+        # instead of one per 0.8s tick.)
+        for grow in (7000, 8000):
+            r = await adapter.edit_message("123", "456", "x" * grow, finalize=False)
+            assert r.success is True
+            assert r.message_id == "456"
+        assert adapter._bot.edit_message_text.await_count == 1, (
+            "identical saturated previews must not be re-sent"
+        )
+        # Chunk-count boundary: marker changes (1/2 → 1/3) — one real edit.
+        await adapter.edit_message("123", "456", "x" * 9000, finalize=False)
+        assert adapter._bot.edit_message_text.await_count == 2
+        # ...and saturates again at the new marker.
+        await adapter.edit_message("123", "456", "x" * 9100, finalize=False)
+        assert adapter._bot.edit_message_text.await_count == 2
+
+        # A DIFFERENT oversized prefix (content changed within the first 4096)
+        # must still go through.
+        await adapter.edit_message("123", "456", "y" * 9100, finalize=False)
+        assert adapter._bot.edit_message_text.await_count == 3
+
+    @pytest.mark.asyncio
+    async def test_saturated_preview_state_cleared_on_finalize(self):
+        """finalize=True delivers full content (split) and clears saturation
+        state, so a reused message id can't be masked by stale dedup."""
+        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
+        adapter._bot = MagicMock()
+        adapter._bot.edit_message_text = AsyncMock()
+        _next_id = [1000]
+
+        async def _fake_send(**kwargs):
+            _next_id[0] += 1
+            return SimpleNamespace(message_id=_next_id[0])
+
+        adapter._bot.send_message = AsyncMock(side_effect=_fake_send)
+
+        await adapter.edit_message("123", "456", "x" * 6000, finalize=False)
+        assert ("123", "456") in adapter._last_overflow_preview
+
+        result = await adapter.edit_message("123", "456", "x" * 6000, finalize=True)
+        assert result.success is True
+        # Finalize split-delivered (edit + continuation) and cleared the state.
+        assert adapter._bot.send_message.await_count >= 1
+        assert ("123", "456") not in adapter._last_overflow_preview
+
+    @pytest.mark.asyncio
+    async def test_saturation_state_cleared_when_content_shrinks(self):
+        """A same-id edit back under the cap (segment reset) clears saturation
+        state so later oversized edits aren't wrongly deduped."""
+        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
+        adapter._bot = MagicMock()
+        adapter._bot.edit_message_text = AsyncMock()
+        adapter._bot.send_message = AsyncMock()
+
+        await adapter.edit_message("123", "456", "x" * 6000, finalize=False)
+        assert ("123", "456") in adapter._last_overflow_preview
+        await adapter.edit_message("123", "456", "short", finalize=False)
+        assert ("123", "456") not in adapter._last_overflow_preview
+        # Oversized again → must be delivered, not deduped.
+        await adapter.edit_message("123", "456", "x" * 6000, finalize=False)
+        assert adapter._bot.edit_message_text.await_count == 3
+
     @pytest.mark.asyncio
     async def test_mid_stream_reactive_overflow_retries_truncated_edit(self):
         """If Telegram rejects a streaming edit as too long, retry with a