fix(live): preserve caller queues and prevent uncertain WebSocket replay (openai#3980)

If a Live connection opened with an empty manager queue, iterator
recovery lost the caller's queued messages and byte budget. A typed
command could also be sent again after the first socket actually
received it but its send raised. Keep the passed queue even when empty
and drop only the attempted message on a failed send/flush, leaving
unattempted messages ordered. This matches the merged Realtime handling.

Primary, fork and sideband fixes are separate commits for review. No
startup, auth, default reconnect, imports or public method signatures
change. Direct send errors still raise; queued flush failures still log
the existing warning. The Live helper notes describe the recovery and
recording-error policy.

Validation:
- Baseline: 337 Live/redirect tests pass under both Pydantic 1 and 2.
- Reproduced all 18 queue/replay failures using actual loopback wire
delivery, then 42 role tests pass, including per-role
error/unknown-event/close ordering and helper disposal.
- Final 399 Live, protected dial, Realtime and send-queue tests pass in
each Pydantic lane; Ruff and full SDK Mypy (1,865 source files) pass.
- Trusted main budget/isolation pass: 7,826/10,000 with unchanged
generated snapshot. Independent review found only the queued-error doc
clarification, which is included.

Author: markstuart-oai

src/openai/lib/live/README.md

@@ -0,0 +1,33 @@
+# Live WebSocket lifecycle
+
+`client.live.connect()` and `client.live.forks.connect(session_id=...)` leave
+startup to the caller: send `connection.session.start(...)` and wait for
+`session.started`. A sideband connection created by
+`client.live.sideband.connect(session_id=...)` attaches to an existing session.
+It does not send a start or require a new `session.started`; an attachment's
+short replay is not a complete session snapshot.
+
+Direct `recv()` and `recv_bytes()` calls report transport errors to their
+caller. Iterator reconnection is available only when an `on_reconnecting`
+callback was explicitly supplied. The callback controls retry and can update
+credentials or query parameters. A new socket is not proof that the previous
+Live session, recording, or application state was restored. Applications own
+their recovery decision and any necessary startup or state reconstruction.
+Don't use multiple physical readers: a dispatcher owns the read loop while it
+runs. Detaching or closing one transcript grouper doesn't cancel another
+observer or the connection.
+
+All modes preserve the manager's caller-configured `max_queue_size`, including
+if the queue was empty at connection time. Only messages that haven't been
+attempted on a socket remain eligible for flushing after a retry. A direct send
+exception is raised to its caller and cannot prove the server didn't receive
+that message. A failure while flushing an already queued message still logs a
+warning, as before; it cannot be raised at the original queueing call. Neither
+failed attempt is automatically retried. Unattempted pre-open messages and
+messages explicitly queued during recovery remain in order within their
+existing budget. No automatic session reopening or restoration is implied.
+
+Treat each wire error as an event to handle, not as a successful session result.
+In particular, preserve `session_storage_failed` if `session.closed` follows;
+the close doesn't mean that recording storage succeeded. Unknown events and
+fields are preserved for callers that need them.

src/openai/resources/live/forks.py

@@ -139,7 +139,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 = AsyncForksSessionResource(self)
@@ -215,11 +215,7 @@ async def send(self, event: ForkClientEvent | ForkClientEventParam) -> None:
         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:
@@ -324,7 +320,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")
 
@@ -638,7 +634,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 = ForksSessionResource(self)
@@ -714,11 +710,7 @@ def send(self, event: ForkClientEvent | ForkClientEventParam) -> 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:
@@ -817,7 +809,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")
 

src/openai/resources/live/live.py

@@ -360,7 +360,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 = AsyncLiveSessionResource(self)
@@ -436,11 +436,7 @@ async def send(self, event: ClientEvent | ClientEventParam) -> None:
         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:
@@ -545,7 +541,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")
 
@@ -856,7 +852,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 = LiveSessionResource(self)
@@ -932,11 +928,7 @@ def send(self, event: ClientEvent | ClientEventParam) -> 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:
@@ -1035,7 +1027,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")
 

src/openai/resources/live/sideband.py

@@ -142,7 +142,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 = AsyncSidebandSessionResource(self)
@@ -218,11 +218,7 @@ async def send(self, event: ConnectClientEvent | ConnectClientEventParam) -> Non
         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:
@@ -329,7 +325,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")
 
@@ -646,7 +642,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 = SidebandSessionResource(self)
@@ -722,11 +718,7 @@ def send(self, event: ConnectClientEvent | ConnectClientEventParam) -> 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:
@@ -827,7 +819,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")