diff --git a/python/scenario/voice/__init__.py b/python/scenario/voice/__init__.py index 5852e4f1..df9b6583 100644 --- a/python/scenario/voice/__init__.py +++ b/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", diff --git a/python/scenario/voice/adapter.py b/python/scenario/voice/adapter.py index 26a9b534..ac6dddc8 100644 --- a/python/scenario/voice/adapter.py +++ b/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) diff --git a/python/scenario/voice/adapters/__init__.py b/python/scenario/voice/adapters/__init__.py index 8d660ec4..5e20bc03 100644 --- a/python/scenario/voice/adapters/__init__.py +++ b/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", diff --git a/python/scenario/voice/adapters/pipecat.py b/python/scenario/voice/adapters/pipecat.py index 7f2f65e4..4b17cede 100644 --- a/python/scenario/voice/adapters/pipecat.py +++ b/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 diff --git a/python/tests/voice/test_drain_timeout_surfacing.py b/python/tests/voice/test_drain_timeout_surfacing.py index deaa4c80..1f258619 100644 --- a/python/tests/voice/test_drain_timeout_surfacing.py +++ b/python/tests/voice/test_drain_timeout_surfacing.py @@ -211,3 +211,164 @@ async def test_tail_silence_timeout_ends_drain_without_raising(): "returned audio must contain the first chunk's data after the " f"tail-silence cutoff; got {len(result.data)} bytes" ) + + +# ---------------------------------------------------------------- # +# Part 2 — AgentStreamEndedError propagation (#498 diagnostic slice) # +# ---------------------------------------------------------------- # +# +# Two tests pin the contract for when the adapter's transport terminates: +# +# A. recv_audio raises AgentStreamEndedError on the FIRST call +# (loop crashed / peer closed before any audio) → +# _drain_agent_response PROPAGATES the error unchanged, does NOT +# wrap it in FirstChunkTimeoutError. +# +# B. recv_audio returns one real chunk THEN raises AgentStreamEndedError +# (clean close after turn-1 audio) → +# _drain_agent_response treats it like tail-silence and RETURNS the +# collected audio without raising. +# +# Both tests reference new symbols via _adapter_mod so a missing symbol +# fails only the dereferencing assertion, not import-time collection. + + +class _StreamEndedOnFirstAdapter(VoiceAgentAdapter): + """Transport terminates before any audio chunk — recv_audio raises + AgentStreamEndedError on the very first call. + + __cause__ is set to a ConnectionResetError so we can verify the cause + is threaded through unchanged by _drain_agent_response. + """ + + capabilities = AdapterCapabilities() + + _SENTINEL_MSG = "sentinel stream ended message" + _CAUSE = ConnectionResetError("peer reset") + + async def connect(self) -> None: + pass + + async def disconnect(self) -> None: + pass + + async def send_audio(self, chunk: AudioChunk) -> None: + pass + + async def recv_audio(self, timeout: float) -> AudioChunk: + # Raise the new contract exception — exactly what PipecatAgentAdapter + # will raise when its _recv_loop has terminated and the queue is drained. + # We raise it with __cause__ so the propagation test can check it too. + exc = _adapter_mod.AgentStreamEndedError(self._SENTINEL_MSG) + exc.__cause__ = self._CAUSE + raise exc + + +class _StreamEndedAfterFirstChunkAdapter(VoiceAgentAdapter): + """Transport delivers one real chunk then closes cleanly. + + Call 1: returns a non-empty AudioChunk. + Call 2+: raises AgentStreamEndedError — transport is gone. + + This mirrors the #498 scenario: turn-1 audio OK, then on the next + drain's tail recv the loop has exited. Expected: drain returns the + collected audio, same as a tail-silence asyncio.TimeoutError. + """ + + FIRST_CHUNK = AudioChunk(data=b"\x03\x04" * 600, transcript="world") + + capabilities = AdapterCapabilities() + + def __init__(self) -> None: + super().__init__() + self._calls = 0 + + async def connect(self) -> None: + pass + + async def disconnect(self) -> None: + pass + + async def send_audio(self, chunk: AudioChunk) -> None: + pass + + async def recv_audio(self, timeout: float) -> AudioChunk: + self._calls += 1 + if self._calls == 1: + return self.FIRST_CHUNK + raise _adapter_mod.AgentStreamEndedError("stream closed after first chunk") + + +@pytest.mark.asyncio +async def test_first_chunk_stream_ended_propagates_unchanged(): + """When the transport ends BEFORE the first chunk, _drain_agent_response + must propagate AgentStreamEndedError UNCHANGED — not re-wrap it as + FirstChunkTimeoutError (which would bury the real cause). + + RED against current code: AgentStreamEndedError does not yet exist in + scenario.voice.adapter, so dereferencing _adapter_mod.AgentStreamEndedError + raises AttributeError at the assertion line (not at collection time). + + The two falsifiable properties: + 1. The raised type IS AgentStreamEndedError (str matches sentinel). + 2. The raised type is NOT FirstChunkTimeoutError (anti-relabel check). + """ + adapter = _StreamEndedOnFirstAdapter() + adapter.response_timeout = _SENTINEL_TIMEOUT + + # Dereference the not-yet-defined contract symbol; fails RED if absent. + AgentStreamEndedError = _adapter_mod.AgentStreamEndedError # noqa: N806 + + with pytest.raises(AgentStreamEndedError) as excinfo: + await adapter._drain_agent_response() + + # AC1 — message is the sentinel we embedded, passed through verbatim. + assert str(excinfo.value) == _StreamEndedOnFirstAdapter._SENTINEL_MSG, ( + "AgentStreamEndedError message must propagate unchanged; " + f"got: {str(excinfo.value)!r}" + ) + + # AC2 — __cause__ is the ConnectionResetError we set, not lost or replaced. + assert excinfo.value.__cause__ is _StreamEndedOnFirstAdapter._CAUSE, ( + "AgentStreamEndedError.__cause__ must be the original ConnectionResetError; " + f"got: {excinfo.value.__cause__!r}" + ) + + # AC3 — critical anti-relabel: must NOT be a FirstChunkTimeoutError. + # If _drain_agent_response catches AgentStreamEndedError and re-wraps it + # as FirstChunkTimeoutError (as the old asyncio.TimeoutError branch did), + # this assertion catches the regression. + assert not isinstance(excinfo.value, _adapter_mod.FirstChunkTimeoutError), ( + "AgentStreamEndedError on first recv must NOT be re-wrapped as " + "FirstChunkTimeoutError — that hides the real transport failure cause" + ) + + +@pytest.mark.asyncio +async def test_tail_stream_ended_ends_drain_without_raising(): + """When the transport closes AFTER the first chunk, _drain_agent_response + must return the collected audio (treating the close like tail-silence) — + NOT raise AgentStreamEndedError. + + This is the falsifiable guard that the first-chunk propagation fix is + SCOPED to the first recv: if the fix re-raised on every recv_audio + AgentStreamEndedError, this test would catch it as a regression. + + Return shape: single collected chunk → _merge_chunks returns it + unchanged, so result.data equals the first chunk's data verbatim. + """ + adapter = _StreamEndedAfterFirstChunkAdapter() + adapter.response_timeout = _SENTINEL_TIMEOUT + + # Must RETURN, not raise. (The adapter's recv_audio dereferences + # _adapter_mod.AgentStreamEndedError directly, so a missing contract symbol + # still fails this test RED — no separate dereference needed here.) + result = await adapter._drain_agent_response() + + assert isinstance(result, AudioChunk), ( + f"drain must return an AudioChunk after tail-close; got: {result!r}" + ) + assert result.data == _StreamEndedAfterFirstChunkAdapter.FIRST_CHUNK.data, ( + "returned audio must equal the first chunk's data; " + f"got {len(result.data)} bytes" + ) diff --git a/python/tests/voice/test_pipecat_recv_loop_surfacing.py b/python/tests/voice/test_pipecat_recv_loop_surfacing.py new file mode 100644 index 00000000..f95f2d35 --- /dev/null +++ b/python/tests/voice/test_pipecat_recv_loop_surfacing.py @@ -0,0 +1,350 @@ +"""Recv-loop crash/close attributable surfacing — creds-free. + +Pins issue #498: when PipecatAgentAdapter._recv_loop terminates (crash or +clean peer close), recv_audio must raise PipecatRecvError FAST rather than +blocking the full timeout and then raising a bare RuntimeError/TimeoutError. + +Three tests, each wrapping recv_audio(timeout=30.0) in asyncio.wait_for(5s): +- If the impl were still swallowing, recv_audio would block ~30s, the outer + wait_for would fire asyncio.TimeoutError (NOT PipecatRecvError), and the + test fails with an unmistakable "didn't get PipecatRecvError" message. +- If the impl is correct, PipecatRecvError escapes before the 5s guard fires. + +The dual assertion (type + speed) with a single fixture mechanism is the only +way to prove BOTH "attributable" AND "fast fail" without a real pipecat bot. + +Creds-free: websockets.connect is monkeypatched with a scripted fake. +""" + +from __future__ import annotations + +import asyncio +import base64 +import json +from typing import Any, Optional + +import pytest + +from scenario.voice import AudioChunk, PipecatAgentAdapter + +# Import via the full module path (not from scenario.voice) so a missing +# symbol fails only at dereference, not at import-time collection. +from scenario.voice.adapters.pipecat import PipecatRecvError + + +# ---------------------------------------------------------------- # +# Scripted fake WebSocket # +# ---------------------------------------------------------------- # +# +# Modelled on _FakeWebSocket in test_pipecat_adapter.py. Additions: +# - ``crash_with`` — if set, __anext__ raises this exception instead of +# serving the next frame, simulating a recv-loop transport error. +# - Clean close: calling ``end_stream()`` or init with no crash_with and +# draining all frames will raise StopAsyncIteration from __anext__. + +_SENTINEL_CLOSE = object() + + +class _ScriptedFakeWebSocket: + """Stand-in for websockets.asyncio.client.ClientConnection. + + Serves queued frames on async iteration, then either crashes or closes + depending on construction arguments. + """ + + def __init__( + self, + *, + crash_with: Optional[Exception] = None, + ) -> None: + self.sent: list[str] = [] + self._inbox: asyncio.Queue[Any] = asyncio.Queue() + self.closed = False + # An exception to raise from __anext__ once the queue empties. + # If None, raises StopAsyncIteration (clean close). + self._crash_with = crash_with + + async def send(self, text: str) -> None: + self.sent.append(text) + + def __aiter__(self) -> "_ScriptedFakeWebSocket": + return self + + async def __anext__(self) -> Any: + item = await self._inbox.get() + if item is _SENTINEL_CLOSE: + # Signal we absorbed the close sentinel; raise stop or crash. + if self._crash_with is not None: + raise self._crash_with + raise StopAsyncIteration + return item + + async def close(self) -> None: + self.closed = True + self._inbox.put_nowait(_SENTINEL_CLOSE) + + def feed(self, frame: str) -> None: + """Enqueue one JSON text frame to serve on the next __anext__ call.""" + self._inbox.put_nowait(frame) + + def end_stream(self) -> None: + """Signal clean close to __anext__.""" + self._inbox.put_nowait(_SENTINEL_CLOSE) + + +# ---------------------------------------------------------------- # +# Helpers # +# ---------------------------------------------------------------- # + +def _make_media_frame(stream_sid: str, mulaw: bytes) -> str: + """Build a Twilio Media Streams JSON media frame from raw µ-law bytes.""" + return json.dumps( + { + "event": "media", + "streamSid": stream_sid, + "media": {"payload": base64.b64encode(mulaw).decode()}, + } + ) + + +# ---------------------------------------------------------------- # +# Fixtures # +# ---------------------------------------------------------------- # + +@pytest.fixture +def scripted_ws(monkeypatch): + """Return the fake WebSocket and monkeypatch websockets.connect. + + Caller sets up crash_with / feeds frames AFTER the fixture hands back + the fake, because the adapter only calls connect() during the test body. + + Usage:: + + fake = scripted_ws # fixture value IS the fake + fake._crash_with = ValueError("boom") # or fake.end_stream() + fake.feed(media_frame_str) + """ + fake = _ScriptedFakeWebSocket() + + async def _fake_connect(url, **_): + return fake + + monkeypatch.setattr("websockets.connect", _fake_connect) + return fake + + +# ---------------------------------------------------------------- # +# Tests # +# ---------------------------------------------------------------- # + +@pytest.mark.asyncio +async def test_recv_loop_crash_surfaces_attributable_error(scripted_ws): + """recv_loop crash → recv_audio raises PipecatRecvError fast. + + The crash exception (ValueError) must be: + - The __cause__ of PipecatRecvError. + - Mentioned in str(PipecatRecvError) so the message is attributable. + + Speed proof: recv_audio(timeout=30.0) wrapped in asyncio.wait_for(5.0). + If the impl blocked the full 30s the outer guard would fire asyncio.TimeoutError, + not PipecatRecvError, and pytest.raises would fail with an unmistakable mismatch. + """ + boom = ValueError("pipeline exploded") + scripted_ws._crash_with = boom + scripted_ws.end_stream() # immediately trigger crash on first __anext__ + + adapter = PipecatAgentAdapter(url="ws://fake/ws") + await adapter.connect() + try: + with pytest.raises(PipecatRecvError) as excinfo: + await asyncio.wait_for( + adapter.recv_audio(timeout=30.0), + timeout=5.0, # speed guard: if blocks >5s the impl is broken + ) + + # AC1 — cause is the original exception object (not a copy). + assert excinfo.value.__cause__ is boom, ( + f"PipecatRecvError.__cause__ must be the original ValueError; " + f"got: {excinfo.value.__cause__!r}" + ) + + # AC2 — message contains the crash type and text. + msg = str(excinfo.value) + assert "pipeline exploded" in msg, ( + f"PipecatRecvError message must mention the crash text; got: {msg!r}" + ) + finally: + await adapter.disconnect() + + +@pytest.mark.asyncio +async def test_recv_loop_clean_close_surfaces_attributable_error(scripted_ws): + """recv_loop clean close (bot hung up) → recv_audio raises PipecatRecvError fast. + + Clean close = bot closed the WS with no crash; __anext__ raises + StopAsyncIteration. PipecatRecvError.__cause__ must be None and the + message (lowercased) must contain 'closed' or 'hung up'. + + Speed proof: same asyncio.wait_for(5.0) guard as the crash test. + """ + # No crash_with — end_stream() signals clean StopAsyncIteration. + scripted_ws.end_stream() + + adapter = PipecatAgentAdapter(url="ws://fake/ws") + await adapter.connect() + try: + with pytest.raises(PipecatRecvError) as excinfo: + await asyncio.wait_for( + adapter.recv_audio(timeout=30.0), + timeout=5.0, + ) + + # AC1 — clean close has no cause. + assert excinfo.value.__cause__ is None, ( + f"PipecatRecvError.__cause__ must be None for a clean close; " + f"got: {excinfo.value.__cause__!r}" + ) + + # AC2 — message signals the peer closed the connection. + msg = str(excinfo.value).lower() + assert "closed" in msg or "hung up" in msg, ( + f"PipecatRecvError message (lowercased) must contain 'closed' or 'hung up'; " + f"got: {msg!r}" + ) + finally: + await adapter.disconnect() + + +@pytest.mark.asyncio +async def test_audio_then_close_returns_turn_then_surfaces_close(scripted_ws): + """Turn 1 audio delivered OK; turn 2 recv raises PipecatRecvError. + + Mirrors the #498 failure mode: + - _recv_loop delivers one 100ms batch of audio (turn 1), then the bot + closes the WebSocket cleanly. + - First recv_audio(timeout=...) returns a non-empty AudioChunk. + - Second recv_audio(timeout=30.0) raises PipecatRecvError fast (not + after a 30s hang). + + The 5s asyncio.wait_for guard on the SECOND recv_audio is the speed proof: + if the impl had not been fixed it would block the full 30s while the + inbound queue is silent, then raise a bare asyncio.TimeoutError. + """ + # 100ms of µ-law silence: 8000 samples/s × 0.1s = 800 bytes. + # This is the same batch size the recv_loop buffers before enqueueing. + mulaw_batch = b"\x7f" * 800 + + # We need the adapter connected to know its stream_sid. Connect first, + # then feed the frame (stream_sid is generated in connect()). + adapter = PipecatAgentAdapter(url="ws://fake/ws") + await adapter.connect() + try: + # Feed one 100ms media frame, then end the stream cleanly. + # connect() fabricates stream_sid; assert to narrow Optional[str] -> str. + assert adapter.stream_sid is not None + scripted_ws.feed(_make_media_frame(adapter.stream_sid, mulaw_batch)) + scripted_ws.end_stream() + + # Turn 1 recv — must succeed and return a non-empty chunk. + chunk = await adapter.recv_audio(timeout=5.0) + assert isinstance(chunk, AudioChunk), ( + f"first recv_audio must return an AudioChunk; got: {chunk!r}" + ) + assert len(chunk.data) > 0, "first AudioChunk must contain PCM16 bytes" + + # Turn 2 recv — transport is gone; must raise PipecatRecvError fast. + with pytest.raises(PipecatRecvError): + await asyncio.wait_for( + adapter.recv_audio(timeout=30.0), + timeout=5.0, # speed guard + ) + + # Turn 3 recv — the queue is now fully drained. This exercises the + # fail-fast branch (loop done + empty queue) that must NOT block or + # re-grow the queue; it still raises the same attributable error. + with pytest.raises(PipecatRecvError): + await asyncio.wait_for( + adapter.recv_audio(timeout=30.0), + timeout=5.0, + ) + finally: + await adapter.disconnect() + + +@pytest.mark.asyncio +async def test_stop_event_terminates_loop_and_surfaces_recv_error(scripted_ws): + """A Twilio ``stop`` event ends the recv loop via its ``return`` path — the + most common real termination (bot signals end-of-call). The ``finally`` must + still mark the loop done and enqueue the sentinel, so recv_audio surfaces an + attributable PipecatRecvError (clean close, no crash cause) rather than + blocking the full timeout. Guards the ``stop``-branch exit, which the other + surfacing tests do not reach. + """ + adapter = PipecatAgentAdapter(url="ws://fake/ws") + await adapter.connect() + try: + assert adapter.stream_sid is not None + scripted_ws.feed( + json.dumps({"event": "stop", "streamSid": adapter.stream_sid}) + ) + with pytest.raises(PipecatRecvError) as excinfo: + await asyncio.wait_for(adapter.recv_audio(timeout=30.0), timeout=5.0) + # stop is a graceful end → clean-close branch (no crash cause). + assert excinfo.value.__cause__ is None, ( + f"stop-event termination must have no crash cause; " + f"got: {excinfo.value.__cause__!r}" + ) + finally: + await adapter.disconnect() + + +@pytest.mark.asyncio +async def test_real_websocket_close_after_turn1_surfaces_recv_error(): + """#498 over a REAL websockets transport (no mock, no creds). + + A real in-process server sends one media turn, then closes the socket + normally. The adapter's FIRST recv_audio returns the turn-1 audio; the NEXT + recv_audio (turn 2's first chunk) must raise PipecatRecvError — proving the + fix fires against genuine websockets close semantics (normal closure ends + async-iteration cleanly in websockets 16.0), not only the unit test's + mocked StopAsyncIteration. Guards against a transport-library upgrade + silently changing close behavior so the clean-close branch stops firing. + """ + import websockets + + async def handler(ws): + # The adapter sends a synthetic `connected` then `start` frame on + # connect(); read both, then send one 100ms µ-law media batch and close + # normally. The recv loop does not filter inbound media by streamSid. + await ws.recv() # connected + await ws.recv() # start + await ws.send(json.dumps({ + "event": "media", + "streamSid": "MZtest", + "media": {"payload": base64.b64encode(b"\x7f" * 800).decode()}, + })) + await ws.close() # normal closure (code 1000) + + server = await websockets.serve(handler, "127.0.0.1", 0) + port = server.sockets[0].getsockname()[1] + a = PipecatAgentAdapter(url=f"ws://127.0.0.1:{port}/stream") + await a.connect() + try: + # Turn 1: audio arrives over the real wire. + first = await asyncio.wait_for(a.recv_audio(timeout=10.0), timeout=8.0) + assert isinstance(first, AudioChunk) and first.data, "turn-1 audio must arrive" + # Turn 2: the server already closed → attributable error, fast (the + # 8s outer guard proves it does not block the 10s response timeout). + with pytest.raises(PipecatRecvError) as ei: + await asyncio.wait_for(a.recv_audio(timeout=10.0), timeout=8.0) + # Normal close → clean-close branch: no crash cause, named in message. + assert ei.value.__cause__ is None, ( + f"normal close must hit the clean-close branch (no __cause__); " + f"got {ei.value.__cause__!r}" + ) + msg = str(ei.value).lower() + assert "closed" in msg or "hung up" in msg, msg + finally: + await a.disconnect() + server.close() + await server.wait_closed()