Telegram streaming finalize and MarkdownV2 formatting fixes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #25923 fix(telegram): streaming finalize — drop preview-deletion race, finalize MarkdownV2, accept nested config (salvage of #13542, #25724, #25694)

  merged to main

  files: gateway/config.py, gateway/platforms/telegram.py, gateway/run.py, gateway/stream_consumer.py, tests/gateway/test_duplicate_reply_suppression.py, tests/test_gateway_streaming_nested_config.py

  Show the diff

  • gateway/config.py

    @@ -735,6 +735,10 @@ def load_gateway_config() -> GatewayConfig:
                     gw_data["thread_sessions_per_user"] = yaml_cfg["thread_sessions_per_user"]
     
                 streaming_cfg = yaml_cfg.get("streaming")
    +            if not isinstance(streaming_cfg, dict):
    +                # Fall back to nested gateway.streaming written by
    +                # ``hermes config set gateway.streaming.*``
    +                streaming_cfg = yaml_cfg.get("gateway", {}).get("streaming")
                 if isinstance(streaming_cfg, dict):
                     gw_data["streaming"] = streaming_cfg
     

  • gateway/platforms/telegram.py

    @@ -332,6 +332,13 @@ class TelegramAdapter(BasePlatformAdapter):
         MEDIA_GROUP_WAIT_SECONDS = 0.8
         _GENERAL_TOPIC_THREAD_ID = "1"
     
    +    # Telegram's edit_message applies MarkdownV2 formatting only on the
    +    # finalize=True path.  Without this flag, stream_consumer._send_or_edit
    +    # short-circuits when the raw text is unchanged between the last streamed
    +    # edit and the final edit, skipping the plain-text → MarkdownV2 conversion.
    +    # Fixes #25710.
    +    REQUIRES_EDIT_FINALIZE: bool = True
    +
         # Adaptive text-batch ingress: short messages need a tighter delay so the
         # first token reaches the agent fast.  Numbers tuned for "feels instant":
         # ≤320 codepoints (one short paragraph) settles in ~180ms; ≤1024

  • gateway/run.py

    @@ -16131,6 +16131,7 @@ async def _notify_long_running():
                         _already_streamed = bool(
                             (_sc and getattr(_sc, "final_response_sent", False))
                             or _previewed
    +                        or (_sc and getattr(_sc, "final_content_delivered", False))
                         )
                         first_response = result.get("final_response", "")
                         if first_response and not _already_streamed:
    @@ -16292,12 +16293,16 @@ async def _notify_long_running():
                 # response_previewed means the interim_assistant_callback already
                 # sent the final text via the adapter (non-streaming path).
                 _previewed = bool(response.get("response_previewed"))
    -            if not _is_empty_sentinel and (_streamed or _previewed):
    +            _content_delivered = bool(
    +                _sc and getattr(_sc, "final_content_delivered", False)
    +            )
    +            if not _is_empty_sentinel and (_streamed or _previewed or _content_delivered):
                     logger.info(
    -                    "Suppressing normal final send for session %s: final delivery already confirmed (streamed=%s previewed=%s).",
    +                    "Suppressing normal final send for session %s: final delivery already confirmed (streamed=%s previewed=%s content_delivered=%s).",
                         session_key or "?",
                         _streamed,
                         _previewed,
    +                    _content_delivered,
                     )
                     response["already_sent"] = True
     

  • gateway/stream_consumer.py

    @@ -150,6 +150,10 @@ def __init__(
             self._flood_strikes = 0         # Consecutive flood-control edit failures
             self._current_edit_interval = self.cfg.edit_interval  # Adaptive backoff
             self._final_response_sent = False
    +        # Set when the final response content was sent to the user via
    +        # streaming, even if the final edit (cursor removal etc.)
    +        # subsequently failed.
    +        self._final_content_delivered = False
             # Cache adapter lifecycle capability: only platforms that need an
             # explicit finalize call (e.g. DingTalk AI Cards) force us to make
             # a redundant final edit.  Everyone else keeps the fast path.
    @@ -187,6 +191,12 @@ def final_response_sent(self) -> bool:
             """True when the stream consumer delivered the final assistant reply."""
             return self._final_response_sent
     
    +    @property
    +    def final_content_delivered(self) -> bool:
    +        """True when the final response content reached the user, even if
    +        the subsequent cosmetic edit (cursor removal) failed."""
    +        return self._final_content_delivered
    +
         def on_segment_break(self) -> None:
             """Finalize the current stream segment and start a fresh message."""
             self._queue.put(_NEW_SEGMENT)
    @@ -455,6 +465,8 @@ async def run(self) -> None:
                                 # tool-progress edits or fallback-mode promotion (#10748)
                                 # — that doesn't mean the final answer reached the user.
                                 self._final_response_sent = chunks_delivered
    +                            if chunks_delivered:
    +                                self._final_content_delivered = True
                                 return
                             if got_segment_break:
                                 self._message_id = None
    @@ -505,6 +517,11 @@ async def run(self) -> None:
                         self._last_edit_time = time.monotonic()
     
                     if got_done:
    +                    # Record that the final content reached the user even
    +                    # if the cosmetic final edit below fails.
    +                    if current_update_visible and self._accumulated:
    +                        self._final_content_delivered = True
    +
                         # Final edit without cursor. If progressive editing failed
                         # mid-stream, send a single continuation/fallback message
                         # here instead of letting the base gateway path send the

  • tests/gateway/test_duplicate_reply_suppression.py

    @@ -467,3 +467,59 @@ def test_old_behavior_would_have_promoted_partial(self):
                 final_response_sent = True
     
             assert final_response_sent is True  # the bug: partial promoted to final
    +
    +
    +class TestFinalContentDeliveredSuppression:
    +    """When stream consumer delivered the final content but the cosmetic
    +    final edit (cursor removal) failed, the gateway must suppress the
    +    fallback send to prevent duplicate messages.
    +
    +    Covers the scenario not handled by final_response_sent alone:
    +    content reached the user via _send_or_edit, but the subsequent edit
    +    that clears a typing cursor or streaming marker failed, leaving
    +    final_response_sent=False even though the user already saw the text.
    +    """
    +
    +    def test_content_delivered_but_final_edit_failed_suppresses(self):
    +        """final_content_delivered=True + final_response_sent=False
    +        must suppress (content already visible to user)."""
    +        sc = SimpleNamespace(
    +            already_sent=True,
    +            final_response_sent=False,
    +            final_content_delivered=True,
    +        )
    +        response = {"final_response": "Hello!", "response_previewed": False}
    +
    +        _streamed = bool(getattr(sc, "final_response_sent", False))
    +        _previewed = bool(response.get("response_previewed"))
    +        _content_delivered = bool(getattr(sc, "final_content_delivered", False))
    +        _is_empty_sentinel = (
    +            not response.get("final_response")
    +            or response.get("final_response") == "(empty)"
    +        )
    +        if not _is_empty_sentinel and (_streamed or _previewed or _content_delivered):
    +            response["already_sent"] = True
    +
    +        assert response.get("already_sent") is True
    +
    +    def test_intermediate_text_only_does_not_suppress(self):
    +        """already_sent=True from intermediate text + final_content_delivered=False
    +        must NOT suppress (user still needs the real final answer)."""
    +        sc = SimpleNamespace(
    +            already_sent=True,
    +            final_response_sent=False,
    +            final_content_delivered=False,
    +        )
    +        response = {"final_response": "Real answer", "response_previewed": False}
    +
    +        _streamed = bool(getattr(sc, "final_response_sent", False))
    +        _previewed = bool(response.get("response_previewed"))
    +        _content_delivered = bool(getattr(sc, "final_content_delivered", False))
    +        _is_empty_sentinel = (
    +            not response.get("final_response")
    +            or response.get("final_response") == "(empty)"
    +        )
    +        if not _is_empty_sentinel and (_streamed or _previewed or _content_delivered):
    +            response["already_sent"] = True
    +
    +        assert "already_sent" not in response

  • tests/test_gateway_streaming_nested_config.py

    @@ -0,0 +1,46 @@
    +"""Regression test for #25676 — nested gateway.streaming config must be loaded."""
    +from pathlib import Path
    +from unittest.mock import patch, MagicMock
    +import json
    +
    +import pytest
    +import yaml
    +
    +
    +def _load_with_yaml_dict(yaml_dict: dict):
    +    """Patch filesystem so load_gateway_config() sees *yaml_dict* as config.yaml."""
    +    from gateway.config import load_gateway_config
    +
    +    fake_home = Path("/tmp/fake_hermes_home_25676")
    +
    +    def fake_exists(self):
    +        return str(self).endswith("config.yaml")
    +
    +    with patch("gateway.config.get_hermes_home", return_value=fake_home), \
    +         patch.object(Path, "exists", fake_exists), \
    +         patch("builtins.open", create=True) as mock_file:
    +        mock_file.return_value.__enter__ = lambda s: s
    +        mock_file.return_value.__exit__ = MagicMock(return_value=False)
    +        with patch("yaml.safe_load", return_value=yaml_dict):
    +            return load_gateway_config()
    +
    +
    +class TestStreamingConfigNested:
    +    def test_top_level_streaming(self):
    +        cfg = _load_with_yaml_dict({"streaming": {"enabled": True, "transport": "draft"}})
    +        assert cfg.streaming.enabled is True
    +        assert cfg.streaming.transport == "draft"
    +
    +    def test_nested_gateway_streaming(self):
    +        """Regression for #25676."""
    +        cfg = _load_with_yaml_dict({"gateway": {"streaming": {"enabled": True, "transport": "draft"}}})
    +        assert cfg.streaming.enabled is True
    +        assert cfg.streaming.transport == "draft"
    +
    +    def test_top_level_takes_precedence(self):
    +        cfg = _load_with_yaml_dict({
    +            "streaming": {"enabled": True, "transport": "edit"},
    +            "gateway": {"streaming": {"enabled": False, "transport": "draft"}},
    +        })
    +        assert cfg.streaming.enabled is True
    +        assert cfg.streaming.transport == "edit"