fix(gateway): preserve pending voice media semantics

gateway/run.py

@@ -2180,6 +2180,17 @@ def _event_media_is_audio(event, index: int) -> bool:
     return getattr(event, "message_type", None) in {MessageType.VOICE, MessageType.AUDIO}
 
 
+def _event_media_is_stt_input(event, index: int) -> bool:
+    """True when an audio attachment should enter the automatic STT pipeline."""
+    message_type = getattr(event, "message_type", None)
+    if message_type in {MessageType.AUDIO, MessageType.DOCUMENT}:
+        return False
+    return (
+        message_type == MessageType.VOICE
+        or _event_media_type_at(event, index).startswith("audio/")
+    )
+
+
 def _event_media_is_video(event, index: int) -> bool:
     """True if the attachment at *index* is video (per-attachment MIME first)."""
     mtype = _event_media_type_at(event, index)
@@ -11299,7 +11310,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         already run and images are represented in-text.
         """
         history = history or []
-        message_text = event.text or ""
+        _pending_stt_prepared = hasattr(event, "_gateway_pending_stt_text")
+        message_text = (
+            getattr(event, "_gateway_pending_stt_text", None)
+            if _pending_stt_prepared
+            else event.text
+        ) or ""
         _group_sessions_per_user = getattr(self.config, "group_sessions_per_user", True)
         _thread_sessions_per_user = getattr(self.config, "thread_sessions_per_user", False)
         # Prefer the already resolved session key from the caller so this write
@@ -11353,10 +11369,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 # MessageType.VOICE = voice message (Opus/OGG) — always STT
                 if event.message_type == MessageType.AUDIO:
                     audio_file_paths.append(path)
-                elif event.message_type == MessageType.VOICE or (
-                    mtype.startswith("audio/")
-                    and event.message_type not in {MessageType.AUDIO, MessageType.DOCUMENT}
-                ):
+                elif not _pending_stt_prepared and _event_media_is_stt_input(event, i):
                     audio_paths.append(path)
                 if mtype.startswith("video/") or (not mtype and event.message_type == MessageType.VIDEO):
                     video_paths.append(path)
@@ -16373,17 +16386,11 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         return user_text, successful_transcripts
 
     def _pending_event_audio_paths(self, event) -> List[str]:
-        """Return audio paths that the pending-message interrupt path transcribes."""
+        """Return STT-eligible paths from a pending voice message."""
         audio_paths: List[str] = []
         media_urls = getattr(event, "media_urls", None) or []
-        media_types = getattr(event, "media_types", None) or []
         for i, path in enumerate(media_urls):
-            mtype = media_types[i] if i < len(media_types) else ""
-            is_audio = (
-                mtype.startswith("audio/")
-                or getattr(event, "message_type", None) in (MessageType.VOICE, MessageType.AUDIO)
-            )
-            if is_audio:
+            if _event_media_is_stt_input(event, i):
                 audio_paths.append(path)
         return audio_paths
 

tests/gateway/test_telegram_audio_vs_voice.py

@@ -110,6 +110,26 @@ async def test_audio_attachment_skips_stt():
     assert "audio file attachment" in result.lower()
 
 
+@pytest.mark.asyncio
+async def test_pending_audio_attachment_is_not_selected_for_stt():
+    """Pending Telegram AUDIO files retain file semantics during interrupts."""
+    runner = _make_runner(stt_enabled=True)
+    event = _audio_event("/tmp/pending-song.mp3")
+
+    with patch(
+        "tools.transcription_tools.transcribe_audio",
+        side_effect=AssertionError("pending audio attachments must not enter STT"),
+    ):
+        result, transcripts = await runner._transcribe_pending_audio_event_once(
+            event,
+            event.text,
+        )
+
+    assert runner._pending_event_audio_paths(event) == []
+    assert result == ""
+    assert transcripts == []
+
+
 @pytest.mark.asyncio
 async def test_audio_attachment_context_note_format():
     """Context note for audio file attachments should include the file path and guidance."""

tests/gateway/test_telegram_voice_v0_regressions.py

@@ -1,4 +1,7 @@
+import asyncio
 import sys
+import threading
+import types
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -10,8 +13,8 @@ ROOT = Path(__file__).resolve().parents[2]
 if str(ROOT) not in sys.path:
     sys.path.insert(0, str(ROOT))
 
-from gateway.config import Platform
-from gateway.platforms.base import MessageEvent, MessageType
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType, SendResult
 from plugins.platforms.telegram.adapter import TelegramAdapter
 from gateway.run import GatewayRunner
 from gateway.session import SessionSource
@@ -36,6 +39,88 @@ def _runner(adapter=None):
     return runner
 
 
+class _PendingVoiceAdapter(BasePlatformAdapter):
+    def __init__(self):
+        super().__init__(PlatformConfig(enabled=True, token="test"), Platform.TELEGRAM)
+        self.sent = []
+
+    async def connect(self, *, is_reconnect: bool = False) -> bool:
+        return True
+
+    async def disconnect(self) -> None:
+        self._mark_disconnected()
+
+    async def send(self, chat_id, content, reply_to=None, metadata=None):
+        self.sent.append((chat_id, content, metadata))
+        return SendResult(success=True, message_id="voice-echo")
+
+    async def send_typing(self, chat_id, metadata=None) -> None:
+        return None
+
+    async def stop_typing(self, chat_id) -> None:
+        return None
+
+    async def get_chat_info(self, chat_id):
+        return {"id": chat_id, "type": "dm"}
+
+
+class _PendingVoiceAgent:
+    messages = []
+
+    def __init__(self, **kwargs):
+        self.tools = []
+        self.model = "test-model"
+        self.provider = "test-provider"
+        self._interrupt_requested = False
+        self._interrupt_message = None
+        self._interrupted = threading.Event()
+
+    @property
+    def is_interrupted(self):
+        return self._interrupt_requested
+
+    def interrupt(self, message):
+        self._interrupt_requested = True
+        self._interrupt_message = message
+        self._interrupted.set()
+
+    def run_conversation(self, message, conversation_history=None, task_id=None, **kwargs):
+        type(self).messages.append(message)
+        if len(type(self).messages) == 1:
+            assert self._interrupted.wait(timeout=3), "pending voice interrupt was not delivered"
+            return {
+                "final_response": "interrupted",
+                "messages": [],
+                "api_calls": 1,
+                "interrupted": True,
+                "interrupt_message": self._interrupt_message,
+            }
+        return {
+            "final_response": "follow-up complete",
+            "messages": [],
+            "api_calls": 1,
+            "interrupted": False,
+        }
+
+
+def _run_agent_runner(adapter):
+    runner = _runner(adapter)
+    runner._voice_mode = {}
+    runner._prefill_messages = []
+    runner._ephemeral_system_prompt = ""
+    runner._reasoning_config = None
+    runner._provider_routing = {}
+    runner._fallback_model = None
+    runner._session_db = None
+    runner._running_agents = {}
+    runner._session_run_generation = {}
+    runner._queued_events = {}
+    runner._draining = False
+    runner.hooks = SimpleNamespace(loaded_hooks=False)
+    runner._should_echo_stt_transcripts = lambda: True
+    return runner
+
+
 @pytest.mark.asyncio
 async def test_pending_voice_interrupt_reuses_transcript_and_echo():
     adapter = SimpleNamespace(send=AsyncMock())
@@ -86,6 +171,55 @@ async def test_pending_voice_interrupt_reuses_transcript_and_echo():
     )
 
 
+@pytest.mark.asyncio
+async def test_monitor_to_drain_transcribes_and_echoes_pending_voice_once(
+    monkeypatch,
+    tmp_path,
+):
+    monkeypatch.setenv("HERMES_TOOL_PROGRESS_MODE", "off")
+    monkeypatch.setenv("HERMES_GATEWAY_NOTIFY_INTERVAL", "0")
+    monkeypatch.setitem(sys.modules, "dotenv", types.SimpleNamespace(load_dotenv=lambda: None))
+    monkeypatch.setitem(sys.modules, "run_agent", types.SimpleNamespace(AIAgent=_PendingVoiceAgent))
+
+    adapter = _PendingVoiceAdapter()
+    runner = _run_agent_runner(adapter)
+    source = _source()
+    session_key = "telegram:dm:12345"
+    event = MessageEvent(
+        text="",
+        message_type=MessageType.VOICE,
+        source=source,
+        media_urls=["/tmp/telegram-pending-voice.ogg"],
+        media_types=["audio/ogg"],
+    )
+    adapter._pending_messages[session_key] = event
+    adapter._active_sessions[session_key] = asyncio.Event()
+    adapter._active_sessions[session_key].set()
+    _PendingVoiceAgent.messages = []
+
+    with (
+        patch("gateway.run._hermes_home", tmp_path),
+        patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "fake"}),
+        patch(
+            "tools.transcription_tools.transcribe_audio",
+            return_value={"success": True, "transcript": "hello once", "provider": "mock"},
+        ) as mock_transcribe,
+    ):
+        result = await runner._run_agent(
+            message="initial turn",
+            context_prompt="",
+            history=[],
+            source=source,
+            session_id="pending-voice-session",
+            session_key=session_key,
+        )
+
+    assert result["final_response"] == "follow-up complete"
+    assert _PendingVoiceAgent.messages == ["initial turn", '"hello once"']
+    mock_transcribe.assert_called_once_with("/tmp/telegram-pending-voice.ogg")
+    assert adapter.sent == [("12345", '🎙️ "hello once"', None)]
+
+
 @pytest.mark.asyncio
 async def test_busy_voice_interrupt_transcribes_before_pending_drain(monkeypatch):
     adapter = SimpleNamespace(send=AsyncMock(), _pending_messages={})