(stt): reset the retry budget on any event, not only on a final transcript

livekit-agents/livekit/agents/stt/multi_speaker_adapter.py

@@ -139,9 +139,14 @@ async def _metrics_monitor_task(self, event_aiter: AsyncIterable[SpeechEvent]) -
         # the wrapped stream already reports usage for this audio and the adapter re-emits it,
         # so measuring the forwarded events here would count every recognition twice. the
         # retry-count reset still has to happen, otherwise hiccups spread over a long call
-        # accumulate instead of being forgiven by a successful transcript.
+        # accumulate instead of being forgiven by a successful provider response.
         async for ev in event_aiter:
-            if ev.type == SpeechEventType.FINAL_TRANSCRIPT:
+            if ev.type in (
+                SpeechEventType.INTERIM_TRANSCRIPT,
+                SpeechEventType.PREFLIGHT_TRANSCRIPT,
+                SpeechEventType.FINAL_TRANSCRIPT,
+                SpeechEventType.RECOGNITION_USAGE,
+            ):
                 self._num_retries = 0
 
     async def _run(self) -> None:

livekit-agents/livekit/agents/stt/stt.py

@@ -538,8 +538,15 @@ async def _metrics_monitor_task(self, event_aiter: AsyncIterable[SpeechEvent]) -
                 )
 
                 self._stt.emit("metrics_collected", stt_metrics)
-            elif ev.type == SpeechEventType.FINAL_TRANSCRIPT:
-                # reset the retry count after a successful recognition
+
+            if ev.type in (
+                SpeechEventType.INTERIM_TRANSCRIPT,
+                SpeechEventType.PREFLIGHT_TRANSCRIPT,
+                SpeechEventType.FINAL_TRANSCRIPT,
+                SpeechEventType.RECOGNITION_USAGE,
+            ):
+                # START/END_OF_SPEECH can be synthesized by adapters, so only provider
+                # responses prove the underlying recognition connection has recovered.
                 self._num_retries = 0
 
     def push_frame(self, frame: rtc.AudioFrame) -> None:

tests/test_multi_speaker_adapter.py

@@ -120,7 +120,7 @@ async def _run(self) -> None:
 
 
 class _HiccupSTT(_RecordingSTT):
-    """Its stream delivers a good transcript and then drops the connection, every attempt.
+    """Its stream delivers a provider health event and then drops, every attempt.
 
     ``max_retry=0`` on the inner stream mirrors a plugin that has already spent its own retry
     budget, so each hiccup propagates to the adapter's wrapper.
@@ -141,9 +141,9 @@ class _HiccupStream(RecognizeStream):
     async def _run(self) -> None:
         self._event_ch.send_nowait(
             SpeechEvent(
-                type=SpeechEventType.FINAL_TRANSCRIPT,
+                type=SpeechEventType.RECOGNITION_USAGE,
                 request_id="req-1",
-                alternatives=[SpeechData(language="en", text="hello", speaker_id="A")],
+                recognition_usage=RecognitionUsage(audio_duration=1.0),
             )
         )
         await asyncio.sleep(0)
@@ -273,35 +273,36 @@ async def test_stream_reports_usage_once() -> None:
     assert received[0].audio_duration == 1.0
 
 
-async def test_successful_transcript_forgives_earlier_hiccups() -> None:
+async def test_provider_response_forgives_earlier_hiccups() -> None:
     """Suppressing the adapter's own metrics must not drop the retry-count reset.
 
     ``RecognizeStream._main_task`` gives up once ``_num_retries`` exceeds ``max_retry``, and
-    the base metrics monitor resets it on every final transcript. Without that reset,
+    the base metrics monitor resets it on every provider response. Without that reset,
     unrelated brief failures accumulate over a long call until recognition dies for good.
     """
     inner = _HiccupSTT()
     adapter = MultiSpeakerAdapter(stt=inner)
     conn = APIConnectOptions(max_retry=3, retry_interval=0.0, timeout=5.0)
     stream = adapter.stream(conn_options=conn)
 
-    transcripts: list[str] = []
+    usage_events = 0
     fatal: list[BaseException] = []
 
     async def _read() -> None:
+        nonlocal usage_events
         try:
             async for ev in stream:
-                if ev.type == SpeechEventType.FINAL_TRANSCRIPT and ev.alternatives[0].text:
-                    transcripts.append(ev.alternatives[0].text)
+                if ev.type == SpeechEventType.RECOGNITION_USAGE:
+                    usage_events += 1
         except Exception as e:  # noqa: BLE001 - recorded, asserted on below
             fatal.append(e)
 
     task = asyncio.create_task(_read())
     stream.push_frame(_silence())
-    while len(transcripts) < 5 and not fatal:
+    while usage_events < 5 and not fatal:
         await asyncio.sleep(0)
 
-    assert not fatal, f"recognition died after {len(transcripts)} good transcripts: {fatal}"
+    assert not fatal, f"recognition died after {usage_events} provider responses: {fatal}"
     assert stream._num_retries == 0
 
     await utils.aio.cancel_and_wait(task)

tests/test_stt_base.py

@@ -3,13 +3,15 @@
 from __future__ import annotations
 
 import asyncio
+import dataclasses
 import time
 
 import pytest
 
 from livekit.agents import APIConnectionError, APIStatusError
 from livekit.agents.stt import (
     STT,
+    RecognitionUsage,
     RecognizeStream,
     SpeechData,
     SpeechEvent,
@@ -180,3 +182,82 @@ async def test_stream_adapter_keeps_vad_speech_end_on_delayed_final(
     assert end_event.speech_end_time is not None
     assert final_event.speech_end_time == end_event.speech_end_time
     assert final_event.created_at - end_event.created_at == pytest.approx(0.5, abs=0.01)
+
+
+class _FlappingStream(RecognizeStream):
+    """Every connection succeeds, then drops. A real reconnect, every time."""
+
+    def __init__(
+        self,
+        *,
+        stt: STT,
+        drops: int,
+        emit: SpeechEvent | None,
+    ) -> None:
+        super().__init__(
+            stt=stt,
+            conn_options=dataclasses.replace(DEFAULT_API_CONNECT_OPTIONS, retry_interval=0.0),
+        )
+        self.runs = 0
+        self._drops = drops
+        self._emit = emit
+
+    async def _run(self) -> None:
+        self.runs += 1
+        if self._emit is not None:
+            self._event_ch.send_nowait(self._emit)
+            await asyncio.sleep(0)
+        if self.runs <= self._drops:
+            raise APIConnectionError("socket dropped")
+        await asyncio.sleep(3600)  # healthy at last
+
+
+async def _survives(
+    emit: SpeechEvent | None,
+    *,
+    drops: int = 10,
+) -> bool:
+    stream = _FlappingStream(
+        stt=_DummySTT(),
+        drops=drops,
+        emit=emit,
+    )
+    try:
+        for _ in range(500):
+            if stream._task.done() or stream.runs > drops:
+                break
+            await asyncio.sleep(0.01)
+        if stream._task.done():
+            stream._task.exception()
+            return False
+        return True
+    finally:
+        if not stream._task.done():
+            await stream.aclose()
+
+
+@pytest.mark.parametrize(
+    "event",
+    [
+        SpeechEvent(type=SpeechEventType.INTERIM_TRANSCRIPT),
+        SpeechEvent(type=SpeechEventType.PREFLIGHT_TRANSCRIPT),
+        SpeechEvent(type=SpeechEventType.FINAL_TRANSCRIPT),
+        SpeechEvent(
+            type=SpeechEventType.RECOGNITION_USAGE,
+            recognition_usage=RecognitionUsage(audio_duration=5.0),
+        ),
+    ],
+)
+async def test_retry_budget_resets_after_provider_response(event: SpeechEvent) -> None:
+    """The budget counts consecutive failures, not the lifetime of the stream.
+
+    A response proves the connection came up, even when the caller has not produced
+    a final transcript. START/END_OF_SPEECH are excluded because adapters may emit
+    them without contacting the STT provider.
+    """
+    assert await _survives(event)
+
+
+async def test_retry_budget_still_gives_up_when_nothing_is_ever_delivered() -> None:
+    """A connection that has never delivered anything must not retry forever."""
+    assert not await _survives(None)