fix(voice/#498): surface recv-loop termination as attributable PipecatRecvError

python/scenario/voice/__init__.py

@@ -10,6 +10,8 @@
     - AudioChunk — canonical internal audio (PCM16 @ 24kHz mono)
     - AdapterCapabilities / UnsupportedCapabilityError — capability matrix
     - FirstChunkTimeoutError — attributable first-chunk recv timeout
+    - AgentStreamEndedError — recv_audio's transport terminated (crash/clean close)
+    - PipecatRecvError — Pipecat recv-loop ended (attributable; subclass of above)
     - VoiceRecording / VoiceEvent / LatencyMetrics — result-side types
     - AudioSegment — per-speaker slice of the recording
     - synthesize / STTProvider / set_stt_provider / get_stt_provider —
@@ -22,7 +24,7 @@
 
 from __future__ import annotations
 
-from .adapter import FirstChunkTimeoutError, VoiceAgentAdapter
+from .adapter import AgentStreamEndedError, FirstChunkTimeoutError, VoiceAgentAdapter
 from .adapters import (
     ComposableVoiceAgent,
     ElevenLabsAgentAdapter,
@@ -31,6 +33,7 @@
     LiveKitAgentAdapter,
     OpenAIRealtimeAgentAdapter,
     PipecatAgentAdapter,
+    PipecatRecvError,
     TwilioAgentAdapter,
     VapiAgentAdapter,
     WebRTCAgentAdapter,
@@ -57,6 +60,7 @@
 
 __all__ = [
     "AdapterCapabilities",
+    "AgentStreamEndedError",
     "AudioChunk",
     "AudioSegment",
     "CONTEXTUAL_PROMPT",
@@ -72,6 +76,7 @@
     "OpenAIRealtimeAgentAdapter",
     "OpenAISTTProvider",
     "PipecatAgentAdapter",
+    "PipecatRecvError",
     "STTProvider",
     "TwilioAgentAdapter",
     "UnsupportedCapabilityError",

python/scenario/voice/adapter.py

@@ -37,6 +37,22 @@
 """Phase marker for the first-chunk recv timeout (used in FirstChunkTimeoutError)."""
 
 
+class AgentStreamEndedError(Exception):
+    """An adapter's recv_audio raises this when the agent's audio transport has
+    TERMINATED — a background read-loop crash or a clean close by the peer — so
+    no further audio can arrive on this connection.
+
+    WHY distinct from asyncio.TimeoutError: a timeout is TRANSIENT (the agent may
+    still be mid-think; audio could still arrive), a stream-ended is TERMINAL (the
+    connection is done). _drain_agent_response treats them differently: on the
+    FIRST chunk it propagates this unchanged (it already names the real cause —
+    this is the #498 diagnostic fix), on TAIL chunks it ends the turn normally
+    (the peer closed after the agent finished speaking). Subclasses (e.g.
+    PipecatRecvError) carry a transport-specific message and chain the underlying
+    cause via __cause__.
+    """
+
+
 class FirstChunkTimeoutError(asyncio.TimeoutError):
     """Raised when the agent fails to send its first audio chunk within ``response_timeout``.
 
@@ -302,6 +318,11 @@ async def _drain_agent_response(
             first = await self.recv_audio(timeout=self.response_timeout)
         except asyncio.TimeoutError as err:
             raise FirstChunkTimeoutError(timeout=self.response_timeout) from err
+        # An AgentStreamEndedError is intentionally NOT caught here: it is not a
+        # TimeoutError, so it propagates past this handler unchanged. That
+        # preserves the real terminal cause (recv-loop crash or clean peer close)
+        # for the #498 diagnostic fix instead of masking it as a first-chunk
+        # timeout. Do NOT add a bare ``except Exception`` here.
         # First chunk arrived → agent is now speaking. Wakes anyone awaiting
         # _agent_speaking_event (the interruption path).
         if first.data and on_first_chunk is not None:
@@ -314,6 +335,11 @@ async def _drain_agent_response(
                 nxt = await self.recv_audio(timeout=self.response_tail_silence)
             except asyncio.TimeoutError:
                 break
+            except AgentStreamEndedError:
+                # The stream ended after the agent already spoke — a normal
+                # end-of-turn (the peer closed once it finished). Return the
+                # audio collected so far instead of surfacing the terminal cause.
+                break
             if not nxt.data:
                 break
             chunks.append(nxt)

python/scenario/voice/adapters/__init__.py

@@ -15,7 +15,7 @@
 from .gemini_live import GeminiLiveAgentAdapter
 from .livekit import LiveKitAgentAdapter
 from .openai_realtime import OpenAIRealtimeAgentAdapter
-from .pipecat import PipecatAgentAdapter
+from .pipecat import PipecatAgentAdapter, PipecatRecvError
 from .twilio import TwilioAgentAdapter
 from .vapi import VapiAgentAdapter
 from .webrtc import WebRTCAgentAdapter
@@ -30,6 +30,7 @@
     "OpenAIRealtimeAgentAdapter",
     "PendingTransportError",
     "PipecatAgentAdapter",
+    "PipecatRecvError",
     "TwilioAgentAdapter",
     "VapiAgentAdapter",
     "WebRTCAgentAdapter",

python/scenario/voice/adapters/pipecat.py

@@ -24,7 +24,7 @@
 import uuid
 from typing import Any, ClassVar, Literal, Optional
 
-from ..adapter import VoiceAgentAdapter
+from ..adapter import AgentStreamEndedError, VoiceAgentAdapter
 from ..audio_chunk import AudioChunk
 from ..capabilities import AdapterCapabilities
 from ._twilio_shared import (
@@ -42,6 +42,22 @@
 logger = logging.getLogger("scenario.voice.pipecat")
 
 
+_RECV_LOOP_DONE = object()
+"""Sentinel pushed onto the inbound queue when _recv_loop terminates, so a
+waiting recv_audio wakes immediately and surfaces the terminal cause instead of
+blocking until its caller's timeout fires on a queue nothing will fill (#498)."""
+
+
+class PipecatRecvError(AgentStreamEndedError):
+    """recv_audio could get no audio because the background _recv_loop ended.
+
+    Names the real reason — a crash in the read loop (decode/transport error,
+    chained via __cause__) or a clean WebSocket close by the bot — so the #498
+    2nd-turn hang surfaces an attributable error instead of a blind
+    response_timeout with an empty body.
+    """
+
+
 class PipecatAgentAdapter(VoiceAgentAdapter):
     """
     Test a running Pipecat bot via its exposed WebSocket endpoint.
@@ -103,7 +119,15 @@ def __init__(
 
         self._ws: Any = None
         self._recv_task: Optional[asyncio.Task] = None
-        self._inbound_queue: Optional[asyncio.Queue[AudioChunk]] = None
+        # Carries AudioChunks plus the _RECV_LOOP_DONE sentinel (hence Any), so a
+        # waiting recv_audio learns the loop ended instead of blocking forever.
+        self._inbound_queue: Optional[asyncio.Queue[Any]] = None
+        # Set by _recv_loop when it crashes; recv_audio reads it to name the root
+        # cause (chained via __cause__) on the PipecatRecvError it raises (#498).
+        self._recv_loop_exc: Optional[BaseException] = None
+        # Set True when _recv_loop terminates (crash or clean close). Lets
+        # recv_audio fail fast on a drained queue without re-reading it (#498).
+        self._recv_loop_done: bool = False
         # Serialises concurrent send_audio() calls — without it two paced
         # senders would interleave 20-ms mulaw frames on the wire and the
         # bot would receive corrupted audio. Used for the interruption case
@@ -135,6 +159,8 @@ async def connect(self) -> None:
             self.url, ping_interval=None, ping_timeout=None
         )
         self._inbound_queue = asyncio.Queue()
+        self._recv_loop_exc = None  # reset per fresh connection
+        self._recv_loop_done = False
         self._send_lock = asyncio.Lock()
 
         # Send the synthetic `start` event that pipecat's TwilioFrameSerializer
@@ -232,7 +258,31 @@ async def send_audio(self, chunk: AudioChunk) -> None:
     async def recv_audio(self, timeout: float) -> AudioChunk:
         self._assert_connected()
         assert self._inbound_queue is not None
-        return await asyncio.wait_for(self._inbound_queue.get(), timeout=timeout)
+        # The loop already terminated and its queue is drained → fail fast with
+        # the terminal cause instead of blocking for `timeout` on a queue nothing
+        # will fill. The flag + drained-queue check is the terminal state for the
+        # rest of this connection (no await, no re-pushed sentinel to leak).
+        if self._recv_loop_done and self._inbound_queue.empty():
+            raise self._recv_loop_ended_error() from self._recv_loop_exc
+        item = await asyncio.wait_for(self._inbound_queue.get(), timeout=timeout)
+        if item is _RECV_LOOP_DONE:
+            raise self._recv_loop_ended_error() from self._recv_loop_exc
+        return item
+
+    def _recv_loop_ended_error(self) -> PipecatRecvError:
+        # Chaining is done at the raise site (``raise ... from self._recv_loop_exc``)
+        # so ``__suppress_context__`` is set correctly and the clean-close branch
+        # (exc is None → ``from None``) gets a true empty cause.
+        exc = self._recv_loop_exc
+        if exc is not None:
+            return PipecatRecvError(
+                "pipecat recv loop crashed; no further audio will arrive: "
+                f"{type(exc).__name__}: {exc}"
+            )
+        return PipecatRecvError(
+            "pipecat bot closed the WebSocket; no further audio will arrive — the "
+            "bot hung up or its pipeline stopped without responding"
+        )
 
     async def interrupt(self) -> None:
         """Send a Twilio ``clear`` frame — the bot drops all buffered outbound
@@ -252,6 +302,7 @@ async def interrupt(self) -> None:
     async def _recv_loop(self) -> None:
         """Read frames from pipecat, decode µ-law → PCM16 24k, enqueue."""
         assert self._ws is not None and self._inbound_queue is not None
+        queue = self._inbound_queue
         buffered_mulaw = bytearray()
         BATCH_MS = 100
 
@@ -264,7 +315,7 @@ async def _recv_loop(self) -> None:
                     if len(buffered_mulaw) >= (BATCH_MS * 8):
                         pcm = mulaw8k_to_pcm16_24k(bytes(buffered_mulaw))
                         buffered_mulaw.clear()
-                        await self._inbound_queue.put(AudioChunk(data=pcm))
+                        await queue.put(AudioChunk(data=pcm))
                     continue
 
                 frame = parse_media_stream_frame(raw)
@@ -275,17 +326,32 @@ async def _recv_loop(self) -> None:
                     if len(buffered_mulaw) >= (BATCH_MS * 8):
                         pcm = mulaw8k_to_pcm16_24k(bytes(buffered_mulaw))
                         buffered_mulaw.clear()
-                        await self._inbound_queue.put(AudioChunk(data=pcm))
+                        await queue.put(AudioChunk(data=pcm))
                 elif frame.event == "stop":
                     if buffered_mulaw:
                         pcm = mulaw8k_to_pcm16_24k(bytes(buffered_mulaw))
                         buffered_mulaw.clear()
-                        await self._inbound_queue.put(AudioChunk(data=pcm))
+                        await queue.put(AudioChunk(data=pcm))
                     return
         except asyncio.CancelledError:
             raise
-        except Exception:
-            logger.warning("PipecatAgentAdapter: recv loop exited with error", exc_info=True)
+        except Exception as exc:
+            # #498: do NOT swallow. A crash here (decode error, transport reset,
+            # bot pipeline failure) used to only log + fall through, leaving the
+            # inbound queue silent so recv_audio blocked the full response_timeout
+            # with no attributable cause. Record it; recv_audio raises
+            # PipecatRecvError naming this as the root cause (chained via __cause__).
+            self._recv_loop_exc = exc
+            logger.warning("PipecatAgentAdapter: recv loop crashed", exc_info=True)
+        finally:
+            # Mark terminal, then wake any pending recv_audio: no more audio will
+            # arrive on this connection. The flag lets later recv_audio calls fail
+            # fast on the drained queue; the sentinel unblocks a getter currently
+            # awaiting an empty queue. Together they turn an indefinite wait into
+            # an immediate, attributable PipecatRecvError. (No await between the
+            # two lines, so a waiter can't observe a half-set terminal state.)
+            self._recv_loop_done = True
+            queue.put_nowait(_RECV_LOOP_DONE)
 
     # ------------------------------------------------------------------ assertions