fix(voice): stop recreating STT stream on non-retryable errors

livekit-agents/livekit/agents/voice/audio_recognition.py

@@ -188,8 +188,9 @@ async def _stt_pump(self) -> None:
         """Iterate the STT node and forward events into *event_ch*.
 
         Owns the generator lifecycle — never cancelled during handoff, only the
-        consumer is swapped. On a connection failure the long-lived stream is
-        recreated after a backoff; the session tolerance is what closes it.
+        consumer is swapped. On a retryable failure the long-lived stream is
+        recreated after a backoff; the session tolerance is what closes it. A
+        non-retryable failure closes *audio_ch*, since nothing will read it again.
         """
         from .agent import ModelSettings
 
@@ -208,13 +209,21 @@ async def _stt_pump(self) -> None:
                         f"STT node must yield SpeechEvent, got: {type(ev)}"
                     )
                     self._event_ch.send_nowait(ev)
-            except APIError:
-                # only a connection failure is retried (it was emitted and counted by the
-                # session); any other error propagates and stops the pump
+            except APIError as e:
+                # only a retryable failure is recreated (it was emitted and counted by
+                # the session); a non-retryable one fails the same way on every stream
                 if self._is_closing():
                     return
+                if not e.retryable:
+                    # the stream already emitted this as an unrecoverable STTError and
+                    # the session counted it; recreating would just hot-loop the failure
+                    logger.warning("STT stream ended on a non-retryable error, not recreating")
+                    # nothing will read audio_ch again, so close it rather than let
+                    # _push_audio keep filling an unbounded queue for the whole session
+                    self._audio_ch.close()
+                    return
                 logger.warning(
-                    "STT stream ended on an unrecoverable error, recreating",
+                    "STT stream ended on a retryable error, recreating",
                     exc_info=True,
                 )
                 await asyncio.sleep(_STT_RECONNECT_INTERVAL)
@@ -737,7 +746,7 @@ def _push_audio(
         speech). VAD, AMD and the interruption channel always receive ``frame``.
         """
         self._sample_rate = frame.sample_rate
-        if self._stt_pipeline is not None:
+        if self._stt_pipeline is not None and not self._stt_pipeline.audio_ch.closed:
             # stamp the wall-clock anchor on the first frame to reach the pipeline
             if self._stt_pipeline.input_started_at is None:
                 self._stt_pipeline.input_started_at = time.time() - frame.duration

tests/test_agent_session.py

@@ -17,6 +17,7 @@
     AgentSession,
     AgentStateChangedEvent,
     APIConnectionError,
+    APIStatusError,
     ConversationItemAddedEvent,
     FlushSentinel,
     LanguageCode,
@@ -1648,6 +1649,117 @@ async def _collect() -> None:
         await pipeline.aclose()
 
 
+async def test_stt_pipeline_does_not_recreate_on_non_retryable_api_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from livekit.agents.voice import audio_recognition
+    from livekit.agents.voice.audio_recognition import _STTPipeline
+
+    monkeypatch.setattr(audio_recognition, "_STT_RECONNECT_INTERVAL", 0.0)
+
+    attempts = 0
+
+    async def stt_node(audio, model_settings):  # type: ignore[no-untyped-def]
+        nonlocal attempts
+        attempts += 1
+        if attempts > 2:
+            # end the stream rather than raise, so a pump that keeps recreating
+            # fails this test on the assertion instead of looping forever
+            return
+
+        # a bad request cannot succeed on a new stream; APIStatusError forces
+        # retryable=False for a 4xx outside 408/429/499
+        raise APIStatusError("invalid request", status_code=400)
+        yield  # pragma: no cover - makes this an async generator
+
+    # recreating cannot help: the same request fails the same way on every stream
+    pipeline = _STTPipeline(stt_node)
+    try:
+        events: list[SpeechEvent] = []
+
+        async def _collect() -> None:
+            async for ev in pipeline.event_ch:
+                events.append(ev)
+
+        await asyncio.wait_for(_collect(), timeout=5)
+        assert events == []
+        assert attempts == 1
+    finally:
+        await pipeline.aclose()
+
+
+async def test_stt_pipeline_closes_audio_input_after_non_retryable_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from livekit.agents.voice import audio_recognition
+    from livekit.agents.voice.audio_recognition import _STTPipeline
+
+    monkeypatch.setattr(audio_recognition, "_STT_RECONNECT_INTERVAL", 0.0)
+
+    attempts = 0
+
+    async def stt_node(audio, model_settings):  # type: ignore[no-untyped-def]
+        nonlocal attempts
+        attempts += 1
+        if attempts > 2:
+            # end the stream rather than raise, so a pump that keeps recreating
+            # reaches the assertion instead of looping forever
+            return
+
+        raise APIStatusError("invalid request", status_code=400)
+        yield  # pragma: no cover - makes this an async generator
+
+    pipeline = _STTPipeline(stt_node)
+    try:
+        # wait for the pump to finish, which closes event_ch
+        async def _drain() -> None:
+            async for _ in pipeline.event_ch:
+                pass
+
+        await asyncio.wait_for(_drain(), timeout=5)
+
+        # the pump owns the only reader, so leaving audio_ch open would let the
+        # caller keep filling an unbounded queue for the rest of the session
+        assert pipeline.audio_ch.closed
+    finally:
+        await pipeline.aclose()
+
+
+async def test_stt_pipeline_recreates_stream_after_retryable_status_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from livekit.agents.voice import audio_recognition
+    from livekit.agents.voice.audio_recognition import _STTPipeline
+
+    monkeypatch.setattr(audio_recognition, "_STT_RECONNECT_INTERVAL", 0.0)
+
+    attempts = 0
+
+    async def stt_node(audio, model_settings):  # type: ignore[no-untyped-def]
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            # a 5xx stays retryable, unlike a 4xx outside 408/429/499
+            raise APIStatusError("stt upstream failure", status_code=503)
+
+        yield SpeechEvent(
+            type=SpeechEventType.FINAL_TRANSCRIPT,
+            alternatives=[SpeechData(text="recovered", language="en")],
+        )
+        # stay open like a live stream until the pipeline is closed
+        async for _ in audio:
+            pass
+
+    pipeline = _STTPipeline(stt_node)
+    try:
+        ev = await asyncio.wait_for(pipeline.event_ch.recv(), timeout=5)
+        assert ev.type == SpeechEventType.FINAL_TRANSCRIPT
+        assert ev.alternatives[0].text == "recovered"
+        assert attempts == 2
+    finally:
+        await pipeline.aclose()
+
+
 async def test_stt_pipeline_does_not_recreate_stream_while_closing(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:

tests/test_audio_recognition_push_audio.py

@@ -27,6 +27,9 @@ def _make_recognition() -> AudioRecognition:
     ar._stt_pipeline = MagicMock()  # type: ignore[attr-defined]
     # the input anchor lives on the pipeline (see _STTPipeline.input_started_at)
     ar._stt_pipeline.input_started_at = None  # type: ignore[attr-defined]
+    # the pump closes audio_ch when it stops reading; a MagicMock attribute is
+    # truthy, so it has to be set explicitly for the open case
+    ar._stt_pipeline.audio_ch.closed = False  # type: ignore[attr-defined]
     ar._vad_ch = MagicMock()  # type: ignore[attr-defined]
     ar._interruption_ch = MagicMock()  # type: ignore[attr-defined]
     ar._session = MagicMock()  # type: ignore[attr-defined]
@@ -71,6 +74,23 @@ def test_push_audio_skips_optional_consumers_when_unset() -> None:
     ar._push_audio(_make_frame())
 
 
+def test_push_audio_skips_a_closed_stt_input() -> None:
+    """The pump closes audio_ch when it gives up on a non-retryable error.
+
+    Nothing reads it after that, so forwarding would fill an unbounded queue for
+    the rest of the session — and send_nowait on a closed Chan raises.
+    """
+    ar = _make_recognition()
+    ar._stt_pipeline.audio_ch.closed = True  # type: ignore[attr-defined]
+    frame = _make_frame()
+
+    ar._push_audio(frame)
+
+    ar._stt_pipeline.audio_ch.send_nowait.assert_not_called()  # type: ignore[attr-defined]
+    # the other consumers are unaffected
+    ar._vad_ch.send_nowait.assert_called_once_with(frame)  # type: ignore[attr-defined]
+
+
 def test_push_audio_records_sample_rate_and_input_start() -> None:
     ar = _make_recognition()
     frame = _make_frame(sample_rate=24000)