From c599c5a455b8f5243c3b6b955dd0b77d9c643269 Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Mon, 3 Aug 2026 09:31:54 -0700 Subject: [PATCH 1/2] MAINT: Unify OpenAI realtime event routing Centralize OpenAI realtime event aliases and response delta accumulation while keeping atomic soft-finish and streaming barge-in termination policies in their existing owners. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: be81055b-39ed-494c-94ca-c7656200a344 --- .../openai/_openai_realtime_dispatcher.py | 37 +++--- .../openai/_openai_realtime_event_router.py | 106 ++++++++++++++++++ .../openai/openai_realtime_target.py | 68 +++++------ .../target/test_realtime_target.py | 70 ++++++++++++ 4 files changed, 225 insertions(+), 56 deletions(-) create mode 100644 pyrit/prompt_target/openai/_openai_realtime_event_router.py diff --git a/pyrit/prompt_target/openai/_openai_realtime_dispatcher.py b/pyrit/prompt_target/openai/_openai_realtime_dispatcher.py index 92fe9324f1..db3f7123ff 100644 --- a/pyrit/prompt_target/openai/_openai_realtime_dispatcher.py +++ b/pyrit/prompt_target/openai/_openai_realtime_dispatcher.py @@ -3,7 +3,6 @@ """Concrete OpenAI Realtime event dispatcher for streaming sessions.""" -import base64 import logging from typing import Any, ClassVar @@ -13,6 +12,10 @@ RealtimeTargetResult, RealtimeTurnState, ) +from pyrit.prompt_target.openai._openai_realtime_event_router import ( + _OpenAIRealtimeEventKind, + _OpenAIRealtimeEventRouter, +) logger = logging.getLogger(__name__) @@ -32,18 +35,19 @@ class _OpenAIRealtimeDispatcher(RealtimeEventDispatcher): async def _route_event_async(self, *, event: Any, state: RealtimeTurnState | None) -> None: """Route an OpenAI Realtime event to the active turn or to an input-side callback.""" event_type = getattr(event, "type", "") + event_kind = _OpenAIRealtimeEventRouter.classify_event(event_type) # Capture audio_start_ms from speech_started for the next committed event. # The server reports it reliably here but omits it from the commit event itself. # Do not return — the downstream state-aware branch still needs to fire the # barge-in cancel when speech starts mid-response. - if event_type == "input_audio_buffer.speech_started": + if event_kind is _OpenAIRealtimeEventKind.SPEECH_STARTED: speech_start = getattr(event, "audio_start_ms", None) if speech_start is not None: self._pending_speech_start_ms = speech_start # Input-side events fire callbacks regardless of whether a turn is registered. - if event_type == "input_audio_buffer.committed": + if event_kind is _OpenAIRealtimeEventKind.INPUT_COMMITTED: item_id = getattr(event, "item_id", None) if item_id is None: return @@ -63,32 +67,33 @@ async def _route_event_async(self, *, event: Any, state: RealtimeTurnState | Non if state is None or state.completion.done(): return - if event_type == "response.created": + _OpenAIRealtimeEventRouter.collect_response_delta( + event=event, + event_kind=event_kind, + audio_buffer=state.delivered_audio, + transcripts=state.delivered_transcripts, + ) + + if event_kind is _OpenAIRealtimeEventKind.RESPONSE_CREATED: state.is_responding = True response = getattr(event, "response", None) if response is not None: state.last_response_id = getattr(response, "id", None) return - if event_type in ("response.output_item.added", "response.output_item.created"): + if event_kind is _OpenAIRealtimeEventKind.OUTPUT_ITEM: item = getattr(event, "item", None) if item is not None: state.current_item_id = getattr(item, "id", None) return - if event_type in ("response.audio.delta", "response.output_audio.delta"): - delta = getattr(event, "delta", "") - if delta: - state.delivered_audio.extend(base64.b64decode(delta)) + if event_kind is _OpenAIRealtimeEventKind.AUDIO_DELTA: return - if event_type in ("response.audio_transcript.delta", "response.output_audio_transcript.delta"): - delta = getattr(event, "delta", "") - if delta: - state.delivered_transcripts.append(delta) + if event_kind is _OpenAIRealtimeEventKind.TRANSCRIPT_DELTA: return - if event_type == "response.done": + if event_kind is _OpenAIRealtimeEventKind.RESPONSE_DONE: response = getattr(event, "response", None) done_response_id = getattr(response, "id", None) if response is not None else None if state.last_response_id is not None and done_response_id != state.last_response_id: @@ -103,7 +108,7 @@ async def _route_event_async(self, *, event: Any, state: RealtimeTurnState | Non ) return - if event_type == "input_audio_buffer.speech_started" and state.is_responding: + if event_kind is _OpenAIRealtimeEventKind.SPEECH_STARTED and state.is_responding: await self._cancel_async(state=state) state.is_responding = False state.completion.set_result( @@ -115,7 +120,7 @@ async def _route_event_async(self, *, event: Any, state: RealtimeTurnState | Non ) return - if event_type == "error": + if event_kind is _OpenAIRealtimeEventKind.ERROR: error = getattr(event, "error", None) code = getattr(error, "code", None) if error is not None else None message = getattr(error, "message", "unknown") if error is not None else "unknown" diff --git a/pyrit/prompt_target/openai/_openai_realtime_event_router.py b/pyrit/prompt_target/openai/_openai_realtime_event_router.py new file mode 100644 index 0000000000..8deb710616 --- /dev/null +++ b/pyrit/prompt_target/openai/_openai_realtime_event_router.py @@ -0,0 +1,106 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Shared OpenAI Realtime event classification and response accumulation.""" + +import base64 +from enum import Enum, auto +from typing import Any, ClassVar + + +class _OpenAIRealtimeEventKind(Enum): + """Provider event categories shared by atomic and streaming receive policies.""" + + RESPONSE_DONE = auto() + ERROR = auto() + AUDIO_DELTA = auto() + AUDIO_DONE = auto() + TRANSCRIPT_DELTA = auto() + OUTPUT_TEXT_DONE = auto() + RESPONSE_CREATED = auto() + OUTPUT_ITEM = auto() + SPEECH_STARTED = auto() + INPUT_COMMITTED = auto() + LIFECYCLE = auto() + OTHER = auto() + + +class _OpenAIRealtimeEventRouter: + """Classify provider events and apply response deltas to caller-owned buffers.""" + + _KINDS_BY_EVENT_TYPE: ClassVar[dict[str, _OpenAIRealtimeEventKind]] = { + "response.done": _OpenAIRealtimeEventKind.RESPONSE_DONE, + "error": _OpenAIRealtimeEventKind.ERROR, + "response.audio.delta": _OpenAIRealtimeEventKind.AUDIO_DELTA, + "response.output_audio.delta": _OpenAIRealtimeEventKind.AUDIO_DELTA, + "response.audio.done": _OpenAIRealtimeEventKind.AUDIO_DONE, + "response.output_audio.done": _OpenAIRealtimeEventKind.AUDIO_DONE, + "response.audio_transcript.delta": _OpenAIRealtimeEventKind.TRANSCRIPT_DELTA, + "response.output_audio_transcript.delta": _OpenAIRealtimeEventKind.TRANSCRIPT_DELTA, + "response.output_text.done": _OpenAIRealtimeEventKind.OUTPUT_TEXT_DONE, + "response.created": _OpenAIRealtimeEventKind.RESPONSE_CREATED, + "response.output_item.added": _OpenAIRealtimeEventKind.OUTPUT_ITEM, + "response.output_item.created": _OpenAIRealtimeEventKind.OUTPUT_ITEM, + "input_audio_buffer.speech_started": _OpenAIRealtimeEventKind.SPEECH_STARTED, + "input_audio_buffer.committed": _OpenAIRealtimeEventKind.INPUT_COMMITTED, + } + _LIFECYCLE_EVENT_TYPES: ClassVar[frozenset[str]] = frozenset( + { + "session.created", + "session.updated", + "conversation.created", + "conversation.item.created", + "conversation.item.added", + "conversation.item.done", + "input_audio_buffer.speech_stopped", + "conversation.item.input_audio_transcription.completed", + "response.output_item.done", + "response.content_part.added", + "response.content_part.done", + "response.audio_transcript.done", + "response.output_audio_transcript.done", + "response.output_text.delta", + "rate_limits.updated", + } + ) + _LIFECYCLE_KINDS: ClassVar[frozenset[_OpenAIRealtimeEventKind]] = frozenset( + { + _OpenAIRealtimeEventKind.RESPONSE_CREATED, + _OpenAIRealtimeEventKind.OUTPUT_ITEM, + _OpenAIRealtimeEventKind.SPEECH_STARTED, + _OpenAIRealtimeEventKind.INPUT_COMMITTED, + _OpenAIRealtimeEventKind.LIFECYCLE, + } + ) + + @classmethod + def classify_event(cls, event_type: str) -> _OpenAIRealtimeEventKind: + """Return the normalized category for a provider event type.""" + event_kind = cls._KINDS_BY_EVENT_TYPE.get(event_type) + if event_kind is not None: + return event_kind + if event_type in cls._LIFECYCLE_EVENT_TYPES: + return _OpenAIRealtimeEventKind.LIFECYCLE + return _OpenAIRealtimeEventKind.OTHER + + @classmethod + def is_lifecycle_event(cls, event_kind: _OpenAIRealtimeEventKind) -> bool: + """Return whether atomic receiving should log the event as lifecycle-only.""" + return event_kind in cls._LIFECYCLE_KINDS + + @staticmethod + def collect_response_delta( + *, + event: Any, + event_kind: _OpenAIRealtimeEventKind, + audio_buffer: bytearray, + transcripts: list[str], + ) -> None: + """Apply an audio or transcript delta to caller-owned response buffers.""" + delta = getattr(event, "delta", "") + if not delta: + return + if event_kind is _OpenAIRealtimeEventKind.AUDIO_DELTA: + audio_buffer.extend(base64.b64decode(delta)) + elif event_kind is _OpenAIRealtimeEventKind.TRANSCRIPT_DELTA: + transcripts.append(delta) diff --git a/pyrit/prompt_target/openai/openai_realtime_target.py b/pyrit/prompt_target/openai/openai_realtime_target.py index 3894ddc623..2785720f4f 100644 --- a/pyrit/prompt_target/openai/openai_realtime_target.py +++ b/pyrit/prompt_target/openai/openai_realtime_target.py @@ -24,6 +24,10 @@ from pyrit.prompt_target.common.target_capabilities import TargetCapabilities from pyrit.prompt_target.common.target_configuration import TargetConfiguration from pyrit.prompt_target.common.utils import limit_requests_per_minute +from pyrit.prompt_target.openai._openai_realtime_event_router import ( + _OpenAIRealtimeEventKind, + _OpenAIRealtimeEventRouter, +) from pyrit.prompt_target.openai._openai_realtime_streaming_session import ( _OpenAIRealtimeStreamingSession, ) @@ -575,6 +579,7 @@ async def receive_events_async(self, conversation_id: str) -> RealtimeTargetResu connection = self._get_connection(conversation_id=conversation_id) result = RealtimeTargetResult() + audio_buffer = bytearray() audio_done_received = False current_turn_event_count = 0 grace_period_sec = 1.0 # Wait 1 second after audio.done before soft-finishing @@ -595,7 +600,7 @@ async def receive_events_async(self, conversation_id: str) -> RealtimeTargetResu if audio_done_received: logger.warning( f"Soft-finishing: No response.done {grace_period_sec}s after audio.done. " - f"Audio bytes: {len(result.audio_bytes)}" + f"Audio bytes: {len(audio_buffer)}" ) break # Should not happen if timeout is None, but re-raise if it does @@ -606,22 +611,30 @@ async def receive_events_async(self, conversation_id: str) -> RealtimeTargetResu break except Exception as conn_err: # Handle websockets connection errors as soft-finish if we have audio - if "ConnectionClosed" in str(type(conn_err).__name__) and result.audio_bytes: + if "ConnectionClosed" in str(type(conn_err).__name__) and audio_buffer: logger.warning( f"Connection closed without response.done (likely API issue). " - f"Audio bytes received: {len(result.audio_bytes)}. Soft-finishing." + f"Audio bytes received: {len(audio_buffer)}. Soft-finishing." ) break # Re-raise if not a connection close or no audio received raise event_type = event.type + event_kind = _OpenAIRealtimeEventRouter.classify_event(event_type) current_turn_event_count += 1 logger.debug(f"Processing event type: {event_type}") - - if event_type == "response.done": + audio_size_before = len(audio_buffer) + _OpenAIRealtimeEventRouter.collect_response_delta( + event=event, + event_kind=event_kind, + audio_buffer=audio_buffer, + transcripts=result.transcripts, + ) + + if event_kind is _OpenAIRealtimeEventKind.RESPONSE_DONE: self._handle_response_done_event(event=event, result=result) - if result.audio_bytes or current_turn_event_count > 1: + if audio_buffer or current_turn_event_count > 1: # Legitimate response.done: either we have audio, or other events # (e.g. response.created) preceded it, confirming it belongs to this turn. logger.debug("Received response.done - finishing normally") @@ -635,53 +648,27 @@ async def receive_events_async(self, conversation_id: str) -> RealtimeTargetResu "likely a stale event from a prior turn's soft-finish. Skipping." ) - elif event_type == "error": + elif event_kind is _OpenAIRealtimeEventKind.ERROR: error_message = event.error.message if hasattr(event.error, "message") else str(event.error) error_type = event.error.type if hasattr(event.error, "type") else "unknown" logger.error(f"Received 'error' event: [{error_type}] {error_message}") raise RuntimeError(f"Server error: [{error_type}] {error_message}") - elif event_type in ["response.audio.delta", "response.output_audio.delta"]: - audio_data = base64.b64decode(event.delta) - result.audio_bytes += audio_data - logger.debug(f"Decoded {len(audio_data)} bytes of audio data") + elif event_kind is _OpenAIRealtimeEventKind.AUDIO_DELTA: + logger.debug(f"Decoded {len(audio_buffer) - audio_size_before} bytes of audio data") - elif event_type in ["response.audio.done", "response.output_audio.done"]: + elif event_kind is _OpenAIRealtimeEventKind.AUDIO_DONE: logger.debug(f"Received audio.done - will soft-finish in {grace_period_sec}s if no response.done") audio_done_received = True - elif event_type in ["response.audio_transcript.delta", "response.output_audio_transcript.delta"]: - # Capture transcript deltas as they arrive (needed when response.done never comes) - if hasattr(event, "delta") and event.delta: - result.transcripts.append(event.delta) + elif event_kind is _OpenAIRealtimeEventKind.TRANSCRIPT_DELTA: + if getattr(event, "delta", ""): logger.debug(f"Captured transcript delta: {event.delta[:50]}...") - elif event_type in ["response.output_text.done"]: + elif event_kind is _OpenAIRealtimeEventKind.OUTPUT_TEXT_DONE: logger.debug("Received text.done") - # Handle lifecycle events that we can safely log - elif event_type in [ - "session.created", - "session.updated", - "conversation.created", - "conversation.item.created", - "conversation.item.added", - "conversation.item.done", - "input_audio_buffer.committed", - "input_audio_buffer.speech_started", - "input_audio_buffer.speech_stopped", - "conversation.item.input_audio_transcription.completed", - "response.created", - "response.output_item.added", - "response.output_item.created", - "response.output_item.done", - "response.content_part.added", - "response.content_part.done", - "response.audio_transcript.done", - "response.output_audio_transcript.done", - "response.output_text.delta", - "rate_limits.updated", - ]: + elif _OpenAIRealtimeEventRouter.is_lifecycle_event(event_kind): logger.debug(f"Lifecycle event '{event_type}'") else: @@ -691,6 +678,7 @@ async def receive_events_async(self, conversation_id: str) -> RealtimeTargetResu logger.error(f"An unexpected error occurred for conversation {conversation_id}: {e}") raise + result.audio_bytes = bytes(audio_buffer) logger.debug( f"Completed receive_events with {len(result.transcripts)} transcripts " f"and {len(result.audio_bytes)} bytes of audio" diff --git a/tests/unit/prompt_target/target/test_realtime_target.py b/tests/unit/prompt_target/target/test_realtime_target.py index 6a4959e76d..b25067823d 100644 --- a/tests/unit/prompt_target/target/test_realtime_target.py +++ b/tests/unit/prompt_target/target/test_realtime_target.py @@ -20,6 +20,10 @@ from pyrit.prompt_target.openai._openai_realtime_dispatcher import ( _OpenAIRealtimeDispatcher, ) +from pyrit.prompt_target.openai._openai_realtime_event_router import ( + _OpenAIRealtimeEventKind, + _OpenAIRealtimeEventRouter, +) # Env vars that may leak from .env files loaded by other tests in parallel workers. _CLEAN_UNDERLYING_MODEL_ENV = { @@ -417,6 +421,26 @@ async def test_receive_events_with_audio_and_transcript(target): assert result.transcripts[1] == "this is a test transcript." +async def test_receive_events_soft_finishes_after_audio_done(target): + """Atomic receiving returns accumulated deltas when audio.done is followed by its grace-period timeout.""" + mock_connection = AsyncMock() + conversation_id = "test_soft_finish" + target._existing_conversation[conversation_id] = mock_connection + + async def _events(): + yield _scripted_event("response.output_audio.delta", delta=base64.b64encode(b"audio").decode("ascii")) + yield _scripted_event("response.output_audio_transcript.delta", delta="partial") + yield _scripted_event("response.output_audio.done") + raise asyncio.TimeoutError + + mock_connection.__aiter__.side_effect = _events + + result = await target.receive_events_async(conversation_id) + + assert result.audio_bytes == b"audio" + assert result.transcripts == ["partial"] + + async def test_multi_turn_reuses_connection(target): """Test that multiple turns in the same conversation reuse the same connection. @@ -582,6 +606,42 @@ def _make_dispatcher(connection): return _OpenAIRealtimeDispatcher(connection=connection) +@pytest.mark.parametrize( + ("event_type", "expected_kind"), + [ + ("response.audio.delta", _OpenAIRealtimeEventKind.AUDIO_DELTA), + ("response.output_audio.delta", _OpenAIRealtimeEventKind.AUDIO_DELTA), + ("response.audio.done", _OpenAIRealtimeEventKind.AUDIO_DONE), + ("response.output_audio.done", _OpenAIRealtimeEventKind.AUDIO_DONE), + ("response.audio_transcript.delta", _OpenAIRealtimeEventKind.TRANSCRIPT_DELTA), + ("response.output_audio_transcript.delta", _OpenAIRealtimeEventKind.TRANSCRIPT_DELTA), + ], +) +def test_realtime_event_router_normalizes_response_aliases(event_type, expected_kind): + assert _OpenAIRealtimeEventRouter.classify_event(event_type) is expected_kind + + +def test_realtime_event_router_collects_audio_and_transcript_deltas(): + audio_buffer = bytearray() + transcripts: list[str] = [] + + _OpenAIRealtimeEventRouter.collect_response_delta( + event=_scripted_event("response.output_audio.delta", delta=base64.b64encode(b"audio").decode("ascii")), + event_kind=_OpenAIRealtimeEventKind.AUDIO_DELTA, + audio_buffer=audio_buffer, + transcripts=transcripts, + ) + _OpenAIRealtimeEventRouter.collect_response_delta( + event=_scripted_event("response.output_audio_transcript.delta", delta="hello"), + event_kind=_OpenAIRealtimeEventKind.TRANSCRIPT_DELTA, + audio_buffer=audio_buffer, + transcripts=transcripts, + ) + + assert bytes(audio_buffer) == b"audio" + assert transcripts == ["hello"] + + async def test_cancel_does_not_send_response_cancel(): """_cancel_async must NOT send response.cancel (server auto-cancels on speech detection).""" connection = AsyncMock() @@ -685,6 +745,16 @@ async def test_route_event_happy_path_resolves_completion_with_assembled_result( assert state.interrupted is False +async def test_route_event_audio_done_does_not_complete_streaming_turn(): + """Streaming waits for response.done or barge-in instead of using the atomic soft-finish policy.""" + dispatcher = _make_dispatcher(AsyncMock()) + state = RealtimeTurnState(completion=asyncio.get_event_loop().create_future()) + + await dispatcher._route_event_async(event=_scripted_event("response.output_audio.done"), state=state) + + assert not state.completion.done() + + async def test_route_event_speech_started_while_responding_cancels_and_resolves_interrupted(): """speech_started during a response triggers cancel and resolves with interrupted=True.""" connection = AsyncMock() From 0b014aabf36709b5661e4e1ff8cd4ea6656fc37c Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Mon, 3 Aug 2026 12:35:58 -0700 Subject: [PATCH 2/2] TEST: Cover realtime routing policies Pin shared event classification, provider aliases, atomic termination fallbacks, and streaming state guards with focused regression cases. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: be81055b-39ed-494c-94ca-c7656200a344 --- .../target/test_realtime_target.py | 245 +++++++++++++++++- 1 file changed, 244 insertions(+), 1 deletion(-) diff --git a/tests/unit/prompt_target/target/test_realtime_target.py b/tests/unit/prompt_target/target/test_realtime_target.py index b25067823d..f7d3b86852 100644 --- a/tests/unit/prompt_target/target/test_realtime_target.py +++ b/tests/unit/prompt_target/target/test_realtime_target.py @@ -441,6 +441,82 @@ async def _events(): assert result.transcripts == ["partial"] +async def test_receive_events_connection_close_soft_finishes_with_audio(target): + """Atomic receiving returns accumulated audio when the provider closes before response.done.""" + + class ConnectionClosedTestError(Exception): + pass + + mock_connection = AsyncMock() + conversation_id = "test_connection_close_with_audio" + target._existing_conversation[conversation_id] = mock_connection + + async def _events(): + yield _scripted_event("response.audio.delta", delta=base64.b64encode(b"partial").decode("ascii")) + raise ConnectionClosedTestError("closed") + + mock_connection.__aiter__.side_effect = _events + + result = await target.receive_events_async(conversation_id) + + assert result.audio_bytes == b"partial" + + +async def test_receive_events_connection_close_without_audio_raises(target): + """Atomic receiving must not hide a connection failure before any response audio arrives.""" + + class ConnectionClosedTestError(Exception): + pass + + mock_connection = AsyncMock() + conversation_id = "test_connection_close_without_audio" + target._existing_conversation[conversation_id] = mock_connection + + async def _events(): + raise ConnectionClosedTestError("closed") + yield # pragma: no cover + + mock_connection.__aiter__.side_effect = _events + + with pytest.raises(ConnectionClosedTestError, match="closed"): + await target.receive_events_async(conversation_id) + + +async def test_receive_events_timeout_before_audio_done_raises(target): + """Atomic receiving only treats a timeout as completion after an audio.done event.""" + mock_connection = AsyncMock() + conversation_id = "test_timeout_without_audio_done" + target._existing_conversation[conversation_id] = mock_connection + + async def _events(): + raise asyncio.TimeoutError + yield # pragma: no cover + + mock_connection.__aiter__.side_effect = _events + + with pytest.raises(asyncio.TimeoutError): + await target.receive_events_async(conversation_id) + + +async def test_receive_events_ignores_non_response_and_empty_delta_events(target): + """Lifecycle, unknown, text-done, and empty transcript events do not mutate an atomic result.""" + mock_connection = AsyncMock() + conversation_id = "test_ignored_events" + target._existing_conversation[conversation_id] = mock_connection + + mock_connection.__aiter__.return_value = [ + _scripted_event("response.audio_transcript.delta", delta=""), + _scripted_event("response.output_text.done"), + _scripted_event("provider.new_event"), + _scripted_event("response.done", **{"response.status": "success"}), + ] + + result = await target.receive_events_async(conversation_id) + + assert result.audio_bytes == b"" + assert result.transcripts == [] + + async def test_multi_turn_reuses_connection(target): """Test that multiple turns in the same conversation reuse the same connection. @@ -609,18 +685,44 @@ def _make_dispatcher(connection): @pytest.mark.parametrize( ("event_type", "expected_kind"), [ + ("response.done", _OpenAIRealtimeEventKind.RESPONSE_DONE), + ("error", _OpenAIRealtimeEventKind.ERROR), ("response.audio.delta", _OpenAIRealtimeEventKind.AUDIO_DELTA), ("response.output_audio.delta", _OpenAIRealtimeEventKind.AUDIO_DELTA), ("response.audio.done", _OpenAIRealtimeEventKind.AUDIO_DONE), ("response.output_audio.done", _OpenAIRealtimeEventKind.AUDIO_DONE), ("response.audio_transcript.delta", _OpenAIRealtimeEventKind.TRANSCRIPT_DELTA), ("response.output_audio_transcript.delta", _OpenAIRealtimeEventKind.TRANSCRIPT_DELTA), + ("response.output_text.done", _OpenAIRealtimeEventKind.OUTPUT_TEXT_DONE), + ("response.created", _OpenAIRealtimeEventKind.RESPONSE_CREATED), + ("response.output_item.added", _OpenAIRealtimeEventKind.OUTPUT_ITEM), + ("response.output_item.created", _OpenAIRealtimeEventKind.OUTPUT_ITEM), + ("input_audio_buffer.speech_started", _OpenAIRealtimeEventKind.SPEECH_STARTED), + ("input_audio_buffer.committed", _OpenAIRealtimeEventKind.INPUT_COMMITTED), + ("session.updated", _OpenAIRealtimeEventKind.LIFECYCLE), + ("provider.new_event", _OpenAIRealtimeEventKind.OTHER), ], ) -def test_realtime_event_router_normalizes_response_aliases(event_type, expected_kind): +def test_realtime_event_router_classifies_provider_events(event_type, expected_kind): assert _OpenAIRealtimeEventRouter.classify_event(event_type) is expected_kind +@pytest.mark.parametrize( + ("event_kind", "expected"), + [ + (_OpenAIRealtimeEventKind.RESPONSE_CREATED, True), + (_OpenAIRealtimeEventKind.OUTPUT_ITEM, True), + (_OpenAIRealtimeEventKind.SPEECH_STARTED, True), + (_OpenAIRealtimeEventKind.INPUT_COMMITTED, True), + (_OpenAIRealtimeEventKind.LIFECYCLE, True), + (_OpenAIRealtimeEventKind.RESPONSE_DONE, False), + (_OpenAIRealtimeEventKind.OTHER, False), + ], +) +def test_realtime_event_router_identifies_atomic_lifecycle_events(event_kind, expected): + assert _OpenAIRealtimeEventRouter.is_lifecycle_event(event_kind) is expected + + def test_realtime_event_router_collects_audio_and_transcript_deltas(): audio_buffer = bytearray() transcripts: list[str] = [] @@ -642,6 +744,29 @@ def test_realtime_event_router_collects_audio_and_transcript_deltas(): assert transcripts == ["hello"] +@pytest.mark.parametrize( + ("event_kind", "delta"), + [ + (_OpenAIRealtimeEventKind.AUDIO_DELTA, ""), + (_OpenAIRealtimeEventKind.TRANSCRIPT_DELTA, ""), + (_OpenAIRealtimeEventKind.OTHER, base64.b64encode(b"ignored").decode("ascii")), + ], +) +def test_realtime_event_router_ignores_empty_or_unrelated_deltas(event_kind, delta): + audio_buffer = bytearray(b"existing") + transcripts = ["existing"] + + _OpenAIRealtimeEventRouter.collect_response_delta( + event=_scripted_event("test", delta=delta), + event_kind=event_kind, + audio_buffer=audio_buffer, + transcripts=transcripts, + ) + + assert bytes(audio_buffer) == b"existing" + assert transcripts == ["existing"] + + async def test_cancel_does_not_send_response_cancel(): """_cancel_async must NOT send response.cancel (server auto-cancels on speech detection).""" connection = AsyncMock() @@ -702,6 +827,18 @@ async def test_cancel_marks_interrupted_when_truncate_raises(caplog): ) +async def test_cancel_without_current_item_only_marks_interrupted(): + """A turn interrupted before an output item exists cannot be truncated but is still marked.""" + connection = AsyncMock() + dispatcher = _make_dispatcher(connection) + state = _turn_state(item_id=None) + + await dispatcher._cancel_async(state=state) + + connection.conversation.item.truncate.assert_not_awaited() + assert state.interrupted is True + + def _scripted_event(event_type, **fields): """Build a MagicMock event with the named type plus any extra attribute paths.""" event = MagicMock() @@ -745,6 +882,31 @@ async def test_route_event_happy_path_resolves_completion_with_assembled_result( assert state.interrupted is False +@pytest.mark.parametrize( + ("audio_event_type", "transcript_event_type"), + [ + ("response.audio.delta", "response.audio_transcript.delta"), + ("response.output_audio.delta", "response.output_audio_transcript.delta"), + ], +) +async def test_route_event_accumulates_response_aliases(audio_event_type, transcript_event_type): + dispatcher = _make_dispatcher(AsyncMock()) + state = RealtimeTurnState(completion=asyncio.get_event_loop().create_future()) + + await dispatcher._route_event_async( + event=_scripted_event(audio_event_type, delta=base64.b64encode(b"audio").decode("ascii")), + state=state, + ) + await dispatcher._route_event_async( + event=_scripted_event(transcript_event_type, delta="transcript"), + state=state, + ) + + assert bytes(state.delivered_audio) == b"audio" + assert state.delivered_transcripts == ["transcript"] + assert not state.completion.done() + + async def test_route_event_audio_done_does_not_complete_streaming_turn(): """Streaming waits for response.done or barge-in instead of using the atomic soft-finish policy.""" dispatcher = _make_dispatcher(AsyncMock()) @@ -755,6 +917,52 @@ async def test_route_event_audio_done_does_not_complete_streaming_turn(): assert not state.completion.done() +async def test_route_event_missing_optional_payloads_leave_ids_unset(): + """Missing speech timing, response, and output item payloads do not synthesize state identifiers.""" + dispatcher = _make_dispatcher(AsyncMock()) + state = RealtimeTurnState(completion=asyncio.get_event_loop().create_future()) + + await dispatcher._route_event_async( + event=_scripted_event("input_audio_buffer.speech_started", audio_start_ms=None), + state=state, + ) + await dispatcher._route_event_async( + event=_scripted_event("response.created", response=None), + state=state, + ) + await dispatcher._route_event_async( + event=_scripted_event("response.output_item.added", item=None), + state=state, + ) + + assert dispatcher._pending_speech_start_ms is None + assert state.is_responding is True + assert state.last_response_id is None + assert state.current_item_id is None + + +async def test_route_event_drops_output_without_active_turn(): + dispatcher = _make_dispatcher(AsyncMock()) + + await dispatcher._route_event_async( + event=_scripted_event("response.audio.delta", delta=base64.b64encode(b"ignored").decode("ascii")), + state=None, + ) + + +async def test_route_event_drops_output_for_completed_turn(): + dispatcher = _make_dispatcher(AsyncMock()) + state = RealtimeTurnState(completion=asyncio.get_event_loop().create_future()) + state.completion.set_result(RealtimeTargetResult()) + + await dispatcher._route_event_async( + event=_scripted_event("response.audio.delta", delta=base64.b64encode(b"ignored").decode("ascii")), + state=state, + ) + + assert state.delivered_audio == bytearray() + + async def test_route_event_speech_started_while_responding_cancels_and_resolves_interrupted(): """speech_started during a response triggers cancel and resolves with interrupted=True.""" connection = AsyncMock() @@ -796,6 +1004,24 @@ async def test_route_event_stale_response_done_after_cancel_is_dropped(): await dispatcher._route_event_async(event=_scripted_event("response.done", **{"response.id": "r1"}), state=state) +async def test_route_event_stale_response_done_does_not_resolve_active_turn(): + """An active turn ignores response.done from a different response id.""" + dispatcher = _make_dispatcher(AsyncMock()) + state = RealtimeTurnState( + completion=asyncio.get_event_loop().create_future(), + is_responding=True, + last_response_id="current", + ) + + await dispatcher._route_event_async( + event=_scripted_event("response.done", **{"response.id": "stale"}), + state=state, + ) + + assert not state.completion.done() + assert state.is_responding is True + + async def test_route_event_error_resolves_with_exception(): """error events resolve the completion future via set_exception.""" connection = AsyncMock() @@ -878,6 +1104,23 @@ async def test_route_event_committed_event_without_callback_is_noop(): ) +async def test_route_event_committed_without_item_id_is_ignored(): + received: list[CommittedEvent] = [] + + async def on_committed(event: CommittedEvent) -> None: + received.append(event) + + dispatcher = _OpenAIRealtimeDispatcher(connection=AsyncMock(), on_user_audio_committed=on_committed) + + await dispatcher._route_event_async( + event=_scripted_event("input_audio_buffer.committed", item_id=None), + state=None, + ) + await asyncio.sleep(0) + + assert received == [] + + async def test_route_event_speech_started_audio_start_propagates_to_commit(): """speech_started's audio_start_ms is captured and attached to the next CommittedEvent.