fix(gateway): clean up stale telegram stream previews after fresh fin…

gateway/run.py

@@ -19243,6 +19243,68 @@ async def _notify_long_running():
                             "Failed to edit streamed message for session %s: %s",
                             session_key or "?", _edit_err,
                         )
+            elif (
+                not _is_empty_sentinel
+                and _sc is not None
+                and getattr(_sc, "already_sent", False)
+                and getattr(_sc, "message_id", None)
+                and getattr(_sc, "message_id", None) != "__no_edit__"
+                and _status_adapter is not None
+                and hasattr(_status_adapter, "delete_message")
+                and session_key
+                and hasattr(_status_adapter, "register_post_delivery_callback")
+            ):
+                # A streamed preview bubble is visible, but final delivery was
+                # not confirmed (e.g. Telegram finalize edit hit long flood
+                # control and the gateway will now send a fresh final message).
+                # Delete the stale preview AFTER the fresh final message lands
+                # so the user sees one canonical answer instead of a preview
+                # bubble plus a near-duplicate final bubble.
+                _stale_preview_id = _sc.message_id
+                _adapter_snapshot = _status_adapter
+                _chat_id_snapshot = source.chat_id
+                _loop_snapshot = asyncio.get_running_loop()
+
+                def _cleanup_stale_stream_preview() -> None:
+                    async def _delete_preview() -> None:
+                        try:
+                            await _adapter_snapshot.delete_message(
+                                _chat_id_snapshot,
+                                _stale_preview_id,
+                            )
+                            logger.info(
+                                "Deleted stale streamed preview %s for session %s after fresh final send.",
+                                _stale_preview_id,
+                                session_key or "?",
+                            )
+                        except Exception as _delete_err:
+                            logger.debug(
+                                "Stale streamed preview cleanup failed for session %s (%s): %s",
+                                session_key or "?",
+                                _stale_preview_id,
+                                _delete_err,
+                            )
+
+                    try:
+                        safe_schedule_threadsafe(
+                            _delete_preview(),
+                            _loop_snapshot,
+                            logger=logger,
+                            log_message="Stale streamed preview cleanup scheduling error",
+                        )
+                    except Exception:
+                        pass
+
+                try:
+                    _status_adapter.register_post_delivery_callback(
+                        session_key,
+                        _cleanup_stale_stream_preview,
+                        generation=run_generation,
+                    )
+                except Exception as _rpe:
+                    logger.debug(
+                        "Stale preview cleanup registration failed: %s", _rpe
+                    )
 
         # Schedule deletion of tracked temporary progress bubbles after the
         # final response lands. Failed runs skip this so bubbles remain as

tests/gateway/test_run_progress_topics.py

@@ -121,6 +121,61 @@ async def edit_message(self, chat_id, message_id, content) -> SendResult:
         raise AssertionError("non-editable adapters should not receive edit_message calls")
 
 
+class FailingFinalizePreviewAdapter(ProgressCaptureAdapter):
+    """Simulate a visible streamed preview whose finalize edit fails.
+
+    This matches the Telegram failure mode behind duplicate near-final bubbles:
+    a preview message is already visible, but the final edit hits long flood
+    control so the gateway falls back to a fresh final send.
+    """
+
+    REQUIRES_EDIT_FINALIZE = True
+
+    def __init__(self, platform=Platform.TELEGRAM):
+        super().__init__(platform=platform)
+        self._next_id = 0
+        self.callbacks = {}
+        self.deleted = []
+
+    def _mint_id(self):
+        self._next_id += 1
+        return f"progress-{self._next_id}"
+
+    async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult:
+        self.sent.append(
+            {
+                "chat_id": chat_id,
+                "content": content,
+                "reply_to": reply_to,
+                "metadata": metadata,
+            }
+        )
+        return SendResult(success=True, message_id=self._mint_id())
+
+    async def edit_message(
+        self, chat_id, message_id, content, *, finalize: bool = False, metadata=None
+    ) -> SendResult:
+        self.edits.append(
+            {
+                "chat_id": chat_id,
+                "message_id": message_id,
+                "content": content,
+                "finalize": finalize,
+                "metadata": metadata,
+            }
+        )
+        if finalize:
+            return SendResult(success=False, error="flood_control:150.0", retryable=False)
+        return SendResult(success=True, message_id=message_id)
+
+    def register_post_delivery_callback(self, session_key, callback, *, generation=None):
+        self.callbacks[session_key] = (generation, callback)
+
+    async def delete_message(self, chat_id, message_id) -> SendResult:
+        self.deleted.append({"chat_id": chat_id, "message_id": message_id})
+        return SendResult(success=True, message_id=message_id)
+
+
 class FakeAgent:
     def __init__(self, **kwargs):
         # Capture anything passed via kwargs (older code path) but don't
@@ -620,6 +675,22 @@ def run_conversation(self, message, conversation_history=None, task_id=None):
         }
 
 
+class StreamingFinalizeFailureAgent:
+    def __init__(self, **kwargs):
+        self.stream_delta_callback = kwargs.get("stream_delta_callback")
+        self.tools = []
+
+    def run_conversation(self, message, conversation_history=None, task_id=None):
+        if self.stream_delta_callback:
+            self.stream_delta_callback("Partial streamed preview")
+        time.sleep(0.05)
+        return {
+            "final_response": "Partial streamed preview — completed final answer.",
+            "messages": [],
+            "api_calls": 1,
+        }
+
+
 class QueuedCommentaryAgent:
     calls = 0
 
@@ -918,6 +989,37 @@ async def test_run_agent_previewed_final_marks_already_sent(monkeypatch, tmp_pat
     assert [call["content"] for call in adapter.sent] == ["You're welcome."]
 
 
+@pytest.mark.asyncio
+async def test_run_agent_registers_cleanup_for_stale_stream_preview(monkeypatch, tmp_path):
+    adapter, result = await _run_with_agent(
+        monkeypatch,
+        tmp_path,
+        StreamingFinalizeFailureAgent,
+        session_id="sess-stale-stream-preview",
+        config_data={
+            "display": {"tool_progress": "off", "interim_assistant_messages": False},
+            "streaming": {"enabled": True, "edit_interval": 0.01, "buffer_threshold": 1},
+        },
+        platform=Platform.TELEGRAM,
+        chat_id="123456789",
+        chat_type="dm",
+        thread_id=None,
+        adapter_cls=FailingFinalizePreviewAdapter,
+    )
+
+    session_key = "agent:main:telegram:dm:123456789"
+    assert result.get("already_sent") is not True
+    assert session_key in adapter.callbacks, "expected stale preview cleanup callback"
+
+    _generation, callback = adapter.callbacks[session_key]
+    callback()
+    await asyncio.sleep(0.05)
+
+    assert adapter.deleted == [
+        {"chat_id": "123456789", "message_id": "progress-1"}
+    ]
+
+
 @pytest.mark.asyncio
 async def test_run_agent_matrix_streaming_omits_cursor(monkeypatch, tmp_path):
     adapter, result = await _run_with_agent(