feat(gateway): stream Telegram edits safely

cli-config.yaml.example

@@ -500,6 +500,7 @@ group_sessions_per_user: true
 # Stream tokens to messaging platforms in real-time. The bot sends a message
 # on first token, then progressively edits it as more tokens arrive.
 # Disabled by default — enable to try the streaming UX on Telegram/Discord/Slack.
+# For Telegram, partial edits are sent as plain text and only the final edit uses MarkdownV2.
 streaming:
   enabled: false
   # transport: edit           # "edit" = progressive editMessageText

gateway/platforms/telegram.py

@@ -1515,6 +1515,14 @@ class TelegramAdapter(BasePlatformAdapter):
         if not self._bot:
             return SendResult(success=False, error="Not connected")
         try:
+            if not finalize:
+                await self._bot.edit_message_text(
+                    chat_id=int(chat_id),
+                    message_id=int(message_id),
+                    text=content,
+                )
+                return SendResult(success=True, message_id=message_id)
+
             formatted = self.format_message(content)
             try:
                 await self._bot.edit_message_text(

tests/gateway/test_telegram_format.py

@@ -716,3 +716,44 @@ async def test_send_escapes_chunk_indicator_for_markdownv2(adapter):
     assert len(sent_texts) > 1
     assert re.search(r" \\\([0-9]+/[0-9]+\\\)$", sent_texts[0])
     assert re.search(r" \\\([0-9]+/[0-9]+\\\)$", sent_texts[-1])
+
+
+# =========================================================================
+# edit_message — streaming Markdown safety
+# =========================================================================
+
+
+class TestEditMessageStreamingSafety:
+    @pytest.mark.asyncio
+    async def test_non_final_edit_uses_plain_text_without_markdown(self):
+        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
+        adapter._bot = MagicMock()
+        adapter._bot.edit_message_text = AsyncMock()
+
+        result = await adapter.edit_message("123", "456", "partial **bold", finalize=False)
+
+        assert result.success is True
+        adapter._bot.edit_message_text.assert_awaited_once_with(
+            chat_id=123,
+            message_id=456,
+            text="partial **bold",
+        )
+
+    @pytest.mark.asyncio
+    async def test_final_edit_uses_markdownv2_with_plain_fallback(self):
+        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
+        adapter._bot = MagicMock()
+        adapter._bot.edit_message_text = AsyncMock(side_effect=[Exception("bad markdown"), None])
+
+        result = await adapter.edit_message("123", "456", "final **bold**", finalize=True)
+
+        assert result.success is True
+        first_call = adapter._bot.edit_message_text.await_args_list[0].kwargs
+        second_call = adapter._bot.edit_message_text.await_args_list[1].kwargs
+        assert "parse_mode" in first_call
+        assert first_call["text"] == "final *bold*"
+        assert second_call == {
+            "chat_id": 123,
+            "message_id": 456,
+            "text": "final **bold**",
+        }