tts.FallbackAdapter: one shared recovering_task slot lets one path suppress the other's recovery probe

[LHMQ878]

Summary

_TTSStatus has a single recovering_task slot, but it is written by both of the TTS fallback adapter's recovery paths — FallbackChunkedStream._try_recovery and FallbackSynthesizeStream._try_recovery. Because each path guards on if tts_status.recovering_task is None or tts_status.recovering_task.done(), a probe in flight on one path silently suppresses probing on the other. A provider that is up again therefore stays marked unavailable for longer than it should — or, in a process that keeps one path continuously busy recovering, indefinitely.

_STTStatus deliberately does not do this: it keeps recovering_recognize_task and recovering_stream_task as separate fields, and aclose() cancels both. The TTS adapter is the odd one out.

Where

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

@dataclass
class _TTSStatus:
    available: bool
    recovering_task: asyncio.Task[None] | None   # <-- one slot, two writers
    needs_resampling: bool

Writers:

• FallbackChunkedStream._try_recovery — guard at :185, assignment at :201
• FallbackSynthesizeStream._try_recovery — guard at :452, assignment at :484

Compare livekit-agents/livekit/agents/stt/fallback_adapter.py:

@dataclass
class _STTStatus:
    available: bool
    recovering_recognize_task: asyncio.Task[None] | None
    recovering_stream_task: asyncio.Task[None] | None

Consequences

1. Suppression (confirmed below). While one path's probe is in flight, the other path's _try_recovery sees a non-done task in the shared slot and returns without probing. A recovered provider is not detected on that path.
2. Overwrite / lost cancellation. When the second path does pass the guard (the first task having completed), it overwrites the slot. aclose() only cancels whatever is in the slot at that moment, so any task the slot no longer references cannot be cancelled by the adapter. This is the same class of problem as fix: cancel recovery tasks in LLM FallbackAdapter.aclose() #4921 on the LLM adapter, but reachable here without aclose() racing anything — the reference is simply dropped.

Reproduction

Both scripts run against plain bbf163f with no other patches. This does not depend on #6680 — the suppression is triggered entirely from the chunked path plus one stream() call.

probe_slot.py:

import asyncio, logging
from livekit.agents import APIConnectionError
from livekit.agents.tts import FallbackAdapter
logging.basicConfig(level=logging.CRITICAL)
from tests.fake_tts import FakeTTS

async def main() -> None:
    # fails without pushing audio, so _try_recovery is actually reached;
    # fake_exception_count must cover the in-stream retry or it is never marked unavailable
    fake1 = FakeTTS(fake_exception=APIConnectionError("fake1 down"), fake_exception_count=99)
    fake2 = FakeTTS(fake_audio_duration=1.0)
    adapter = FallbackAdapter([fake1, fake2], max_retry_per_tts=0)
    status = adapter._status[0]

    try:
        async with adapter.synthesize("hello one") as s:
            async for _ in s: pass
    except Exception: pass

    assert not status.available, "PROBE INVALID: fake1 was never marked unavailable"

    # make the next chunked probe hang, so it is unambiguously in flight
    fake1.update_options(fake_exception=None, fake_timeout=100.0)
    try:
        async with adapter.synthesize("hello two") as s:
            async for _ in s: pass
    except Exception: pass
    hanging = status.recovering_task
    assert hanging is not None and not hanging.done(), "PROBE INVALID: no in-flight probe"

    # streamed request on the SAME adapter and provider index
    try:
        async with adapter.stream() as st:
            st.push_text("hello three"); st.end_input()
            async for _ in st: pass
    except Exception: pass

    print("slot unchanged (streamed path did not probe):", status.recovering_task is hanging)

asyncio.run(main())

Output:

after chunked request : recovering_task = <Task finished ...FallbackChunkedStream._try_recovery...>
in-flight chunked task: <Task pending ...fallback_adapter.py:189> done = False
after streamed request: recovering_task = <Task pending ...fallback_adapter.py:189>
>>> (B) SUPPRESSION: streamed path skipped probing because the chunked
    probe still occupies the single shared slot.

The slot still holds the chunked task (fallback_adapter.py:189) after the streamed request — the streamed path never created one.

Control: the STT adapter, with separate slots, behaves differently

Same scenario against stt.FallbackAdapter (probe_slot_stt.py, using FakeSTT; note it has no fake_exception_count, and with max_retry_per_stt=0 it does not need one):

available after r1     : False
recovering_recognize   : pending=True
recovering_stream      : None
after streamed request :
   recovering_recognize_task = pending
   recovering_stream_task    = <Task pending ...stt/fallback_adapter.py:442>
>>> STT DID probe on the streamed path while the recognize probe was in flight.

So the behaviour is not a global policy that recovery should be serialized per provider — the sibling adapter explicitly allows the two paths to probe concurrently. TTS differs only because of the shared field.

Suggested fix

Mirror _STTStatus: split the field into recovering_synthesize_task and recovering_stream_task, have each _try_recovery guard and assign its own, and cancel both in aclose(). That keeps the existing "one probe per path per provider" behaviour while removing the cross-path interference, and makes the two adapters structurally identical.

I mentioned this while filing #6678/#6680 as a follow-up I'd hold back rather than fold in; opening it separately now that I've confirmed it reproduces on its own at bbf163f. Happy to send the patch — it is a small, self-contained change, and it does not conflict with #6680 (which touches retry_text, not the slot). Let me know if you'd rather have it as part of that PR.

Environment

livekit-agents at bbf163f, Python 3.12, Windows. Probes use the in-repo tests/fake_tts.py / tests/fake_stt.py; no network and no API keys.

[LHMQ878]

Correction to my own reproduction above — the defect is real, but the script I posted did not prove it. Flagging this before it costs anyone else time.

While writing the regression test I checked whether the separate-slot fix actually changes the scenario I posted. It does not:

chunked slot : pending
stream slot  : None
>>> even with a free slot the streamed path did NOT probe

The reason is FallbackSynthesizeStream._try_recovery's first statement:

retry_text = self._pushed_tokens.copy()
if not retry_text:
    return

In my scenario fake1 was already unavailable when the streamed request ran, so _run skipped it, never forwarded a token, and _try_recovery returned at that guard — before ever looking at the slot. The slot was irrelevant. My conclusion happened to be right for the wrong reason, which is worse than being wrong.

The direction that does isolate the slot

FallbackChunkedStream._try_recovery has no such early return, so drive it the other way: let the streamed path leave a probe in flight, then let the chunked path try. The observable is provider-level and names no private field — does fake1 receive a chunked synthesize() call at all?

To keep the probe pending deterministically I gate it on an asyncio.Event that is never set, rather than a timer that could race:

class _GateTTS(FakeTTS):
    """FakeTTS whose recovery probe (the second `stream()`) blocks until `gate` is set."""
    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)
        self.gate = asyncio.Event()
        self._stream_calls = 0

    def stream(self, *, conn_options=DEFAULT_API_CONNECT_OPTIONS):
        self._stream_calls += 1
        if self._stream_calls == 1:
            return super().stream(conn_options=conn_options)
        gate = self.gate

        class _Blocking(FakeSynthesizeStream):
            async def _run(self, output_emitter) -> None:
                await gate.wait()
                await super()._run(output_emitter)

        stream = _Blocking(tts=self, conn_options=conn_options)
        self._stream_ch.send_nowait(stream)
        return stream

fake1 = _GateTTS(fake_exception=APIConnectionError("fake1 failed"), fake_exception_count=99)
fake2 = FakeTTS(fake_audio_duration=1.0)
adapter = FallbackAdapter([fake1, fake2], max_retry_per_tts=0)

# streamed request: fake1 fails only *after* tokens were pushed, so the streamed recovery
# clears the `retry_text` guard and then parks on the gate.
async with adapter.stream() as st:
    st.push_text("hello test"); st.end_input()
    async for _ in st: pass

await asyncio.wait_for(fake1.stream_ch.recv(), 1.0)   # the failing attempt
await asyncio.wait_for(fake1.stream_ch.recv(), 1.0)   # the recovery probe, now gated

# chunked request: no guard on this path, so only the slot can stop it probing.
async with adapter.synthesize("hello test") as s:
    async for _ in s: pass

await asyncio.wait_for(fake1.synthesize_ch.recv(), 1.0)  # times out at bbf163f

A/B at bbf163f vs. the split-field fix, same script:

bbf163f      : chunked path never probed fake1 -- the streamed probe still holds the shared slot
split fields : chunked path DID probe fake1

So the reported behaviour stands, in the streamed → chunked direction; the chunked → streamed direction I originally posted is masked by the retry_text guard and is not evidence either way. (The STT control is unaffected — I re-checked, and FallbackRecognizeStream._try_recovery has no analogous early return, so it really was comparing like with like.)

Consequence 2 in the original post (overwrite / lost cancellation) is unchanged and is now covered by an assertion that aclose() cancels the still-gated streamed probe after the chunked path has written its own slot.

Two things worth noting for whoever picks this up:

• This interacts with fix(tts): probe a skipped provider so the streamed path can recover it #6680. That PR removes the if not retry_text: return guard, which un-masks the chunked → streamed direction — so after it lands, the shared slot becomes reachable from both directions rather than one. The two changes are still independent (fix(tts): probe a skipped provider so the streamed path can recover it #6680 touches retry_text, this touches the slot) and don't conflict textually.
• Sending the patch now as a PR, with the regression test written in the direction that actually discriminates: failing at bbf163f on the behavioural assertion, passing with the fix.