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

[biztex]

Fixes #6678.

Problem

FallbackSynthesizeStream._try_recovery starts by copying _pushed_tokens and returns when it is empty. _pushed_tokens is filled by _forward_input_task, which is created just above the provider loop but has not run yet: for an instance skipped by if tts_status.available or all_failed, nothing between the create_task and the _try_recovery(tts) call awaits, so the probe is dropped for want of text.

The instance is only skipped when it is already unavailable, which is exactly when it needs probing — so once a provider fails, a process that only uses stream() never re-probes it and stays pinned to its fallback for the rest of its life. The chunked path is unaffected because it has the text up front, which is why the same provider recovers there on the very next request.

Thanks @LHMQ878 for the diagnosis — it was precisely right, down to the line.

Fix

The probes are started after the input has been consumed, in the finally that already awaits input_task, so they have the text they need. Instances are collected during the loop instead of being probed inside it; the success path returns through the same finally, so a provider skipped ahead of the one that worked is still probed.

Moving the call into finally introduced a second problem, also found by @LHMQ878: aclose() cancels the recovery slots exactly once, and a finally also runs on cancellation — so a stream still in flight when the adapter closes could create a probe in a slot nothing was left to sweep, leaving a live synthesis against a provider that was just closed. FallbackAdapter._closed is set before the sweep and refused in _try_recovery; because _try_recovery is synchronous, a probe is either already in a slot (and cancelled by the sweep) or refused by the flag, with no window between. Both stream types carry the guard so they don't drift.

Ordering is otherwise unchanged: an instance that fails during a request already had text by the time it was probed, and now gets probed a moment later with the same text. The STT adapter calls _try_recovery from its loop body rather than a finally, so it has no equivalent post-close path and is untouched.

Verification

• test_tts_recover_on_streamed_path drives two streamed requests: the first marks the primary unavailable, the second skips it and must still probe it. Times out waiting for the recovery event on main, passes with the fix.
• test_no_recovery_probe_after_close leaves a request mid-flight (no end_input()), closes the adapter, and asserts no slot holds a live task. Fails without the close guard.

ruff format --check, ruff check, check_types.py (mypy strict) and the full pytest --unit suite pass locally.

Merge order

This overlaps #6683, which splits _TTSStatus.recovering_task into recovering_synthesize_task / recovering_stream_task. Whichever lands second needs a small rebase: _closed and the two guards are unaffected, but aclose() sweeps two slots instead of one, and test_no_recovery_probe_after_close checks both rather than one. Happy for #6683 to go first — I'll rebase.

[devin-ai-integration]

✅ Devin Review: No Issues Found

Devin Review analyzed this PR and found no bugs or issues to report.

[LHMQ878]

Thanks for picking this up so fast — and for the extra detail about it not being only the first provider. That's a better framing than mine: it's any instance reached before something has awaited.

The fix is correct for the bug, and your regression test fails without it and passes with it (I ran tests/test_tts_fallback.py at ed27026: 9 passed).

One thing moving the call into finally introduces, though: a probe task can now be created after aclose() has already swept the slots, and nothing is left to cancel it.

aclose() cancels recovering_task exactly once, at close time. A finally also runs on cancellation, so if the adapter is closed while a streamed request is still in flight — agent shutdown racing an active synthesis — the ordering is:

aclose() sweeps the slots  ->  returns
stream unwinds  ->  finally  ->  _try_recovery()  ->  new task in a swept slot

A/B on the same probe, with fake_timeout=30.0 so a surviving probe is unmistakable rather than a timing artifact:

                     slot after adapter.aclose() + stream unwind
bbf163fe (base)      None                      -- no probe created
ed27026  (this PR)   <Task pending ...>        -- still running 1s later

The probe drives one failing request to mark fake1 unavailable, then starts a second request and never calls end_input() — so it is genuinely mid-flight, _pushed_tokens is populated, and fake1 has already been skipped into pending_recovery. Then adapter.aclose(), then cancel. The end_input() version does not reproduce: the request completes inside the sleep window, so the finally fires before aclose() and the sweep still catches the task. That's why this is easy to miss.

What survives is a real synthesis against the provider you just closed, plus an availability_changed emit on a closed adapter.

A guard that holds because _try_recovery is synchronous — a probe is therefore either already in a slot (and cancelled by the sweep) or refused by the flag, with no window in between:

     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:

     def _try_recovery(self, tts: TTS) -> None:
         assert isinstance(self._tts, FallbackAdapter)
 
+        if self._tts._closed:
+            return
+

plus self._closed = False in __init__. Worth applying to both _try_recovery methods — FallbackChunkedStream and FallbackSynthesizeStream. The chunked one isn't reachable via your finally change, but it has the same close race through its own path, and leaving one guarded and one not is the kind of asymmetry that gets re-broken later.

With that added: the leak probe reports slot=None / no probe task created after aclose, and tests/test_tts_fallback.py is still 9 passed, including your new test_tts_recover_on_streamed_path.

The STT sibling does not need this — stt/fallback_adapter.py calls _try_recovery in the loop body rather than a finally, so there's no post-close path there. This is specific to the restructuring here.

Happy to push this as a commit to your branch if that's easier, or as a follow-up PR once this lands — your call. One note on the base you're on: _TTSStatus here still has a single recovering_task slot, so if #6683 (per-path slots) lands first this guard applies to both recovering_synthesize_task and recovering_stream_task; the flag itself is unaffected either way.

[biztex]

You're right, and thank you — that's a regression my change introduced, not a pre-existing one. Applied in cba3fd1.

I reproduced it before taking the patch: with the guard stubbed out, the probe survives aclose() and the leaked-task check fires; with it, the slot is clean. Your point about end_input() masking it is what makes it nasty — the obvious version of the test passes and tells you nothing.

The reasoning for the flag holds for exactly the reason you give: _try_recovery is synchronous, so between "already in a slot" and "refused" there is no third state. That's what makes a plain bool sufficient here rather than needing to re-sweep.

Guarded both stream types as you suggested. FallbackChunkedStream isn't reachable through the finally, but the asymmetry is the kind of thing that gets re-broken by the next person who touches this, and the guard costs nothing there.

Also confirmed your read on the STT sibling — both _try_recovery call sites in stt/fallback_adapter.py are in loop bodies, so there's no post-close path to close. Left it alone.

Added test_no_recovery_probe_after_close, which is your probe: fail one request to mark fake1 unavailable, leave a second mid-flight with no end_input(), close the adapter, then assert no slot holds a live task. It fails without the guard.

No need for a follow-up PR — it's in this one. Appreciate you running the branch and doing the A/B; that was a genuinely good catch.

[LHMQ878]

Verified cba3fd18 — the guard is right and the test earns its keep.

• Suite at cba3fd18: 10 passed.
• My independent leak probe now reports slot=None / no probe task created after aclose.
• A/B'd your test_no_recovery_probe_after_close by stubbing out only the streamed guard and leaving everything else identical: 1 failed, 9 passed, failing on exactly that assertion. So it's testing the guard rather than passing incidentally — worth checking, since the end_input() variant of this test passes with or without the fix.

Nothing further from me. Thanks for the quick turnaround, and for confirming the STT read rather than just taking it.

One unrelated heads-up, since it touches the same slot this guard reads: _TTSStatus on this base has a single recovering_task, and #6683 splits it into recovering_synthesize_task / recovering_stream_task so a probe in flight on one path can't suppress the other or orphan its task. Whichever of the two lands second will need a trivial rebase — the guard becomes if self._tts._closed: return unchanged, but aclose() sweeps two slots instead of one. No action needed here; just flagging it so it isn't a surprise at merge time.

[biztex]

Thanks for re-running it, and for A/B'ing my test rather than just the fix — stubbing only the streamed guard is the right way to check it isn't passing incidentally, especially given the end_input() variant passes either way.

Noted on #6683, and one detail to add to your merge-order flag: test_no_recovery_probe_after_close reads the slot too (tts_status.recovering_task), so the rebase touches the test as well as aclose(), not just the sweep. Trivial either way — it becomes a check over both slots — but it'd fail with an AttributeError rather than an assertion, which is a confusing way to find out.

I've put a note in the PR description so whoever merges sees the interaction. Happy for #6683 to land first if that's easier for the maintainers; I'll rebase this one.

[longcw]

@biztex #6683 was merged, and btw @LHMQ878 could you close the #6680 if it's duplicated with this one?

[LHMQ878]

@longcw #6680 is already closed on my side — it was the earlier attempt at the same recovery probe and got superseded by this thread / #6683. Nothing left open from me there.

[biztex]

@LHMQ878 #6683 landed first, so this is the rebase I owed you. It's done and verified locally; I'm held up pushing it by a token-scope problem on my side (my OAuth token lacks workflow, and every base containing #6683 also contains the .github/workflows/deploy-examples.yml change from #6649, so the push is refused). Sorting that out — nothing needed from anyone here.

The resolution is exactly as small as you predicted, and your heads-up meant nothing was a surprise:

• _closed is set before the sweep, which now cancels both recovering_synthesize_task and recovering_stream_task rather than one slot. No change to the flag itself.
• Both _try_recovery guards are untouched.
• test_no_recovery_probe_after_close now checks both slots — the part that would otherwise have failed with an AttributeError rather than an assertion.
• The test conflict was purely additive: your _GateTTS block and my streamed-recovery test both landed at the end of the file, so both are kept.

tests/test_tts_fallback.py is 11 passed on the rebased branch (your two new tests included), and stubbing out the close guard still fails test_no_recovery_probe_after_close, so it hasn't gone slack across the move.

[LHMQ878]

@biztex thanks for doing the rebase — dual-slot sweep + both guards unchanged is exactly what I expected from your note. Good to know the push blocker is just the workflow scope on your token; nothing needed from me here.

[biztex]

Rebase pushed (e35eb00) — the conflict with #6683 is resolved and this is mergeable again. Details are in my note above; nothing changed since, the token-scope issue on my side is sorted.

[LHMQ878]

@biztex thanks for getting the rebase through — good to hear e35eb00 is up and mergeable again.

[LHMQ878]

Thanks @longcw for the merge, and @biztex for carrying the streamed-path fix through the rebase — glad this landed.