feat(responses): add WebSocket lane overflow isolation

src/openai/lib/responses_websocket/README.md

@@ -186,14 +186,32 @@ it. After reconnect releases a closed default lane, `session.lane()` (or
 `session.lane(None)`) registers a fresh default and updates `session.default`.
 
 Limits apply only to this opt-in session. Event queues are bounded per lane and
-across the session; overflow reports `ResponsesWebSocketBufferError` and closes
+across the session. By default, overflow reports `ResponsesWebSocketBufferError` and closes
 the owned connection so ordered events are not silently dropped. Queue byte
 counts measure received message bytes before parsing, not Python object overhead or
 transport decoding memory. Response accumulation is uncapped by default:
 omit `max_response_bytes` or set it to `None`. A positive integer opts into a
 budget for the cumulative compatible event bytes passed to an accumulator;
 exceeding it disables that response's collection, while raw events and other
 lanes remain usable. Queue and lane limits still require positive integers.
+
+To keep healthy lanes usable when another lane falls behind, pass
+`isolate_lane_overflow=True` to either session constructor. The default is `False`.
+An overflowing lane loses its queued events and response collection; its reads
+and final-response collection raise the same first `ResponsesWebSocketBufferError`.
+Later events for that failed ID are dropped, and its reservation lasts until
+reconnect. Manual lane detachment and unregistered-stream inspection keep their
+existing default-lane behavior.
+
+For aggregate queue pressure, the session fails the largest existing backlog
+in the pressured unit (bytes first, otherwise events), excluding the incoming
+event. Registration order breaks ties. An individually oversized event fails
+its target lane. The default lane can also fail under this policy: subsequent
+default events, including API error/control events, are then unavailable, while
+registered named lanes remain usable. Physical connection failures and explicit
+session close still end the session. After reconnect, register a new default
+explicitly with `session.lane()` if the old default failed.
+
 Choose budgets for large legitimate outputs. These are application budgets,
 not protocol limits.
 The examples disable the transport's separate 1 MiB message limit with

src/openai/lib/responses_websocket/_session.py

@@ -157,6 +157,7 @@ def __init__(self, stream_id: str | None, limit: int | None) -> None:
         self.queue: deque[tuple[ResponsesServerEvent, int]] = deque()
         self.size = 0
         self.closed = False
+        self.failure: ResponsesWebSocketBufferError | None = None
         self.in_flight = False
         self.closed_connection: object | None = None
         self.received_generation = 0
@@ -167,10 +168,15 @@ def __init__(self, stream_id: str | None, limit: int | None) -> None:
 
 class _Router:
     def __init__(
-        self, limits: ResponsesWebSocketLimits, connection: ResponsesConnection | AsyncResponsesConnection
+        self,
+        limits: ResponsesWebSocketLimits,
+        connection: ResponsesConnection | AsyncResponsesConnection,
+        *,
+        isolate_lane_overflow: bool = False,
     ) -> None:
         self.connection = connection
         self.limits = limits
+        self.isolate_lane_overflow = isolate_lane_overflow
         self.condition = threading.Condition()
         self.lanes: dict[str | None, _LaneState] = {}
         self.events = 0
@@ -213,6 +219,11 @@ def detach(self, lane: _LaneState) -> None:
             lane.wake.set()
             self.condition.notify_all()
 
+    def _fail(self, lane: _LaneState) -> None:
+        if not lane.closed:
+            lane.failure = ResponsesWebSocketBufferError("The lane exceeded its buffered event limits")
+            self.detach(lane)
+
     def reconnected(self, connection: object) -> None:
         with self.condition:
             for stream_id, lane in list(self.lanes.items()):
@@ -227,11 +238,21 @@ def dispatch(self, event: ResponsesServerEvent, size: int) -> None:
                 return
             stream_id = _field(event, "stream_id")
             lane = self.lanes.get(stream_id, self.default)
+            if lane.failure is not None:
+                return
             if lane.closed:
                 lane = self.default
             if lane.closed:
                 return
-            if (
+            if self.isolate_lane_overflow:
+                if (
+                    len(lane.queue) >= self.limits.max_events_per_lane
+                    or lane.size + size > self.limits.max_bytes_per_lane
+                    or size > self.limits.max_bytes
+                ):
+                    self._fail(lane)
+                    return
+            elif (
                 len(lane.queue) >= self.limits.max_events_per_lane
                 or self.events >= self.limits.max_events
                 or lane.size + size > self.limits.max_bytes_per_lane
@@ -245,6 +266,18 @@ def dispatch(self, event: ResponsesServerEvent, size: int) -> None:
             lane.size += size
             self.size += size
             self.events += 1
+            if self.isolate_lane_overflow:
+                while self.events > self.limits.max_events or self.size > self.limits.max_bytes:
+                    byte_pressure = self.size > self.limits.max_bytes
+                    largest, largest_size = lane, 0
+                    for candidate in self.lanes.values():
+                        # Exclude this incoming event once; ties follow registration order.
+                        backlog = candidate.size if byte_pressure else len(candidate.queue)
+                        if candidate is lane:
+                            backlog -= size if byte_pressure else 1
+                        if backlog > largest_size:
+                            largest, largest_size = candidate, backlog
+                    self._fail(largest)
             lane.wake.set()
             self.condition.notify_all()
 
@@ -259,6 +292,8 @@ def finish(self, error: BaseException | None = None) -> None:
 
     def pop(self, lane: _LaneState) -> ResponsesServerEvent | None:
         # The caller holds the condition. No await occurs after dequeueing.
+        if lane.failure is not None:
+            raise lane.failure
         if lane.closed:
             raise EOFError("The WebSocket lane is detached")
         if lane.queue:
@@ -353,11 +388,17 @@ def get_final_response(self, *, timeout: float | None = None) -> Response:
             remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
             self.recv(timeout=remaining)
         while self._state.accumulator.final is None:
-            if self._state.accumulator.error is not None:
-                raise self._state.accumulator.error
+            with self._session._router.condition:
+                if self._state.failure is not None:
+                    raise self._state.failure
+                if self._state.accumulator.error is not None:
+                    raise self._state.accumulator.error
             remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
             self.recv(timeout=remaining)
-        return self._state.accumulator.final
+        with self._session._router.condition:
+            if self._state.failure is not None:
+                raise self._state.failure
+            return self._state.accumulator.final
 
     def close(self) -> None:
         self._session._router.detach(self._state)
@@ -375,13 +416,19 @@ class ResponsesWebSocketSession:
     Keep a lane open for sequential responses. Reconnect releases closed lanes;
     open lanes remain registered on the new socket.
 
+    By default queue overflow closes the connection. ``isolate_lane_overflow=True``
+    instead fails the affected lane and drops its later events. A failed default
+    lane also loses later default/API controls; physical failures remain global.
+
     Recovery remains the existing connection's opt-in policy. Reconnecting a
     transport does not restore server-side response state or replay lane sends.
     """
 
-    def __init__(self, connection: ResponsesConnection, *, limits: ResponsesWebSocketLimits) -> None:
+    def __init__(
+        self, connection: ResponsesConnection, *, limits: ResponsesWebSocketLimits, isolate_lane_overflow: bool = False
+    ) -> None:
         self.connection = connection
-        self._router = _Router(limits, connection)
+        self._router = _Router(limits, connection, isolate_lane_overflow=isolate_lane_overflow)
         self._thread: threading.Thread | None = None
         self._closed = False
         self._owns_connection = False
@@ -515,10 +562,16 @@ async def get_final_response(self) -> Response:
         ):
             await self.recv()
         while self._state.accumulator.final is None:
-            if self._state.accumulator.error is not None:
-                raise self._state.accumulator.error
+            with self._session._router.condition:
+                if self._state.failure is not None:
+                    raise self._state.failure
+                if self._state.accumulator.error is not None:
+                    raise self._state.accumulator.error
             await self.recv()
-        return self._state.accumulator.final
+        with self._session._router.condition:
+            if self._state.failure is not None:
+                raise self._state.failure
+            return self._state.accumulator.final
 
     def close(self) -> None:
         self._session._router.detach(self._state)
@@ -527,9 +580,15 @@ def close(self) -> None:
 class AsyncResponsesWebSocketSession:
     """Async counterpart of ResponsesWebSocketSession, owning the connection lifetime."""
 
-    def __init__(self, connection: AsyncResponsesConnection, *, limits: ResponsesWebSocketLimits) -> None:
+    def __init__(
+        self,
+        connection: AsyncResponsesConnection,
+        *,
+        limits: ResponsesWebSocketLimits,
+        isolate_lane_overflow: bool = False,
+    ) -> None:
         self.connection = connection
-        self._router = _Router(limits, connection)
+        self._router = _Router(limits, connection, isolate_lane_overflow=isolate_lane_overflow)
         self._task: asyncio.Task[None] | None = None
         self._close_task: asyncio.Task[None] | None = None
         self._closed = False

tests/lib/responses/test_websocket_session.py

@@ -90,6 +90,7 @@ async def session_for(
     connection_options: dict[str, Any] | None = None,
     client_headers: dict[str, str] | None = None,
     connection_headers: Headers | None = None,
+    session_options: dict[str, Any] | None = None,
 ) -> AsyncGenerator[Driver]:
     options = connection_options or {}
     headers = {"X-Contract-Test": "synthetic", **(connection_headers or {})}
@@ -102,7 +103,7 @@ async def session_for(
                 http_client=httpx2.Client(trust_env=False),
             ) as client,
             client.responses.connect(extra_query={"contract": "1"}, extra_headers=headers, **options) as connection,
-            ResponsesWebSocketSession(connection, limits=limits) as session,
+            ResponsesWebSocketSession(connection, limits=limits, **(session_options or {})) as session,
         ):
             yield Driver(session, mode)
     else:
@@ -116,7 +117,7 @@ async def session_for(
             async_client.responses.connect(
                 extra_query={"contract": "1"}, extra_headers=headers, **options
             ) as async_connection,
-            AsyncResponsesWebSocketSession(async_connection, limits=limits) as async_session,
+            AsyncResponsesWebSocketSession(async_connection, limits=limits, **(session_options or {})) as async_session,
         ):
             yield Driver(async_session, mode)