fix(openai): stop realtime reconnect loop on fatal server errors (insufficient_quota, invalid_api_key)

[ByteMaster-1]

Addresses the production failure in #6205 . Rescoped per review (https://github.com/livekit/agents/pull/6352#issuecomment-4920965533)— see "Scoped out" below.

Motivation
On an OpenAI account whose quota ran out mid-service, the realtime session reconnect-loops forever while the caller hears dead air. From the production capture, every ~300ms cycle is:

error=APIError('OpenAI Realtime API returned an error',
body=RealtimeError(message='You exceeded your current quota, ...',
type='insufficient_quota', code='insufficient_quota'), retryable=True) recoverable=True
error=APIConnectionError('OpenAI S2S connection closed unexpectedly', body=None, retryable=True) recoverable=True
Two things combine to cause this, and neither is handled today:

The quota error is emitted with recoverable=True (the handlers optimistically mark everything recoverable), so nothing — including RealtimeModelFallbackAdapter, whose swap is triggered by recoverable=False — treats it as terminal.
The socket itself always reconnects fine (the key is valid; only generation fails, then the server closes). _main_task resets num_retries on every successful connect, so the give-up branch is unreachable and the loop never ends.
Meanwhile a pending generate_reply() sits out its full 10s timeout and fails with a generic "generate_reply timed out" that says nothing about quota.

Changes (OpenAI plugin only)
_FATAL_ERROR_CODES (insufficient_quota, invalid_api_key, account_deactivated, billing_hard_limit_reached) — codes that can never succeed on retry — classified where errors are parsed: _handle_error (server error events) and _handle_response_done_but_not_complete (failed response.done). These now emit recoverable=False instead of the hardcoded True; everything else keeps the optimistic default.
When a fatal code is seen, the session records it and _main_task stops reconnecting (raises the fatal error instead of retrying), ending the infinite loop. No retry-budget heuristics needed — the fatal error event arrives on every cycle, so classification alone is sufficient.
Pending generate_reply futures fail immediately with the actual code ("openai returned a fatal error: insufficient_quota") instead of waiting out the timeout.
With recoverable=False now emitted correctly, the existing recovery path works as designed: RealtimeModelFallbackAdapter swaps to the next model (and re-issues the reply via regenerate_on_swap), or a bare session fails fast and loud instead of spinning silently.

Scoped out (per review)
Earlier revisions of this PR added a framework-level reconnect state machine and an automatic generate_reply retry in the turn loop; both were dropped — speech retry belongs to RealtimeModelFallbackAdapter / the application. Two small follow-up PRs planned:

Make the pending-reply drop in _reconnect() observable via SpeechHandle.exception() (building on #6304.
Graceful handling of the update_chat_ctx timeout inside generate_reply (today an unhandled RealtimeError).
Testing
8 new unit tests in tests/test_realtime/test_openai_realtime_model.py: fatal-code matcher, classification at both parse sites (fatal + transient), fail-fast of pending reply futures, transient errors left untouched, Cancellation failed early-return.
ruff format --check, ruff check, full mypy (agents + all plugins), and pytest tests/test_realtime --unit all green.

[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!

[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?

[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().

[ByteMaster-1]

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().

@longcw
Confirmed — it arrived as an error event with a code, followed ~150ms later by the socket close, repeating every ~300ms. From the production capture:

error=APIError('OpenAI Realtime API returned an error',
body=RealtimeError(message='You exceeded your current quota, please check your plan and billing details. ...',
type='insufficient_quota', code='insufficient_quota'), retryable=True) recoverable=True
error=APIConnectionError('OpenAI S2S connection closed unexpectedly', body=None, retryable=True) recoverable=True
Three full error→close→reconnect cycles between t=…131.74 and t=…132.34 — the loop was hammering the API ~3x/sec, every error labeled recoverable=True.

Since a parseable error event arrives on every cycle, agreed on all points — I've rescoped this PR to exactly what you described, and dropped everything else:

Kept: fatal codes (insufficient_quota, invalid_api_key, account_deactivated, billing_hard_limit_reached) are classified as recoverable=False where they're parsed (_handle_error and failed response.done), and _main_task stops reconnecting once one has been seen. Pending generate_reply futures fail immediately with the actual code instead of waiting out the 10s timeout. Diff is now ~170 lines, OpenAI plugin + tests only.
Dropped: the framework reconnect state machine, the num_retries heuristic, and the turn-loop auto-retry — speech retry stays with RealtimeModelFallbackAdapter (regenerate_on_swap) and the app.
I'll open two small follow-up PRs: (1) making the pending-reply drop in _reconnect() observable via SpeechHandle.exception() as you suggested, and (2) graceful handling of the update_chat_ctx timeout inside generate_reply.

[ByteMaster-1]

@longcw can we please check this if possible ,
thanks

[longcw]

we can raise the APIError with retryable=False from the handler and re-raise from recv_task with

except Exception as e:
    # terminal server errors (e.g. insufficient_quota) must break the recv loop so
    # _main_task stops reconnecting; every other handler failure is logged and skipped
    if isinstance(e, APIError) and not e.retryable:
        raise
    if event["type"] == "response.output_audio.delta":
        event["delta"] = event["delta"][:10] + "..."
    logger.exception("failed to handle event", extra={"event": event})

self._fatal_error is not needed then, and actually setting this won't break the run_ws.

[ByteMaster-1]

Done — the handlers now raise APIError(retryable=False) and the recv loop re-raises non-retryable APIErrors (used your snippet); _main_task's existing non-retryable branch handles the emit + stop. _fatal_error removed.

[longcw]

if raising APIError from the run_ws, this is not needed, the retryable=False case is already handled.

[ByteMaster-1]

Removed — not needed with the raise-based flow.

[longcw]

same as above

[ByteMaster-1]

Removed — not needed with the raise-based flow.

[longcw]

maybe move this to the finally of _main_task

[ByteMaster-1]

Done — moved to _main_task's finally, so pending generate_reply futures now fail fast on every terminal exit (fatal error, retries exhausted, close), not just on fatal error events.