fix(realtime): retain configured queues without replaying attempted sends (openai#3978)

An empty Realtime manager queue was replaced on connect, losing its
configured byte limit and later messages. A typed or queued send could
also be replayed on reconnect even after reaching the server. Preserve
the manager's queue and release only attempted writes on failure or
cancellation; never-attempted messages remain in order. Other endpoints
retain the shared queue's existing defaults.

Six sync/async real-socket regressions fail on the base and pass with
this change, covering empty queues and interrupted pre-open/direct
sends. A focused cancellation regression verifies in-flight bounds and
retained messages. Validation: 456 affected tests per Pydantic 1/2, 45
final queue/Realtime tests per version, Ruff, mypy (1,865 files),
Pyright, and the trusted custom-code check (7,766/10,000).

Author: markstuart-oai

src/openai/_send_queue.py

@@ -37,26 +37,34 @@ def enqueue(self, data: str) -> None:
             self._queue.append((data, byte_length))
             self._bytes += byte_length
 
-    def flush_sync(self, send: typing.Callable[[str], object]) -> None:
+    def flush_sync(self, send: typing.Callable[[str], object], *, requeue_failed: bool = True) -> None:
         """Send every queued message via *send*.
 
         If *send* raises, the failing message and all subsequent messages
-        are re-queued and the error is re-raised.
+        are re-queued and the error is re-raised. When `requeue_failed` is
+        false, release the attempted message even on failure or interruption.
         """
         while isinstance(pending := self._begin_flush(), threading.Event):
             pending.wait()
 
         try:
             while pending:
                 data, byte_length = pending[0]
-                send(data)
-                with self._lock:
-                    pending.popleft()
-                    self._bytes -= byte_length
+                sent = False
+                try:
+                    send(data)
+                    sent = True
+                finally:
+                    if sent or not requeue_failed:
+                        with self._lock:
+                            pending.popleft()
+                            self._bytes -= byte_length
         finally:
             self._end_flush(pending)
 
-    async def flush_async(self, send: typing.Callable[[str], typing.Awaitable[object]]) -> None:
+    async def flush_async(
+        self, send: typing.Callable[[str], typing.Awaitable[object]], *, requeue_failed: bool = True
+    ) -> None:
         """Async variant of :meth:`flush_sync`."""
         while isinstance(pending := self._begin_flush(), threading.Event):
             # Waiting in a worker keeps the event loop responsive. Cancellation
@@ -66,10 +74,15 @@ async def flush_async(self, send: typing.Callable[[str], typing.Awaitable[object
         try:
             while pending:
                 data, byte_length = pending[0]
-                await send(data)
-                with self._lock:
-                    pending.popleft()
-                    self._bytes -= byte_length
+                sent = False
+                try:
+                    await send(data)
+                    sent = True
+                finally:
+                    if sent or not requeue_failed:
+                        with self._lock:
+                            pending.popleft()
+                            self._bytes -= byte_length
         finally:
             self._end_flush(pending)
 

src/openai/resources/realtime/realtime.py

@@ -294,7 +294,7 @@ def __init__(
         self._extra_headers = extra_headers
         self._intentionally_closed = False
         self._is_reconnecting = False
-        self._send_queue = send_queue or SendQueue()
+        self._send_queue = send_queue if send_queue is not None else SendQueue()
         self._event_handler_registry = EventHandlerRegistry(use_lock=False)
 
         self.session = AsyncRealtimeSessionResource(self)
@@ -373,11 +373,7 @@ async def send(self, event: RealtimeClientEvent | RealtimeClientEventParam) -> N
         if self._is_reconnecting:
             self._send_queue.enqueue(data)
             return
-        try:
-            await self._connection.send(data)
-        except Exception:
-            self._send_queue.enqueue(data)
-            raise
+        await self._connection.send(data)
 
     async def send_raw(self, data: bytes | str) -> None:
         if self._is_reconnecting:
@@ -484,7 +480,7 @@ async def _send(data: str) -> None:
             await self._connection.send(data)
 
         try:
-            await self._send_queue.flush_async(_send)
+            await self._send_queue.flush_async(_send, requeue_failed=False)
         except Exception:
             log.warning("Failed to flush send queue after reconnect")
 
@@ -818,7 +814,7 @@ def __init__(
         self._extra_headers = extra_headers
         self._intentionally_closed = False
         self._is_reconnecting = False
-        self._send_queue = send_queue or SendQueue()
+        self._send_queue = send_queue if send_queue is not None else SendQueue()
         self._event_handler_registry = EventHandlerRegistry(use_lock=True)
 
         self.session = RealtimeSessionResource(self)
@@ -897,11 +893,7 @@ def send(self, event: RealtimeClientEvent | RealtimeClientEventParam) -> None:
         if self._is_reconnecting:
             self._send_queue.enqueue(data)
             return
-        try:
-            self._connection.send(data)
-        except Exception:
-            self._send_queue.enqueue(data)
-            raise
+        self._connection.send(data)
 
     def send_raw(self, data: bytes | str) -> None:
         if self._is_reconnecting:
@@ -1002,7 +994,7 @@ def _reconnect(self, exc: Exception) -> bool:
     def _flush_send_queue(self) -> None:
         """Send all queued messages over the current connection."""
         try:
-            self._send_queue.flush_sync(lambda data: self._connection.send(data))
+            self._send_queue.flush_sync(lambda data: self._connection.send(data), requeue_failed=False)
         except Exception:
             log.warning("Failed to flush send queue after reconnect")
 

tests/lib/test_realtime_websocket_contract.py

@@ -6,9 +6,13 @@
 
 import httpx2
 import pytest
+from websockets.sync.client import ClientConnection
 from websockets.sync.server import ServerConnection
+from websockets.asyncio.client import ClientConnection as AsyncClientConnection
 
 from openai import OpenAI, AsyncOpenAI
+from openai._exceptions import WebSocketQueueFullError
+from openai.types.websocket_reconnection import ReconnectingEvent
 from openai.types.realtime.realtime_error_event import RealtimeErrorEvent
 from openai.types.realtime.input_audio_buffer_cleared_event import InputAudioBufferClearedEvent
 
@@ -133,3 +137,136 @@ def script(socket: ServerConnection) -> None:
                         "event_id": "future-1",
                         "text": "東京🙂",
                     }
+
+
+@pytest.mark.parametrize("mode", ["sync", "async"])
+async def test_realtime_empty_manager_queue_keeps_budget_after_open(mode: str) -> None:
+    opened = 0
+
+    def script(socket: ServerConnection) -> None:
+        nonlocal opened
+        opened += 1
+        if opened == 1:
+            socket.close(code=1011, reason="synthetic restart")
+            return
+        # Let the iterator resume even if its connection lost the manager's
+        # queue. Server-side inspection still proves exactly what was sent.
+        socket.send('{"type": "input_audio_buffer.cleared", "event_id": "ready"}')
+        assert json.loads(socket.recv(timeout=5)) == {"type": "input_audio_buffer.clear", "event_id": "queued"}
+
+    with script_server(script, expected_connections=2) as url:
+        if mode == "sync":
+            with OpenAI(
+                api_key="fake-realtime-key", base_url=url, http_client=httpx2.Client(trust_env=False)
+            ) as client:
+
+                def on_retry(_event: ReconnectingEvent) -> None:
+                    manager.send({"type": "input_audio_buffer.clear", "event_id": "queued"})
+                    with pytest.raises(WebSocketQueueFullError):
+                        manager.send({"type": "input_audio_buffer.append", "audio": "AAAA" * 100})
+
+                manager = client.realtime.connect(
+                    model="gpt-realtime", max_queue_size=96, on_reconnecting=on_retry, initial_delay=0
+                )
+                with manager as connection:
+                    event = next(iter(connection))
+                    assert event.type == "input_audio_buffer.cleared"
+        else:
+            async with AsyncOpenAI(
+                api_key="fake-realtime-key", base_url=url, http_client=httpx2.AsyncClient(trust_env=False)
+            ) as client:
+
+                def on_async_retry(_event: ReconnectingEvent) -> None:
+                    async_manager.send({"type": "input_audio_buffer.clear", "event_id": "queued"})
+                    with pytest.raises(WebSocketQueueFullError):
+                        async_manager.send({"type": "input_audio_buffer.append", "audio": "AAAA" * 100})
+
+                async_manager = client.realtime.connect(
+                    model="gpt-realtime", max_queue_size=96, on_reconnecting=on_async_retry, initial_delay=0
+                )
+                async with async_manager as connection:
+                    event = await asyncio.wait_for(anext(aiter(connection)), 5)
+                    assert event.type == "input_audio_buffer.cleared"
+
+
+@pytest.mark.parametrize("mode", ["sync", "async"])
+@pytest.mark.parametrize("preopen", [True, False], ids=["preopen-flush", "direct"])
+async def test_realtime_recovery_never_replays_an_attempted_command(
+    mode: str, preopen: bool, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    opened = 0
+    recorded: list[tuple[int, str]] = []
+
+    def script(socket: ServerConnection) -> None:
+        nonlocal opened
+        opened += 1
+        current = opened
+        if current == 1:
+            for _ in range(2 if preopen else 1):
+                recorded.append((current, json.loads(socket.recv(timeout=5))["event_id"]))
+            socket.close(code=1011, reason="synthetic restart")
+            return
+        socket.send('{"type": "input_audio_buffer.cleared", "event_id": "ready"}')
+        for _ in range(2 if preopen else 1):
+            recorded.append((current, json.loads(socket.recv(timeout=5))["event_id"]))
+
+    # Fail after the real wire send: an exception cannot establish non-delivery.
+    original = ClientConnection.send
+    original_async = AsyncClientConnection.send
+
+    def send_then_interrupt(self: ClientConnection, message: object, *args: object, **kwargs: object) -> None:
+        original(self, message, *args, **kwargs)  # type: ignore[arg-type]
+        if json.loads(message)["event_id"] == "attempted":  # type: ignore[arg-type]
+            raise OSError("synthetic interruption after send")
+
+    async def async_send_then_interrupt(
+        self: AsyncClientConnection, message: object, *args: object, **kwargs: object
+    ) -> None:
+        await original_async(self, message, *args, **kwargs)  # type: ignore[arg-type]
+        if json.loads(message)["event_id"] == "attempted":  # type: ignore[arg-type]
+            raise OSError("synthetic interruption after send")
+
+    if mode == "sync":
+        monkeypatch.setattr(ClientConnection, "send", send_then_interrupt)
+    else:
+        monkeypatch.setattr(AsyncClientConnection, "send", async_send_then_interrupt)
+    with script_server(script, expected_connections=2) as url:
+        if mode == "sync":
+            with OpenAI(
+                api_key="fake-realtime-key", base_url=url, http_client=httpx2.Client(trust_env=False)
+            ) as client:
+
+                def on_retry(_event: ReconnectingEvent) -> None:
+                    manager.send({"type": "input_audio_buffer.clear", "event_id": "during-recovery"})
+
+                manager = client.realtime.connect(model="gpt-realtime", on_reconnecting=on_retry, initial_delay=0)
+                if preopen:
+                    for identity in ("first", "attempted", "unattempted"):
+                        manager.send({"type": "input_audio_buffer.clear", "event_id": identity})
+                with manager as connection:
+                    if not preopen:
+                        with pytest.raises(OSError, match="synthetic interruption after send"):
+                            connection.send({"type": "input_audio_buffer.clear", "event_id": "attempted"})
+                    assert next(iter(connection)).type == "input_audio_buffer.cleared"
+        else:
+            async with AsyncOpenAI(
+                api_key="fake-realtime-key", base_url=url, http_client=httpx2.AsyncClient(trust_env=False)
+            ) as client:
+
+                def on_async_retry(_event: ReconnectingEvent) -> None:
+                    async_manager.send({"type": "input_audio_buffer.clear", "event_id": "during-recovery"})
+
+                async_manager = client.realtime.connect(
+                    model="gpt-realtime", on_reconnecting=on_async_retry, initial_delay=0
+                )
+                if preopen:
+                    for identity in ("first", "attempted", "unattempted"):
+                        async_manager.send({"type": "input_audio_buffer.clear", "event_id": identity})
+                async with async_manager as connection:
+                    if not preopen:
+                        with pytest.raises(OSError, match="synthetic interruption after send"):
+                            await connection.send({"type": "input_audio_buffer.clear", "event_id": "attempted"})
+                    event = await asyncio.wait_for(anext(aiter(connection)), 5)
+                    assert event.type == "input_audio_buffer.cleared"
+    expected = [(1, "first"), (1, "attempted"), (2, "unattempted"), (2, "during-recovery")]
+    assert recorded == (expected if preopen else [(1, "attempted"), (2, "during-recovery")])