From de17599ea5f2ce3334d6bfe2ad8f92da3b7c177c Mon Sep 17 00:00:00 2001
From: "mateo@berri.ai" <277851410+mateo-berri@users.noreply.github.com>
Date: Thu, 16 Jul 2026 20:46:14 +0000
Subject: [PATCH] test(e2e/realtime): fail fast on provider error events with
 the real reason

When the realtime provider rejects the socket (OpenAI insufficient_quota is
the motivating case), the proxy forwards an `error` event but collect_until
kept waiting for the lifecycle event and then raised a bare
"no session.created within 20s; got ['error']", discarding the cause.

collect_until now raises RealtimeError as soon as an `error` event arrives
before the awaited stop event, surfacing the provider's code and message so a
quota rejection is diagnosable from CI logs instead of an opaque timeout. The
session connection is typed as a WebSocketConnection Protocol so the collect
loop is driven with a scripted connection in harness coverage (no live socket,
no monkeypatching).

Co-authored-by: linear-code[bot] <222613912+linear-code[bot]@users.noreply.github.com>
---
 .../realtime/realtime_client.py               | 60 +++++++++++-
 .../realtime/test_realtime_client.py          | 95 +++++++++++++++++++
 2 files changed, 152 insertions(+), 3 deletions(-)
 create mode 100644 tests/e2e/llm_translation/realtime/test_realtime_client.py

diff --git a/tests/e2e/llm_translation/realtime/realtime_client.py b/tests/e2e/llm_translation/realtime/realtime_client.py
index b50160e3538..e2d5e0b937a 100644
--- a/tests/e2e/llm_translation/realtime/realtime_client.py
+++ b/tests/e2e/llm_translation/realtime/realtime_client.py
@@ -14,12 +14,11 @@ import time
 from collections.abc import Generator, Mapping
 from contextlib import contextmanager
 from dataclasses import dataclass
-from typing import TypeVar
+from typing import Protocol, TypeVar
 from urllib.parse import urlencode
 
 from pydantic import BaseModel, ConfigDict
 from websockets.sync.client import connect
-from websockets.sync.connection import Connection
 
 from e2e_config import PROXY_BASE_URL, unique_marker
 from e2e_gateway import Gateway, build_gateway
@@ -226,6 +225,47 @@ class ResponseDone(BaseModel):
     response: ResponsePayload
 
 
+class ErrorDetail(BaseModel):
+    model_config = ConfigDict(extra="allow")
+    type: str | None = None
+    code: str | None = None
+    message: str | None = None
+
+
+class ErrorEvent(BaseModel):
+    model_config = ConfigDict(extra="allow")
+    type: str = "error"
+    error: ErrorDetail | None = None
+
+
+def error_summary(payload: str) -> str:
+    """Human-readable one-liner for an `error` event payload. OpenAI's realtime
+    quota rejection arrives as {"type":"error","error":{"code":"insufficient_quota",
+    "message":"..."}}, so surfacing code + message turns an opaque socket failure
+    into the actual reason. Falls back to the raw payload when it is not shaped
+    that way (a provider whose error envelope the proxy has not normalized)."""
+    detail = ErrorEvent.model_validate_json(payload).error
+    if detail is None:
+        return payload
+    label = detail.code or detail.type
+    if label and detail.message:
+        return f"{label}: {detail.message}"
+    return detail.message or label or payload
+
+
+class RealtimeError(RuntimeError):
+    """Raised when the provider emits an `error` event while a test is waiting for
+    a lifecycle or response event, instead of letting the wait time out with the
+    cause discarded. `payload` is the raw error event for further inspection."""
+
+    def __init__(self, stop_type: str, payload: str, preceding: tuple[str, ...]) -> None:
+        self.payload = payload
+        super().__init__(
+            f"realtime provider returned an error before {stop_type!r}: "
+            f"{error_summary(payload)} (preceding events: {list(preceding)})"
+        )
+
+
 @dataclass(frozen=True, slots=True)
 class ReceivedEvent:
     type: str
@@ -297,9 +337,19 @@ def _as_text(message: str | bytes) -> str:
     return message.decode("utf-8") if isinstance(message, bytes) else message
 
 
+class WebSocketConnection(Protocol):
+    """The slice of websockets.sync.connection.Connection the session uses, kept a
+    Protocol so the collect loop can be driven with a scripted connection in harness
+    coverage without a live socket (dependency injection over monkeypatching)."""
+
+    def send(self, message: str) -> None: ...
+
+    def recv(self, timeout: float | None = None) -> str | bytes: ...
+
+
 @dataclass(frozen=True, slots=True)
 class RealtimeSession:
-    connection: Connection
+    connection: WebSocketConnection
 
     def send(self, event: BaseModel) -> None:
         self.connection.send(event.model_dump_json(by_alias=True, exclude_none=True))
@@ -322,6 +372,10 @@ class RealtimeSession:
             collected.append(event)
             if event.type == stop_type:
                 return tuple(collected)
+            if event.type == "error":
+                raise RealtimeError(
+                    stop_type, text, tuple(e.type for e in collected[:-1])
+                )
         raise TimeoutError(
             f"no {stop_type!r} within {timeout}s; got {[e.type for e in collected]}"
         )
diff --git a/tests/e2e/llm_translation/realtime/test_realtime_client.py b/tests/e2e/llm_translation/realtime/test_realtime_client.py
new file mode 100644
index 00000000000..e353fa215c0
--- /dev/null
+++ b/tests/e2e/llm_translation/realtime/test_realtime_client.py
@@ -0,0 +1,95 @@
+"""Harness coverage for realtime_client.RealtimeSession.collect_until.
+
+Not an e2e test (no live proxy): it drives the collect loop with a scripted
+connection injected in place of the websocket, so it runs regardless of whether a
+proxy is up. The behavior under test is the diagnostic contract: when the provider
+emits an `error` event (OpenAI realtime quota rejection is the motivating case)
+while a test waits for a lifecycle or response event, the wait must fail fast and
+surface the provider's reason, not time out with the cause discarded.
+"""
+
+from collections.abc import Iterator
+from dataclasses import dataclass
+
+import pytest
+
+from realtime_client import RealtimeError, RealtimeSession, error_summary
+
+QUOTA_ERROR = (
+    '{"type":"error","event_id":"event_1","error":{"type":"insufficient_quota",'
+    '"code":"insufficient_quota","message":"You exceeded your current quota, '
+    'please check your plan and billing details."}}'
+)
+SESSION_CREATED = '{"type":"session.created","session":{"id":"sess_1"}}'
+
+
+@dataclass(frozen=True, slots=True)
+class _ScriptedConnection:
+    """A WebSocketConnection that replays a fixed list of frames, then behaves like
+    a silent socket (recv times out). send is a no-op; the collect loop only recvs."""
+
+    frames: Iterator[str]
+
+    def send(self, message: str) -> None:
+        return None
+
+    def recv(self, timeout: float | None = None) -> str:
+        try:
+            return next(self.frames)
+        except StopIteration:
+            raise TimeoutError("no more frames") from None
+
+
+def _session(*frames: str) -> RealtimeSession:
+    return RealtimeSession(connection=_ScriptedConnection(iter(frames)))
+
+
+def test_collect_until_raises_realtime_error_on_error_event() -> None:
+    session = _session(QUOTA_ERROR)
+
+    with pytest.raises(RealtimeError) as excinfo:
+        session.collect_until("session.created", timeout=20)
+
+    message = str(excinfo.value)
+    assert "insufficient_quota" in message
+    assert "exceeded your current quota" in message
+    assert "session.created" in message
+    assert excinfo.value.payload == QUOTA_ERROR
+
+
+def test_collect_until_reports_events_seen_before_the_error() -> None:
+    session = _session(SESSION_CREATED, QUOTA_ERROR)
+
+    with pytest.raises(RealtimeError) as excinfo:
+        session.collect_until("response.done", timeout=20)
+
+    assert "session.created" in str(excinfo.value)
+
+
+def test_collect_until_returns_on_stop_event_without_raising() -> None:
+    session = _session(SESSION_CREATED)
+
+    events = session.collect_until("session.created", timeout=20)
+
+    assert [e.type for e in events] == ["session.created"]
+
+
+def test_collect_until_times_out_when_no_stop_or_error_arrives() -> None:
+    session = _session(SESSION_CREATED)
+
+    with pytest.raises(TimeoutError) as excinfo:
+        session.collect_until("session.updated", timeout=5)
+
+    assert "session.updated" in str(excinfo.value)
+
+
+def test_error_summary_extracts_openai_quota_reason() -> None:
+    assert error_summary(QUOTA_ERROR) == (
+        "insufficient_quota: You exceeded your current quota, "
+        "please check your plan and billing details."
+    )
+
+
+def test_error_summary_falls_back_to_raw_payload_when_not_normalized() -> None:
+    payload = '{"type":"error","message":"boom"}'
+    assert error_summary(payload) == payload