fix(realtime): honor native options for legacy beta sockets

src/openai/resources/beta/realtime/realtime.py

@@ -44,7 +44,7 @@
     TranscriptionSessionsWithStreamingResponse,
     AsyncTranscriptionSessionsWithStreamingResponse,
 )
-from ....types.websocket_connection_options import WebsocketConnectionOptions
+from ....types.websocket_connection_options import WebsocketConnectionOptions, SyncWebSocketConnectionOptions
 from ....types.beta.realtime.realtime_client_event import RealtimeClientEvent
 from ....types.beta.realtime.realtime_server_event import RealtimeServerEvent
 from ....types.beta.realtime.conversation_item_param import ConversationItemParam
@@ -95,7 +95,7 @@ def connect(
         model: str,
         extra_query: Query = {},
         extra_headers: Headers = {},
-        websocket_connection_options: WebsocketConnectionOptions = {},
+        websocket_connection_options: SyncWebSocketConnectionOptions = {},
     ) -> RealtimeConnectionManager:
         """
         The Realtime API enables you to build low-latency, multi-modal conversational experiences. It currently supports text and audio as both input and output, as well as function calling.
@@ -382,6 +382,7 @@ async def __aenter__(self) -> AsyncRealtimeConnection:
         if self.__websocket_connection_options:
             log.debug("Custom WebSocket connection options provided")
 
+        options: WebsocketConnectionOptions = {"max_size": None, **self.__websocket_connection_options}
         self.__connection = AsyncRealtimeConnection(
             await connect(
                 str(url),
@@ -393,7 +394,7 @@ async def __aenter__(self) -> AsyncRealtimeConnection:
                     },
                     self.__extra_headers,
                 ),
-                **self.__websocket_connection_options,
+                **options,
             )
         )
 
@@ -524,7 +525,7 @@ def __init__(
         model: str,
         extra_query: Query,
         extra_headers: Headers,
-        websocket_connection_options: WebsocketConnectionOptions,
+        websocket_connection_options: SyncWebSocketConnectionOptions,
     ) -> None:
         self.__client = client
         self.__model = model
@@ -575,6 +576,7 @@ def __enter__(self) -> RealtimeConnection:
         if self.__websocket_connection_options:
             log.debug("Custom WebSocket connection options provided")
 
+        options: SyncWebSocketConnectionOptions = {"max_size": None, **self.__websocket_connection_options}
         self.__connection = RealtimeConnection(
             connect(
                 str(url),
@@ -586,7 +588,7 @@ def __enter__(self) -> RealtimeConnection:
                     },
                     self.__extra_headers,
                 ),
-                **self.__websocket_connection_options,
+                **options,
             )
         )
 

tests/lib/test_beta_realtime_websocket_options.py

@@ -0,0 +1,84 @@
+import json
+import asyncio
+from typing import Literal, Iterator, Optional
+from threading import Thread
+
+import pytest
+from websockets.exceptions import ConnectionClosed, ConnectionClosedError
+from websockets.sync.server import ServerConnection, serve
+
+from openai import OpenAI, AsyncOpenAI
+from openai.types.websocket_connection_options import WebSocketConnectionOptions
+
+
+@pytest.fixture(autouse=True)
+def bypass_loopback_proxy(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("NO_PROXY", "*")
+    monkeypatch.setenv("no_proxy", "*")
+
+
+@pytest.fixture
+def realtime_peer(payload_size: int) -> Iterator[tuple[str, bytes]]:
+    wire = json.dumps({"type": "future.event", "text": "x" * payload_size}).encode()
+
+    def accept(socket: ServerConnection) -> None:
+        socket.send(wire)
+        try:
+            socket.recv()
+        except ConnectionClosed:
+            # The client closes after consuming or rejecting the test event.
+            pass
+
+    with serve(accept, "127.0.0.1", 0) as server:
+        thread = Thread(target=server.serve_forever)
+        thread.start()
+        try:
+            yield f"ws://127.0.0.1:{server.socket.getsockname()[1]}", wire
+        finally:
+            server.shutdown()
+            thread.join(3)
+            assert not thread.is_alive()
+
+
+@pytest.mark.parametrize("mode", ["sync", "async"])
+@pytest.mark.parametrize(
+    ("payload_size", "max_size", "reject"),
+    [
+        pytest.param(128, "default", False, id="default-small"),
+        pytest.param(2**20 + 32, "default", False, id="default-over-native"),
+        pytest.param(2**20 + 32, None, False, id="explicit-unlimited"),
+        pytest.param(128, 512, False, id="explicit-limit-under"),
+        pytest.param(900, 512, True, id="explicit-limit-over"),
+        pytest.param(1, 0, True, id="explicit-zero"),
+    ],
+)
+async def test_beta_realtime_preserves_receive_limit_choice(
+    realtime_peer: tuple[str, bytes], mode: Literal["sync", "async"], max_size: Optional[int | str], reject: bool
+) -> None:
+    url, expected = realtime_peer
+    options: WebSocketConnectionOptions = {}
+    if isinstance(max_size, int) or max_size is None:
+        options["max_size"] = max_size
+
+    if mode == "sync":
+
+        def receive() -> bytes:
+            with OpenAI(api_key="local", websocket_base_url=url) as client:
+                with client.beta.realtime.connect(model="test", websocket_connection_options=options) as conn:
+                    return conn.recv_bytes()
+
+        if reject:
+            with pytest.raises(ConnectionClosedError) as exc:
+                await asyncio.wait_for(asyncio.to_thread(receive), 10)
+            assert exc.value.sent is not None and exc.value.sent.code == 1009
+        else:
+            assert await asyncio.wait_for(asyncio.to_thread(receive), 10) == expected
+    else:
+        async with AsyncOpenAI(api_key="local", websocket_base_url=url) as client:
+            async with client.beta.realtime.connect(model="test", websocket_connection_options=options) as conn:
+                if reject:
+                    with pytest.raises(ConnectionClosedError) as exc:
+                        await asyncio.wait_for(conn.recv_bytes(), 10)
+                    assert exc.value.sent is not None and exc.value.sent.code == 1009
+                else:
+                    assert await asyncio.wait_for(conn.recv_bytes(), 10) == expected

tests/test_websocket_connection_options.py

@@ -25,7 +25,7 @@ def test_beta_realtime_import_still_works_with_old_alias() -> None:
 
 
 if TYPE_CHECKING:
-    from openai import AsyncOpenAI
+    from openai import OpenAI, AsyncOpenAI
     from openai.types.websocket_connection_options import SyncWebSocketConnectionOptions
 
     def native_options_contract(
@@ -49,3 +49,8 @@ def native_options_contract(
         client.realtime.connect(model="test", websocket_connection_options=original)
         client.responses.connect(websocket_connection_options=legacy)
         return valid_sync, invalid_sync
+
+    def beta_realtime_options_contract(client: OpenAI, async_client: AsyncOpenAI) -> None:
+        client.beta.realtime.connect(model="test", websocket_connection_options={"max_size": 512})
+        client.beta.realtime.connect(model="test", websocket_connection_options={"write_limit": 65536})  # pyright: ignore[reportArgumentType]
+        async_client.beta.realtime.connect(model="test", websocket_connection_options={"write_limit": 65536})