diff --git a/gateway/run.py b/gateway/run.py index 339954f5bef40..28e0a182d303f 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2786,9 +2786,27 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str): if is_audio: audio_paths.append(path) if audio_paths: - message_text = await self._enrich_message_with_transcription( + message_text, _successful_transcripts = await self._enrich_message_with_transcription( message_text, audio_paths ) + # Echo each successful transcript back to the user immediately, + # before the agent loop runs. Lets the user verify STT quality + # in real-time and see the raw whisper output verbatim. + if _successful_transcripts: + _echo_adapter = self.adapters.get(source.platform) + _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, + ) # If STT failed, send a direct message to the user so they # know voice isn't configured β€” don't rely on the agent to # relay the error clearly. @@ -6018,7 +6036,7 @@ async def _enrich_message_with_transcription( self, user_text: str, audio_paths: List[str], - ) -> str: + ) -> tuple[str, List[str]]: """ Auto-transcribe user voice/audio messages using the configured STT provider and prepend the transcript to the message text. @@ -6028,7 +6046,13 @@ async def _enrich_message_with_transcription( audio_paths: List of local file paths to cached audio files. Returns: - The enriched message string with transcriptions prepended. + A tuple of ``(enriched_text, successful_transcripts)``: + - ``enriched_text``: the message string with transcription wrappers + prepended (same as before). + - ``successful_transcripts``: the raw transcript strings for audio + clips that were successfully transcribed, in input order. Empty + list if every clip failed or STT is disabled. Callers can use + this to echo transcripts back to the user before the agent loop. """ if not getattr(self.config, "stt_enabled", True): disabled_note = "[The user sent voice message(s), but transcription is disabled in config." @@ -6039,8 +6063,8 @@ async def _enrich_message_with_transcription( ) disabled_note += "]" if user_text: - return f"{disabled_note}\n\n{user_text}" - return disabled_note + return f"{disabled_note}\n\n{user_text}", [] + return disabled_note, [] from tools.transcription_tools import transcribe_audio, get_stt_model_from_config import asyncio @@ -6048,12 +6072,14 @@ async def _enrich_message_with_transcription( stt_model = get_stt_model_from_config() enriched_parts = [] + successful_transcripts: List[str] = [] for path in audio_paths: try: logger.debug("Transcribing user voice: %s", path) result = await asyncio.to_thread(transcribe_audio, path, model=stt_model) if result["success"]: transcript = result["transcript"] + successful_transcripts.append(transcript) enriched_parts.append( f'[The user sent a voice message~ ' f'Here\'s what they said: "{transcript}"]' @@ -6098,9 +6124,75 @@ async def _enrich_message_with_transcription( if user_text and user_text.strip() == _placeholder: return prefix if user_text: - return f"{prefix}\n\n{user_text}" - return prefix - return user_text + return f"{prefix}\n\n{user_text}", successful_transcripts + return prefix, successful_transcripts + return user_text, successful_transcripts + + async def _dequeue_pending_with_transcription( + self, + 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 + + text = event.text or "" + + 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: + audio_paths.append(path) + + if audio_paths: + enriched_text, successful_transcripts = await self._enrich_message_with_transcription( + text, audio_paths, + ) + # Echo raw transcripts back to the user so voice interrupts + # feel identical to fresh voice messages. + if successful_transcripts: + echo_adapter = self.adapters.get(source.platform) + 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 async def _run_process_watcher(self, watcher: dict) -> None: """ @@ -7096,7 +7188,52 @@ async def monitor_for_interrupt(): agent = agent_holder[0] if agent: pending_event = adapter.get_pending_message(session_key) - pending_text = pending_event.text if pending_event else None + pending_text = None + if pending_event is not None: + pending_text = pending_event.text or "" + # Transcribe audio media BEFORE signaling the + # agent, so voice messages interrupt with the + # real transcript instead of an empty string + # (or file-path placeholder). Matches the UX + # of fresh voice messages including the + # πŸŽ™οΈ echo back to the user. + _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 _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: + _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(pending_event) logger.debug("Interrupt detected from adapter, signaling agent...") agent.interrupt(pending_text) break @@ -7299,11 +7436,15 @@ async def _notify_long_running(): pending = None if result and adapter and session_key: if result.get("interrupted"): - pending = _dequeue_pending_text(adapter, session_key) + pending = await self._dequeue_pending_with_transcription( + adapter, session_key, source, + ) if not pending and result.get("interrupt_message"): pending = result.get("interrupt_message") else: - pending = _dequeue_pending_text(adapter, session_key) + pending = await self._dequeue_pending_with_transcription( + adapter, session_key, source, + ) if pending: logger.debug("Processing queued message after agent completion: '%s...'", pending[:40]) diff --git a/run_agent.py b/run_agent.py index b8ed44ef78b01..37cd0d69db56a 100644 --- a/run_agent.py +++ b/run_agent.py @@ -4256,6 +4256,13 @@ def _interruptible_api_call(self, api_kwargs: dict): """ result = {"response": None, "error": None} request_client_holder = {"client": None} + # Request-local cancellation flag. Distinct from self._interrupt_requested + # because that flag is cleared at run_conversation() turn boundaries, but + # this daemon worker thread can outlive the turn. Tracks whether THIS + # specific request was cancelled by the main thread's interrupt handler, + # so transport errors that are the expected consequence of our own + # force-close aren't misread as a network bug. + _request_cancelled = {"value": False} def _call(): try: @@ -4272,6 +4279,17 @@ def _call(): request_client_holder["client"] = self._create_request_openai_client(reason="chat_completion_request") result["response"] = request_client_holder["client"].chat.completions.create(**api_kwargs) except Exception as e: + # If the request was cancelled by the main thread's interrupt + # handler, the transport error is the expected consequence of + # our own force-close, NOT a network bug. Swallow it instead + # of surfacing β€” the main thread will raise InterruptedError. + if _request_cancelled["value"]: + logger.debug( + "Non-streaming worker caught %s after request cancellation β€” " + "exiting without surfacing a network error.", + type(e).__name__, + ) + return result["error"] = e finally: request_client = request_client_holder.get("client") @@ -4283,6 +4301,12 @@ def _call(): while t.is_alive(): t.join(timeout=0.3) if self._interrupt_requested: + # Mark the request cancelled so the worker's exception handler + # recognizes the forced close and doesn't surface the error. + _request_cancelled["value"] = True + logger.debug( + "Force-closing httpx client due to interrupt (not a network error)." + ) # Force-close the in-flight worker-local HTTP connection to stop # token generation without poisoning the shared client used to # seed future retries. @@ -4393,6 +4417,14 @@ def _interruptible_streaming_api_call( # poll loop uses this to detect stale connections that keep receiving # SSE keep-alive pings but no actual data. last_chunk_time = {"t": time.time()} + # Request-local cancellation flag. Set by the outer poll loop when it + # observes self._interrupt_requested and force-closes the httpx client. + # The worker's retry loop checks this flag to distinguish our own + # interrupt-induced close from a genuine transient transport error β€” + # so we exit cleanly instead of burning 2*180s retry cycles on a dead + # request, which was the root cause of the 7-minute cascading-interrupt + # hang observed in production. + _request_cancelled = {"value": False} def _fire_first_delta(): if not first_delta_fired["done"] and on_first_delta: @@ -4652,6 +4684,16 @@ def _call(): try: for _stream_attempt in range(_max_stream_retries + 1): + # Check cancellation before each new attempt. If an interrupt + # fired between iterations (e.g. during cleanup/retry wait), + # the new attempt would just get force-closed again β€” exit + # now instead of wasting a request. + if _request_cancelled["value"]: + logger.debug( + "Streaming worker saw request cancellation before attempt %s β€” exiting.", + _stream_attempt + 1, + ) + return try: if self.api_mode == "anthropic_messages": self._try_refresh_anthropic_client_credentials() @@ -4660,6 +4702,20 @@ def _call(): result["response"] = _call_chat_completions() return # success except Exception as e: + # If this request was cancelled by the outer poll loop, + # the transport error is the expected consequence of our + # own force-close β€” NOT a transient network error. Exit + # without retrying, falling back, or emitting reconnect + # status. Preserve the error for diagnostics. + if _request_cancelled["value"]: + logger.debug( + "Streaming worker caught %s after request cancellation β€” " + "exiting without retry. (This error is the expected " + "consequence of interrupt force-close, not a network bug.)", + type(e).__name__, + ) + result["error"] = e + return if deltas_were_sent["yes"]: # Streaming failed AFTER some tokens were already # delivered. Don't retry or fall back β€” partial @@ -4711,6 +4767,14 @@ def _call(): # Transient network / timeout error. Retry the # streaming request with a fresh connection first. if _stream_attempt < _max_stream_retries: + # Final cancel check before emitting user-facing + # "Reconnecting…" status β€” the cancel flag may + # have been set during exception handling above. + if _request_cancelled["value"]: + logger.debug( + "Streaming worker cancelled before reconnect status/retry β€” exiting." + ) + return logger.info( "Streaming attempt %s/%s failed (%s: %s), " "retrying with fresh connection...", @@ -4770,6 +4834,13 @@ def _call(): e, ) + # Cancel check before falling back β€” avoid spinning up + # a non-streaming request that will just get force-closed. + if _request_cancelled["value"]: + logger.debug( + "Streaming worker cancelled before fallback β€” exiting." + ) + return try: # Reset stale timer β€” the non-streaming fallback # uses its own client; prevent the stale detector @@ -4845,6 +4916,17 @@ def _call(): last_chunk_time["t"] = time.time() if self._interrupt_requested: + # Mark the request cancelled so the worker's retry loop exits + # cleanly instead of treating the forced RemoteProtocolError + # as a transient network error and burning 2*180s on doomed + # retry attempts. This is the primary fix for the cascading + # interrupt hang bug. + _request_cancelled["value"] = True + logger.debug( + "Interrupt during streaming API call β€” marking request cancelled " + "and force-closing client. Subsequent transport errors on this " + "request are the expected consequence of this close, not a network bug." + ) try: if self.api_mode == "anthropic_messages": from agent.anthropic_adapter import build_anthropic_client diff --git a/tests/gateway/test_stt_config.py b/tests/gateway/test_stt_config.py index 436afd7c175d9..700ea295cbbdd 100644 --- a/tests/gateway/test_stt_config.py +++ b/tests/gateway/test_stt_config.py @@ -44,13 +44,14 @@ async def test_enrich_message_with_transcription_skips_when_stt_disabled(): "tools.transcription_tools.get_stt_model_from_config", return_value=None, ): - result = await runner._enrich_message_with_transcription( + result, transcripts = await runner._enrich_message_with_transcription( "caption", ["/tmp/voice.ogg"], ) assert "transcription is disabled" in result.lower() assert "caption" in result + assert transcripts == [] @pytest.mark.asyncio @@ -67,7 +68,7 @@ async def test_enrich_message_with_transcription_avoids_bogus_no_provider_messag "tools.transcription_tools.get_stt_model_from_config", return_value=None, ): - result = await runner._enrich_message_with_transcription( + result, transcripts = await runner._enrich_message_with_transcription( "caption", ["/tmp/voice.ogg"], ) @@ -75,3 +76,4 @@ async def test_enrich_message_with_transcription_avoids_bogus_no_provider_messag assert "No STT provider is configured" not in result assert "trouble transcribing" in result assert "caption" in result + assert transcripts == [] diff --git a/tests/test_cascading_interrupt.py b/tests/test_cascading_interrupt.py new file mode 100644 index 0000000000000..e4a01efa2f1db --- /dev/null +++ b/tests/test_cascading_interrupt.py @@ -0,0 +1,279 @@ +import logging +import threading +import time +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import httpx +import pytest + +from tools.interrupt import set_interrupt + + +def _make_agent(): + from run_agent import AIAgent + + with ( + patch("run_agent.get_tool_definitions", return_value=[]), + patch("run_agent.check_toolset_requirements", return_value={}), + patch("run_agent.OpenAI"), + ): + agent = AIAgent( + model="test/model", + api_key="test-key", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + ) + agent.api_mode = "chat_completions" + agent._emit_status = MagicMock() + agent._replace_primary_openai_client = MagicMock(return_value=True) + return agent + + +def _make_stream_chunk( + content=None, finish_reason=None, model="test-model", usage=None +): + delta = SimpleNamespace( + content=content, + tool_calls=None, + reasoning_content=None, + reasoning=None, + ) + choice = SimpleNamespace(index=0, delta=delta, finish_reason=finish_reason) + return SimpleNamespace(choices=[choice], model=model, usage=usage) + + +def _make_empty_chunk(model="test-model", usage=None): + return SimpleNamespace(choices=[], model=model, usage=usage) + + +@pytest.fixture(autouse=True) +def _reset_interrupt_state(): + set_interrupt(False) + yield + set_interrupt(False) + + +def _run_in_thread(target): + holder = {} + + def _wrapped(): + try: + holder["result"] = target() + except BaseException as exc: # noqa: BLE001 + holder["error"] = exc + + thread = threading.Thread(target=_wrapped, daemon=True) + thread.start() + return thread, holder + + +def test_interrupt_during_stream_does_not_retry(): + agent = _make_agent() + started = threading.Event() + closed = threading.Event() + release = threading.Event() + mock_client = MagicMock() + + def create(**kwargs): + started.set() + assert kwargs["stream"] is True + release.wait(timeout=2) + raise httpx.RemoteProtocolError("interrupt-induced close") + + mock_client.chat.completions.create.side_effect = create + agent._create_request_openai_client = MagicMock(return_value=mock_client) + + def close_request(client, *, reason): + if reason == "stream_interrupt_abort": + closed.set() + + agent._close_request_openai_client = MagicMock(side_effect=close_request) + + start = time.monotonic() + thread, holder = _run_in_thread( + lambda: agent._interruptible_streaming_api_call( + {"model": "test", "messages": []} + ) + ) + + assert started.wait(timeout=1) + agent.interrupt("stop") + assert closed.wait(timeout=1) + release.set() + thread.join(timeout=2) + elapsed = time.monotonic() - start + + assert not thread.is_alive() + assert isinstance(holder.get("error"), InterruptedError) + assert mock_client.chat.completions.create.call_count == 1 + assert not any( + "Reconnecting" in call.args[0] for call in agent._emit_status.call_args_list + ) + assert elapsed < 2 + + +def test_cached_agent_after_interrupt_second_turn_clean(): + agent = _make_agent() + first_started = threading.Event() + first_closed = threading.Event() + first_release = threading.Event() + mock_client = MagicMock() + + def create(**kwargs): + if mock_client.chat.completions.create.call_count == 0: + raise AssertionError("unexpected mock state") + if mock_client.chat.completions.create.call_count == 1: + first_started.set() + first_release.wait(timeout=2) + raise httpx.RemoteProtocolError("interrupt-induced close") + return iter( + [ + _make_stream_chunk(content="second turn", finish_reason="stop"), + _make_empty_chunk(), + ] + ) + + mock_client.chat.completions.create.side_effect = create + agent._create_request_openai_client = MagicMock(return_value=mock_client) + + def close_request(client, *, reason): + if reason == "stream_interrupt_abort": + first_closed.set() + + agent._close_request_openai_client = MagicMock(side_effect=close_request) + + start = time.monotonic() + thread, holder = _run_in_thread( + lambda: agent._interruptible_streaming_api_call( + {"model": "test", "messages": []} + ) + ) + assert first_started.wait(timeout=1) + agent.interrupt("stop") + assert first_closed.wait(timeout=1) + first_release.set() + thread.join(timeout=2) + + assert isinstance(holder.get("error"), InterruptedError) + + agent.clear_interrupt() + second = agent._interruptible_streaming_api_call({"model": "test", "messages": []}) + elapsed = time.monotonic() - start + + assert second.choices[0].message.content == "second turn" + assert elapsed < 3 + assert not any( + "Reconnecting" in call.args[0] for call in agent._emit_status.call_args_list + ) + + +def test_interrupt_during_non_streaming_does_not_leak_error(): + agent = _make_agent() + started = threading.Event() + closed = threading.Event() + release = threading.Event() + mock_client = MagicMock() + + def create(**kwargs): + started.set() + release.wait(timeout=2) + raise httpx.RemoteProtocolError("interrupt-induced close") + + mock_client.chat.completions.create.side_effect = create + agent._create_request_openai_client = MagicMock(return_value=mock_client) + + def close_request(client, *, reason): + if reason == "interrupt_abort": + closed.set() + + agent._close_request_openai_client = MagicMock(side_effect=close_request) + + start = time.monotonic() + thread, holder = _run_in_thread( + lambda: agent._interruptible_api_call({"model": "test", "messages": []}) + ) + + assert started.wait(timeout=1) + agent.interrupt("stop") + assert closed.wait(timeout=1) + release.set() + thread.join(timeout=2) + elapsed = time.monotonic() - start + + assert not thread.is_alive() + assert isinstance(holder.get("error"), InterruptedError) + assert not isinstance(holder.get("error"), httpx.RemoteProtocolError) + assert elapsed < 2 + + +def test_logged_as_cancellation_not_reconnect(caplog): + agent = _make_agent() + started = threading.Event() + closed = threading.Event() + release = threading.Event() + mock_client = MagicMock() + + def create(**kwargs): + started.set() + release.wait(timeout=2) + raise httpx.RemoteProtocolError("interrupt-induced close") + + mock_client.chat.completions.create.side_effect = create + agent._create_request_openai_client = MagicMock(return_value=mock_client) + + def close_request(client, *, reason): + if reason == "stream_interrupt_abort": + closed.set() + + agent._close_request_openai_client = MagicMock(side_effect=close_request) + + caplog.set_level(logging.DEBUG) + caplog.set_level(logging.DEBUG, logger="run_agent") + thread, holder = _run_in_thread( + lambda: agent._interruptible_streaming_api_call( + {"model": "test", "messages": []} + ) + ) + + assert started.wait(timeout=1) + agent.interrupt("stop") + assert closed.wait(timeout=1) + release.set() + thread.join(timeout=2) + + assert isinstance(holder.get("error"), InterruptedError) + messages = [record.getMessage() for record in caplog.records] + assert any( + "not a network bug" in message.lower() + and ("interrupt" in message.lower() or "cancellation" in message.lower()) + for message in messages + ) + assert not any(message.startswith("Streaming attempt ") for message in messages) + + +def test_normal_transient_error_still_retries(): + agent = _make_agent() + mock_client = MagicMock() + mock_client.chat.completions.create.side_effect = [ + httpx.RemoteProtocolError("socket closed"), + iter( + [ + _make_stream_chunk(content="ok", finish_reason="stop"), + _make_empty_chunk(), + ] + ), + ] + agent._create_request_openai_client = MagicMock(return_value=mock_client) + agent._close_request_openai_client = MagicMock() + + response = agent._interruptible_streaming_api_call( + {"model": "test", "messages": []} + ) + + assert mock_client.chat.completions.create.call_count == 2 + assert response.choices[0].message.content == "ok" + assert any( + "Reconnecting" in call.args[0] for call in agent._emit_status.call_args_list + )