diff --git a/gateway/run.py b/gateway/run.py index a2625f4e986a..7d59e54042ba 100644 --- a/gateway/run.py +++ b/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 diff --git a/tests/gateway/test_telegram_audio_vs_voice.py b/tests/gateway/test_telegram_audio_vs_voice.py index be9cc37678fc..5245155a09fa 100644 --- a/tests/gateway/test_telegram_audio_vs_voice.py +++ b/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.""" diff --git a/tests/gateway/test_telegram_voice_v0_regressions.py b/tests/gateway/test_telegram_voice_v0_regressions.py index aa36cf654762..438039411d8f 100644 --- a/tests/gateway/test_telegram_voice_v0_regressions.py +++ b/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={})