fix(gateway): cover all finalized stream sends

Apply the same metadata invariant to sealed split chunks, keep expect_edits on live previews, and exercise the real Telegram adapter path with rich messages enabled but rich drafts disabled.\n\nCredits PR #78525 by @Slobaka for the reproduced rich_messages/rich_drafts combination.

gateway/stream_consumer.py

@@ -1257,7 +1257,10 @@ class GatewayStreamConsumer:
                 chat_id=self.chat_id,
                 content=text,
                 reply_to=reply_to_id,
-                metadata=self._metadata_for_send(final=final, expect_edits=True),
+                metadata=self._metadata_for_send(
+                    final=final,
+                    expect_edits=not final,
+                ),
             )
             if result.success and result.message_id:
                 self._message_id = str(result.message_id)

tests/gateway/test_stream_consumer_draft.py

@@ -140,6 +140,37 @@ class TestDraftStreamingHappyPath:
         assert final_metadata.get("notify") is True
         assert "expect_edits" not in final_metadata
 
+    @pytest.mark.asyncio
+    async def test_edit_preview_still_marks_expect_edits(self):
+        adapter = _make_draft_capable_adapter(supports_draft=False)
+        cfg = StreamConsumerConfig(transport="edit", chat_type="dm", cursor="")
+        consumer = GatewayStreamConsumer(adapter, "12345", cfg)
+
+        delivered = await consumer._send_or_edit("Preview", finalize=False)
+
+        assert delivered is True
+        send_mock = adapter.__dict__["send"]
+        metadata = send_mock.call_args.kwargs.get("metadata") or {}
+        assert metadata.get("expect_edits") is True
+        assert "notify" not in metadata
+
+    @pytest.mark.asyncio
+    async def test_final_split_chunk_does_not_mark_expect_edits(self):
+        adapter = _make_draft_capable_adapter(supports_draft=False)
+        consumer = GatewayStreamConsumer(
+            adapter,
+            "12345",
+            StreamConsumerConfig(transport="edit", chat_type="dm", cursor=""),
+        )
+
+        message_id = await consumer._send_new_chunk("Sealed head", None, final=True)
+
+        assert message_id == "msg_real"
+        send_mock = adapter.__dict__["send"]
+        metadata = send_mock.call_args.kwargs.get("metadata") or {}
+        assert metadata.get("notify") is True
+        assert "expect_edits" not in metadata
+
 
 class TestDraftFallbackOnFailure:
     """When a draft frame fails, the consumer disables drafts for the rest

tests/gateway/test_telegram_rich_messages.py

@@ -18,6 +18,7 @@ import pytest
 
 from gateway.config import PlatformConfig
 from gateway.platforms.base import SendResult
+from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig
 from plugins.platforms.telegram.adapter import TelegramAdapter
 from telegram.error import BadRequest, NetworkError, TimedOut
 
@@ -384,6 +385,29 @@ def test_prefers_fresh_final_streaming_stays_disabled_when_rich_enabled():
     assert adapter.prefers_fresh_final_streaming(RICH_CONTENT) is False
 
 
+@pytest.mark.asyncio
+async def test_legacy_draft_stream_finalizes_with_persistent_rich_message():
+    """A MarkdownV2 draft must not force the persistent final to MarkdownV2."""
+    adapter = _make_adapter()  # rich messages on, rich drafts off
+    assert adapter.supports_draft_streaming(chat_type="dm") is True
+
+    consumer = GatewayStreamConsumer(
+        adapter,
+        "12345",
+        StreamConsumerConfig(transport="auto", chat_type="dm", cursor=""),
+    )
+    consumer._use_draft_streaming = True
+
+    delivered = await consumer._send_or_edit(RICH_CONTENT, finalize=True)
+
+    assert delivered is True
+    bot = adapter._bot
+    assert bot is not None
+    bot.do_api_request.assert_awaited_once()
+    assert bot.do_api_request.call_args.args[0] == "sendRichMessage"
+    bot.send_message.assert_not_called()
+
+
 # ----------------------------------------------------------------------
 # streaming_overflow_limit: with rich on, the stream consumer may accumulate up
 # to the 32,768-char rich cap before splitting, so a reply that fits one