From a5a6f80b56b834bf524acfda149bdfd95c68fe53 Mon Sep 17 00:00:00 2001 From: markstuart-oai Date: Sun, 27 Sep 2026 18:13:09 +0000 Subject: [PATCH 1/2] fix(realtime): preserve base URL queries during WebSocket setup --- src/openai/_base_client.py | 5 +- src/openai/resources/realtime/realtime.py | 6 +- tests/lib/test_realtime_websocket_contract.py | 135 ++++++++++++++++++ 3 files changed, 142 insertions(+), 4 deletions(-) create mode 100644 tests/lib/test_realtime_websocket_contract.py diff --git a/src/openai/_base_client.py b/src/openai/_base_client.py index c8830a6a9f..883bed6e19 100644 --- a/src/openai/_base_client.py +++ b/src/openai/_base_client.py @@ -475,9 +475,10 @@ def __init__( self._validate_max_retries(max_retries) def _enforce_trailing_slash(self, url: URL) -> URL: - if url.raw_path.endswith(b"/"): + path, separator, query = url.raw_path.partition(b"?") + if path.endswith(b"/"): return url - return url.copy_with(raw_path=url.raw_path + b"/") + return url.copy_with(raw_path=path + b"/" + separator + query) def _make_status_error_from_response( self, diff --git a/src/openai/resources/realtime/realtime.py b/src/openai/resources/realtime/realtime.py index fab966898e..aef7757cb0 100644 --- a/src/openai/resources/realtime/realtime.py +++ b/src/openai/resources/realtime/realtime.py @@ -772,7 +772,8 @@ def _prepare_url(self) -> httpx2.URL: ws_scheme = "ws" if scheme == "http" else "wss" base_url = self.__client._base_url.copy_with(scheme=ws_scheme) - merge_raw_path = base_url.raw_path.rstrip(b"/") + b"/realtime" + path, separator, query = base_url.raw_path.partition(b"?") + merge_raw_path = path.rstrip(b"/") + b"/realtime" + separator + query return base_url.copy_with(raw_path=merge_raw_path) async def __aexit__( @@ -1281,7 +1282,8 @@ def _prepare_url(self) -> httpx2.URL: ws_scheme = "ws" if scheme == "http" else "wss" base_url = self.__client._base_url.copy_with(scheme=ws_scheme) - merge_raw_path = base_url.raw_path.rstrip(b"/") + b"/realtime" + path, separator, query = base_url.raw_path.partition(b"?") + merge_raw_path = path.rstrip(b"/") + b"/realtime" + separator + query return base_url.copy_with(raw_path=merge_raw_path) def __exit__( diff --git a/tests/lib/test_realtime_websocket_contract.py b/tests/lib/test_realtime_websocket_contract.py new file mode 100644 index 0000000000..5dddcd2707 --- /dev/null +++ b/tests/lib/test_realtime_websocket_contract.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +import json +import asyncio +from urllib.parse import parse_qs, urlsplit + +import httpx2 +import pytest +from websockets.sync.server import ServerConnection + +from openai import OpenAI, AsyncOpenAI +from openai.types.realtime.realtime_error_event import RealtimeErrorEvent +from openai.types.realtime.input_audio_buffer_cleared_event import InputAudioBufferClearedEvent + +from .responses.test_websocket_session import script_server + + +@pytest.fixture(autouse=True) +def bypass_loopback_proxy(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("NO_PROXY", "127.0.0.1") + monkeypatch.setenv("no_proxy", "127.0.0.1") + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize("base_query", ["", "?tenant=sample"], ids=["custom-path", "custom-path-and-query"]) +async def test_realtime_error_keeps_connection_open(mode: str, base_query: str) -> None: + def script(socket: ServerConnection) -> None: + assert socket.request is not None + target = urlsplit(socket.request.path) + assert target.path == "/v1/customer/realtime" + expected_query = {"model": ["gpt-realtime"], "contract": ["socket"]} + if base_query: + expected_query["tenant"] = ["sample"] + assert parse_qs(target.query) == expected_query + assert socket.request.headers.get_all("Authorization") == ["Bearer ek_fake_realtime"] + assert socket.request.headers["X-Realtime-Test"] == "connection" + assert socket.request.headers["X-Client-Only"] == "preserved" + assert socket.request.headers.get("Sec-WebSocket-Extensions") is None + assert json.loads(socket.recv(timeout=5)) == { + "type": "input_audio_buffer.append", + "event_id": "append-1", + "audio": "AA==", + } + socket.send( + json.dumps( + { + "type": "error", + "event_id": "error-1", + "error": { + "type": "invalid_request_error", + "code": "invalid_audio", + "param": "audio", + "message": "Synthetic audio rejected", + "event_id": "append-1", + }, + } + ) + ) + # A documented recoverable error must not close or replace the physical socket. + assert json.loads(socket.recv(timeout=5)) == {"type": "input_audio_buffer.clear", "event_id": "clear-1"} + socket.send(json.dumps({"type": "input_audio_buffer.cleared", "event_id": "clear-1", "future": {"kept": True}})) + socket.send(' { "type": "realtime.future", "event_id": "future-1", "text": "東京🙂" }\n') + # script_server verifies one connection, a clean caller close, and no extra sends. + + with script_server(script) as url: + base_url = f"{url}/customer{base_query}" + client_headers = {"X-Realtime-Test": "client", "X-Client-Only": "preserved"} + if mode == "sync": + with OpenAI( + api_key="ek_fake_realtime", + base_url=base_url, + default_headers=client_headers, + http_client=httpx2.Client(trust_env=False), + ) as client: + with client.realtime.connect( + model="gpt-realtime", + extra_query={"contract": "socket"}, + extra_headers={"X-Realtime-Test": "connection"}, + websocket_connection_options={"compression": None}, + ) as connection: + connection.input_audio_buffer.append(audio="AA==", event_id="append-1") + error = connection.recv() + assert isinstance(error, RealtimeErrorEvent) + assert (error.error.code, error.error.param, error.error.event_id) == ( + "invalid_audio", + "audio", + "append-1", + ) + connection.input_audio_buffer.clear(event_id="clear-1") + cleared = connection.recv() + assert isinstance(cleared, InputAudioBufferClearedEvent) + assert cleared.to_dict(exclude_unset=True) == { + "type": "input_audio_buffer.cleared", + "event_id": "clear-1", + "future": {"kept": True}, + } + assert json.loads(connection.recv_bytes()) == { + "type": "realtime.future", + "event_id": "future-1", + "text": "東京🙂", + } + else: + async with AsyncOpenAI( + api_key="ek_fake_realtime", + base_url=base_url, + default_headers=client_headers, + http_client=httpx2.AsyncClient(trust_env=False), + ) as async_client: + async with async_client.realtime.connect( + model="gpt-realtime", + extra_query={"contract": "socket"}, + extra_headers={"X-Realtime-Test": "connection"}, + websocket_connection_options={"compression": None}, + ) as async_connection: + await async_connection.input_audio_buffer.append(audio="AA==", event_id="append-1") + async_error = await asyncio.wait_for(async_connection.recv(), timeout=5) + assert isinstance(async_error, RealtimeErrorEvent) + assert (async_error.error.code, async_error.error.param, async_error.error.event_id) == ( + "invalid_audio", + "audio", + "append-1", + ) + await async_connection.input_audio_buffer.clear(event_id="clear-1") + async_cleared = await asyncio.wait_for(async_connection.recv(), timeout=5) + assert isinstance(async_cleared, InputAudioBufferClearedEvent) + assert async_cleared.to_dict(exclude_unset=True) == { + "type": "input_audio_buffer.cleared", + "event_id": "clear-1", + "future": {"kept": True}, + } + assert json.loads(await asyncio.wait_for(async_connection.recv_bytes(), timeout=5)) == { + "type": "realtime.future", + "event_id": "future-1", + "text": "東京🙂", + } From 6665a3ec66a37a45525ac20b281f3f7ee474a6d2 Mon Sep 17 00:00:00 2001 From: markstuart-oai Date: Sun, 27 Sep 2026 18:19:01 +0000 Subject: [PATCH 2/2] fix(live): preserve base URL queries across WebSocket roles --- src/openai/resources/live/forks.py | 12 +- src/openai/resources/live/live.py | 6 +- src/openai/resources/live/sideband.py | 12 +- tests/lib/live/test_websocket_roles.py | 159 +++++++++++++++++++++++++ 4 files changed, 175 insertions(+), 14 deletions(-) create mode 100644 tests/lib/live/test_websocket_roles.py diff --git a/src/openai/resources/live/forks.py b/src/openai/resources/live/forks.py index 9c1a243238..e27846290e 100644 --- a/src/openai/resources/live/forks.py +++ b/src/openai/resources/live/forks.py @@ -594,9 +594,9 @@ def _prepare_url(self) -> httpx2.URL: ws_scheme = "ws" if scheme == "http" else "wss" base_url = self.__client._base_url.copy_with(scheme=ws_scheme) - merge_raw_path = base_url.raw_path.rstrip(b"/") + path_template( - "/live/sessions/{session_id}/fork", session_id=self.__session_id - ).encode("utf-8") + path, separator, query = base_url.raw_path.partition(b"?") + endpoint = path_template("/live/sessions/{session_id}/fork", session_id=self.__session_id).encode("utf-8") + merge_raw_path = path.rstrip(b"/") + endpoint + separator + query return base_url.copy_with(raw_path=merge_raw_path) async def __aexit__( @@ -1081,9 +1081,9 @@ def _prepare_url(self) -> httpx2.URL: ws_scheme = "ws" if scheme == "http" else "wss" base_url = self.__client._base_url.copy_with(scheme=ws_scheme) - merge_raw_path = base_url.raw_path.rstrip(b"/") + path_template( - "/live/sessions/{session_id}/fork", session_id=self.__session_id - ).encode("utf-8") + path, separator, query = base_url.raw_path.partition(b"?") + endpoint = path_template("/live/sessions/{session_id}/fork", session_id=self.__session_id).encode("utf-8") + merge_raw_path = path.rstrip(b"/") + endpoint + separator + query return base_url.copy_with(raw_path=merge_raw_path) def __exit__( diff --git a/src/openai/resources/live/live.py b/src/openai/resources/live/live.py index 36c08a6e30..1af46d4e3f 100644 --- a/src/openai/resources/live/live.py +++ b/src/openai/resources/live/live.py @@ -813,7 +813,8 @@ def _prepare_url(self) -> httpx2.URL: ws_scheme = "ws" if scheme == "http" else "wss" base_url = self.__client._base_url.copy_with(scheme=ws_scheme) - merge_raw_path = base_url.raw_path.rstrip(b"/") + b"/live/sessions" + path, separator, query = base_url.raw_path.partition(b"?") + merge_raw_path = path.rstrip(b"/") + b"/live/sessions" + separator + query return base_url.copy_with(raw_path=merge_raw_path) async def __aexit__( @@ -1295,7 +1296,8 @@ def _prepare_url(self) -> httpx2.URL: ws_scheme = "ws" if scheme == "http" else "wss" base_url = self.__client._base_url.copy_with(scheme=ws_scheme) - merge_raw_path = base_url.raw_path.rstrip(b"/") + b"/live/sessions" + path, separator, query = base_url.raw_path.partition(b"?") + merge_raw_path = path.rstrip(b"/") + b"/live/sessions" + separator + query return base_url.copy_with(raw_path=merge_raw_path) def __exit__( diff --git a/src/openai/resources/live/sideband.py b/src/openai/resources/live/sideband.py index e15a836139..043a7233e0 100644 --- a/src/openai/resources/live/sideband.py +++ b/src/openai/resources/live/sideband.py @@ -602,9 +602,9 @@ def _prepare_url(self) -> httpx2.URL: ws_scheme = "ws" if scheme == "http" else "wss" base_url = self.__client._base_url.copy_with(scheme=ws_scheme) - merge_raw_path = base_url.raw_path.rstrip(b"/") + path_template( - "/live/sessions/{session_id}/attach", session_id=self.__session_id - ).encode("utf-8") + path, separator, query = base_url.raw_path.partition(b"?") + endpoint = path_template("/live/sessions/{session_id}/attach", session_id=self.__session_id).encode("utf-8") + merge_raw_path = path.rstrip(b"/") + endpoint + separator + query return base_url.copy_with(raw_path=merge_raw_path) async def __aexit__( @@ -1094,9 +1094,9 @@ def _prepare_url(self) -> httpx2.URL: ws_scheme = "ws" if scheme == "http" else "wss" base_url = self.__client._base_url.copy_with(scheme=ws_scheme) - merge_raw_path = base_url.raw_path.rstrip(b"/") + path_template( - "/live/sessions/{session_id}/attach", session_id=self.__session_id - ).encode("utf-8") + path, separator, query = base_url.raw_path.partition(b"?") + endpoint = path_template("/live/sessions/{session_id}/attach", session_id=self.__session_id).encode("utf-8") + merge_raw_path = path.rstrip(b"/") + endpoint + separator + query return base_url.copy_with(raw_path=merge_raw_path) def __exit__( diff --git a/tests/lib/live/test_websocket_roles.py b/tests/lib/live/test_websocket_roles.py new file mode 100644 index 0000000000..e3b0ae5cdc --- /dev/null +++ b/tests/lib/live/test_websocket_roles.py @@ -0,0 +1,159 @@ +from __future__ import annotations + +import json +import asyncio +from urllib.parse import parse_qs, urlsplit + +import httpx2 +import pytest +from websockets.sync.server import ServerConnection + +from openai import OpenAI, AsyncOpenAI +from openai.types.live import SessionStartedEvent, SessionUpdatedEvent, OutputTranscriptDeltaEvent +from openai.resources.live.live import LiveConnection, AsyncLiveConnection +from openai.resources.live.forks import ForksConnection, AsyncForksConnection + +from ..responses.test_websocket_session import script_server + + +@pytest.fixture(autouse=True) +def bypass_loopback_proxy(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("NO_PROXY", "127.0.0.1") + monkeypatch.setenv("no_proxy", "127.0.0.1") + + +@pytest.mark.parametrize("role", ["primary", "fork", "sideband"]) +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize("base_query", ["", "?tenant=sample"], ids=["custom-path", "custom-path-and-query"]) +async def test_live_role_startup_and_routing(role: str, mode: str, base_query: str) -> None: + session = {"id": "live_fixture", "model": "gpt-live-1", "status": "active", "expires_at": 123} + + def script(socket: ServerConnection) -> None: + assert socket.request is not None + target = urlsplit(socket.request.path) + expected_path = "/v1/customer/live/sessions" + if role != "primary": + expected_path += "/stored%20%2F%3F%23%25/" + ("fork" if role == "fork" else "attach") + assert target.path == expected_path + expected_query = {"trace": ["role-contract"]} + if base_query: + expected_query["tenant"] = ["sample"] + if role == "sideband": + expected_query["graceful_close"] = ["true"] + assert parse_qs(target.query) == expected_query + assert socket.request.headers.get_all("Authorization") == ["Bearer ek_fake_live"] + + if role == "sideband": + # Attach is already active and need not send a fresh session.started. + socket.send( + json.dumps( + { + "type": "session.output_transcript.delta", + "event_id": "first", + "delta": "Attached", + "start_ms": 0, + "end_ms": 120, + } + ) + ) + else: + # This first and only start must come from the caller, with its identifier. + expected_session = {"model": "gpt-live-1"} if role == "primary" else {} + assert json.loads(socket.recv(timeout=5)) == { + "type": "session.start", + "event_id": "caller-start", + "session": expected_session, + } + socket.send( + json.dumps( + { + "type": "session.started", + "event_id": "started", + "client_event_id": "caller-start", + "session": session, + } + ) + ) + + assert json.loads(socket.recv(timeout=5)) == { + "type": "session.update", + "event_id": "caller-update", + "session": {}, + } + socket.send( + json.dumps( + { + "type": "session.updated", + "event_id": "updated", + "client_event_id": "caller-update", + "session": session, + } + ) + ) + # script_server also checks for extra sends, one socket, and caller close. + + with script_server(script) as url: + base_url = f"{url}/customer{base_query}" + extra_query = {"trace": "role-contract"} + if mode == "sync": + with OpenAI( + api_key="ek_fake_live", + base_url=base_url, + http_client=httpx2.Client(trust_env=False), + ) as client: + if role == "primary": + manager = client.live.connect(extra_query=extra_query) + elif role == "fork": + manager = client.live.forks.connect(session_id="stored /?#%", extra_query=extra_query) + else: + manager = client.live.sideband.connect( + session_id="stored /?#%", graceful_close=True, extra_query=extra_query + ) + with manager as connection: + if isinstance(connection, LiveConnection): + connection.session.start(session={"model": "gpt-live-1"}, event_id="caller-start") + elif isinstance(connection, ForksConnection): + connection.session.start(session={}, event_id="caller-start") + first = connection.recv() + if role == "sideband": + assert isinstance(first, OutputTranscriptDeltaEvent) + assert first.delta == "Attached" + else: + assert isinstance(first, SessionStartedEvent) + assert first.client_event_id == "caller-start" + connection.session.update(session={}, event_id="caller-update") + updated = connection.recv() + assert isinstance(updated, SessionUpdatedEvent) + assert updated.client_event_id == "caller-update" + assert updated.session.id == "live_fixture" + else: + async with AsyncOpenAI( + api_key="ek_fake_live", + base_url=base_url, + http_client=httpx2.AsyncClient(trust_env=False), + ) as async_client: + if role == "primary": + async_manager = async_client.live.connect(extra_query=extra_query) + elif role == "fork": + async_manager = async_client.live.forks.connect(session_id="stored /?#%", extra_query=extra_query) + else: + async_manager = async_client.live.sideband.connect( + session_id="stored /?#%", graceful_close=True, extra_query=extra_query + ) + async with async_manager as async_connection: + if isinstance(async_connection, AsyncLiveConnection): + await async_connection.session.start(session={"model": "gpt-live-1"}, event_id="caller-start") + elif isinstance(async_connection, AsyncForksConnection): + await async_connection.session.start(session={}, event_id="caller-start") + async_first = await asyncio.wait_for(async_connection.recv(), timeout=5) + if role == "sideband": + assert isinstance(async_first, OutputTranscriptDeltaEvent) + assert async_first.delta == "Attached" + else: + assert isinstance(async_first, SessionStartedEvent) + assert async_first.client_event_id == "caller-start" + await async_connection.session.update(session={}, event_id="caller-update") + async_updated = await asyncio.wait_for(async_connection.recv(), timeout=5) + assert isinstance(async_updated, SessionUpdatedEvent) + assert async_updated.client_event_id == "caller-update" + assert async_updated.session.id == "live_fixture"