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

livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/realtime/realtime_model.py

@@ -281,6 +281,24 @@ class _DiscardedGeneration:
     pass
 
 
+# error codes that can never succeed on retry: the socket reconnects fine (the key is
+# valid), but every generation fails and the server closes the connection again, so
+# retrying loops forever. When one of these is received, the reconnect loop must stop.
+_FATAL_ERROR_CODES = frozenset(
+    {
+        "insufficient_quota",
+        "invalid_api_key",
+        "account_deactivated",
+        "billing_hard_limit_reached",
+    }
+)
+
+
+def _is_fatal_error(error: object | None) -> bool:
+    code = getattr(error, "code", None) or getattr(error, "type", None)
+    return isinstance(code, str) and code in _FATAL_ERROR_CODES
+
+
 class RealtimeModel(llm.RealtimeModel):
     @overload
     def __init__(
@@ -835,6 +853,9 @@ def __init__(self, realtime_model: RealtimeModel) -> None:
         self._instructions: str | None = None
         # set on aclose; trailing server events are ignored while it's set
         self._closing = False
+        # set when the server reports an error that cannot succeed on retry (e.g.
+        # insufficient_quota); _main_task stops reconnecting once this is set
+        self._fatal_error: APIError | None = None
         self._main_atask = asyncio.create_task(self._main_task(), name="RealtimeSession._main_task")
         self.send_event(self._create_session_update_event())
 
@@ -934,6 +955,10 @@ async def _reconnect() -> None:
 
         reconnecting = False
         while not self._msg_ch.closed:
+            if self._fatal_error is not None:
+                # also reachable when _run_ws exits normally (e.g. max_session_duration
+                # refresh) after a fatal error, not only via a connection failure below
+                raise self._fatal_error
             try:
                 ws_conn = await self._create_ws_conn()
                 if reconnecting:
@@ -942,6 +967,12 @@ async def _reconnect() -> None:
                 await self._run_ws(ws_conn)
 
             except APIError as e:
+                if self._fatal_error is not None:
+                    # the server reported a terminal error (e.g. insufficient_quota) before
+                    # closing the connection: reconnecting succeeds (the key is valid) but
+                    # every generation fails and the server closes again, so retrying would
+                    # loop forever. The error was already emitted with recoverable=False.
+                    raise self._fatal_error from e
                 if max_retries == 0 or not e.retryable:
                     self._emit_error(e, recoverable=False)
                     raise
@@ -2138,16 +2169,17 @@ def _handle_response_done_but_not_complete(self, event: ResponseDoneEvent) -> No
             else:
                 error_body = None
                 message = f"{provider_label} response failed with unknown error"
-            self._emit_error(
-                APIError(
-                    message=message,
-                    body=error_body,
-                    retryable=True,
-                ),
-                # all possible faulures undocumented by openai,
-                # so we assume optimistically all retryable/recoverable
-                recoverable=True,
+            # failures are largely undocumented by openai, so we assume optimistically
+            # recoverable unless the code is a known-fatal one (quota / auth / billing)
+            recoverable = not _is_fatal_error(error_body)
+            error = APIError(
+                message=message,
+                body=error_body,
+                retryable=recoverable,
             )
+            self._emit_error(error, recoverable=recoverable)
+            if not recoverable:
+                self._fatal_error = error
         elif event.response.status in {"cancelled", "incomplete"}:
             status_details = event.response.status_details
             if isinstance(status_details, str):
@@ -2181,17 +2213,27 @@ def _handle_error(self, event: RealtimeErrorEvent) -> None:
             f"{provider_label} returned an error: {event.error}",
             extra={"error": event.error},
         )
-        self._emit_error(
-            APIError(
-                message=f"{provider_label} returned an error",
-                body=event.error,
-                retryable=True,
-            ),
-            recoverable=True,
+        recoverable = not _is_fatal_error(event.error)
+        error = APIError(
+            message=f"{provider_label} returned an error",
+            body=event.error,
+            retryable=recoverable,
         )
+        self._emit_error(error, recoverable=recoverable)
 
         # response errors are handled by _handle_response_done via _done_fut.
         # error events here are for non-response errors (e.g. invalid request).
+        if not recoverable:
+            self._fatal_error = error
+            # a pending generate_reply can never be served now; fail its future
+            # immediately instead of letting it wait out the timeout
+            code = getattr(event.error, "code", None) or getattr(event.error, "type", None)
+            for fut in self._response_created_futures.values():
+                if not fut.done():
+                    fut.set_exception(
+                        llm.RealtimeError(f"{provider_label} returned a fatal error: {code}")
+                    )
+            self._response_created_futures.clear()
 
     def _emit_error(self, error: Exception, recoverable: bool) -> None:
         self.emit(

tests/test_realtime/test_openai_realtime_model.py

@@ -1,13 +1,15 @@
 from __future__ import annotations
 
+import asyncio
 from types import SimpleNamespace
 from typing import cast
 
 import pytest
 
 from livekit.agents import llm
+from livekit.agents._exceptions import APIError
 from livekit.agents.llm.remote_chat_context import RemoteChatContext
-from livekit.plugins.openai.realtime.realtime_model import RealtimeSession
+from livekit.plugins.openai.realtime.realtime_model import RealtimeSession, _is_fatal_error
 
 pytestmark = pytest.mark.unit
 
@@ -31,3 +33,116 @@ def test_update_chat_ctx_deletes_empty_remote_items() -> None:
         if getattr(event, "type", None) == "conversation.item.delete"
     ]
     assert delete_ids == ["audio_item"]
+
+
+# --------------------------------------------------------------------------- #
+# fatal error classification: a fatal error must stop the reconnect loop
+# --------------------------------------------------------------------------- #
+
+
+def test_is_fatal_error_matches_known_codes() -> None:
+    assert _is_fatal_error(SimpleNamespace(code="insufficient_quota"))
+    assert _is_fatal_error(SimpleNamespace(code=None, type="invalid_api_key"))
+    assert not _is_fatal_error(SimpleNamespace(code="server_error"))
+    assert not _is_fatal_error(SimpleNamespace())
+    assert not _is_fatal_error(None)
+
+
+def _handle_error_session(capture: dict[str, object]) -> RealtimeSession:
+    return cast(
+        RealtimeSession,
+        SimpleNamespace(
+            _realtime_model=SimpleNamespace(_provider_label="openai"),
+            _emit_error=lambda error, recoverable: capture.update(recoverable=recoverable),
+            _response_created_futures={},
+            _fatal_error=None,
+        ),
+    )
+
+
+def test_handle_error_marks_fatal_non_recoverable() -> None:
+    captured: dict[str, object] = {}
+    session = _handle_error_session(captured)
+    event = SimpleNamespace(
+        error=SimpleNamespace(message="quota exceeded", code="insufficient_quota")
+    )
+    RealtimeSession._handle_error(session, event)
+    assert captured["recoverable"] is False
+    assert isinstance(session._fatal_error, APIError)  # _main_task stops reconnecting
+
+
+def test_handle_error_keeps_transient_recoverable() -> None:
+    captured: dict[str, object] = {}
+    session = _handle_error_session(captured)
+    event = SimpleNamespace(error=SimpleNamespace(message="server hiccup", code="server_error"))
+    RealtimeSession._handle_error(session, event)
+    assert captured["recoverable"] is True
+    assert session._fatal_error is None
+
+
+def test_handle_error_ignores_cancellation_failed() -> None:
+    captured: dict[str, object] = {}
+    event = SimpleNamespace(error=SimpleNamespace(message="Cancellation failed: no response"))
+    RealtimeSession._handle_error(_handle_error_session(captured), event)
+    assert captured == {}  # early return, nothing emitted
+
+
+async def test_handle_error_fails_pending_reply_future_on_fatal() -> None:
+    # a fatal error means a pending generate_reply will never be served; its future must
+    # fail immediately instead of waiting out the timeout
+    fut: asyncio.Future[llm.GenerationCreatedEvent] = asyncio.get_running_loop().create_future()
+    session = _handle_error_session({})
+    session._response_created_futures["ev"] = fut
+
+    event = SimpleNamespace(error=SimpleNamespace(message="quota", code="insufficient_quota"))
+    RealtimeSession._handle_error(session, event)
+
+    assert fut.done()
+    assert isinstance(fut.exception(), llm.RealtimeError)
+    assert not session._response_created_futures
+
+
+async def test_handle_error_leaves_pending_future_on_transient() -> None:
+    fut: asyncio.Future[llm.GenerationCreatedEvent] = asyncio.get_running_loop().create_future()
+    session = _handle_error_session({})
+    session._response_created_futures["ev"] = fut
+
+    event = SimpleNamespace(error=SimpleNamespace(message="hiccup", code="server_error"))
+    RealtimeSession._handle_error(session, event)
+
+    assert not fut.done()  # transient error: let it resolve or time out normally
+    fut.cancel()  # cleanup
+
+
+def test_response_done_failed_fatal_sets_fatal_error() -> None:
+    captured: dict[str, object] = {}
+    session = _handle_error_session(captured)
+    event = SimpleNamespace(
+        response=SimpleNamespace(
+            id="resp_1",
+            status="failed",
+            status_details=SimpleNamespace(
+                error=SimpleNamespace(type="insufficient_quota", code="insufficient_quota")
+            ),
+        )
+    )
+    RealtimeSession._handle_response_done_but_not_complete(session, event)
+    assert captured["recoverable"] is False
+    assert isinstance(session._fatal_error, APIError)
+
+
+def test_response_done_failed_transient_stays_recoverable() -> None:
+    captured: dict[str, object] = {}
+    session = _handle_error_session(captured)
+    event = SimpleNamespace(
+        response=SimpleNamespace(
+            id="resp_1",
+            status="failed",
+            status_details=SimpleNamespace(
+                error=SimpleNamespace(type="invalid_request_error", code="rate_limit_exceeded")
+            ),
+        )
+    )
+    RealtimeSession._handle_response_done_but_not_complete(session, event)
+    assert captured["recoverable"] is True
+    assert session._fatal_error is None