fix(telegram): stop typing after final reply

gateway/platforms/base.py

@@ -6022,6 +6022,14 @@ async def _stop_typing_task() -> None:
                         event.source.chat_id,
                     )
                     _reply_anchor = _reply_anchor_for_event(event)
+                    # Telegram re-triggers typing after each send() to keep the
+                    # bubble alive across intermediate progress/status messages.
+                    # Mark the terminal reply so adapters can skip that post-send
+                    # refresh; otherwise users can receive the final answer while
+                    # the client still shows Hermes as typing for a few extra
+                    # seconds.
+                    _final_thread_metadata["suppress_post_send_typing"] = True
+
                     # Delivery-obligation ledger: durably record the final
                     # response BEFORE the send attempt so a gateway crash
                     # between finalize and platform ACK can redeliver it on

gateway/run.py

@@ -17271,10 +17271,18 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                     try:
                         _foot_adapter = self._adapter_for_source(source)
                         if _foot_adapter:
+                            _footer_meta = self._thread_metadata_for_source(
+                                source, self._reply_anchor_for_event(event)
+                            )
+                            if _footer_meta is not None:
+                                _footer_meta = dict(_footer_meta)
+                            else:
+                                _footer_meta = {}
+                            _footer_meta["suppress_post_send_typing"] = True
                             await _foot_adapter.send(
                                 source.chat_id,
                                 _footer_line,
-                                metadata=self._thread_metadata_for_source(source, self._reply_anchor_for_event(event)),
+                                metadata=_footer_meta,
                             )
                     except Exception as _e:
                         logger.debug("trailing footer send failed: %s", _e)

plugins/platforms/telegram/adapter.py

@@ -4613,12 +4613,20 @@ async def send(
             # so without this the "...typing" bubble disappears mid-response
             # (especially noticeable when the agent sends intermediate progress
             # messages like "Checking:" before running tools).
-            # Skip this on the FINAL reply (metadata["notify"]): the gateway has
-            # already cancelled the typing refresh loop by the time the final
-            # send returns, so re-arming Telegram's ~5s timer here would leave
-            # the indicator lingering after the answer with nothing to cancel
-            # it (Telegram exposes no stop-typing API). See #48678.
-            if not (metadata or {}).get("notify"):
+            #
+            # Skip the refresh on terminal/final sends. Otherwise the final
+            # answer itself creates a brand-new typing bubble with no more work
+            # behind it, making the bot look stuck "typing" for a few seconds
+            # after it already replied.
+            #
+            # Preserve the long-standing ``notify=True`` final-send contract for
+            # existing terminal producers, while also honoring the dedicated
+            # footer-specific suppression flag added for residual post-send paths.
+            suppress_post_send_typing = bool(
+                (metadata or {}).get("notify")
+                or (metadata or {}).get("suppress_post_send_typing")
+            )
+            if not suppress_post_send_typing:
                 try:
                     await self.send_typing(chat_id, metadata=metadata)
                 except Exception:

tests/gateway/test_base_topic_sessions.py

@@ -116,7 +116,7 @@ async def hold_typing(_chat_id, interval=2.0, metadata=None):
                 "chat_id": "-1001",
                 "content": "ack",
                 "reply_to": None,
-                "metadata": {"thread_id": "17585", "notify": True},
+                "metadata": {"thread_id": "17585", "notify": True, "suppress_post_send_typing": True},
             }
         ]
         assert typing_calls == [
@@ -195,7 +195,35 @@ async def test_long_original_with_short_spoken_script_still_sends_full_reply(sel
                 "chat_id": "-1001",
                 "content": long_reply,
                 "reply_to": None,
-                "metadata": {"thread_id": "17585", "notify": True},
+                "metadata": {"thread_id": "17585", "notify": True, "suppress_post_send_typing": True},
             }
         ]
 
+    @pytest.mark.asyncio
+    async def test_telegram_auto_tts_send_failure_keeps_followup_text(self, tmp_path):
+        adapter = DummyTelegramAdapter()
+        adapter._keep_typing = self._hold_typing()
+        adapter._should_auto_tts_for_chat = lambda _chat_id: True
+        adapter.play_tts = AsyncMock(return_value=SendResult(success=False, error="boom"))
+        adapter.set_message_handler(lambda _event: asyncio.sleep(0, result="Short reply"))
+
+        tts_path = tmp_path / "reply.ogg"
+        tts_path.write_text("audio", encoding="utf-8")
+        event = self._make_voice_event()
+
+        with patch("tools.tts_tool.check_tts_requirements", return_value=True), patch(
+            "tools.tts_tool.text_to_speech_tool",
+            return_value=json.dumps({"file_path": str(tts_path)}),
+        ):
+            await adapter._process_message_background(event, build_session_key(event.source))
+
+        adapter.play_tts.assert_awaited_once()
+        assert adapter.play_tts.await_args.kwargs["caption"] == "Short reply"
+        assert adapter.sent == [
+            {
+                "chat_id": "-1001",
+                "content": "Short reply",
+                "reply_to": None,
+                "metadata": {"thread_id": "17585", "notify": True, "suppress_post_send_typing": True},
+            }
+        ]

tests/gateway/test_telegram_footer_post_send_typing.py

@@ -0,0 +1,161 @@
+"""Regression test for streamed Telegram trailing-footer delivery metadata.
+
+When the agent already streamed the main body (``already_sent=True``),
+``GatewayRunner._handle_message_with_agent`` sends the runtime footer as a small
+trailing message. That footer is still part of the terminal reply, so it must
+carry ``suppress_post_send_typing=True`` to avoid re-triggering Telegram's
+typing bubble after the answer already landed.
+"""
+
+from datetime import datetime
+from types import SimpleNamespace
+from typing import Any, cast
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from gateway.config import GatewayConfig, Platform, PlatformConfig
+from gateway.platforms.base import MessageEvent
+from gateway.session import SessionEntry, SessionSource, build_session_key
+
+
+def _make_source() -> SessionSource:
+    return SessionSource(
+        platform=Platform.TELEGRAM,
+        user_id="u1",
+        chat_id="208214988",
+        user_name="tester",
+        chat_type="dm",
+        thread_id="17585",
+    )
+
+
+def _make_event(text: str = "hi") -> MessageEvent:
+    return MessageEvent(text=text, source=_make_source(), message_id="m1")
+
+
+def _make_runner():
+    from gateway.run import GatewayRunner
+
+    runner = object.__new__(GatewayRunner)
+    r = cast(Any, runner)
+    r.config = GatewayConfig(
+        platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")}
+    )
+    adapter = MagicMock()
+    adapter.send = AsyncMock()
+    adapter.stop_typing = AsyncMock()
+    r.adapters = {Platform.TELEGRAM: adapter}
+    r.hooks = SimpleNamespace(
+        emit=AsyncMock(),
+        emit_collect=AsyncMock(return_value=[]),
+        loaded_hooks=False,
+    )
+
+    source = _make_source()
+    session_entry = SessionEntry(
+        session_key=build_session_key(source),
+        session_id="sess-1",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        platform=Platform.TELEGRAM,
+        chat_type="dm",
+    )
+    r.session_store = MagicMock()
+    r.session_store.get_or_create_session.return_value = session_entry
+    r.session_store.load_transcript.return_value = []
+    r.session_store.append_to_transcript = MagicMock()
+    r.session_store.update_session = MagicMock()
+    r.session_store._save = MagicMock()
+
+    r._session_db = None
+    r._voice_mode = {}
+    r._reasoning_config = None
+    r._provider_routing = {}
+    r._fallback_model = None
+    r._show_reasoning = False
+    r._pending_model_notes = {}
+    r._session_model_overrides = {}
+    r._pending_messages = {}
+    r._pending_approvals = {}
+    r._recover_telegram_topic_thread_id = lambda source: None
+    r._cache_session_source = lambda *_a, **_kw: None
+    r._is_telegram_topic_lane = lambda source: False
+    r._set_session_env = lambda context: []
+    r._bind_adapter_run_generation = lambda *_a, **_kw: None
+    r._is_session_run_current = lambda *_a, **_kw: True
+    r._capture_gateway_honcho_if_configured = lambda *args, **kwargs: None
+    r._emit_gateway_run_progress = AsyncMock()
+    r._run_process_watcher = AsyncMock()
+    r._deliver_media_from_response = AsyncMock()
+    r._should_send_voice_reply = lambda *_a, **_kw: False
+    r._clear_restart_failure_count = lambda *_a, **_kw: None
+    r._evict_cached_agent = lambda *_a, **_kw: None
+    r._set_session_reasoning_override = lambda *_a, **_kw: None
+
+    from gateway.run import GatewayRunner as _GR
+
+    r._thread_metadata_for_source = _GR._thread_metadata_for_source.__get__(runner, _GR)
+    r._reply_anchor_for_event = _GR._reply_anchor_for_event
+    return runner
+
+
+@pytest.mark.asyncio
+async def test_streamed_trailing_footer_suppresses_post_send_typing(monkeypatch):
+    import gateway.run as gateway_run
+    import gateway.runtime_footer as runtime_footer
+
+    runner = _make_runner()
+    event = _make_event()
+    session_key = build_session_key(event.source)
+
+    runner._run_agent = AsyncMock(
+        return_value={
+            "final_response": "main body",
+            "messages": [
+                {"role": "user", "content": "hi"},
+                {"role": "assistant", "content": "main body"},
+            ],
+            "history_offset": 0,
+            "already_sent": True,
+            "failed": False,
+            "model": "openai/gpt-5.4",
+            "last_prompt_tokens": 25,
+            "context_length": 100,
+            "api_calls": 1,
+            "session_id": "sess-1",
+            "tools": [],
+        }
+    )
+
+    monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {})
+    monkeypatch.setattr(runtime_footer, "build_footer_line", lambda **_kw: "gpt-5.4 · 25%")
+
+    result = await runner._handle_message_with_agent(
+        event,
+        event.source,
+        session_key,
+        run_generation=1,
+    )
+
+    r = cast(Any, runner)
+    assert result is None
+    r._deliver_media_from_response.assert_awaited_once_with(
+        "main body", event, r.adapters[Platform.TELEGRAM]
+    )
+    assert r.adapters[Platform.TELEGRAM].send.await_count >= 1
+
+    footer_call = r.adapters[Platform.TELEGRAM].send.await_args_list[-1]
+    send_chat_id = footer_call.kwargs.get("chat_id", footer_call.args[0] if footer_call.args else None)
+    send_content = footer_call.kwargs.get("content", footer_call.args[1] if len(footer_call.args) > 1 else None)
+    send_metadata = footer_call.kwargs.get("metadata", footer_call.args[2] if len(footer_call.args) > 2 else None)
+    assert send_chat_id == event.source.chat_id
+    assert send_content == "gpt-5.4 · 25%"
+    assert send_metadata["thread_id"] == "17585"
+    assert send_metadata["suppress_post_send_typing"] is True
+
+    fresh_meta = runner._thread_metadata_for_source(
+        event.source, runner._reply_anchor_for_event(event)
+    )
+    assert fresh_meta is not None
+    assert "suppress_post_send_typing" not in fresh_meta

tests/gateway/test_telegram_post_send_typing.py

@@ -0,0 +1,104 @@
+"""Regression tests for Telegram post-send typing refresh behavior.
+
+Telegram clears the typing bubble when a message is delivered, so Hermes
+re-triggers ``send_chat_action('typing')`` after intermediate progress/status
+messages. That refresh must NOT run for the terminal reply, or the client keeps
+showing the bot as typing for a few extra seconds after the answer already
+landed.
+"""
+
+from __future__ import annotations
+
+import sys
+import types
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from gateway.config import PlatformConfig
+
+
+def _install_fake_telegram(monkeypatch):
+    fake_telegram = types.ModuleType("telegram")
+    setattr(fake_telegram, "Update", SimpleNamespace(ALL_TYPES=()))
+    setattr(fake_telegram, "Bot", object)
+    setattr(fake_telegram, "Message", object)
+    setattr(fake_telegram, "InlineKeyboardButton", object)
+    setattr(fake_telegram, "InlineKeyboardMarkup", object)
+
+    fake_error = types.ModuleType("telegram.error")
+    setattr(fake_error, "NetworkError", type("NetworkError", (Exception,), {}))
+    setattr(fake_error, "BadRequest", type("BadRequest", (Exception,), {}))
+    setattr(fake_error, "TimedOut", type("TimedOut", (Exception,), {}))
+    setattr(fake_telegram, "error", fake_error)
+
+    fake_constants = types.ModuleType("telegram.constants")
+    setattr(fake_constants, "ParseMode", SimpleNamespace(MARKDOWN_V2="MarkdownV2"))
+    setattr(fake_constants, "ChatType", SimpleNamespace(
+        GROUP="group", SUPERGROUP="supergroup", CHANNEL="channel", PRIVATE="private"
+    ))
+    setattr(fake_telegram, "constants", fake_constants)
+
+    fake_ext = types.ModuleType("telegram.ext")
+    setattr(fake_ext, "Application", object)
+    setattr(fake_ext, "CommandHandler", object)
+    setattr(fake_ext, "CallbackQueryHandler", object)
+    setattr(fake_ext, "MessageHandler", object)
+    setattr(fake_ext, "ContextTypes", SimpleNamespace(DEFAULT_TYPE=object))
+    setattr(fake_ext, "filters", object)
+
+    fake_request = types.ModuleType("telegram.request")
+    setattr(fake_request, "HTTPXRequest", object)
+
+    monkeypatch.setitem(sys.modules, "telegram", fake_telegram)
+    monkeypatch.setitem(sys.modules, "telegram.error", fake_error)
+    monkeypatch.setitem(sys.modules, "telegram.constants", fake_constants)
+    monkeypatch.setitem(sys.modules, "telegram.ext", fake_ext)
+    monkeypatch.setitem(sys.modules, "telegram.request", fake_request)
+
+
+@pytest.fixture
+def adapter(monkeypatch):
+    _install_fake_telegram(monkeypatch)
+    sys.modules.pop("plugins.platforms.telegram.adapter", None)
+    from plugins.platforms.telegram.adapter import TelegramAdapter
+
+    a = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
+    a._bot = MagicMock()
+    assert a._bot is not None
+    a._bot.send_message = AsyncMock(return_value=SimpleNamespace(message_id=42))
+    a.send_typing = AsyncMock()
+    return a
+
+
+@pytest.mark.asyncio
+async def test_send_retriggers_typing_by_default(adapter):
+    result = await adapter.send("12345", "intermediate progress")
+
+    assert result.success is True
+    adapter.send_typing.assert_awaited_once_with("12345", metadata=None)
+
+
+@pytest.mark.asyncio
+async def test_send_notify_skips_post_send_typing(adapter):
+    result = await adapter.send(
+        "12345",
+        "final answer",
+        metadata={"notify": True},
+    )
+
+    assert result.success is True
+    adapter.send_typing.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_send_dedicated_suppression_flag_skips_post_send_typing(adapter):
+    result = await adapter.send(
+        "12345",
+        "footer message",
+        metadata={"suppress_post_send_typing": True},
+    )
+
+    assert result.success is True
+    adapter.send_typing.assert_not_awaited()