diff --git a/src/openai/_send_queue.py b/src/openai/_send_queue.py index 56d7824422..b3eb812b0e 100644 --- a/src/openai/_send_queue.py +++ b/src/openai/_send_queue.py @@ -37,11 +37,12 @@ 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() @@ -49,14 +50,21 @@ def flush_sync(self, send: typing.Callable[[str], object]) -> None: 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) diff --git a/src/openai/resources/realtime/realtime.py b/src/openai/resources/realtime/realtime.py index aef7757cb0..f541bdd886 100644 --- a/src/openai/resources/realtime/realtime.py +++ b/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") diff --git a/tests/lib/live/test_websocket_roles.py b/tests/lib/live/test_websocket_roles.py index e3b0ae5cdc..40de79fed2 100644 --- a/tests/lib/live/test_websocket_roles.py +++ b/tests/lib/live/test_websocket_roles.py @@ -9,10 +9,17 @@ from websockets.sync.server import ServerConnection from openai import OpenAI, AsyncOpenAI -from openai.types.live import SessionStartedEvent, SessionUpdatedEvent, OutputTranscriptDeltaEvent +from openai.types.live import ( + ServerEvent, + SessionClosedEvent, + SessionStartedEvent, + SessionUpdatedEvent, + OutputTranscriptDeltaEvent, +) from openai.resources.live.live import LiveConnection, AsyncLiveConnection from openai.resources.live.forks import ForksConnection, AsyncForksConnection +from .helpers import FakeClock, Recording, ClockGrouper, AsyncClockGrouper from ..responses.test_websocket_session import script_server @@ -157,3 +164,137 @@ def script(socket: ServerConnection) -> None: assert isinstance(async_updated, SessionUpdatedEvent) assert async_updated.client_event_id == "caller-update" assert async_updated.session.id == "live_fixture" + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +async def test_disposed_grouper_leaves_dispatcher_other_observers_and_socket_usable(mode: str) -> None: + session = {"id": "live_fixture", "model": "gpt-live-1", "status": "active", "expires_at": 123} + first_clock, second_clock = FakeClock(), FakeClock() + transcripts = [ + { + "type": "session.output_transcript.delta", + "event_id": f"part-{index}", + "delta": value, + "start_ms": index * 200, + "end_ms": (index + 1) * 200, + } + for index, value in enumerate(["One", " two", " three", " four"]) + ] + observed: list[ServerEvent] = [] + typed: list[OutputTranscriptDeltaEvent] = [] + + def script(socket: ServerConnection) -> None: + assert json.loads(socket.recv(timeout=5)) == { + "type": "session.start", + "event_id": "caller-start", + "session": {"model": "gpt-live-1"}, + } + socket.send(json.dumps({"type": "session.started", "event_id": "started", "session": session})) + for event in transcripts[:2]: + socket.send(json.dumps(event)) + # Issued only after the caller detaches and closes its first grouper. + assert json.loads(socket.recv(timeout=5)) == { + "type": "session.update", + "event_id": "after-dispose", + "session": {}, + } + socket.send( + json.dumps( + { + "type": "session.updated", + "event_id": "updated", + "client_event_id": "after-dispose", + "session": session, + "future_metadata": {"explicit_null": None, "nested": [1, "retained"]}, + } + ) + ) + for event in transcripts[2:]: + socket.send(json.dumps(event)) + assert json.loads(socket.recv(timeout=5)) == {"type": "session.close", "event_id": "caller-finish"} + socket.send(json.dumps({"type": "session.closed", "event_id": "closed", "reason": "client_close"})) + # The shared fixture rejects extra writes/reconnects and waits for caller close. + + with script_server(script) as url: + if mode == "sync": + first = ClockGrouper(first_clock) + second = ClockGrouper(second_clock) + first_record, second_record = Recording(first, first_clock), Recording(second, second_clock) + with OpenAI(api_key="ek_fake_live", base_url=url, http_client=httpx2.Client(trust_env=False)) as client: + with client.live.connect() as connection: + connection.on("session.output_transcript.delta", first.push) + connection.on("session.output_transcript.delta", second.push) + connection.on("session.output_transcript.delta", typed.append) + connection.on("event", observed.append) + + def manage(event: OutputTranscriptDeltaEvent) -> None: + if event.event_id == "part-1": + assert first_clock.pending + connection.off("session.output_transcript.delta", first.push) + first.close() + first.close() + assert not first_clock.pending + connection.session.update(session={}, event_id="after-dispose") + elif event.event_id == "part-3": + connection.session.close(event_id="caller-finish") + + def close_connection(_event: SessionClosedEvent) -> None: + connection.close() + + connection.on("session.output_transcript.delta", manage) + connection.on("session.closed", close_connection) + connection.session.start(session={"model": "gpt-live-1"}, event_id="caller-start") + connection.dispatch_events() + second.close() + else: + async_first = AsyncClockGrouper(first_clock) + async_second = AsyncClockGrouper(second_clock) + first_record, second_record = ( + Recording(async_first, first_clock), + Recording(async_second, second_clock), + ) + async with AsyncOpenAI( + api_key="ek_fake_live", base_url=url, http_client=httpx2.AsyncClient(trust_env=False) + ) as async_client: + async with async_client.live.connect() as async_connection: + async_connection.on("session.output_transcript.delta", async_first.push) + async_connection.on("session.output_transcript.delta", async_second.push) + async_connection.on("session.output_transcript.delta", typed.append) + async_connection.on("event", observed.append) + + async def async_manage(event: OutputTranscriptDeltaEvent) -> None: + if event.event_id == "part-1": + assert first_clock.pending + async_connection.off("session.output_transcript.delta", async_first.push) + await async_first.close() + await async_first.close() + assert not first_clock.pending + await async_connection.session.update(session={}, event_id="after-dispose") + elif event.event_id == "part-3": + await async_connection.session.close(event_id="caller-finish") + + async def close_async_connection(_event: SessionClosedEvent) -> None: + await async_connection.close() + + async_connection.on("session.output_transcript.delta", async_manage) + async_connection.on("session.closed", close_async_connection) + await async_connection.session.start(session={"model": "gpt-live-1"}, event_id="caller-start") + await asyncio.wait_for(async_connection.dispatch_events(), timeout=5) + await async_second.close() + + assert [event.to_dict().get("event_id") for event in observed] == [ + "started", + "part-0", + "part-1", + "updated", + "part-2", + "part-3", + "closed", + ] + assert [event.to_dict() for event in typed] == transcripts + updated = observed[3] + assert isinstance(updated, SessionUpdatedEvent) + assert updated.to_dict()["future_metadata"] == {"explicit_null": None, "nested": [1, "retained"]} + assert [(event.segment.text, event.reason) for event in first_record.closed] == [("One two", "manual")] + assert [(event.segment.text, event.reason) for event in second_record.closed] == [("One two three four", "manual")] + assert not first_clock.pending and not second_clock.pending diff --git a/tests/lib/test_realtime_websocket_contract.py b/tests/lib/test_realtime_websocket_contract.py index 5dddcd2707..d1dde5f00c 100644 --- a/tests/lib/test_realtime_websocket_contract.py +++ b/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")]) diff --git a/tests/test_send_queue_reconnect.py b/tests/test_send_queue_reconnect.py index fec12479ab..5d48ed5097 100644 --- a/tests/test_send_queue_reconnect.py +++ b/tests/test_send_queue_reconnect.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio from unittest.mock import AsyncMock, MagicMock import pytest @@ -22,15 +23,17 @@ def test_reconnect_retries_bounded_send_queue( q.enqueue("aaa") ws = MagicMock() attempts = 0 + realtime = connection_type is RealtimeConnection def failing_send(data: str) -> None: nonlocal attempts - assert data == "aaa" + assert data == ("b" if realtime and attempts == 1 else "aaa") if attempts == 0: q.enqueue("b") attempts += 1 - with pytest.raises(WebSocketQueueFullError): - q.enqueue("c") + if not realtime or attempts == 1: + with pytest.raises(WebSocketQueueFullError): + q.enqueue("c") raise RuntimeError("fake send failure") ws.send.side_effect = failing_send @@ -44,10 +47,10 @@ def failing_send(data: str) -> None: ) for expected_attempts in range(1, 4): assert connection._reconnect(RuntimeError("fake disconnect")) - assert q._bytes == 4 - assert attempts == expected_attempts + assert q._bytes == ((1 if expected_attempts == 1 else 0) if realtime else 4) + assert attempts == (min(expected_attempts, 2) if realtime else expected_attempts) assert not connection._reconnect(RuntimeError("fake disconnect")) - assert attempts == 3 + assert attempts == (2 if realtime else 3) # A healthy application event resets the budget; an upgrade alone does not. ws.recv.return_value = '{"type": "response.created"}' @@ -56,7 +59,7 @@ def failing_send(data: str) -> None: sent: list[str] = [] ws.send.side_effect = sent.append assert connection._reconnect(RuntimeError("fake disconnect")) - assert sent == ["aaa", "b"] + assert sent == ([] if realtime else ["aaa", "b"]) assert q._bytes == 0 for _ in range(2): assert connection._reconnect(RuntimeError("fake disconnect")) @@ -76,15 +79,17 @@ async def test_async_reconnect_retries_bounded_send_queue( q.enqueue("aaa") ws = MagicMock() attempts = 0 + realtime = connection_type is AsyncRealtimeConnection async def failing_send(data: str) -> None: nonlocal attempts - assert data == "aaa" + assert data == ("b" if realtime and attempts == 1 else "aaa") if attempts == 0: q.enqueue("b") attempts += 1 - with pytest.raises(WebSocketQueueFullError): - q.enqueue("c") + if not realtime or attempts == 1: + with pytest.raises(WebSocketQueueFullError): + q.enqueue("c") raise RuntimeError("fake send failure") ws.send = AsyncMock(side_effect=failing_send) @@ -98,10 +103,10 @@ async def failing_send(data: str) -> None: ) for expected_attempts in range(1, 4): assert await connection._reconnect(RuntimeError("fake disconnect")) - assert q._bytes == 4 - assert attempts == expected_attempts + assert q._bytes == ((1 if expected_attempts == 1 else 0) if realtime else 4) + assert attempts == (min(expected_attempts, 2) if realtime else expected_attempts) assert not await connection._reconnect(RuntimeError("fake disconnect")) - assert attempts == 3 + assert attempts == (2 if realtime else 3) # A healthy application event resets the budget; an upgrade alone does not. ws.recv = AsyncMock(return_value='{"type": "response.created"}') @@ -110,8 +115,44 @@ async def failing_send(data: str) -> None: sent: list[str] = [] ws.send.side_effect = sent.append assert await connection._reconnect(RuntimeError("fake disconnect")) - assert sent == ["aaa", "b"] + assert sent == ([] if realtime else ["aaa", "b"]) assert q._bytes == 0 for _ in range(2): assert await connection._reconnect(RuntimeError("fake disconnect")) assert not await connection._reconnect(RuntimeError("retry budget exhausted")) + + +@pytest.mark.asyncio +async def test_cancelled_realtime_flush_does_not_replay_active_send() -> None: + q = SendQueue(max_bytes=4) + q.enqueue("é") + q.enqueue("b") + active = asyncio.Event() + socket = MagicMock() + + async def suspended_send(data: str) -> None: + assert data == "é" + active.set() + await asyncio.Event().wait() + + socket.send = AsyncMock(side_effect=suspended_send) + connection = AsyncRealtimeConnection(socket, send_queue=q) + task = asyncio.create_task(connection._flush_send_queue()) + try: + await asyncio.wait_for(active.wait(), 5) + q.enqueue("c") + with pytest.raises(WebSocketQueueFullError): + q.enqueue("d") + finally: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + # Only the active send was attempted; its bytes are released, and the + # queued tail and concurrent enqueue remain in order for the next socket. + q.enqueue("dd") + sent: list[str] = [] + socket.send.side_effect = sent.append + await connection._flush_send_queue() + assert sent == ["b", "c", "dd"] + q.enqueue("1234")