fix(telegram): upgrade streaming preview to rich in place at stream end

The rich branch in send_delta(stream_end=True) was unreachable: it was
guarded by 'not buf.message_id' after an earlier return had already
ensured buf.message_id is set, so sendRichMessage never fired with
streaming enabled and the final message always went through the legacy
HTML editMessageText path.

Bot API 10.1 added a rich_message parameter to editMessageText, which
upgrades an existing message to rich in place. Use it at stream end via
do_api_request so the streaming preview keeps its identity — no
delete-and-resend, so none of the flickering or dropped line breaks
that made rich-at-stream-end fail in #4470.

Capability errors (server older than 10.1) trip the existing
_rich_send_disabled latch and fall back to the legacy HTML path, which
is unchanged.

Fixes #5516

nanobot/channels/telegram/runtime.py

@@ -897,6 +897,47 @@ class TelegramChannel(BaseChannel):
             self.logger.debug("sendRichMessage failed: {}", exc)
             return False
 
+    async def _try_edit_rich(self, chat_id: int, message_id: int, content: str) -> bool:
+        """Upgrade an existing message to rich in place via editMessageText (Bot API 10.1).
+
+        Editing in place keeps the message identity, so the streaming preview is
+        upgraded without the delete-and-resend pattern that caused flickering and
+        dropped line breaks (issue #4470). Returns True on success; on capability
+        errors (server older than Bot API 10.1) the rich latch is tripped so the
+        legacy HTML path is used from then on.
+        """
+        if not self._app:
+            return False
+
+        payload: dict[str, Any] = {
+            "chat_id": chat_id,
+            "message_id": message_id,
+            "rich_message": {
+                "markdown": content,
+            },
+        }
+        try:
+            await self._call_with_retry(
+                self._app.bot.do_api_request,
+                "editMessageText",
+                api_kwargs=payload,
+            )
+            return True
+        except BadRequest as exc:
+            if self._is_rich_capability_error(exc):
+                self.logger.debug("editMessageText rich_message not available, disabling")
+                self._rich_send_disabled = True
+            else:
+                self.logger.debug("editMessageText rich_message rejected: {}", exc)
+            return False
+        except Exception as exc:
+            err_str = str(exc).lower()
+            if "timed out" in err_str or isinstance(exc, TimedOut):
+                self.logger.debug("editMessageText rich_message timeout, falling back to legacy path")
+                return False
+            self.logger.debug("editMessageText rich_message failed: {}", exc)
+            return False
+
     async def send(self, msg: OutboundMessage) -> None:
         """Send a message through Telegram."""
         app = await self._wait_for_app()
@@ -1136,26 +1177,16 @@ class TelegramChannel(BaseChannel):
                 thread_kwargs["message_thread_id"] = message_thread_id
             raw_text = buf.text
 
-            # Try sendRichMessage for final output (Bot API 10.1).
-            # Skip when a streaming preview already exists to avoid the
-            # delete-and-resend pattern that causes flickering and drops
-            # line breaks (issue #4470).
-            if not buf.message_id and self.config.rich_messages and not getattr(self, "_rich_send_disabled", False):
-                reply_params = None
-                if reply_to_message_id := meta.get("message_id"):
-                    reply_params = {"message_id": int(reply_to_message_id), "allow_sending_without_reply": True}
-                rich_ok = await self._try_send_rich(
-                    int_chat_id, raw_text, reply_params, thread_kwargs, None,
-                )
+            # Try upgrading the streaming preview to rich in place (Bot API 10.1:
+            # editMessageText gained a rich_message parameter). Editing in place
+            # keeps the message identity, so there is no delete-and-resend and
+            # none of the flickering / dropped line breaks from issue #4470.
+            # The previous branch here was unreachable: it was guarded by
+            # ``not buf.message_id`` after an early return had already ensured
+            # ``buf.message_id`` is set (issue #5516).
+            if self.config.rich_messages and not getattr(self, "_rich_send_disabled", False):
+                rich_ok = await self._try_edit_rich(int_chat_id, buf.message_id, raw_text)
                 if rich_ok:
-                    # Delete the streaming preview message
-                    try:
-                        await self._call_with_retry(
-                            app.bot.delete_message,
-                            chat_id=int_chat_id, message_id=buf.message_id,
-                        )
-                    except Exception:
-                        pass  # Preview stays if delete fails
                     self._stream_bufs.pop(chat_id, None)
                     return
 

nanobot/channels/telegram/tests/test_telegram_channel.py

@@ -2735,3 +2735,76 @@ def test_markdown_to_html_code_block_same_line_no_newline() -> None:
 
     stripped = _strip_md_block(text)
     assert stripped == "Use <tag> here"
+
+
+@pytest.mark.asyncio
+async def test_send_delta_stream_end_upgrades_preview_to_rich_in_place() -> None:
+    """Rich messages finally work with streaming: the preview is upgraded via
+    editMessageText rich_message (in place), not delete-and-resend (issue #5516)."""
+    from telegram.error import BadRequest
+
+    channel = TelegramChannel(
+        TelegramConfig(enabled=True, token="123:abc", allow_from=["*"], rich_messages=True),
+        MessageBus(),
+    )
+    _install_ready_app(channel)
+    channel._app.bot.do_api_request = AsyncMock()
+    channel._app.bot.edit_message_text = AsyncMock(side_effect=BadRequest("should not be reached"))
+    channel._stream_bufs["123"] = _StreamBuf(text="**hello**", message_id=7, last_edit=0.0)
+
+    await channel.send_delta("123", "", stream_end=True)
+
+    # editMessageText with rich_message payload, in place (same message_id)
+    channel._app.bot.do_api_request.assert_awaited_once()
+    args, kwargs = channel._app.bot.do_api_request.await_args
+    assert args[0] == "editMessageText"
+    assert kwargs["api_kwargs"]["chat_id"] == 123
+    assert kwargs["api_kwargs"]["message_id"] == 7
+    assert kwargs["api_kwargs"]["rich_message"] == {"markdown": "**hello**"}
+    # No delete-and-resend, no legacy HTML edit
+    channel._app.bot.edit_message_text.assert_not_awaited()
+    assert "123" not in channel._stream_bufs
+
+
+@pytest.mark.asyncio
+async def test_send_delta_stream_end_rich_capability_error_latches_and_falls_back() -> None:
+    """On a pre-10.1 Bot API server the rich edit fails, the latch trips, and the
+    legacy HTML edit handles the final output."""
+    from telegram.error import BadRequest
+
+    channel = TelegramChannel(
+        TelegramConfig(enabled=True, token="123:abc", allow_from=["*"], rich_messages=True),
+        MessageBus(),
+    )
+    _install_ready_app(channel)
+    channel._app.bot.do_api_request = AsyncMock(side_effect=BadRequest("Method not found"))
+    channel._app.bot.edit_message_text = AsyncMock()
+    channel._stream_bufs["123"] = _StreamBuf(text="hello", message_id=7, last_edit=0.0)
+
+    await channel.send_delta("123", "", stream_end=True)
+
+    channel._app.bot.do_api_request.assert_awaited_once()
+    # Latch tripped: subsequent sends skip the rich path entirely
+    assert channel._rich_send_disabled is True
+    # Legacy HTML edit handled the final message
+    channel._app.bot.edit_message_text.assert_awaited_once()
+    assert "123" not in channel._stream_bufs
+
+
+@pytest.mark.asyncio
+async def test_send_delta_stream_end_rich_disabled_uses_legacy_html() -> None:
+    """rich_messages=False (the default) keeps the legacy HTML path untouched."""
+    channel = TelegramChannel(
+        TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
+        MessageBus(),
+    )
+    _install_ready_app(channel)
+    channel._app.bot.do_api_request = AsyncMock()
+    channel._app.bot.edit_message_text = AsyncMock()
+    channel._stream_bufs["123"] = _StreamBuf(text="hello", message_id=7, last_edit=0.0)
+
+    await channel.send_delta("123", "", stream_end=True)
+
+    channel._app.bot.do_api_request.assert_not_called()
+    channel._app.bot.edit_message_text.assert_awaited_once()
+    assert "123" not in channel._stream_bufs