Duplicate voice transcript echo deduplication

Fixes on record

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

• #67248 fix(gateway): dedupe pending voice transcript echoes

  merged to main

  files: gateway/platforms/base.py, gateway/run.py, tests/gateway/test_telegram_audio_vs_voice.py, tests/gateway/test_telegram_voice_v0_regressions.py

  Show the diff

  • gateway/platforms/base.py

    @@ -2104,6 +2104,25 @@ def text(self) -> str:
             return str.__str__(self)
     
     
    +def _invalidate_pending_stt_cache(event: MessageEvent) -> None:
    +    """Clear gateway-side STT cache attrs when media is merged into an event.
    +
    +    ``merge_pending_message_event`` extends ``media_urls`` in place when two
    +    media-bearing messages arrive in quick succession.  The gateway runner
    +    caches STT transcripts on the event via ``setattr`` (see
    +    ``_transcribe_pending_audio_event_once``); if the cached event gains new
    +    media after the cache was populated, the stale transcript must be
    +    discarded so the next transcription call picks up the merged attachments.
    +    """
    +    for attr in (
    +        "_gateway_pending_stt_text",
    +        "_gateway_pending_stt_transcripts",
    +        "_gateway_pending_stt_echo_sent",
    +    ):
    +        if hasattr(event, attr):
    +            delattr(event, attr)
    +
    +
     def merge_pending_message_event(
         pending_messages: Dict[str, MessageEvent],
         session_key: str,
    @@ -2134,6 +2153,7 @@ def merge_pending_message_event(
                 existing.media_types.extend(event.media_types)
                 if event.text:
                     existing.text = BasePlatformAdapter._merge_caption(existing.text, event.text)
    +            _invalidate_pending_stt_cache(existing)
                 return
     
             if existing_has_media or incoming_has_media:
    @@ -2152,6 +2172,7 @@ def merge_pending_message_event(
                     and event.message_type != MessageType.TEXT
                 ):
                     existing.message_type = event.message_type
    +            _invalidate_pending_stt_cache(existing)
                 return
     
             if (

  • gateway/run.py

    @@ -1989,6 +1989,9 @@ def _own_policy_open_startup_violation(config) -> Optional[str]:
     # between the guard check and actual agent creation.
     _AGENT_PENDING_SENTINEL = object()
     
    +# Sentinel for "caller did not pass metadata" vs "caller passed None".
    +_UNSET = object()
    +
     
     def _resolve_runtime_agent_kwargs() -> dict:
         """Resolve provider credentials for gateway-created AIAgent instances.
    @@ -2180,6 +2183,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)
    @@ -5799,7 +5813,19 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session
             # at the next check point.
             if effective_mode == "interrupt" and running_agent and running_agent is not _AGENT_PENDING_SENTINEL:
                 try:
    -                running_agent.interrupt(event.text)
    +                _interrupt_text = event.text
    +                _media_urls = getattr(event, "media_urls", None) or []
    +                if self._pending_event_audio_paths(event):
    +                    _interrupt_text, _ = await self._transcribe_and_echo_pending_voice(
    +                        event,
    +                        adapter,
    +                        event.source,
    +                        event.text or "",
    +                        log_context="Voice-busy-interrupt",
    +                    )
    +                elif not _interrupt_text and _media_urls:
    +                    _interrupt_text = _build_media_placeholder(event)
    +                running_agent.interrupt(_interrupt_text)
                 except Exception:
                     pass  # don't let interrupt failure block the ack
     
    @@ -10450,7 +10476,19 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
                     self._queue_or_replace_pending_event(_quick_key, event)
                     return None
                 logger.debug("PRIORITY interrupt for session %s", _quick_key)
    -            running_agent.interrupt(event.text)
    +            _interrupt_text = event.text
    +            _media_urls = getattr(event, "media_urls", None) or []
    +            if self._pending_event_audio_paths(event):
    +                _interrupt_text, _ = await self._transcribe_and_echo_pending_voice(
    +                    event,
    +                    self._adapter_for_source(source),
    +                    source,
    +                    event.text or "",
    +                    log_context="Voice-priority-interrupt",
    +                )
    +            elif not _interrupt_text and _media_urls:
    +                _interrupt_text = _build_media_placeholder(event)
    +            running_agent.interrupt(_interrupt_text)
                 # NOTE: self._pending_messages was write-only (never consumed).
                 # The actual interrupt message is delivered via adapter._pending_messages
                 # which is read by _run_agent. Removed to prevent unbounded growth.
    @@ -11244,7 +11282,12 @@ async def _prepare_inbound_message_text(
             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
    @@ -11298,10 +11341,7 @@ async def _prepare_inbound_message_text(
                     # 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)
    @@ -16317,71 +16357,118 @@ async def _enrich_message_with_transcription(
                 return prefix, successful_transcripts
             return user_text, successful_transcripts
     
    -    async def _dequeue_pending_with_transcription(
    +    def _pending_event_audio_paths(self, event) -> List[str]:
    +        """Return STT-eligible paths from a pending voice message."""
    +        audio_paths: List[str] = []
    +        media_urls = getattr(event, "media_urls", None) or []
    +        for i, path in enumerate(media_urls):
    +            if _event_media_is_stt_input(event, i):
    +                audio_paths.append(path)
    +        return audio_paths
    +
    +    async def _transcribe_pending_audio_event_once(
    +        self,
    +        event,
    +        user_text: Optional[str] = None,
    +    ) -> tuple[str | None, List[str]]:
    +        """Transcribe a pending audio event once and cache the result on the event.
    +
    +        Voice follow-ups can be inspected first by the interrupt monitor and
    +        later consumed by the pending-drain path.  Both need the same transcript,
    +        but only one STT call and one transcript echo should happen for the
    +        platform message.
    +        """
    +        if hasattr(event, "_gateway_pending_stt_text"):
    +            cached_text = getattr(event, "_gateway_pending_stt_text")
    +            cached_transcripts = getattr(event, "_gateway_pending_stt_transcripts", []) or []
    +            return cached_text, list(cached_transcripts)
    +
    +        audio_paths = self._pending_event_audio_paths(event)
    +        if not audio_paths:
    +            return user_text if user_text is not None else (getattr(event, "text", None) or None), []
    +
    +        text = user_text if user_text is not None else (getattr(event, "text", "") or "")
    +        enriched_text, successful_transcripts = await self._enrich_message_with_transcription(
    +            text,
    +            audio_paths,
    +        )
    +        setattr(event, "_gateway_pending_stt_text", enriched_text)
    +        setattr(event, "_gateway_pending_stt_transcripts", list(successful_transcripts))
    +        return enriched_text, successful_transcripts
    +
    +    async def _echo_pending_stt_transcripts_once(
             self,
    +        event,
             adapter,
    -        session_key: str,
             source,
    -    ) -> str | None:
    -        """Dequeue a pending queued message, auto-transcribing audio media.
    -
    -        When a voice/audio message arrives during an active agent run, the
    -        adapter stores the event in its pending queue and signals an interrupt
    -        (see base.BaseAdapter.handle_message). The adapter path bypasses
    -        _handle_message entirely, so the normal STT pipeline at message-receive
    -        time never runs.
    -
    -        This helper fills that gap: when the dequeued event has audio media,
    -        we transcribe inline, echo the raw transcript back to the user (same
    -        "🎙️" format as the fresh-message path), and return enriched text.
    -        Non-audio events fall back to _build_media_placeholder, matching the
    -        original _dequeue_pending_text behavior.
    -        """
    -        event = adapter.get_pending_message(session_key)
    -        if not event:
    -            return None
    +        transcripts: List[str],
    +        *,
    +        metadata=None,
    +        log_context: str = "Transcript",
    +    ) -> None:
    +        """Echo pending-event STT transcripts to the chat at most once."""
    +        if (
    +            not transcripts
    +            or not self._should_echo_stt_transcripts()
    +            or adapter is None
    +            or getattr(event, "_gateway_pending_stt_echo_sent", False)
    +        ):
    +            return
    +        setattr(event, "_gateway_pending_stt_echo_sent", True)
    +        for tx in transcripts:
    +            try:
    +                await adapter.send(
    +                    source.chat_id,
    +                    f'🎙️ "{tx}"',
    +                    metadata=metadata,
    +                )
    +            except Exception as echo_exc:
    +                logger.debug("%s echo failed (non-fatal): %s", log_context, echo_exc)
    +
    +    async def _transcribe_and_echo_pending_voice(
    +        self,
    +        event,
    +        adapter,
    +        source,
    +        text: str,
    +        *,
    +        log_context: str,
    +        metadata=_UNSET,
    +    ) -> tuple[str, List[str]]:
    +        """Transcribe a pending voice event and echo transcripts once.
     
    -        text = event.text or ""
    +        Unified helper for all interrupt/monitor/backup/drain paths that need
    +        to transcribe a pending voice event and echo the transcript to chat.
    +        Returns ``(enriched_text, transcripts)`` so the caller can feed the
    +        enriched text into ``agent.interrupt()`` or the pending-drain flow.
     
    -        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 the event has no STT-eligible media, returns ``(text, [])`` unchanged.
    +        The caller is responsible for the ``_build_media_placeholder`` fallback
    +        when ``text`` is empty and the event has non-audio media.
    +        """
    +        if not self._pending_event_audio_paths(event):
    +            return text, []
    +        try:
    +            enriched_text, transcripts = await self._transcribe_pending_audio_event_once(
    +                event,
    +                text,
                 )
    -            if is_audio:
    -                audio_paths.append(path)
    -
    -        if audio_paths:
    -            enriched_text, successful_transcripts = await self._enrich_message_with_transcription(
    -                text, audio_paths,
    +            echo_meta = self._thread_metadata_for_source(
    +                source,
    +                self._reply_anchor_for_event(event),
    +            ) if metadata is _UNSET else metadata
    +            await self._echo_pending_stt_transcripts_once(
    +                event,
    +                adapter,
    +                source,
    +                transcripts,
    +                metadata=echo_meta,
    +                log_context=log_context,
                 )
    -            # Echo raw transcripts back to the user when configured so voice
    -            # interrupts feel identical to fresh voice messages.
    -            if successful_transcripts and self._should_echo_stt_transcripts():
    -                echo_adapter = self._adapter_for_source(source)
    -                echo_meta = {"thread_id": source.thread_id} if source.thread_id else None
    -                if echo_adapter:
    -                    for tx in successful_transcripts:
    -                        try:
    -                            await echo_adapter.send(
    -                                source.chat_id,
    -                                f'🎙️ "{tx}"',
    -                                metadata=echo_meta,
    -                            )
    -                        except Exception as echo_exc:
    -                            logger.debug(
    -                                "Transcript echo failed (non-fatal): %s", echo_exc,
    -                            )
    -            return enriched_text or None
    -
    -        # Non-audio fallback: preserve original _dequeue_pending_text semantics.
    -        if not text and media_urls:
    -            text = _build_media_placeholder(event)
    -        return text or None
    +            return enriched_text or text, transcripts
    +        except Exception as trans_exc:
    +            logger.warning("%s transcription failed: %s", log_context, trans_exc)
    +            return text, []
     
         def _build_process_event_source(self, evt: dict):
             """Resolve the canonical source for a synthetic background-process event.
    @@ -20658,40 +20745,15 @@ async def monitor_for_interrupt():
                                     # of fresh voice messages including the
                                     # optional 🎙️ echo back to the user.
                                     _media_urls = getattr(_peek_event, "media_urls", None) or []
    -                                _media_types = getattr(_peek_event, "media_types", None) or []
    -                                _audio_paths = []
    -                                for _i, _path in enumerate(_media_urls):
    -                                    _mtype = _media_types[_i] if _i < len(_media_types) else ""
    -                                    _is_audio = (
    -                                        _mtype.startswith("audio/")
    -                                        or getattr(_peek_event, "message_type", None) in (MessageType.VOICE, MessageType.AUDIO)
    +                                if self._pending_event_audio_paths(_peek_event):
    +                                    pending_text, _ = await self._transcribe_and_echo_pending_voice(
    +                                        _peek_event,
    +                                        _adapter,
    +                                        source,
    +                                        pending_text,
    +                                        log_context="Voice-interrupt",
    +                                        metadata={"thread_id": source.thread_id} if source.thread_id else None,
                                         )
    -                                    if _is_audio:
    -                                        _audio_paths.append(_path)
    -                                if _audio_paths:
    -                                    try:
    -                                        _enriched, _transcripts = await self._enrich_message_with_transcription(
    -                                            pending_text, _audio_paths,
    -                                        )
    -                                        pending_text = _enriched
    -                                        if _transcripts and self._should_echo_stt_transcripts():
    -                                            _echo_meta = {"thread_id": source.thread_id} if source.thread_id else None
    -                                            for _tx in _transcripts:
    -                                                try:
    -                                                    await _adapter.send(
    -                                                        source.chat_id,
    -                                                        f'🎙️ "{_tx}"',
    -                                                        metadata=_echo_meta,
    -                                                    )
    -                                                except Exception as _echo_exc:
    -                                                    logger.debug(
    -                                                        "Voice-interrupt echo failed (non-fatal): %s",
    -                                                        _echo_exc,
    -                                                    )
    -                                    except Exception as _trans_exc:
    -                                        logger.warning(
    -                                            "Voice-interrupt transcription failed: %s", _trans_exc,
    -                                        )
                                     elif not pending_text and _media_urls:
                                         pending_text = _build_media_placeholder(_peek_event)
                                 logger.debug("Interrupt detected from adapter, signaling agent...")
    @@ -20876,6 +20938,19 @@ def _stream_confirmed_final_delivery(
                                     and _backup_adapter.has_pending_interrupt(session_key)):
                                 _bp_event = _backup_adapter._pending_messages.get(session_key)
                                 _bp_text = _bp_event.text if _bp_event else None
    +                            if _bp_event is not None:
    +                                _bp_media_urls = getattr(_bp_event, "media_urls", None) or []
    +                                if self._pending_event_audio_paths(_bp_event):
    +                                    _bp_text, _ = await self._transcribe_and_echo_pending_voice(
    +                                        _bp_event,
    +                                        _backup_adapter,
    +                                        source,
    +                                        _bp_text or "",
    +                                        log_context="Voice-backup-interrupt",
    +                                        metadata={"thread_id": source.thread_id} if source.thread_id else None,
    +                                    )
    +                                elif not _bp_text and _bp_media_urls:
    +                                    _bp_text = _build_media_placeholder(_bp_event)
                                 logger.info(
                                     "Backup interrupt detected for session %s "
                                     "(monitor task state: %s)",
    @@ -20936,6 +21011,19 @@ def _stream_confirmed_final_delivery(
                                     and _backup_adapter.has_pending_interrupt(session_key)):
                                 _bp_event = _backup_adapter._pending_messages.get(session_key)
                                 _bp_text = _bp_event.text if _bp_event else None
    +                            if _bp_event is not None:
    +                                _bp_media_urls = getattr(_bp_event, "media_urls", None) or []
    +                                if self._pending_event_audio_paths(_bp_event):
    +                                    _bp_text, _ = await self._transcribe_and_echo_pending_voice(
    +                                        _bp_event,
    +                                        _backup_adapter,
    +                                        source,
    +                                        _bp_text or "",
    +                                        log_context="Voice-backup-interrupt",
    +                                        metadata={"thread_id": source.thread_id} if source.thread_id else None,
    +                                    )
    +                                elif not _bp_text and _bp_media_urls:
    +                                    _bp_text = _build_media_placeholder(_bp_event)
                                 logger.info(
                                     "Backup interrupt detected for session %s "
                                     "(monitor task state: %s)",
    @@ -21080,40 +21168,17 @@ def _stream_confirmed_final_delivery(
                         # fresh voice messages.
                         _pending_text = pending_event.text or ""
                         _media_urls = getattr(pending_event, "media_urls", None) or []
    -                    _media_types = getattr(pending_event, "media_types", None) or []
    -                    _audio_paths = []
    -                    for _i, _path in enumerate(_media_urls):
    -                        _mtype = _media_types[_i] if _i < len(_media_types) else ""
    -                        _is_audio = (
    -                            _mtype.startswith("audio/")
    -                            or getattr(pending_event, "message_type", None) in (MessageType.VOICE, MessageType.AUDIO)
    +                    if self._pending_event_audio_paths(pending_event):
    +                        pending, _ = await self._transcribe_and_echo_pending_voice(
    +                            pending_event,
    +                            adapter,
    +                            source,
    +                            _pending_text,
    +                            log_context="Voice-drain",
    +                            metadata={"thread_id": source.thread_id} if source.thread_id else None,
                             )
    -                        if _is_audio:
    -                            _audio_paths.append(_path)
    -                    if _audio_paths:
    -                        try:
    -                            _enriched, _transcripts = await self._enrich_message_with_transcription(
    -                                _pending_text, _audio_paths,
    -                            )
    -                            pending = _enriched or None
    -                            if _transcripts and self._should_echo_stt_transcripts():
    -                                _echo_meta = {"thread_id": source.thread_id} if source.thread_id else None
    -                                for _tx in _transcripts:
    -                                    try:
    -                                        await adapter.send(
    -                                            source.chat_id,
    -                                            f'🎙️ "{_tx}"',
    -                                            metadata=_echo_meta,
    -                                        )
    -                                    except Exception as _echo_exc:
    -                                        logger.debug(
    -                                            "Voice-drain echo failed (non-fatal): %s", _echo_exc,
    -                                        )
    -                        except Exception as _trans_exc:
    -                            logger.warning(
    -                                "Voice-drain transcription failed: %s", _trans_exc,
    -                            )
    -                            pending = _pending_text or _build_media_placeholder(pending_event)
    +                        if not pending:
    +                            pending = _build_media_placeholder(pending_event)
                         else:
                             pending = _pending_text or _build_media_placeholder(pending_event)
                         if pending:

  • 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()
     
     
    [redacted:email]
    +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,6 +1,10 @@
    +import asyncio
     import sys
    +import threading
    +import types
     from pathlib import Path
     from types import SimpleNamespace
    +from unittest.mock import AsyncMock, MagicMock, patch
     
     import pytest
     
    @@ -9,7 +13,8 @@
     if str(ROOT) not in sys.path:
         sys.path.insert(0, str(ROOT))
     
    -from gateway.config import Platform
    +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
    @@ -34,6 +39,243 @@ 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
    +
    +
    [redacted:email]
    +async def test_pending_voice_interrupt_reuses_transcript_and_echo():
    +    adapter = SimpleNamespace(send=AsyncMock())
    +    runner = _runner(adapter)
    +    source = _source()
    +    event = MessageEvent(
    +        text="",
    +        message_type=MessageType.VOICE,
    +        source=source,
    +        media_urls=["/tmp/telegram-voice.ogg"],
    +        media_types=["audio/ogg"],
    +    )
    +
    +    with patch(
    +        "tools.transcription_tools.transcribe_audio",
    +        return_value={"success": True, "transcript": "hello once", "provider": "mock"},
    +    ) as mock_transcribe:
    +        interrupt_text, interrupt_transcripts = await runner._transcribe_pending_audio_event_once(
    +            event,
    +            event.text,
    +        )
    +        await runner._echo_pending_stt_transcripts_once(
    +            event,
    +            adapter,
    +            source,
    +            interrupt_transcripts,
    +        )
    +
    +        drain_text, drain_transcripts = await runner._transcribe_pending_audio_event_once(
    +            event,
    +            event.text,
    +        )
    +        await runner._echo_pending_stt_transcripts_once(
    +            event,
    +            adapter,
    +            source,
    +            drain_transcripts,
    +        )
    +
    +    assert interrupt_text == '"hello once"'
    +    assert drain_text == interrupt_text
    +    assert drain_transcripts == interrupt_transcripts == ["hello once"]
    +    mock_transcribe.assert_called_once_with("/tmp/telegram-voice.ogg")
    +    adapter.send.assert_awaited_once_with(
    +        "12345",
    +        '🎙️ "hello once"',
    +        metadata=None,
    +    )
    +
    +
    [redacted:email]
    +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)]
    +
    +
    [redacted:email]
    +async def test_busy_voice_interrupt_transcribes_before_pending_drain(monkeypatch):
    +    adapter = SimpleNamespace(send=AsyncMock(), _pending_messages={})
    +    runner = _runner(adapter)
    +    runner._is_user_authorized = lambda _source: True
    +    runner._draining = False
    +    runner._running_agents = {}
    +    runner._busy_input_mode = "interrupt"
    +    runner._busy_text_mode = "interrupt"
    +    runner._busy_ack_ts = {}
    +    runner._queued_events = {}
    +    runner._agent_has_active_subagents = lambda _agent: False
    +    session_key = "telegram:dm:12345"
    +    agent = MagicMock()
    +    runner._running_agents[session_key] = agent
    +    source = _source()
    +    event = MessageEvent(
    +        text="",
    +        message_type=MessageType.VOICE,
    +        source=source,
    +        media_urls=["/tmp/telegram-busy-voice.ogg"],
    +        media_types=["audio/ogg"],
    +    )
    +    monkeypatch.setenv("HERMES_GATEWAY_BUSY_ACK_ENABLED", "false")
    +
    +    with (
    +        patch("tools.approval.has_blocking_approval", return_value=False),
    +        patch(
    +            "tools.transcription_tools.transcribe_audio",
    +            return_value={"success": True, "transcript": "interrupt me", "provider": "mock"},
    +        ) as mock_transcribe,
    +    ):
    +        handled = await runner._handle_active_session_busy_message(event, session_key)
    +        drain_text, drain_transcripts = await runner._transcribe_pending_audio_event_once(
    +            adapter._pending_messages[session_key],
    +            event.text,
    +        )
    +        await runner._echo_pending_stt_transcripts_once(
    +            adapter._pending_messages[session_key],
    +            adapter,
    +            source,
    +            drain_transcripts,
    +        )
    +
    +    assert handled is True
    +    agent.interrupt.assert_called_once_with('"interrupt me"')
    +    assert adapter._pending_messages[session_key] is event
    +    assert drain_text == '"interrupt me"'
    +    mock_transcribe.assert_called_once_with("/tmp/telegram-busy-voice.ogg")
    +    adapter.send.assert_awaited_once_with(
    +        "12345",
    +        '🎙️ "interrupt me"',
    +        metadata={},
    +    )
    +
    +
     def test_telegram_audio_size_gate_rejects_oversized_media_before_download():
         adapter = object.__new__(TelegramAdapter)
         adapter._max_doc_bytes = 1024