fix(telegram): avoid rich final duplicate previews (#46206)

tests/gateway/test_config.py

@@ -186,7 +186,7 @@ class TestStreamingConfig:
         )
         assert restored.edit_interval == 0.8
         assert restored.buffer_threshold == 24
-        assert restored.fresh_final_after_seconds == 60.0
+        assert restored.fresh_final_after_seconds == 0.0
 
 
 class TestGatewayConfigRoundtrip:
@@ -832,7 +832,7 @@ class TestLoadGatewayConfig:
 
         assert config.platforms[Platform.TELEGRAM].extra["rich_messages"] is False
 
-    def test_load_config_default_includes_telegram_rich_messages(self, tmp_path, monkeypatch):
+    def test_load_config_default_disables_telegram_rich_messages(self, tmp_path, monkeypatch):
         hermes_home = tmp_path / ".hermes"
         hermes_home.mkdir()
 
@@ -842,7 +842,7 @@ class TestLoadGatewayConfig:
 
         config = load_config()
 
-        assert config["telegram"]["extra"]["rich_messages"] is True
+        assert config["telegram"]["extra"]["rich_messages"] is False
 
     def test_bridges_telegram_extra_base_url_from_config_yaml(self, tmp_path, monkeypatch):
         hermes_home = tmp_path / ".hermes"

tests/gateway/test_stream_consumer.py

@@ -993,17 +993,16 @@ class TestFinalContentDeliveredGuard:
         requiring a second finalize edit even when content is unchanged."""
         adapter = MagicMock()
         adapter.REQUIRES_EDIT_FINALIZE = True  # Telegram adapter behavior
-        # First send (initial streaming message) succeeds
-        # Mid-stream finalize edit succeeds
-        # Final finalize edit FAILS (e.g. flood control on Telegram)
-        adapter.edit_message = AsyncMock(side_effect=[
-            SimpleNamespace(success=True),   # mid-stream edit
-            SimpleNamespace(success=True),   # finalize edit on line 548
-            SimpleNamespace(success=False),  # final finalize on line 580 (FAILS)
+        # First send (initial streaming message) succeeds.
+        # Mid-stream edit succeeds.
+        # Final finalize edit fails, and the consumer's own fallback send also
+        # fails, so no path has confirmed the complete final response reached
+        # the user.
+        adapter.edit_message = AsyncMock(return_value=SimpleNamespace(success=False))
+        adapter.send = AsyncMock(side_effect=[
+            SimpleNamespace(success=True, message_id="msg_1"),
+            SimpleNamespace(success=False, error="network down"),
         ])
-        adapter.send = AsyncMock(
-            return_value=SimpleNamespace(success=True, message_id="msg_1"),
-        )
         adapter.MAX_MESSAGE_LENGTH = 4096
 
         config = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5)
@@ -1013,6 +1012,10 @@ class TestFinalContentDeliveredGuard:
         consumer.on_delta("Part one of the response...\n")
         task = asyncio.create_task(consumer.run())
         await asyncio.sleep(0.05)
+        # Keep the second delta buffered until finish so the complete answer is
+        # not already visible before the final edit attempt fails.
+        consumer.cfg.buffer_threshold = 10_000
+        consumer._current_edit_interval = 10.0
 
         consumer.on_delta("Part two, the complete final answer.\n")
         await asyncio.sleep(0.05)

tests/gateway/test_stream_consumer_fresh_final.py

@@ -617,15 +617,15 @@ class TestStreamConsumerConfigFreshFinalField:
 class TestStreamingConfigFreshFinalField:
     """The gateway-level StreamingConfig carries the setting."""
 
-    def test_default_enables_with_60s(self):
+    def test_default_is_disabled(self):
         from gateway.config import StreamingConfig
         cfg = StreamingConfig()
-        assert cfg.fresh_final_after_seconds == 60.0
+        assert cfg.fresh_final_after_seconds == 0.0
 
     def test_from_dict_uses_default_when_missing(self):
         from gateway.config import StreamingConfig
         cfg = StreamingConfig.from_dict({"enabled": True})
-        assert cfg.fresh_final_after_seconds == 60.0
+        assert cfg.fresh_final_after_seconds == 0.0
 
     def test_from_dict_respects_explicit_zero(self):
         from gateway.config import StreamingConfig

tests/gateway/test_telegram_rich_messages.py

@@ -48,7 +48,11 @@ PTB_INVALID_TOKEN_404 = InvalidToken(
 
 def _make_adapter(extra=None):
     """Build a TelegramAdapter with a mock bot wired for the rich path."""
-    config = PlatformConfig(enabled=True, token="fake-token", extra=extra or {})
+    config = PlatformConfig(
+        enabled=True,
+        token="fake-token",
+        extra={"rich_messages": True, **(extra or {})},
+    )
     adapter = TelegramAdapter(config)
     bot = MagicMock()
     # do_api_request as an AsyncMock makes inspect.iscoroutinefunction(...) True,
@@ -180,16 +184,22 @@ async def test_rich_messages_opt_out_accepts_string_false():
 
 
 @pytest.mark.asyncio
-async def test_rich_messages_default_is_enabled():
-    adapter = _make_adapter()
+async def test_rich_messages_default_is_disabled():
+    config = PlatformConfig(enabled=True, token="fake-token")
+    adapter = TelegramAdapter(config)
+    bot = MagicMock()
+    bot.do_api_request = AsyncMock(return_value=SimpleNamespace(message_id=123))
+    bot.send_message = AsyncMock(return_value=MagicMock(message_id=1))
+    bot.send_chat_action = AsyncMock()
+    adapter._bot = bot
 
     result = await adapter.send("12345", RICH_CONTENT)
 
     assert result.success is True
     bot = adapter._bot
     assert bot is not None
-    bot.do_api_request.assert_awaited_once()
-    bot.send_message.assert_not_called()
+    bot.do_api_request.assert_not_called()
+    bot.send_message.assert_awaited()
 
 
 @pytest.mark.asyncio
@@ -519,14 +529,13 @@ async def test_rich_draft_oversized_uses_legacy():
 
 
 # ----------------------------------------------------------------------
-# prefers_fresh_final_streaming: the stream consumer asks the adapter whether
-# to finalize a streamed reply by sending a fresh (rich) message + deleting the
-# preview, instead of final-editing the preview through the non-rich edit path.
-# Telegram opts in exactly when the content is rich-eligible.
+# prefers_fresh_final_streaming: Telegram keeps streamed finals on the edit
+# path, even when rich messages are enabled, so users do not briefly see two
+# copies of the answer while the preview cleanup delete races the fresh send.
 # ----------------------------------------------------------------------
-def test_prefers_fresh_final_streaming_when_rich_enabled():
+def test_prefers_fresh_final_streaming_stays_disabled_when_rich_enabled():
     adapter = _make_adapter()
-    assert adapter.prefers_fresh_final_streaming(RICH_CONTENT) is True
+    assert adapter.prefers_fresh_final_streaming(RICH_CONTENT) is False
 
 
 def test_prefers_fresh_final_streaming_honors_rich_opt_out():