fix(tts): probe a skipped provider for recovery on the streamed path

livekit-agents/livekit/agents/tts/fallback_adapter.py

@@ -96,6 +96,7 @@ def __init__(
 
         self._tts_instances = tts
         self._max_retry_per_tts = max_retry_per_tts
+        self._closed = False
 
         self._status: list[_TTSStatus] = []
         for t in tts:
@@ -140,6 +141,12 @@ def _on_metrics_collected(self, *args: Any, **kwargs: Any) -> None:
         self.emit("metrics_collected", *args, **kwargs)
 
     async def aclose(self) -> None:
+        # set before the sweep: _try_recovery is synchronous, so a probe is
+        # either already in a slot (and cancelled below) or refused by this
+        # flag. A stream still in flight runs its finally after this returns,
+        # and must not start a probe that nothing is left to cancel
+        self._closed = True
+
         for tts_status in self._status:
             if tts_status.recovering_synthesize_task is not None:
                 await aio.cancel_and_wait(tts_status.recovering_synthesize_task)
@@ -193,6 +200,10 @@ async def _try_synthesize(
     def _try_recovery(self, tts: TTS) -> None:
         assert isinstance(self._tts, FallbackAdapter)
 
+        if self._tts._closed:
+            # nothing would cancel a probe started from here
+            return
+
         tts_status = self._tts._status[self._tts._tts_instances.index(tts)]
         recovering_task = tts_status.recovering_synthesize_task
         if recovering_task is None or recovering_task.done():
@@ -384,6 +395,12 @@ async def _forward_input_task() -> None:
 
         input_task = asyncio.create_task(_forward_input_task())
 
+        # a probe needs text to synthesize, and _pushed_tokens is only filled
+        # once _forward_input_task has run: an instance that is skipped because
+        # it is already unavailable is reached before that happens, so the
+        # probes are started below, after the input has been consumed
+        pending_recovery: list[TTS] = []
+
         try:
             for i, tts in enumerate(self._fallback_adapter._tts_instances):
                 tts_status = self._fallback_adapter._status[i]
@@ -446,17 +463,25 @@ async def _forward_input_task() -> None:
                             )
                             return
 
-                self._try_recovery(tts)
+                pending_recovery.append(tts)
 
             raise APIConnectionError(
                 f"all TTSs failed ({[tts.label for tts in self._fallback_adapter._tts_instances]}) after {time.time() - start_time} seconds"  # noqa: E501
             )
         finally:
             await utils.aio.cancel_and_wait(input_task)
 
+            for tts in pending_recovery:
+                self._try_recovery(tts)
+
     def _try_recovery(self, tts: TTS) -> None:
         assert isinstance(self._tts, FallbackAdapter)
 
+        if self._tts._closed:
+            # a stream still in flight when the adapter closed runs its finally
+            # after the sweep, and nothing would cancel a probe started here
+            return
+
         retry_text = self._pushed_tokens.copy()
         if not retry_text:
             return

tests/test_tts_fallback.py

@@ -406,3 +406,78 @@ async def test_recovery_is_not_suppressed_across_paths() -> None:
     assert stream_task.done()
 
     fake1.gate.set()
+
+
+async def test_tts_recover_on_streamed_path() -> None:
+    # a provider marked unavailable is skipped on later streamed requests, so
+    # nothing has awaited by the time recovery is considered and the probe used
+    # to be dropped for want of text - leaving the process pinned to its
+    # fallback for good after one transient outage (#6678)
+    fake1 = FakeTTS(fake_exception=APIConnectionError("fake1 failed"))
+    fake2 = FakeTTS(fake_audio_duration=1.0)
+
+    fallback_adapter = FallbackAdapterTester([fake1, fake2])
+
+    async def _drive() -> None:
+        async with fallback_adapter.stream() as stream:
+            stream.push_text("hello test")
+            stream.end_input()
+            async for _ in stream:
+                pass
+
+    # first request: fake1 fails and is marked unavailable
+    await _drive()
+    assert not fallback_adapter.availability_changed_ch(fake1).recv_nowait().available
+
+    fake1.update_options(fake_exception=None, fake_audio_duration=1.0)
+
+    # second request: fake1 is skipped, and must still be probed
+    await _drive()
+
+    assert (
+        await asyncio.wait_for(fallback_adapter.availability_changed_ch(fake1).recv(), 5.0)
+    ).available, "fake1 should have recovered on the streamed path"
+
+    await fallback_adapter.aclose()
+
+
+async def test_no_recovery_probe_after_close() -> None:
+    # aclose() cancels the recovery slots once; a stream still in flight runs
+    # its finally afterwards, and a probe started there would have nothing left
+    # to cancel it - a live synthesis against a provider that was just closed
+    fake1 = FakeTTS(fake_exception=APIConnectionError("fake1 failed"))
+    fake2 = FakeTTS(fake_audio_duration=1.0)
+
+    fallback_adapter = FallbackAdapterTester([fake1, fake2])
+
+    # first request: fake1 fails and is marked unavailable
+    async with fallback_adapter.stream() as stream:
+        stream.push_text("hello test")
+        stream.end_input()
+        async for _ in stream:
+            pass
+
+    assert not fallback_adapter.availability_changed_ch(fake1).recv_nowait().available
+
+    # second request left mid-flight: _pushed_tokens is populated and fake1 has
+    # been skipped, so its probe is pending when the adapter closes
+    fake1.update_options(fake_timeout=30.0, fake_exception=None)
+    stream = fallback_adapter.stream()
+    stream.push_text("hello test")  # deliberately no end_input()
+
+    drain = asyncio.create_task(_drain_stream(stream))
+    await asyncio.sleep(0.5)
+
+    await fallback_adapter.aclose()
+    await utils.aio.cancel_and_wait(drain)
+    await stream.aclose()
+
+    for tts_status in fallback_adapter._status:
+        for task in (tts_status.recovering_synthesize_task, tts_status.recovering_stream_task):
+            assert task is None or task.done(), "a recovery probe outlived aclose()"
+
+
+async def _drain_stream(stream: SynthesizeStream) -> None:
+    with contextlib.suppress(Exception):
+        async for _ in stream:
+            pass