feat(realtime): add response-level retry and recovery for generate_reply

[ByteMaster-1]

Fixes #6205

Motivation
Realtime models have no response-level retry, unlike the pipeline LLM (which retries via LLMStream). When a realtime generate_reply fails for a recoverable reason — a timeout, a transient server error, or a response discarded by an automatic session reconnection — the turn goes silent with no built-in recovery. In production this is dead air: a single dropped OpenAI S2S socket produces a swallowed generate_reply timeout and an APIConnectionError; the SDK reconnects but discards the in-flight reply and never re-issues it.

Builds on #6304 (failures made observable via SpeechHandle.exception()) and adds the missing recovery.

What this changes
Error classification (llm/realtime.py)

RealtimeError.recoverable so the exception path agrees with the event path.
Shared is_fatal_error() + FATAL_REALTIME_ERROR_CODES (quota/auth/billing) — walks the wrapped-error chain; terminal errors are never retried.
Reconnect coordination (llm/realtime.py + OpenAI plugin)

New session_reconnecting event, RealtimeSession.reconnecting, and await wait_reconnected(timeout); the plugin drives _set_reconnecting() / _set_reconnected(). session_reconnected now logged at INFO.
Connection resilience (OpenAI plugin)

Fixes an infinite reconnect: num_retries was reset on every socket open, so a connect-then-drop endpoint looped forever. Now resets only after a connection stays healthy (_MIN_HEALTHY_CONNECTION_DURATION).
generate_reply timeout and response.done/error events tagged recoverable via the classifier.
New cancel_and_wait() cancels the active response and awaits its response.done before re-issuing, avoiding conversation_already_has_active_response.
Retry loop (voice/agent_activity.py)

_realtime_reply_task retries generate_reply for failures before the reply starts (all raise from the returned future): re-issue after cancel_and_wait() and, if reconnecting, after wait_reconnected(). Fatal → fail fast; exhaustion → error recorded on the handle.
update_chat_ctx push failures handled (timeout → proceed; other → surface without replying).
GenerationCreatedEvent.done_fut wired through.
Known limitation / deliberate scope
Post-response.created failures are observable but not yet surfaced or retried. A reply that starts and then fails (response.done: failed, or completes empty) sets GenerationCreatedEvent.done_fut, but the generation/playback task does not yet await it — so today such a failure still ends the turn as if it completed. Closing this requires the playback task to await done_fut, thread through whether any audio played, distinguish cancelled (user interrupt) from failed/incomplete, and either silently re-issue (no audio yet) or surface for an apology-and-continue. That is a larger change to the streaming path and is intentionally left as a follow-up; this PR wires the done_fut groundwork for it.

Non-goals
Retry lives in the voice turn path (all AgentSession usage). Direct RealtimeSession.generate_reply() callers are out of scope: the returned future resolves at response.created, so a complete retry must live where completion is awaited.
UI notification / room teardown are application policy, not included.
Serializing independent overlapping replies (a server-VAD auto-reply racing an app generate_reply) is out of scope — see the separate collision issue. This PR's cancel_and_wait only guards our own retries.
Testing
New tests/test_realtime_retry.py (22): recoverable, is_fatal_error (incl. cycle-safety), reconnect state machine.
Extended tests/test_realtime/test_openai_realtime_model.py (+6): _handle_error classification, cancel_and_wait.
All --unit green; no regressions in existing realtime/speech-handle suites.

[ByteMaster-1]

Hi @longcw @davidzhao — would appreciate a review when you have a moment. This adds response-level retry for realtime generate_reply #6205 , building on @longcw's SpeechHandle.exception() work in #6304

A few design decisions I'd especially like your take on:

Retry placement. I put the retry in the voice turn path (_realtime_reply_task) rather than in RealtimeModel.generate_reply, because the returned future resolves at response.created (needed for streaming), so a complete retry has to live where completion is awaited. This covers all AgentSession usage but not direct RealtimeSession.generate_reply() callers — happy to add a model-level path if you'd prefer that instead.

New base-RealtimeSession surface. I added a session_reconnecting event + reconnecting/wait_reconnected() and a cancel_and_wait() (default falls back to interrupt()), driven by the OpenAI plugin. Want to confirm you're OK extending the base API this way vs. keeping it OpenAI-local.

Deliberately scoped out (called out in the PR): post-response.created failures are made observable (via GenerationCreatedEvent.done_fut) but not yet retried/surfaced — that needs changes to the streaming path and is a follow-up. The response.create collision for independent overlapping replies is tracked separately.
Also fixes an infinite-reconnect bug (num_retries reset on every socket open). Tests are --unit and green. Thanks!

[devin-ai-integration]

Devin Review found 2 new potential issues.

[devin-ai-integration]

🔍 Fallback adapter does not forward 'session_reconnecting' or set reconnecting state

The new "session_reconnecting" event is added to EventTypes (realtime.py:199) but is absent from _FORWARDED_EVENTS (realtime_fallback_adapter.py:57-65). Additionally, _FallbackRealtimeSession._swap() calls self._set_reconnected() at realtime_fallback_adapter.py:338 without ever calling self._set_reconnecting() first. This means:

1. The reconnecting property on a fallback session is always False.
2. The retry loop's wait_reconnected() call at agent_activity.py:3320 is always a no-op for fallback sessions.
3. External listeners for "session_reconnecting" on a fallback session will never fire.

This is likely acceptable today because the fallback adapter handles reconnection via model-swap rather than socket-level reconnection, and the retry loop gracefully handles the reconnecting=False case (it just skips the wait). However, if future code relies on the reconnecting state being set on fallback sessions, this asymmetry could cause issues.

(Refers to lines 57-65)

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[devin-ai-integration]

🔍 NVIDIA plugin calls _set_reconnected() without prior _set_reconnecting()

The NVIDIA plugin calls self._set_reconnected() at line 386 when a restart completes, but never calls _set_reconnecting() before entering its retry loop (lines 388-426). This means the reconnecting property is always False for NVIDIA sessions, and the retry loop in agent_activity.py:3319-3320 will never wait for NVIDIA reconnection. The OpenAI plugin correctly calls _set_reconnecting() at realtime_model.py:970 before retrying. This is a pre-existing inconsistency that this PR surfaces by introducing the reconnect state machine that the retry loop depends on.

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[longcw]

Thanks for the pr. But we already have a RealtimeModelFallbackAdapter that handles the recoverable/non-recoverable split and even re-issues the reply on swap, so the reconnect state machine doesn't need to live at the framework level — is_fatal_error can just feed _emit_error(recoverable=...) and let the existing path handle it.

My bigger concern is the auto-retry of generate_reply: in a live conversation the user may have already moved on, so silently replaying the same (stale) reply seems like a decision that should be left to the developer rather than baked into the turn loop.

[ByteMaster-1]

Thanks for the pr. But we already have a RealtimeModelFallbackAdapter that handles the recoverable/non-recoverable split and even re-issues the reply on swap, so the reconnect state machine doesn't need to live at the framework level — is_fatal_error can just feed _emit_error(recoverable=...) and let the existing path handle it.

My bigger concern is the auto-retry of generate_reply: in a live conversation the user may have already moved on, so silently replaying the same (stale) reply seems like a decision that should be left to the developer rather than baked into the turn loop.

@longcw
Thanks for the detailed review — the "adapter owns recovery" framing makes sense, and I agree the automatic re-speak of a reply is a developer decision, not something to bake into the turn loop. Let me be concrete about the production incident that motivated this, because I don't think the existing paths cover it and I want to make sure the actual bug doesn't get lost in the retry discussion.

The incident — a single OpenAI realtime model, no fallback adapter, on an account whose quota ran out mid-service:

The WebSocket connects fine — the API key is valid, so _create_ws_conn() always succeeds. Connecting was never the problem.
Quota is exhausted, so generation fails and the server tears the S2S socket down shortly after each connect.
_recv_task raises APIConnectionError("… connection closed unexpectedly"), which _main_task treats as retryable and reconnects.
The reconnect succeeds (again — connecting is fine, only generating fails), and _main_task resets num_retries = 0 after every successful reconnect.
Because the connection keeps succeeding, num_retries never reaches max_retries, so the give-up branch is unreachable — the session reconnect-loops forever. The caller hears nothing but dead air, and we keep hammering OpenAI indefinitely on an account that fundamentally can't generate a reply.
Two things combine to cause this, and neither is covered today:

The quota error is tagged recoverable=True (the handlers optimistically mark everything recoverable), so nothing treats it as terminal.
num_retries is reset on every successful socket open, not after the session has proven healthy — so a "connects, then immediately drops" endpoint loops indefinitely.
That's the heart of the PR:

is_fatal_error → _emit_error(recoverable=False) for insufficient_quota / invalid_api_key / billing errors, so the failure is terminal — the adapter can swap (if one is configured) or the session can close cleanly, instead of spinning.
_main_task resets num_retries only after the connection has stayed up long enough to be considered healthy, so connect-then-drop can no longer loop forever.
On recovery and the adapter: I fully agree it should own model-swap recovery. But I want to flag a gap: the adapter re-issues the reply only on the non-recoverable swap path — for a recoverable error it forwards and defers to "the plugin's own reconnect." The plugin's _reconnect() discards the pending reply (RealtimeError("pending response discarded due to session reconnection")) and never re-creates it, so on a transient reconnect the turn silently produces nothing even with the adapter configured. I'm not proposing to auto-replay it — I'd just like these recoverable failures to be observable (surfaced on the SpeechHandle, building on #6304 instead of silently swallowed, so the developer can decide whether to re-prompt.

Proposed scope — happy to reduce the PR to:

the is_fatal_error classification feeding _emit_error(recoverable=…),
the num_retries reconnect-loop fix (the infinite-loop bug above), and
graceful update_chat_ctx timeout handling (today a push timeout raises an unhandled RealtimeError that breaks the turn),
and drop the framework-level reconnect state machine and the turn-loop auto-retry. Would that scope work for you? And for the "recoverable error silently drops the in-flight reply" case — how would you prefer we surface it?

[devin-ai-integration]

Devin Review found 1 new potential issue.

[devin-ai-integration]

🟡 New reconnecting event not forwarded through the fallback adapter, so wrapper consumers never see it

The new "session_reconnecting" event is not included in the forwarded-events list (_FORWARDED_EVENTS at realtime_fallback_adapter.py:57-65), so consumers listening on the fallback wrapper session never receive it when the underlying child session starts reconnecting.

Impact: Any code subscribing to the "session_reconnecting" event on a fallback-adapter session silently receives nothing, breaking observability of reconnection state.

The forwarded-events list includes "session_reconnected" but omits the new counterpart

The PR adds "session_reconnecting" to EventTypes (realtime.py:197) and emits it from _set_reconnecting() (realtime.py:267). The fallback adapter's _bind() method (realtime_fallback_adapter.py:189-192) subscribes to each event in _FORWARDED_EVENTS and re-emits it on the wrapper. Since "session_reconnecting" is missing from this tuple, the child's event is silently dropped. The sibling event "session_reconnected" IS listed at line 62.

(Refers to lines 57-65)

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[longcw]

I'd split this into small PRs. For quota/"can't serve" we don't need is_fatal_error or the num_retries heuristic — just classify those as non-retryable where we parse them so the existing reconnect loop stops (can you confirm the incident arrived as an error/failed response.done with a code, or just a bare socket close?).

For speech retry, the RealtimeModelFallbackAdapter already does it (regenerate_on_swap), so let's rely on that. and yes we can make the drop on reconnect inside the plugin observable via SpeechHandle.exception().