From 2a4b61c46a9dd0cb02607b487d5587d0db143e31 Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Tue, 27 Oct 2026 14:58:46 +0000 Subject: [PATCH 01/19] fix(examples): handle interruptions and disconnects in push-to-talk --- examples/realtime/push_to_talk_app.py | 58 +++++++++++++++++++-------- 1 file changed, 41 insertions(+), 17 deletions(-) diff --git a/examples/realtime/push_to_talk_app.py b/examples/realtime/push_to_talk_app.py index fd0f2efaa0..a5fd1a173a 100755 --- a/examples/realtime/push_to_talk_app.py +++ b/examples/realtime/push_to_talk_app.py @@ -36,7 +36,7 @@ from textual import events from audio_util import CHANNELS, SAMPLE_RATE, AudioPlayerAsync from textual.app import App, ComposeResult -from textual.widgets import Button, Static, RichLog +from textual.widgets import Static, RichLog from textual.reactive import reactive from textual.containers import Container @@ -93,11 +93,6 @@ class RealtimeApp(App[None]): height: 3; /* Explicit height for input */ } - Button { - width: 20%; - height: 3; /* Explicit height for button */ - } - #bottom-pane { width: 100%; height: 82%; /* Reduced to make room for session display */ @@ -141,6 +136,8 @@ def __init__(self) -> None: self.client = AsyncOpenAI() self.audio_player = AudioPlayerAsync() self.last_audio_item_id = None + self._active_response_id: str | None = None + self._interrupted_response_ids: set[str] = set() self.should_send_audio = asyncio.Event() self.connected = asyncio.Event() self._mic_stream: Any = None @@ -151,12 +148,16 @@ def compose(self) -> ComposeResult: with Container(): yield SessionDisplay(id="session-display") yield AudioStatusIndicator(id="status-indicator") - yield RichLog(id="bottom-pane", wrap=True, highlight=True, markup=True) + yield RichLog(id="bottom-pane", wrap=True, highlight=True) async def on_mount(self) -> None: self.run_worker(self.handle_realtime_connection()) self.run_worker(self.send_mic_audio()) + async def on_unmount(self) -> None: + self.audio_player.terminate() + await self.client.close() + async def handle_realtime_connection(self) -> None: async with self.client.realtime.connect(model="gpt-realtime") as conn: self.connection = conn @@ -190,7 +191,30 @@ async def handle_realtime_connection(self) -> None: self.session = event.session continue + if event.type == "error": + self.query_one(RichLog).write(f"Realtime error: {event.error.message}") + continue + + if event.type == "response.created": + self._active_response_id = event.response.id + continue + + if event.type == "response.done": + if self._active_response_id == event.response.id: + self._active_response_id = None + if event.response.id is not None: + self._interrupted_response_ids.discard(event.response.id) + if event.response.status in {"failed", "incomplete"}: + details = event.response.status_details + message = details.error.message if details and details.error else None + reason = message or (details.reason if details else None) + text = f"Response {event.response.status}" + self.query_one(RichLog).write(f"{text}: {reason}" if reason else text) + continue + if event.type == "response.output_audio.delta": + if event.response_id in self._interrupted_response_ids: + continue if event.item_id != self.last_audio_item_id: self.audio_player.reset_frame_count() self.last_audio_item_id = event.item_id @@ -213,6 +237,8 @@ async def handle_realtime_connection(self) -> None: bottom_pane.write(acc_items[event.item_id]) continue + self.exit(return_code=1, message="The Realtime connection closed.") + async def _get_connection(self) -> AsyncRealtimeConnection: await self.connected.wait() assert self.connection is not None @@ -221,8 +247,6 @@ async def _get_connection(self) -> AsyncRealtimeConnection: async def send_mic_audio(self) -> None: import sounddevice as sd # type: ignore - sent_audio = False - device_info = sd.query_devices() print(device_info) @@ -252,10 +276,6 @@ async def send_mic_audio(self) -> None: data, _ = stream.read(read_size) - if not sent_audio: - asyncio.create_task(connection.send({"type": "response.cancel"})) - sent_audio = True - await connection.input_audio_buffer.append(audio=base64.b64encode(cast(Any, data)).decode("utf-8")) await asyncio.sleep(0) @@ -274,10 +294,6 @@ def _stop_mic_stream(self) -> None: async def on_key(self, event: events.Key) -> None: """Handle key press events.""" - if event.key == "enter": - self.query_one(Button).press() - return - if event.key == "q": self.exit() return @@ -304,6 +320,13 @@ async def on_key(self, event: events.Key) -> None: await conn.input_audio_buffer.commit() await conn.response.create() else: + if self.audio_player.playing: + self.audio_player.stop() + if self.connection is not None and self._active_response_id is not None: + response_id = self._active_response_id + self._active_response_id = None + self._interrupted_response_ids.add(response_id) + await self.connection.response.cancel(response_id=response_id) self.should_send_audio.set() status_indicator.is_recording = True @@ -311,3 +334,4 @@ async def on_key(self, event: events.Key) -> None: if __name__ == "__main__": app = RealtimeApp() app.run() + raise SystemExit(app.return_code) From 00609af408ab75ccffc08e52a95a2cdf0576d3cb Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 03:04:28 +0000 Subject: [PATCH 02/19] Synchronize interrupted Realtime playback and capture promptly --- .github/workflows/ci.yml | 8 + examples/realtime/audio_util.py | 44 ++++- examples/realtime/push_to_talk_app.py | 76 +++++--- examples/realtime/test_push_to_talk_app.py | 195 +++++++++++++++++++++ 4 files changed, 290 insertions(+), 33 deletions(-) create mode 100644 examples/realtime/test_push_to_talk_app.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 76fc0cb8c4..4ce71cb064 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -279,6 +279,14 @@ jobs: OPENAI_TEST_HTTP_CLIENT: httpx2 run: ./scripts/test + - name: Test Realtime push-to-talk sample + run: >- + uv run --locked --all-extras + --with 'textual==8.2.8' + --with 'pydub==0.25.1' + --with 'audioop-lts==0.2.2; python_version >= "3.13"' + python -m pytest -o addopts= --noconftest -q examples/realtime/test_push_to_talk_app.py + examples: timeout-minutes: 10 name: examples diff --git a/examples/realtime/audio_util.py b/examples/realtime/audio_util.py index 990093505e..0c7fce3801 100644 --- a/examples/realtime/audio_util.py +++ b/examples/realtime/audio_util.py @@ -43,6 +43,9 @@ def __init__(self): ) self.playing = False self._frame_count = 0 + self._played_by_item: dict[tuple[str, str, int], int] = {} + self._last_played_item: tuple[str, str, int] | None = None + self._finished_responses: set[str] = set() def callback(self, outdata, frames, time, status): # noqa with self.lock: @@ -50,13 +53,17 @@ def callback(self, outdata, frames, time, status): # noqa # get next item from queue if there is still space in the buffer while len(data) < frames and len(self.queue) > 0: - item = self.queue.pop(0) + item, item_key = self.queue.pop(0) frames_needed = frames - len(data) - data = np.concatenate((data, item[:frames_needed])) + played = item[:frames_needed] + data = np.concatenate((data, played)) + self._played_by_item[item_key] = self._played_by_item.get(item_key, 0) + len(played) + self._last_played_item = item_key if len(item) > frames_needed: - self.queue.insert(0, item[frames_needed:]) + self.queue.insert(0, (item[frames_needed:], item_key)) self._frame_count += len(data) + self._discard_finished_counts() # fill the rest of the frames with zeros if there is no more data if len(data) < frames: @@ -64,17 +71,26 @@ def callback(self, outdata, frames, time, status): # noqa outdata[:] = data.reshape(-1, 1) - def reset_frame_count(self): - self._frame_count = 0 - def get_frame_count(self): return self._frame_count - def add_data(self, data: bytes): + def finish_response(self, response_id: str): + with self.lock: + self._finished_responses.add(response_id) + self._discard_finished_counts() + + def _discard_finished_counts(self): + pending = {key for _, key in self.queue} + for key in list(self._played_by_item): + if key[0] in self._finished_responses and key not in pending: + del self._played_by_item[key] + self._finished_responses.intersection_update(key[0] for key in pending) + + def add_data(self, data: bytes, *, response_id: str, item_id: str, content_index: int): with self.lock: # bytes is pcm16 single channel audio data, convert to numpy array np_data = np.frombuffer(data, dtype=np.int16) - self.queue.append(np_data) + self.queue.append((np_data, (response_id, item_id, content_index))) if not self.playing: self.start() @@ -82,11 +98,21 @@ def start(self): self.playing = True self.stream.start() - def stop(self): + def stop(self, *, interrupted_response_id: str | None = None): self.playing = False self.stream.stop() with self.lock: + # Stop the callback first so these counts cannot advance during interruption. + pending = dict.fromkeys(key for _, key in self.queue) + if self._last_played_item is not None and self._last_played_item[0] == interrupted_response_id: + pending[self._last_played_item] = None + unheard = [(key, self._played_by_item.get(key, 0)) for key in pending] self.queue = [] + self._played_by_item.clear() + self._finished_responses.clear() + self._last_played_item = None + self._frame_count = 0 + return unheard def terminate(self): self.stream.close() diff --git a/examples/realtime/push_to_talk_app.py b/examples/realtime/push_to_talk_app.py index a5fd1a173a..c226433796 100755 --- a/examples/realtime/push_to_talk_app.py +++ b/examples/realtime/push_to_talk_app.py @@ -39,8 +39,9 @@ from textual.widgets import Static, RichLog from textual.reactive import reactive from textual.containers import Container +from websockets.exceptions import ConnectionClosed -from openai import AsyncOpenAI +from openai import AsyncOpenAI, WebSocketConnectionClosedError from openai.resources.realtime.realtime import AsyncRealtimeConnection from openai.types.realtime.session_created_event import Session @@ -124,7 +125,6 @@ class RealtimeApp(App[None]): client: AsyncOpenAI should_send_audio: asyncio.Event audio_player: AudioPlayerAsync - last_audio_item_id: str | None connection: AsyncRealtimeConnection | None session: Session | None connected: asyncio.Event @@ -135,10 +135,10 @@ def __init__(self) -> None: self.session = None self.client = AsyncOpenAI() self.audio_player = AudioPlayerAsync() - self.last_audio_item_id = None self._active_response_id: str | None = None self._interrupted_response_ids: set[str] = set() self.should_send_audio = asyncio.Event() + self._mic_send_lock = asyncio.Lock() self.connected = asyncio.Event() self._mic_stream: Any = None @@ -159,6 +159,13 @@ async def on_unmount(self) -> None: await self.client.close() async def handle_realtime_connection(self) -> None: + try: + await self._receive_realtime_events() + except (ConnectionClosed, WebSocketConnectionClosedError): + pass + self.exit(return_code=1, message="The Realtime connection closed.") + + async def _receive_realtime_events(self) -> None: async with self.client.realtime.connect(model="gpt-realtime") as conn: self.connection = conn self.connected.set() @@ -196,6 +203,8 @@ async def handle_realtime_connection(self) -> None: continue if event.type == "response.created": + if self._active_response_id is not None and self._active_response_id != event.response.id: + self.audio_player.finish_response(self._active_response_id) self._active_response_id = event.response.id continue @@ -204,6 +213,7 @@ async def handle_realtime_connection(self) -> None: self._active_response_id = None if event.response.id is not None: self._interrupted_response_ids.discard(event.response.id) + self.audio_player.finish_response(event.response.id) if event.response.status in {"failed", "incomplete"}: details = event.response.status_details message = details.error.message if details and details.error else None @@ -215,12 +225,13 @@ async def handle_realtime_connection(self) -> None: if event.type == "response.output_audio.delta": if event.response_id in self._interrupted_response_ids: continue - if event.item_id != self.last_audio_item_id: - self.audio_player.reset_frame_count() - self.last_audio_item_id = event.item_id - bytes_data = base64.b64decode(event.delta) - self.audio_player.add_data(bytes_data) + self.audio_player.add_data( + bytes_data, + response_id=event.response_id, + item_id=event.item_id, + content_index=event.content_index, + ) continue if event.type == "response.output_audio_transcript.delta": @@ -237,8 +248,6 @@ async def handle_realtime_connection(self) -> None: bottom_pane.write(acc_items[event.item_id]) continue - self.exit(return_code=1, message="The Realtime connection closed.") - async def _get_connection(self) -> AsyncRealtimeConnection: await self.connected.wait() assert self.connection is not None @@ -274,9 +283,11 @@ async def send_mic_audio(self) -> None: await asyncio.sleep(0) continue - data, _ = stream.read(read_size) - - await connection.input_audio_buffer.append(audio=base64.b64encode(cast(Any, data)).decode("utf-8")) + async with self._mic_send_lock: + if stream is not self._mic_stream or not self.should_send_audio.is_set(): + continue + data, _ = stream.read(read_size) + await connection.input_audio_buffer.append(audio=base64.b64encode(cast(Any, data)).decode("utf-8")) await asyncio.sleep(0) except KeyboardInterrupt: @@ -317,18 +328,35 @@ async def on_key(self, event: events.Key) -> None: # However if we're in manual `turn_detection` mode then we need to # manually tell the model to commit the audio buffer and start responding. conn = await self._get_connection() - await conn.input_audio_buffer.commit() - await conn.response.create() + async with self._mic_send_lock: + await conn.input_audio_buffer.commit() + await conn.response.create() else: - if self.audio_player.playing: - self.audio_player.stop() - if self.connection is not None and self._active_response_id is not None: - response_id = self._active_response_id - self._active_response_id = None - self._interrupted_response_ids.add(response_id) - await self.connection.response.cancel(response_id=response_id) - self.should_send_audio.set() - status_indicator.is_recording = True + async with self._mic_send_lock: + self.should_send_audio.set() + status_indicator.is_recording = True + unheard = ( + self.audio_player.stop(interrupted_response_id=self._active_response_id) + if self.audio_player.playing + else [] + ) + if self.connection is not None: + if self._active_response_id is not None: + response_id = self._active_response_id + self._active_response_id = None + self._interrupted_response_ids.add(response_id) + await self.connection.response.cancel(response_id=response_id) + for (_, item_id, content_index), frames in unheard: + if content_index != 0: + self.query_one(RichLog).write( + "Cannot truncate audio: the API only supports content part 0." + ) + continue + await self.connection.conversation.item.truncate( + item_id=item_id, + content_index=content_index, + audio_end_ms=frames * 1000 // SAMPLE_RATE, + ) if __name__ == "__main__": diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py new file mode 100644 index 0000000000..e70805d662 --- /dev/null +++ b/examples/realtime/test_push_to_talk_app.py @@ -0,0 +1,195 @@ +"""Run with pytest after installing the push_to_talk_app.py sample dependencies.""" + +from __future__ import annotations + +import sys +import asyncio +from types import ModuleType, SimpleNamespace +from typing import Any +from pathlib import Path + +import pytest + +np = pytest.importorskip("numpy") +pytest.importorskip("textual") +pytest.importorskip("pydub") + + +@pytest.fixture() +def app_module(monkeypatch: pytest.MonkeyPatch) -> Any: + started = asyncio.Event() + + class InputStream: + def __init__(self, **_kwargs: Any): + self.open = True + self.samples = 480 + + def start(self) -> None: + started.set() + + @property + def read_available(self) -> int: + return self.samples + + def read(self, frames: int) -> tuple[Any, bool]: + assert self.open + self.samples -= frames + return np.zeros((frames, 1), dtype=np.int16), False + + def abort(self) -> None: + self.open = False + + def close(self) -> None: + self.open = False + + class OutputStream: + def __init__(self, callback: Any, **_kwargs: Any): + self.callback = callback + + def start(self) -> None: + pass + + def stop(self) -> None: + pass + + def close(self) -> None: + pass + + # Replace only audio device bindings; import the actual UI, player, and SDK. + device = ModuleType("sounddevice") + monkeypatch.setattr(device, "InputStream", InputStream, raising=False) + monkeypatch.setattr(device, "OutputStream", OutputStream, raising=False) + monkeypatch.setattr(device, "query_devices", lambda: {}, raising=False) + pyaudio = ModuleType("pyaudio") + monkeypatch.setattr(pyaudio, "paInt16", 8, raising=False) + monkeypatch.setitem(sys.modules, "sounddevice", device) + monkeypatch.setitem(sys.modules, "pyaudio", pyaudio) + monkeypatch.syspath_prepend(str(Path(__file__).parent)) + monkeypatch.setenv("OPENAI_API_KEY", "test-only") + # Offline tests never need the user's configured proxy. + for name in ("ALL_PROXY", "HTTPS_PROXY", "HTTP_PROXY", "all_proxy", "https_proxy", "http_proxy"): + monkeypatch.delenv(name, raising=False) + monkeypatch.delitem(sys.modules, "audio_util", raising=False) + monkeypatch.delitem(sys.modules, "push_to_talk_app", raising=False) + import push_to_talk_app + + yield push_to_talk_app, started + sys.modules.pop("audio_util", None) + sys.modules.pop("push_to_talk_app", None) + + +def test_player_truncates_partial_and_queued_items(app_module: Any) -> None: + module, _ = app_module + player = module.AudioPlayerAsync() + player.add_data(bytes(1920), response_id="r1", item_id="first", content_index=0) + player.add_data(bytes(1920), response_id="r2", item_id="second", content_index=0) + out = np.empty((480, 1), dtype=np.int16) + player.stream.callback(out, 480, None, None) + player.finish_response("r1") + player.finish_response("r2") + assert player.stop() == [(("r1", "first", 0), 480), (("r2", "second", 0), 0)] + assert player.stop() == [] + + +def test_finished_playback_does_not_accumulate(app_module: Any) -> None: + module, _ = app_module + player = module.AudioPlayerAsync() + out = np.empty((480, 1), dtype=np.int16) + for index in range(500): + response_id = f"r{index}" + player.add_data(bytes(960), response_id=response_id, item_id=f"i{index}", content_index=0) + player.finish_response(response_id) + player.stream.callback(out, 480, None, None) + assert not player._played_by_item + assert not player._finished_responses + assert player.stop() == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stop_before_cancel", [False, True]) +async def test_capture_starts_before_blocked_cancel_but_sends_after(app_module: Any, stop_before_cancel: bool) -> None: + module, mic_started = app_module + app = module.RealtimeApp() + status = SimpleNamespace(is_recording=False) + app.query_one = lambda *_args, **_kwargs: status + sent: list[str] = [] + waiting = asyncio.Event() + release = asyncio.Event() + + async def cancel(**_kwargs: Any) -> None: + waiting.set() + await release.wait() + sent.append("cancel") + + async def truncate(**kwargs: Any) -> None: + sent.append("truncate") + assert kwargs == {"item_id": "assistant", "content_index": 0, "audio_end_ms": 20} + + async def append(**_kwargs: Any) -> None: + sent.append("append") + + app.connection = SimpleNamespace( + response=SimpleNamespace(cancel=cancel), + conversation=SimpleNamespace(item=SimpleNamespace(truncate=truncate)), + input_audio_buffer=SimpleNamespace(append=append), + ) + app.connected.set() + app._active_response_id = "response" + app.audio_player.add_data(bytes(1920), response_id="response", item_id="assistant", content_index=0) + app.audio_player.stream.callback(np.empty((480, 1), dtype=np.int16), 480, None, None) + mic = asyncio.create_task(app.send_mic_audio()) + key = asyncio.create_task(app.on_key(SimpleNamespace(key="k"))) + try: + await asyncio.wait_for(waiting.wait(), 2) + await asyncio.wait_for(mic_started.wait(), 2) + assert status.is_recording + assert not sent + if stop_before_cancel: + await app.on_key(SimpleNamespace(key="k")) + assert app._mic_stream is None + release.set() + await asyncio.wait_for(key, 2) + for _ in range(100): + if "append" in sent: + break + await asyncio.sleep(0) + assert sent == (["cancel", "truncate"] if stop_before_cancel else ["cancel", "truncate", "append"]) + finally: + release.set() + mic.cancel() + await asyncio.gather(mic, return_exceptions=True) + await app.client.close() + assert app._mic_stream is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", ["normal", "abnormal", "unsent", "cancellation"]) +async def test_disconnect_exits_and_user_cancellation_is_preserved(app_module: Any, failure: str) -> None: + from websockets.exceptions import ConnectionClosedError + + from openai import WebSocketConnectionClosedError + + module, _ = app_module + app = module.RealtimeApp() + exited: list[dict[str, Any]] = [] + app.exit = lambda **kwargs: exited.append(kwargs) + + async def events() -> None: + if failure == "abnormal": + raise ConnectionClosedError(None, None) + if failure == "unsent": + raise WebSocketConnectionClosedError("fixture", unsent_messages=[]) + if failure == "cancellation": + raise asyncio.CancelledError() + + app._receive_realtime_events = events + try: + if failure == "cancellation": + with pytest.raises(asyncio.CancelledError): + await app.handle_realtime_connection() + assert not exited + else: + await app.handle_realtime_connection() + assert exited == [{"return_code": 1, "message": "The Realtime connection closed."}] + finally: + await app.client.close() From 022c50de8e7fa694f9bec85654f7686176671ced Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 03:09:25 +0000 Subject: [PATCH 03/19] Keep optional Realtime sample test imports scoped and typed --- examples/realtime/test_push_to_talk_app.py | 24 ++++++++++++++++------ 1 file changed, 18 insertions(+), 6 deletions(-) diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py index e70805d662..7cf087ca47 100644 --- a/examples/realtime/test_push_to_talk_app.py +++ b/examples/realtime/test_push_to_talk_app.py @@ -4,6 +4,7 @@ import sys import asyncio +import importlib from types import ModuleType, SimpleNamespace from typing import Any from pathlib import Path @@ -57,22 +58,25 @@ def close(self) -> None: # Replace only audio device bindings; import the actual UI, player, and SDK. device = ModuleType("sounddevice") + + def query_devices() -> dict[str, str]: + return {} + monkeypatch.setattr(device, "InputStream", InputStream, raising=False) monkeypatch.setattr(device, "OutputStream", OutputStream, raising=False) - monkeypatch.setattr(device, "query_devices", lambda: {}, raising=False) + monkeypatch.setattr(device, "query_devices", query_devices, raising=False) pyaudio = ModuleType("pyaudio") monkeypatch.setattr(pyaudio, "paInt16", 8, raising=False) monkeypatch.setitem(sys.modules, "sounddevice", device) monkeypatch.setitem(sys.modules, "pyaudio", pyaudio) - monkeypatch.syspath_prepend(str(Path(__file__).parent)) + monkeypatch.setattr(sys, "path", [str(Path(__file__).parent), *sys.path]) monkeypatch.setenv("OPENAI_API_KEY", "test-only") # Offline tests never need the user's configured proxy. for name in ("ALL_PROXY", "HTTPS_PROXY", "HTTP_PROXY", "all_proxy", "https_proxy", "http_proxy"): monkeypatch.delenv(name, raising=False) monkeypatch.delitem(sys.modules, "audio_util", raising=False) monkeypatch.delitem(sys.modules, "push_to_talk_app", raising=False) - import push_to_talk_app - + push_to_talk_app = importlib.import_module("push_to_talk_app") yield push_to_talk_app, started sys.modules.pop("audio_util", None) sys.modules.pop("push_to_talk_app", None) @@ -111,7 +115,11 @@ async def test_capture_starts_before_blocked_cancel_but_sends_after(app_module: module, mic_started = app_module app = module.RealtimeApp() status = SimpleNamespace(is_recording=False) - app.query_one = lambda *_args, **_kwargs: status + + def query_one(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + return status + + app.query_one = query_one sent: list[str] = [] waiting = asyncio.Event() release = asyncio.Event() @@ -172,7 +180,11 @@ async def test_disconnect_exits_and_user_cancellation_is_preserved(app_module: A module, _ = app_module app = module.RealtimeApp() exited: list[dict[str, Any]] = [] - app.exit = lambda **kwargs: exited.append(kwargs) + + def capture_exit(**kwargs: Any) -> None: + exited.append(kwargs) + + app.exit = capture_exit async def events() -> None: if failure == "abnormal": From bee4bb508c7c851e58201b017ec5f6d807959464 Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 03:17:25 +0000 Subject: [PATCH 04/19] Clear interrupted assistant transcripts and ignore their late deltas --- examples/realtime/push_to_talk_app.py | 13 ++++ examples/realtime/test_push_to_talk_app.py | 71 +++++++++++++++++++++- 2 files changed, 83 insertions(+), 1 deletion(-) diff --git a/examples/realtime/push_to_talk_app.py b/examples/realtime/push_to_talk_app.py index c226433796..c4a4b3b310 100755 --- a/examples/realtime/push_to_talk_app.py +++ b/examples/realtime/push_to_talk_app.py @@ -136,6 +136,7 @@ def __init__(self) -> None: self.client = AsyncOpenAI() self.audio_player = AudioPlayerAsync() self._active_response_id: str | None = None + self._displayed_item_id: str | None = None self._interrupted_response_ids: set[str] = set() self.should_send_audio = asyncio.Event() self._mic_send_lock = asyncio.Lock() @@ -162,6 +163,7 @@ async def handle_realtime_connection(self) -> None: try: await self._receive_realtime_events() except (ConnectionClosed, WebSocketConnectionClosedError): + # End the app with the same message for normal and abnormal socket closure. pass self.exit(return_code=1, message="The Realtime connection closed.") @@ -235,6 +237,8 @@ async def _receive_realtime_events(self) -> None: continue if event.type == "response.output_audio_transcript.delta": + if event.response_id in self._interrupted_response_ids: + continue try: text = acc_items[event.item_id] except KeyError: @@ -246,6 +250,7 @@ async def _receive_realtime_events(self) -> None: bottom_pane = self.query_one("#bottom-pane", RichLog) bottom_pane.clear() bottom_pane.write(acc_items[event.item_id]) + self._displayed_item_id = event.item_id continue async def _get_connection(self) -> AsyncRealtimeConnection: @@ -340,6 +345,14 @@ async def on_key(self, event: events.Key) -> None: if self.audio_player.playing else [] ) + if any( + item_id == self._displayed_item_id and content_index == 0 + for (_, item_id, content_index), _ in unheard + ): + bottom_pane = self.query_one("#bottom-pane", RichLog) + bottom_pane.clear() + bottom_pane.write("Response interrupted.") + self._displayed_item_id = None if self.connection is not None: if self._active_response_id is not None: response_id = self._active_response_id diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py index 7cf087ca47..40b231c161 100644 --- a/examples/realtime/test_push_to_talk_app.py +++ b/examples/realtime/test_push_to_talk_app.py @@ -115,9 +115,19 @@ async def test_capture_starts_before_blocked_cancel_but_sends_after(app_module: module, mic_started = app_module app = module.RealtimeApp() status = SimpleNamespace(is_recording=False) + display = ["spoken and unspoken output"] + + class Log: + def clear(self) -> None: + display.clear() + + def write(self, message: str) -> None: + display.append(message) def query_one(*_args: Any, **_kwargs: Any) -> SimpleNamespace: - return status + return ( + status if _args[0] is module.AudioStatusIndicator else SimpleNamespace(clear=Log().clear, write=Log().write) + ) app.query_one = query_one sent: list[str] = [] @@ -143,6 +153,7 @@ async def append(**_kwargs: Any) -> None: ) app.connected.set() app._active_response_id = "response" + app._displayed_item_id = "assistant" app.audio_player.add_data(bytes(1920), response_id="response", item_id="assistant", content_index=0) app.audio_player.stream.callback(np.empty((480, 1), dtype=np.int16), 480, None, None) mic = asyncio.create_task(app.send_mic_audio()) @@ -152,6 +163,7 @@ async def append(**_kwargs: Any) -> None: await asyncio.wait_for(mic_started.wait(), 2) assert status.is_recording assert not sent + assert display == ["Response interrupted."] if stop_before_cancel: await app.on_key(SimpleNamespace(key="k")) assert app._mic_stream is None @@ -205,3 +217,60 @@ async def events() -> None: assert exited == [{"return_code": 1, "message": "The Realtime connection closed."}] finally: await app.client.close() + + +@pytest.mark.asyncio +async def test_late_transcript_deltas_from_interrupted_response_do_not_replace_display(app_module: Any) -> None: + module, _ = app_module + app = module.RealtimeApp() + display: list[str] = [] + + class Log: + def clear(self) -> None: + display.clear() + + def write(self, message: str) -> None: + display.append(message) + + def query_one(*_args: Any, **_kwargs: Any) -> Log: + return Log() + + async def update(**_kwargs: Any) -> None: + pass + + class Connection: + session = SimpleNamespace(update=update) + + async def __aenter__(self) -> Any: + return self + + async def __aexit__(self, *_args: Any) -> None: + pass + + async def __aiter__(self) -> Any: + yield SimpleNamespace( + type="response.output_audio_transcript.delta", response_id="old", item_id="old", delta="first" + ) + assert display == ["first"] + app._interrupted_response_ids.add("old") + yield SimpleNamespace( + type="response.output_audio_transcript.delta", response_id="old", item_id="old", delta="late" + ) + assert display == ["first"] + yield SimpleNamespace( + type="response.output_audio_transcript.delta", response_id="new", item_id="new", delta="new speech" + ) + assert display == ["new speech"] + + app.query_one = query_one + original_client = app.client + + def connect(**_kwargs: Any) -> Connection: + return Connection() + + app.client = SimpleNamespace(realtime=SimpleNamespace(connect=connect)) + try: + await app._receive_realtime_events() + assert app._displayed_item_id == "new" + finally: + await original_client.close() From 5663b7c6ed6bdc945ef3b52ab12e7d254f511c9d Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 03:25:07 +0000 Subject: [PATCH 05/19] Verify pinned hashes for Realtime sample test dependencies --- .github/workflows/ci.yml | 11 ++-- examples/realtime/test-requirements.txt | 83 +++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 6 deletions(-) create mode 100644 examples/realtime/test-requirements.txt diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4ce71cb064..da091bc8c5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -280,12 +280,11 @@ jobs: run: ./scripts/test - name: Test Realtime push-to-talk sample - run: >- - uv run --locked --all-extras - --with 'textual==8.2.8' - --with 'pydub==0.25.1' - --with 'audioop-lts==0.2.2; python_version >= "3.13"' - python -m pytest -o addopts= --noconftest -q examples/realtime/test_push_to_talk_app.py + run: | + uv sync --locked --all-extras + uv pip install --python .venv/bin/python --require-hashes --only-binary :all: -r examples/realtime/test-requirements.txt + uv pip check --python .venv/bin/python + .venv/bin/python -m pytest -o addopts= --noconftest -q examples/realtime/test_push_to_talk_app.py examples: timeout-minutes: 10 diff --git a/examples/realtime/test-requirements.txt b/examples/realtime/test-requirements.txt new file mode 100644 index 0000000000..a275754168 --- /dev/null +++ b/examples/realtime/test-requirements.txt @@ -0,0 +1,83 @@ +audioop-lts==0.2.2 ; python_full_version >= '3.13' \ + --hash=sha256:0337d658f9b81f4cd0fdb1f47635070cc084871a3d4646d9de74fdf4e7c3d24a \ + --hash=sha256:03f061a1915538fd96272bac9551841859dbb2e3bf73ebe4a23ef043766f5449 \ + --hash=sha256:068aa17a38b4e0e7de771c62c60bbca2455924b67a8814f3b0dee92b5820c0b3 \ + --hash=sha256:088327f00488cdeed296edd9215ca159f3a5a5034741465789cad403fcf4bec0 \ + --hash=sha256:0d9385e96f9f6da847f4d571ce3cb15b5091140edf3db97276872647ce37efd7 \ + --hash=sha256:106753a83a25ee4d6f473f2be6b0966fc1c9af7e0017192f5531a3e7463dce58 \ + --hash=sha256:143fad0311e8209ece30a8dbddab3b65ab419cbe8c0dde6e8828da25999be911 \ + --hash=sha256:15ab25dd3e620790f40e9ead897f91e79c0d3ce65fe193c8ed6c26cffdd24be7 \ + --hash=sha256:167d3b62586faef8b6b2275c3218796b12621a60e43f7e9d5845d627b9c9b80e \ + --hash=sha256:2b267b70747d82125f1a021506565bdc5609a2b24bcb4773c16d79d2bb260bbd \ + --hash=sha256:3bcddaaf6cc5935a300a8387c99f7a7fbbe212a11568ec6cf6e4bc458c048636 \ + --hash=sha256:3fc38008969796f0f689f1453722a0f463da1b8a6fbee11987830bfbb664f623 \ + --hash=sha256:47eba38322370347b1c47024defbd36374a211e8dd5b0dcbce7b34fdb6f8847b \ + --hash=sha256:48159d96962674eccdca9a3df280e864e8ac75e40a577cc97c5c42667ffabfc5 \ + --hash=sha256:49ee1a41738a23e98d98b937a0638357a2477bc99e61b0f768a8f654f45d9b7a \ + --hash=sha256:4a53aa7c16a60a6857e6b0b165261436396ef7293f8b5c9c828a3a203147ed4a \ + --hash=sha256:4b4cd51a57b698b2d06cb9993b7ac8dfe89a3b2878e96bc7948e9f19ff51dba6 \ + --hash=sha256:51c916108c56aa6e426ce611946f901badac950ee2ddaf302b7ed35d9958970d \ + --hash=sha256:550c114a8df0aafe9a05442a1162dfc8fec37e9af1d625ae6060fed6e756f303 \ + --hash=sha256:58cf54380c3884fb49fdd37dfb7a772632b6701d28edd3e2904743c5e1773602 \ + --hash=sha256:5b00be98ccd0fc123dcfad31d50030d25fcf31488cde9e61692029cd7394733b \ + --hash=sha256:5f93a5db13927a37d2d09637ccca4b2b6b48c19cd9eda7b17a2e9f77edee6a6f \ + --hash=sha256:64d0c62d88e67b98a1a5e71987b7aa7b5bcffc7dcee65b635823dbdd0a8dbbd0 \ + --hash=sha256:73f80bf4cd5d2ca7814da30a120de1f9408ee0619cc75da87d0641273d202a09 \ + --hash=sha256:752d76472d9804ac60f0078c79cdae8b956f293177acd2316cd1e15149aee132 \ + --hash=sha256:83c381767e2cc10e93e40281a04852facc4cd9334550e0f392f72d1c0a9c5753 \ + --hash=sha256:8fefe5868cd082db1186f2837d64cfbfa78b548ea0d0543e9b28935ccce81ce9 \ + --hash=sha256:9191d68659eda01e448188f60364c7763a7ca6653ed3f87ebb165822153a8547 \ + --hash=sha256:96f19de485a2925314f5020e85911fb447ff5fbef56e8c7c6927851b95533a1c \ + --hash=sha256:9a13dc409f2564de15dd68be65b462ba0dde01b19663720c68c1140c782d1d75 \ + --hash=sha256:a2c2a947fae7d1062ef08c4e369e0ba2086049a5e598fda41122535557012e9e \ + --hash=sha256:a2d4f1513d63c795e82948e1305f31a6d530626e5f9f2605408b300ae6095093 \ + --hash=sha256:a5bf613e96f49712073de86f20dbdd4014ca18efd4d34ed18c75bd808337851b \ + --hash=sha256:a6d2e0f9f7a69403e388894d4ca5ada5c47230716a03f2847cfc7bd1ecb589d6 \ + --hash=sha256:b492c3b040153e68b9fdaff5913305aaaba5bb433d8a7f73d5cf6a64ed3cc1dd \ + --hash=sha256:ba7c3a7e5f23e215cb271516197030c32aef2e754252c4c70a50aaff7031a2c8 \ + --hash=sha256:c0022283e9556e0f3643b7c3c03f05063ca72b3063291834cca43234f20c60bb \ + --hash=sha256:c174e322bb5783c099aaf87faeb240c8d210686b04bd61dfd05a8e5a83d88969 \ + --hash=sha256:c9c8e68d8b4a56fda8c025e538e639f8c5953f5073886b596c93ec9b620055e7 \ + --hash=sha256:cfcac6aa6f42397471e4943e0feb2244549db5c5d01efcd02725b96af417f3fe \ + --hash=sha256:d5e73fa573e273e4f2e5ff96f9043858a5e9311e94ffefd88a3186a910c70917 \ + --hash=sha256:def246fe9e180626731b26e89816e79aae2276f825420a07b4a647abaa84becc \ + --hash=sha256:dfbbc74ec68a0fd08cfec1f4b5e8cca3d3cd7de5501b01c4b5d209995033cde9 \ + --hash=sha256:e160bf9df356d841bb6c180eeeea1834085464626dc1b68fa4e1d59070affdc3 \ + --hash=sha256:e541c3ef484852ef36545f66209444c48b28661e864ccadb29daddb6a4b8e5f5 \ + --hash=sha256:f9b0b8a03ef474f56d1a842af1a2e01398b8f7654009823c6d9e0ecff4d5cfbf \ + --hash=sha256:f9ee9b52f5f857fbaf9d605a360884f034c92c1c23021fb90b2e39b8e64bede6 \ + --hash=sha256:fbdd522624141e40948ab3e8cdae6e04c748d78710e9f0f8d4dae2750831de19 \ + --hash=sha256:fd3d4602dc64914d462924a08c1a9816435a2155d74f325853c1f1ac3b2d9800 +linkify-it-py==2.2.0 \ + --hash=sha256:3adc40eb5af300b2605fcfdb968c24e1d780a90f1f2221af7c15e5111e94d443 \ + --hash=sha256:907acd2d17ac1fbb9ddb62c8957ccbd6158cac602231a15c3b0cd1e215f03cee +markdown-it-py==4.2.0 \ + --hash=sha256:04a21681d6fbb623de53f6f364d352309d4094dd4194040a10fd51833e418d49 \ + --hash=sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a +mdit-py-plugins==0.6.1 \ + --hash=sha256:214c82fb2ac524472ab6a5bcab1de80f73b50443e187f401bfd77efbc7c6481d \ + --hash=sha256:a2bca0f039f39dbd35fb74ae1b5f998608c437463371f0ff7f49a19a17a114d0 +mdurl==0.1.2 \ + --hash=sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8 \ + --hash=sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba +platformdirs==4.12.4 ; python_full_version < '3.11' \ + --hash=sha256:63743c02414e755de4e31b8f68125c1407495b86c5a006e203c01ff8b9924250 \ + --hash=sha256:78bfb9db2a8471ed7eebe3c3c932da413911042994e699b384fbb4493fa872d7 +platformdirs==4.13.5 ; python_full_version >= '3.11' \ + --hash=sha256:05831adf87d392760f4b5c3a0a95579ab5a10b18b9b9bd43680dcb47544788a0 \ + --hash=sha256:53d3ec4280b815c1e0b02edad7f9ee8fda1ba9c0865dd9c2e6e63849ce82fc1e +pydub==0.25.1 \ + --hash=sha256:65617e33033874b59d87db603aa1ed450633288aefead953b30bded59cb599a6 \ + --hash=sha256:980a33ce9949cab2a569606b65674d748ecbca4f0796887fd6f46173a7b0d30f +pygments==2.21.0 \ + --hash=sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9 \ + --hash=sha256:610ca751c9bc2492b38eb9a38a7fbc93edbbb2d7182edaf34e66ae493dee5c8c +rich==15.0.0 \ + --hash=sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb \ + --hash=sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36 +textual==8.2.8 \ + --hash=sha256:267375fd402dc8d981457212efa71f0e3365fd17bba144ba9bb3ed7563cb374a \ + --hash=sha256:3f106a9fbc73e39dd266c9712432087de78a6d644084c7c241d6a25c3169115b +typing-extensions==4.16.0 \ + --hash=sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8 \ + --hash=sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5 From b4eec740c61789640bcb409c1b9928ff0c32d888 Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 03:31:01 +0000 Subject: [PATCH 06/19] test(realtime): install the complete test lock without re-resolving the editable SDK --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index da091bc8c5..343e59ca2e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -282,7 +282,7 @@ jobs: - name: Test Realtime push-to-talk sample run: | uv sync --locked --all-extras - uv pip install --python .venv/bin/python --require-hashes --only-binary :all: -r examples/realtime/test-requirements.txt + uv pip install --python .venv/bin/python --no-deps --require-hashes --only-binary :all: -r examples/realtime/test-requirements.txt uv pip check --python .venv/bin/python .venv/bin/python -m pytest -o addopts= --noconftest -q examples/realtime/test_push_to_talk_app.py From 896e50a12ada43a16dc2bd8bccc22c1d746eca5e Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 03:35:44 +0000 Subject: [PATCH 07/19] fix(realtime): guard empty manual audio and delayed responses before input --- examples/realtime/push_to_talk_app.py | 24 +++- examples/realtime/test_push_to_talk_app.py | 130 ++++++++++++++++++++- 2 files changed, 147 insertions(+), 7 deletions(-) diff --git a/examples/realtime/push_to_talk_app.py b/examples/realtime/push_to_talk_app.py index c4a4b3b310..cc44920ef4 100755 --- a/examples/realtime/push_to_talk_app.py +++ b/examples/realtime/push_to_talk_app.py @@ -140,6 +140,7 @@ def __init__(self) -> None: self._interrupted_response_ids: set[str] = set() self.should_send_audio = asyncio.Event() self._mic_send_lock = asyncio.Lock() + self._awaiting_new_input = False self.connected = asyncio.Event() self._mic_stream: Any = None @@ -207,6 +208,14 @@ async def _receive_realtime_events(self) -> None: if event.type == "response.created": if self._active_response_id is not None and self._active_response_id != event.response.id: self.audio_player.finish_response(self._active_response_id) + if self._awaiting_new_input and event.response.id is not None: + # Nothing from a new recording has been offered since K + # interrupted playback, so this response belongs to an + # earlier turn (even if K was already released). + self._active_response_id = None + self._interrupted_response_ids.add(event.response.id) + await conn.response.cancel(response_id=event.response.id) + continue self._active_response_id = event.response.id continue @@ -292,7 +301,14 @@ async def send_mic_audio(self) -> None: if stream is not self._mic_stream or not self.should_send_audio.is_set(): continue data, _ = stream.read(read_size) - await connection.input_audio_buffer.append(audio=base64.b64encode(cast(Any, data)).decode("utf-8")) + self._awaiting_new_input = False + try: + await connection.input_audio_buffer.append( + audio=base64.b64encode(cast(Any, data)).decode("utf-8") + ) + except BaseException: + self._awaiting_new_input = True + raise await asyncio.sleep(0) except KeyboardInterrupt: @@ -334,10 +350,12 @@ async def on_key(self, event: events.Key) -> None: # manually tell the model to commit the audio buffer and start responding. conn = await self._get_connection() async with self._mic_send_lock: - await conn.input_audio_buffer.commit() - await conn.response.create() + if not self._awaiting_new_input: + await conn.input_audio_buffer.commit() + await conn.response.create() else: async with self._mic_send_lock: + self._awaiting_new_input = True self.should_send_audio.set() status_indicator.is_recording = True unheard = ( diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py index 40b231c161..a902a3061b 100644 --- a/examples/realtime/test_push_to_talk_app.py +++ b/examples/realtime/test_push_to_talk_app.py @@ -111,7 +111,10 @@ def test_finished_playback_does_not_accumulate(app_module: Any) -> None: @pytest.mark.asyncio @pytest.mark.parametrize("stop_before_cancel", [False, True]) -async def test_capture_starts_before_blocked_cancel_but_sends_after(app_module: Any, stop_before_cancel: bool) -> None: +@pytest.mark.parametrize("manual_turn", [False, True]) +async def test_capture_starts_before_blocked_cancel_but_sends_after( + app_module: Any, stop_before_cancel: bool, manual_turn: bool +) -> None: module, mic_started = app_module app = module.RealtimeApp() status = SimpleNamespace(is_recording=False) @@ -146,11 +149,19 @@ async def truncate(**kwargs: Any) -> None: async def append(**_kwargs: Any) -> None: sent.append("append") + async def commit() -> None: + sent.append("commit") + + async def create() -> None: + sent.append("create") + app.connection = SimpleNamespace( - response=SimpleNamespace(cancel=cancel), + response=SimpleNamespace(cancel=cancel, create=create), conversation=SimpleNamespace(item=SimpleNamespace(truncate=truncate)), - input_audio_buffer=SimpleNamespace(append=append), + input_audio_buffer=SimpleNamespace(append=append, commit=commit), ) + if manual_turn: + app.session = SimpleNamespace(audio=SimpleNamespace(input=SimpleNamespace(turn_detection=None))) app.connected.set() app._active_response_id = "response" app._displayed_item_id = "assistant" @@ -158,6 +169,7 @@ async def append(**_kwargs: Any) -> None: app.audio_player.stream.callback(np.empty((480, 1), dtype=np.int16), 480, None, None) mic = asyncio.create_task(app.send_mic_audio()) key = asyncio.create_task(app.on_key(SimpleNamespace(key="k"))) + stop: asyncio.Task[None] | None = None try: await asyncio.wait_for(waiting.wait(), 2) await asyncio.wait_for(mic_started.wait(), 2) @@ -165,15 +177,24 @@ async def append(**_kwargs: Any) -> None: assert not sent assert display == ["Response interrupted."] if stop_before_cancel: - await app.on_key(SimpleNamespace(key="k")) + stop = asyncio.create_task(app.on_key(SimpleNamespace(key="k"))) + for _ in range(100): + if app._mic_stream is None: + break + await asyncio.sleep(0) assert app._mic_stream is None release.set() await asyncio.wait_for(key, 2) + if stop is not None: + await asyncio.wait_for(stop, 2) for _ in range(100): if "append" in sent: break await asyncio.sleep(0) assert sent == (["cancel", "truncate"] if stop_before_cancel else ["cancel", "truncate", "append"]) + if not stop_before_cancel: + await app.on_key(SimpleNamespace(key="k")) + assert sent == ["cancel", "truncate", "append"] + (["commit", "create"] if manual_turn else []) finally: release.set() mic.cancel() @@ -274,3 +295,104 @@ def connect(**_kwargs: Any) -> Connection: assert app._displayed_item_id == "new" finally: await original_client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("new_audio_sent", [False, True]) +@pytest.mark.parametrize("released", [False, True]) +async def test_response_created_while_recording_before_or_after_first_append( + app_module: Any, new_audio_sent: bool, released: bool +) -> None: + module, mic_started = app_module + app = module.RealtimeApp() + status = SimpleNamespace(is_recording=False) + display: list[str] = [] + sent: list[tuple[str, Any]] = [] + + class Log: + def clear(self) -> None: + display.clear() + + def write(self, message: str) -> None: + display.append(message) + + def query_one(*args: Any, **_kwargs: Any) -> Any: + return status if args[0] is module.AudioStatusIndicator else Log() + + async def update(**_kwargs: Any) -> None: + pass + + async def cancel(**kwargs: Any) -> None: + sent.append(("cancel", kwargs)) + + async def append(**_kwargs: Any) -> None: + sent.append(("append", None)) + + class Connection: + session = SimpleNamespace(update=update) + response = SimpleNamespace(cancel=cancel) + input_audio_buffer = SimpleNamespace(append=append) + + async def __aenter__(self) -> Any: + return self + + async def __aexit__(self, *_args: Any) -> None: + pass + + async def __aiter__(self) -> Any: + await app.on_key(SimpleNamespace(key="k")) + if new_audio_sent: + mic = asyncio.create_task(app.send_mic_audio()) + try: + await asyncio.wait_for(mic_started.wait(), 2) + for _ in range(100): + if any(name == "append" for name, _ in sent): + break + await asyncio.sleep(0) + assert ("append", None) in sent + finally: + mic.cancel() + await asyncio.gather(mic, return_exceptions=True) + if released: + await app.on_key(SimpleNamespace(key="k")) + yield SimpleNamespace(type="response.created", response=SimpleNamespace(id="response")) + yield SimpleNamespace( + type="response.output_audio.delta", + response_id="response", + item_id="item", + content_index=0, + delta="AAAAAAAAAAA=", + ) + yield SimpleNamespace( + type="response.output_audio_transcript.delta", + response_id="response", + item_id="item", + delta="a response", + ) + if new_audio_sent: + # Server VAD may legitimately answer a pause during the current recording. + assert sent == [("append", None)] + assert app.audio_player.playing + assert display == ["a response"] + else: + # No new audio has been offered, so this can only be an older response. + assert sent == [("cancel", {"response_id": "response"})] + assert not app.audio_player.playing + assert display == [] + yield SimpleNamespace( + type="response.done", + response=SimpleNamespace(id="response", status="completed", status_details=None), + ) + assert not app._interrupted_response_ids + + app.query_one = query_one + original_client = app.client + + def connect(**_kwargs: Any) -> Connection: + return Connection() + + app.client = SimpleNamespace(realtime=SimpleNamespace(connect=connect)) + try: + await app._receive_realtime_events() + finally: + await original_client.close() From 7c8b1497227853e58a6cd29e2518d8dd60d0f3db Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 03:46:06 +0000 Subject: [PATCH 08/19] test(realtime): constrain test packages to the verified SDK lock --- .github/workflows/ci.yml | 3 ++- examples/realtime/test-requirements.txt | 12 ++++++------ 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 343e59ca2e..dac69e2667 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -282,7 +282,8 @@ jobs: - name: Test Realtime push-to-talk sample run: | uv sync --locked --all-extras - uv pip install --python .venv/bin/python --no-deps --require-hashes --only-binary :all: -r examples/realtime/test-requirements.txt + uv export --locked --all-extras --no-emit-project --format requirements-txt --output-file "$RUNNER_TEMP/realtime-sdk-lock.txt" >/dev/null + uv pip install --directory "$RUNNER_TEMP" --python "$GITHUB_WORKSPACE/.venv/bin/python" --no-deps --require-hashes --only-binary :all: --constraint "$RUNNER_TEMP/realtime-sdk-lock.txt" -r "$GITHUB_WORKSPACE/examples/realtime/test-requirements.txt" uv pip check --python .venv/bin/python .venv/bin/python -m pytest -o addopts= --noconftest -q examples/realtime/test_push_to_talk_app.py diff --git a/examples/realtime/test-requirements.txt b/examples/realtime/test-requirements.txt index a275754168..c9e3a134c7 100644 --- a/examples/realtime/test-requirements.txt +++ b/examples/realtime/test-requirements.txt @@ -51,9 +51,9 @@ audioop-lts==0.2.2 ; python_full_version >= '3.13' \ linkify-it-py==2.2.0 \ --hash=sha256:3adc40eb5af300b2605fcfdb968c24e1d780a90f1f2221af7c15e5111e94d443 \ --hash=sha256:907acd2d17ac1fbb9ddb62c8957ccbd6158cac602231a15c3b0cd1e215f03cee -markdown-it-py==4.2.0 \ - --hash=sha256:04a21681d6fbb623de53f6f364d352309d4094dd4194040a10fd51833e418d49 \ - --hash=sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a +markdown-it-py==3.0.0 \ + --hash=sha256:355216845c60bd96232cd8d8c40e8f9765cc86f46880e43a8fd22dc1a1a8cab1 \ + --hash=sha256:e3f60a94fa066dc52ec76661e37c851cb232d92f9886b15cb560aaada2df8feb mdit-py-plugins==0.6.1 \ --hash=sha256:214c82fb2ac524472ab6a5bcab1de80f73b50443e187f401bfd77efbc7c6481d \ --hash=sha256:a2bca0f039f39dbd35fb74ae1b5f998608c437463371f0ff7f49a19a17a114d0 @@ -69,9 +69,9 @@ platformdirs==4.13.5 ; python_full_version >= '3.11' \ pydub==0.25.1 \ --hash=sha256:65617e33033874b59d87db603aa1ed450633288aefead953b30bded59cb599a6 \ --hash=sha256:980a33ce9949cab2a569606b65674d748ecbca4f0796887fd6f46173a7b0d30f -pygments==2.21.0 \ - --hash=sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9 \ - --hash=sha256:610ca751c9bc2492b38eb9a38a7fbc93edbbb2d7182edaf34e66ae493dee5c8c +pygments==2.20.0 \ + --hash=sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f \ + --hash=sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176 rich==15.0.0 \ --hash=sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb \ --hash=sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36 From 0390b4fb74d7e9b330c71836e22a2d1028c1194c Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 03:53:55 +0000 Subject: [PATCH 09/19] test(realtime): preserve cancellation and skip empty manual commit --- examples/realtime/test_push_to_talk_app.py | 46 ++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py index a902a3061b..463204d600 100644 --- a/examples/realtime/test_push_to_talk_app.py +++ b/examples/realtime/test_push_to_talk_app.py @@ -203,6 +203,52 @@ async def create() -> None: assert app._mic_stream is None +@pytest.mark.asyncio +async def test_cancelling_first_append_does_not_commit_empty_manual_audio(app_module: Any) -> None: + module, _ = app_module + app = module.RealtimeApp() + status = SimpleNamespace(is_recording=False) + + def query_one(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + return status + + app.query_one = query_one + app.session = SimpleNamespace(audio=SimpleNamespace(input=SimpleNamespace(turn_detection=None))) + begun = asyncio.Event() + sent: list[str] = [] + + async def append(**_kwargs: Any) -> None: + begun.set() + await asyncio.Future[None]() + sent.append("append") + + async def commit() -> None: + sent.append("commit") + + async def create() -> None: + sent.append("create") + + app.connection = SimpleNamespace( + input_audio_buffer=SimpleNamespace(append=append, commit=commit), + response=SimpleNamespace(create=create), + ) + app.connected.set() + await app.on_key(SimpleNamespace(key="k")) + mic = asyncio.create_task(app.send_mic_audio()) + try: + await asyncio.wait_for(begun.wait(), 2) + mic.cancel() + with pytest.raises(asyncio.CancelledError): + await mic + assert app._mic_stream is None + await app.on_key(SimpleNamespace(key="k")) + assert sent == [] + finally: + mic.cancel() + await asyncio.gather(mic, return_exceptions=True) + await app.client.close() + + @pytest.mark.asyncio @pytest.mark.parametrize("failure", ["normal", "abnormal", "unsent", "cancellation"]) async def test_disconnect_exits_and_user_cancellation_is_preserved(app_module: Any, failure: str) -> None: From 3d19c6843c07142a14938454c446f852ab0f166c Mon Sep 17 00:00:00 2001 From: markstuart-oai Date: Wed, 28 Oct 2026 05:46:16 +0000 Subject: [PATCH 10/19] fix(realtime): keep recording responsive through manual turn backpressure --- examples/realtime/audio_util.py | 2 +- examples/realtime/push_to_talk_app.py | 120 +++++++++++---------- examples/realtime/test_push_to_talk_app.py | 104 +++++++++++++++++- 3 files changed, 169 insertions(+), 57 deletions(-) diff --git a/examples/realtime/audio_util.py b/examples/realtime/audio_util.py index 0c7fce3801..f62255694a 100644 --- a/examples/realtime/audio_util.py +++ b/examples/realtime/audio_util.py @@ -100,7 +100,7 @@ def start(self): def stop(self, *, interrupted_response_id: str | None = None): self.playing = False - self.stream.stop() + self.stream.abort() with self.lock: # Stop the callback first so these counts cannot advance during interruption. pending = dict.fromkeys(key for _, key in self.queue) diff --git a/examples/realtime/push_to_talk_app.py b/examples/realtime/push_to_talk_app.py index cc44920ef4..37b7199546 100755 --- a/examples/realtime/push_to_talk_app.py +++ b/examples/realtime/push_to_talk_app.py @@ -31,6 +31,7 @@ import base64 import asyncio from typing import Any, cast +from dataclasses import dataclass from typing_extensions import override from textual import events @@ -69,6 +70,11 @@ def render(self) -> str: return status +@dataclass +class Recording: + has_audio: bool = False + + class RealtimeApp(App[None]): CSS = """ Screen { @@ -143,6 +149,7 @@ def __init__(self) -> None: self._awaiting_new_input = False self.connected = asyncio.Event() self._mic_stream: Any = None + self._recording = Recording() @override def compose(self) -> ComposeResult: @@ -268,11 +275,6 @@ async def _get_connection(self) -> AsyncRealtimeConnection: return self.connection async def send_mic_audio(self) -> None: - import sounddevice as sd # type: ignore - - device_info = sd.query_devices() - print(device_info) - read_size = int(SAMPLE_RATE * 0.02) try: @@ -282,17 +284,10 @@ async def send_mic_audio(self) -> None: if not self.should_send_audio.is_set(): continue - if self._mic_stream is None: - # Use a fresh stream for each recording so buffered audio from - # a previous recording cannot be sent when recording resumes. - self._mic_stream = sd.InputStream( - channels=CHANNELS, - samplerate=SAMPLE_RATE, - dtype="int16", - ) - self._mic_stream.start() - stream = self._mic_stream + recording = self._recording + if stream is None: + continue if stream.read_available < read_size: await asyncio.sleep(0) continue @@ -306,8 +301,10 @@ async def send_mic_audio(self) -> None: await connection.input_audio_buffer.append( audio=base64.b64encode(cast(Any, data)).decode("utf-8") ) + recording.has_audio = True except BaseException: - self._awaiting_new_input = True + if recording is self._recording: + self._awaiting_new_input = True raise await asyncio.sleep(0) @@ -348,46 +345,59 @@ async def on_key(self, event: events.Key) -> None: # # However if we're in manual `turn_detection` mode then we need to # manually tell the model to commit the audio buffer and start responding. - conn = await self._get_connection() - async with self._mic_send_lock: - if not self._awaiting_new_input: - await conn.input_audio_buffer.commit() - await conn.response.create() + self.run_worker(self._finish_manual_recording(self._recording)) else: - async with self._mic_send_lock: - self._awaiting_new_input = True - self.should_send_audio.set() - status_indicator.is_recording = True - unheard = ( - self.audio_player.stop(interrupted_response_id=self._active_response_id) - if self.audio_player.playing - else [] - ) - if any( - item_id == self._displayed_item_id and content_index == 0 - for (_, item_id, content_index), _ in unheard - ): - bottom_pane = self.query_one("#bottom-pane", RichLog) - bottom_pane.clear() - bottom_pane.write("Response interrupted.") - self._displayed_item_id = None - if self.connection is not None: - if self._active_response_id is not None: - response_id = self._active_response_id - self._active_response_id = None - self._interrupted_response_ids.add(response_id) - await self.connection.response.cancel(response_id=response_id) - for (_, item_id, content_index), frames in unheard: - if content_index != 0: - self.query_one(RichLog).write( - "Cannot truncate audio: the API only supports content part 0." - ) - continue - await self.connection.conversation.item.truncate( - item_id=item_id, - content_index=content_index, - audio_end_ms=frames * 1000 // SAMPLE_RATE, - ) + import sounddevice as sd # type: ignore + + self._recording = Recording() + self._awaiting_new_input = True + # Open each recording at K, even while an earlier send is blocked. + self._mic_stream = sd.InputStream(channels=CHANNELS, samplerate=SAMPLE_RATE, dtype="int16") + self._mic_stream.start() + status_indicator.is_recording = True + response_id = self._active_response_id + self._active_response_id = None + unheard = ( + self.audio_player.stop(interrupted_response_id=response_id) if self.audio_player.playing else [] + ) + if response_id is not None: + self._interrupted_response_ids.add(response_id) + if any( + item_id == self._displayed_item_id and content_index == 0 + for (_, item_id, content_index), _ in unheard + ): + bottom_pane = self.query_one("#bottom-pane", RichLog) + bottom_pane.clear() + bottom_pane.write("Response interrupted.") + self._displayed_item_id = None + # Queue ordering work before waking the mic sender. Textual owns + # worker cancellation on exit; no WebSocket writes hold the UI. + if response_id is not None or unheard: + self.run_worker(self._interrupt_response(response_id, unheard)) + self.should_send_audio.set() + + async def _finish_manual_recording(self, recording: Recording) -> None: + conn = await self._get_connection() + async with self._mic_send_lock: + if recording.has_audio: + await conn.input_audio_buffer.commit() + await conn.response.create() + + async def _interrupt_response( + self, response_id: str | None, unheard: list[tuple[tuple[str, str, int], int]] + ) -> None: + async with self._mic_send_lock: + if self.connection is None: + return + if response_id is not None: + await self.connection.response.cancel(response_id=response_id) + for (_, item_id, content_index), frames in unheard: + if content_index != 0: + self.query_one(RichLog).write("Cannot truncate audio: the API only supports content part 0.") + continue + await self.connection.conversation.item.truncate( + item_id=item_id, content_index=content_index, audio_end_ms=frames * 1000 // SAMPLE_RATE + ) if __name__ == "__main__": diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py index 463204d600..d554387249 100644 --- a/examples/realtime/test_push_to_talk_app.py +++ b/examples/realtime/test_push_to_talk_app.py @@ -46,12 +46,19 @@ def close(self) -> None: class OutputStream: def __init__(self, callback: Any, **_kwargs: Any): self.callback = callback + self.buffered = b"" + self.emitted = b"" def start(self) -> None: pass def stop(self) -> None: - pass + # PortAudio stop plays pending buffers; abort discards them. + self.emitted += self.buffered + self.buffered = b"" + + def abort(self) -> None: + self.buffered = b"" def close(self) -> None: pass @@ -109,6 +116,100 @@ def test_finished_playback_does_not_accumulate(app_module: Any) -> None: assert player.stop() == [] +def test_interrupt_discards_device_buffered_audio(app_module: Any) -> None: + module, _ = app_module + player = module.AudioPlayerAsync() + player.add_data(bytes(1920), response_id="r1", item_id="first", content_index=0) + player.stream.buffered = b"assistant audio that has not reached the speaker" + assert player.stop(interrupted_response_id="r1") == [(("r1", "first", 0), 0)] + assert not player.stream.emitted + assert not player.stream.buffered + + +@pytest.mark.asyncio +@pytest.mark.parametrize("blocked", ["commit", "create"]) +@pytest.mark.parametrize("continuous", [False, True]) +async def test_manual_send_backpressure_does_not_block_next_key_capture( + app_module: Any, monkeypatch: pytest.MonkeyPatch, blocked: str, continuous: bool +) -> None: + module, mic_started = app_module + if continuous: + # A real input device can always have another frame ready. This leaves + # the sender contending for the lock when key release starts a commit. + device = sys.modules["sounddevice"] + input_type = device.InputStream + + class ContinuousInputStream(input_type): + @property + def read_available(self) -> int: + return 480 + + monkeypatch.setattr(device, "InputStream", ContinuousInputStream) + app = module.RealtimeApp() + waiting = asyncio.Event() + release = asyncio.Event() + appended = asyncio.Event() + sent: list[str] = [] + + async def append(**_kwargs: Any) -> None: + sent.append("append") + appended.set() + + async def write(name: str) -> None: + if name == blocked: + waiting.set() + await release.wait() + sent.append(name) + + async def commit() -> None: + await write("commit") + + async def create() -> None: + await write("create") + + async def already_connected() -> None: + await asyncio.Future[None]() + + monkeypatch.setattr(app, "handle_realtime_connection", already_connected) + app.connection = SimpleNamespace( + input_audio_buffer=SimpleNamespace(append=append, commit=commit), + response=SimpleNamespace(create=create), + ) + app.connected.set() + app.session = SimpleNamespace(audio=SimpleNamespace(input=SimpleNamespace(turn_detection=None))) + async with app.run_test(): + app.post_message(module.events.Key("k", "k")) + await asyncio.wait_for(appended.wait(), 2) + mic_started.clear() + appended.clear() + # Real Textual key dispatch must remain available during a blocked write. + app.post_message(module.events.Key("k", "k")) + try: + await asyncio.wait_for(waiting.wait(), 2) + # The send lock is held, so later appends can only be from the + # next recording after commit/create have finished. + appended.clear() + app.post_message(module.events.Key("k", "k")) + await asyncio.wait_for(mic_started.wait(), 2) + assert app.query_one(module.AudioStatusIndicator).is_recording + prior = list(sent) + if blocked == "commit": + assert prior and set(prior) == {"append"} + else: + assert prior[-1] == "commit" and set(prior[:-1]) == {"append"} + release.set() + await asyncio.wait_for(appended.wait(), 2) + if continuous: + commit_index = sent.index("commit") + assert set(sent[:commit_index]) == {"append"} + assert sent[commit_index : commit_index + 2] == ["commit", "create"] + assert sent[commit_index + 2 :] and set(sent[commit_index + 2 :]) == {"append"} + else: + assert sent == ["append", "commit", "create", "append"] + finally: + release.set() + + @pytest.mark.asyncio @pytest.mark.parametrize("stop_before_cancel", [False, True]) @pytest.mark.parametrize("manual_turn", [False, True]) @@ -194,6 +295,7 @@ async def create() -> None: assert sent == (["cancel", "truncate"] if stop_before_cancel else ["cancel", "truncate", "append"]) if not stop_before_cancel: await app.on_key(SimpleNamespace(key="k")) + await app.workers.wait_for_complete() assert sent == ["cancel", "truncate", "append"] + (["commit", "create"] if manual_turn else []) finally: release.set() From 90dce81bf5a9771b37bb9d82e2648be5a9641894 Mon Sep 17 00:00:00 2001 From: markstuart-oai Date: Wed, 28 Oct 2026 05:59:46 +0000 Subject: [PATCH 11/19] fix(realtime): retain unread audio until device playback --- examples/realtime/audio_util.py | 49 ++++-- examples/realtime/push_to_talk_app.py | 16 +- examples/realtime/test_push_to_talk_app.py | 170 +++++++++++++++++++-- 3 files changed, 199 insertions(+), 36 deletions(-) diff --git a/examples/realtime/audio_util.py b/examples/realtime/audio_util.py index f62255694a..1c6e3db817 100644 --- a/examples/realtime/audio_util.py +++ b/examples/realtime/audio_util.py @@ -44,11 +44,15 @@ def __init__(self): self.playing = False self._frame_count = 0 self._played_by_item: dict[tuple[str, str, int], int] = {} - self._last_played_item: tuple[str, str, int] | None = None + # Item, scheduled DAC time, frames submitted before this span, span length. + self._scheduled: list[tuple[tuple[str, str, int], float, int, int]] = [] self._finished_responses: set[str] = set() def callback(self, outdata, frames, time, status): # noqa with self.lock: + if not self.playing: + outdata.fill(0) + return data = np.empty(0, dtype=np.int16) # get next item from queue if there is still space in the buffer @@ -56,14 +60,17 @@ def callback(self, outdata, frames, time, status): # noqa item, item_key = self.queue.pop(0) frames_needed = frames - len(data) played = item[:frames_needed] + before = self._played_by_item.get(item_key, 0) + self._scheduled.append( + (item_key, time.outputBufferDacTime + len(data) / SAMPLE_RATE, before, len(played)) + ) data = np.concatenate((data, played)) - self._played_by_item[item_key] = self._played_by_item.get(item_key, 0) + len(played) - self._last_played_item = item_key + self._played_by_item[item_key] = before + len(played) if len(item) > frames_needed: self.queue.insert(0, (item[frames_needed:], item_key)) self._frame_count += len(data) - self._discard_finished_counts() + self._discard_finished_counts(time.currentTime) # fill the rest of the frames with zeros if there is no more data if len(data) < frames: @@ -77,15 +84,21 @@ def get_frame_count(self): def finish_response(self, response_id: str): with self.lock: self._finished_responses.add(response_id) - self._discard_finished_counts() + self._discard_finished_counts(self.stream.time) - def _discard_finished_counts(self): + def _discard_finished_counts(self, now: float): + self._scheduled = [span for span in self._scheduled if span[1] + span[3] / SAMPLE_RATE > now] pending = {key for _, key in self.queue} + pending.update(key for key, _, _, _ in self._scheduled) for key in list(self._played_by_item): if key[0] in self._finished_responses and key not in pending: del self._played_by_item[key] self._finished_responses.intersection_update(key[0] for key in pending) + def note_transcript(self, response_id: str, item_id: str, content_index: int): + with self.lock: + self._played_by_item.setdefault((response_id, item_id, content_index), 0) + def add_data(self, data: bytes, *, response_id: str, item_id: str, content_index: int): with self.lock: # bytes is pcm16 single channel audio data, convert to numpy array @@ -99,20 +112,28 @@ def start(self): self.stream.start() def stop(self, *, interrupted_response_id: str | None = None): - self.playing = False - self.stream.abort() with self.lock: - # Stop the callback first so these counts cannot advance during interruption. + # PortAudio supplies both times on the same clock. Exclude later callbacks + # before aborting; abort must not hold the lock a callback may be waiting on. + now = self.stream.time + was_playing = self.playing + self.playing = False + self._discard_finished_counts(now) pending = dict.fromkeys(key for _, key in self.queue) - if self._last_played_item is not None and self._last_played_item[0] == interrupted_response_id: - pending[self._last_played_item] = None - unheard = [(key, self._played_by_item.get(key, 0)) for key in pending] + pending.update(dict.fromkeys(key for key, _, _, _ in self._scheduled)) + pending.update(dict.fromkeys(key for key in self._played_by_item if key[0] == interrupted_response_id)) + heard = {key: self._played_by_item.get(key, 0) for key in pending} + for key, start, before, frames in self._scheduled: + played = max(0, min(frames, int((now - start) * SAMPLE_RATE))) + heard[key] = min(heard[key], before + played) self.queue = [] self._played_by_item.clear() + self._scheduled.clear() self._finished_responses.clear() - self._last_played_item = None self._frame_count = 0 - return unheard + if was_playing: + self.stream.abort() + return list(heard.items()) def terminate(self): self.stream.close() diff --git a/examples/realtime/push_to_talk_app.py b/examples/realtime/push_to_talk_app.py index 37b7199546..6156be977b 100755 --- a/examples/realtime/push_to_talk_app.py +++ b/examples/realtime/push_to_talk_app.py @@ -142,7 +142,7 @@ def __init__(self) -> None: self.client = AsyncOpenAI() self.audio_player = AudioPlayerAsync() self._active_response_id: str | None = None - self._displayed_item_id: str | None = None + self._displayed_item: tuple[str, str, int] | None = None self._interrupted_response_ids: set[str] = set() self.should_send_audio = asyncio.Event() self._mic_send_lock = asyncio.Lock() @@ -266,7 +266,8 @@ async def _receive_realtime_events(self) -> None: bottom_pane = self.query_one("#bottom-pane", RichLog) bottom_pane.clear() bottom_pane.write(acc_items[event.item_id]) - self._displayed_item_id = event.item_id + self.audio_player.note_transcript(event.response_id, event.item_id, event.content_index) + self._displayed_item = (event.response_id, event.item_id, event.content_index) continue async def _get_connection(self) -> AsyncRealtimeConnection: @@ -357,19 +358,14 @@ async def on_key(self, event: events.Key) -> None: status_indicator.is_recording = True response_id = self._active_response_id self._active_response_id = None - unheard = ( - self.audio_player.stop(interrupted_response_id=response_id) if self.audio_player.playing else [] - ) + unheard = self.audio_player.stop(interrupted_response_id=response_id) if response_id is not None: self._interrupted_response_ids.add(response_id) - if any( - item_id == self._displayed_item_id and content_index == 0 - for (_, item_id, content_index), _ in unheard - ): + if any(key == self._displayed_item for key, _ in unheard): bottom_pane = self.query_one("#bottom-pane", RichLog) bottom_pane.clear() bottom_pane.write("Response interrupted.") - self._displayed_item_id = None + self._displayed_item = None # Queue ordering work before waking the mic sender. Textual owns # worker cancellation on exit; no WebSocket writes hold the UI. if response_id is not None or unheard: diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py index d554387249..83103fdc21 100644 --- a/examples/realtime/test_push_to_talk_app.py +++ b/examples/realtime/test_push_to_talk_app.py @@ -45,10 +45,19 @@ def close(self) -> None: class OutputStream: def __init__(self, callback: Any, **_kwargs: Any): - self.callback = callback + self._callback = callback + self.time = 0.0 self.buffered = b"" self.emitted = b"" + def callback(self, out: Any, frames: int, timing: Any, status: Any) -> None: + if timing is None: + # The existing tests model immediately played callback audio. + start = self.time + self.time += frames / module_rate + timing = SimpleNamespace(outputBufferDacTime=start, currentTime=self.time) + self._callback(out, frames, timing, status) + def start(self) -> None: pass @@ -64,6 +73,7 @@ def close(self) -> None: pass # Replace only audio device bindings; import the actual UI, player, and SDK. + module_rate = 24000 device = ModuleType("sounddevice") def query_devices() -> dict[str, str]: @@ -113,6 +123,7 @@ def test_finished_playback_does_not_accumulate(app_module: Any) -> None: player.stream.callback(out, 480, None, None) assert not player._played_by_item assert not player._finished_responses + assert not player._scheduled assert player.stop() == [] @@ -126,6 +137,131 @@ def test_interrupt_discards_device_buffered_audio(app_module: Any) -> None: assert not player.stream.buffered +@pytest.mark.parametrize("at,expected_frames", [(0.99, 0), (1.01, 240), (1.03, None)]) +def test_final_device_buffer_remains_owned_until_dac_finishes( + app_module: Any, at: float, expected_frames: int | None +) -> None: + module, _ = app_module + player = module.AudioPlayerAsync() + player.add_data(bytes(960), response_id="done", item_id="last", content_index=0) + player.stream.time = 0.98 + player.stream.callback( + np.empty((480, 1), dtype=np.int16), + 480, + SimpleNamespace(currentTime=0.98, outputBufferDacTime=1.0), + None, + ) + player.finish_response("done") + player.stream.time = at + assert player.stop() == ([] if expected_frames is None else [(("done", "last", 0), expected_frames)]) + assert not player._played_by_item + assert not player._finished_responses + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stage", ["transcript-first", "done-dac-pending", "done-dac-played"]) +async def test_transcript_interruption_tracks_only_unheard_content(app_module: Any, stage: str) -> None: + module, _ = app_module + app = module.RealtimeApp() + status = SimpleNamespace(is_recording=False) + display: list[str] = [] + sent: list[tuple[str, dict[str, Any]]] = [] + + class Log: + def clear(self) -> None: + display.clear() + + def write(self, text: str) -> None: + display.append(text) + + def query_one(*args: Any, **_kwargs: Any) -> Any: + return status if args[0] is module.AudioStatusIndicator else Log() + + async def update(**_kwargs: Any) -> None: + pass + + async def cancel(**kwargs: Any) -> None: + sent.append(("cancel", kwargs)) + + async def truncate(**kwargs: Any) -> None: + sent.append(("truncate", kwargs)) + + class Connection: + session = SimpleNamespace(update=update) + response = SimpleNamespace(cancel=cancel) + conversation = SimpleNamespace(item=SimpleNamespace(truncate=truncate)) + + async def __aenter__(self) -> Any: + return self + + async def __aexit__(self, *_args: Any) -> None: + pass + + async def __aiter__(self) -> Any: + yield SimpleNamespace(type="response.created", response=SimpleNamespace(id="r1")) + yield SimpleNamespace( + type="response.output_audio_transcript.delta", + response_id="r1", + item_id="only-text", + content_index=0, + delta="Unheard answer", + ) + assert display == ["Unheard answer"] + if stage != "transcript-first": + yield SimpleNamespace( + type="response.output_audio.delta", + response_id="r1", + item_id="only-text", + content_index=0, + delta=module.base64.b64encode(bytes(960)), + ) + player = app.audio_player + player.stream.time = 0.98 + player.stream.callback( + np.empty((480, 1), dtype=np.int16), + 480, + SimpleNamespace(currentTime=0.98, outputBufferDacTime=1.0), + None, + ) + yield SimpleNamespace( + type="response.done", + response=SimpleNamespace(id="r1", status="completed"), + ) + if stage == "done-dac-played": + player.stream.time = 1.03 + await app.on_key(SimpleNamespace(key="k")) + await app.workers.wait_for_complete() + if stage == "done-dac-played": + assert display == ["Unheard answer"] + assert sent == [] + return + assert display == ["Response interrupted."] + expected: list[tuple[str, dict[str, Any]]] = ( + [("cancel", {"response_id": "r1"})] if stage == "transcript-first" else [] + ) + expected.append(("truncate", {"item_id": "only-text", "content_index": 0, "audio_end_ms": 0})) + assert sent == expected + if stage == "transcript-first": + yield SimpleNamespace( + type="response.output_audio.delta", + response_id="r1", + item_id="only-text", + content_index=0, + delta="AAAAAAAAAAA=", + ) + assert not app.audio_player.playing + assert sent == expected + + app.query_one = query_one + client = app.client + app.client = SimpleNamespace(realtime=SimpleNamespace(connect=lambda **_: Connection())) + try: + await app._receive_realtime_events() + finally: + app._stop_mic_stream() + await client.close() + + @pytest.mark.asyncio @pytest.mark.parametrize("blocked", ["commit", "create"]) @pytest.mark.parametrize("continuous", [False, True]) @@ -137,14 +273,11 @@ async def test_manual_send_backpressure_does_not_block_next_key_capture( # A real input device can always have another frame ready. This leaves # the sender contending for the lock when key release starts a commit. device = sys.modules["sounddevice"] - input_type = device.InputStream - class ContinuousInputStream(input_type): - @property - def read_available(self) -> int: - return 480 + def read_available(_stream: Any) -> int: + return 480 - monkeypatch.setattr(device, "InputStream", ContinuousInputStream) + monkeypatch.setattr(device.InputStream, "read_available", property(read_available)) app = module.RealtimeApp() waiting = asyncio.Event() release = asyncio.Event() @@ -265,7 +398,7 @@ async def create() -> None: app.session = SimpleNamespace(audio=SimpleNamespace(input=SimpleNamespace(turn_detection=None))) app.connected.set() app._active_response_id = "response" - app._displayed_item_id = "assistant" + app._displayed_item = ("response", "assistant", 0) app.audio_player.add_data(bytes(1920), response_id="response", item_id="assistant", content_index=0) app.audio_player.stream.callback(np.empty((480, 1), dtype=np.int16), 480, None, None) mic = asyncio.create_task(app.send_mic_audio()) @@ -418,16 +551,28 @@ async def __aexit__(self, *_args: Any) -> None: async def __aiter__(self) -> Any: yield SimpleNamespace( - type="response.output_audio_transcript.delta", response_id="old", item_id="old", delta="first" + type="response.output_audio_transcript.delta", + response_id="old", + item_id="old", + content_index=0, + delta="first", ) assert display == ["first"] app._interrupted_response_ids.add("old") yield SimpleNamespace( - type="response.output_audio_transcript.delta", response_id="old", item_id="old", delta="late" + type="response.output_audio_transcript.delta", + response_id="old", + item_id="old", + content_index=0, + delta="late", ) assert display == ["first"] yield SimpleNamespace( - type="response.output_audio_transcript.delta", response_id="new", item_id="new", delta="new speech" + type="response.output_audio_transcript.delta", + response_id="new", + item_id="new", + content_index=0, + delta="new speech", ) assert display == ["new speech"] @@ -440,7 +585,7 @@ def connect(**_kwargs: Any) -> Connection: app.client = SimpleNamespace(realtime=SimpleNamespace(connect=connect)) try: await app._receive_realtime_events() - assert app._displayed_item_id == "new" + assert app._displayed_item == ("new", "new", 0) finally: await original_client.close() @@ -515,6 +660,7 @@ async def __aiter__(self) -> Any: type="response.output_audio_transcript.delta", response_id="response", item_id="item", + content_index=0, delta="a response", ) if new_audio_sent: From 514d1051af2aac72a80f71adbd47d68485b47a6e Mon Sep 17 00:00:00 2001 From: markstuart-oai Date: Wed, 28 Oct 2026 06:07:36 +0000 Subject: [PATCH 12/19] test(realtime): type test connection factory for static checks --- examples/realtime/test_push_to_talk_app.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py index 83103fdc21..5083e1ac4e 100644 --- a/examples/realtime/test_push_to_talk_app.py +++ b/examples/realtime/test_push_to_talk_app.py @@ -254,7 +254,11 @@ async def __aiter__(self) -> Any: app.query_one = query_one client = app.client - app.client = SimpleNamespace(realtime=SimpleNamespace(connect=lambda **_: Connection())) + + def connect(**_kwargs: Any) -> Connection: + return Connection() + + app.client = SimpleNamespace(realtime=SimpleNamespace(connect=connect)) try: await app._receive_realtime_events() finally: From 8e1d1bd1af81dfc9a9f1a9026a618e768b4cb90c Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 06:31:59 +0000 Subject: [PATCH 13/19] chore: remove sample CI changes from push-to-talk fix --- .github/workflows/ci.yml | 8 --- examples/realtime/test-requirements.txt | 83 ------------------------- 2 files changed, 91 deletions(-) delete mode 100644 examples/realtime/test-requirements.txt diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index dac69e2667..76fc0cb8c4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -279,14 +279,6 @@ jobs: OPENAI_TEST_HTTP_CLIENT: httpx2 run: ./scripts/test - - name: Test Realtime push-to-talk sample - run: | - uv sync --locked --all-extras - uv export --locked --all-extras --no-emit-project --format requirements-txt --output-file "$RUNNER_TEMP/realtime-sdk-lock.txt" >/dev/null - uv pip install --directory "$RUNNER_TEMP" --python "$GITHUB_WORKSPACE/.venv/bin/python" --no-deps --require-hashes --only-binary :all: --constraint "$RUNNER_TEMP/realtime-sdk-lock.txt" -r "$GITHUB_WORKSPACE/examples/realtime/test-requirements.txt" - uv pip check --python .venv/bin/python - .venv/bin/python -m pytest -o addopts= --noconftest -q examples/realtime/test_push_to_talk_app.py - examples: timeout-minutes: 10 name: examples diff --git a/examples/realtime/test-requirements.txt b/examples/realtime/test-requirements.txt deleted file mode 100644 index c9e3a134c7..0000000000 --- a/examples/realtime/test-requirements.txt +++ /dev/null @@ -1,83 +0,0 @@ -audioop-lts==0.2.2 ; python_full_version >= '3.13' \ - --hash=sha256:0337d658f9b81f4cd0fdb1f47635070cc084871a3d4646d9de74fdf4e7c3d24a \ - --hash=sha256:03f061a1915538fd96272bac9551841859dbb2e3bf73ebe4a23ef043766f5449 \ - --hash=sha256:068aa17a38b4e0e7de771c62c60bbca2455924b67a8814f3b0dee92b5820c0b3 \ - --hash=sha256:088327f00488cdeed296edd9215ca159f3a5a5034741465789cad403fcf4bec0 \ - --hash=sha256:0d9385e96f9f6da847f4d571ce3cb15b5091140edf3db97276872647ce37efd7 \ - --hash=sha256:106753a83a25ee4d6f473f2be6b0966fc1c9af7e0017192f5531a3e7463dce58 \ - --hash=sha256:143fad0311e8209ece30a8dbddab3b65ab419cbe8c0dde6e8828da25999be911 \ - --hash=sha256:15ab25dd3e620790f40e9ead897f91e79c0d3ce65fe193c8ed6c26cffdd24be7 \ - --hash=sha256:167d3b62586faef8b6b2275c3218796b12621a60e43f7e9d5845d627b9c9b80e \ - --hash=sha256:2b267b70747d82125f1a021506565bdc5609a2b24bcb4773c16d79d2bb260bbd \ - --hash=sha256:3bcddaaf6cc5935a300a8387c99f7a7fbbe212a11568ec6cf6e4bc458c048636 \ - --hash=sha256:3fc38008969796f0f689f1453722a0f463da1b8a6fbee11987830bfbb664f623 \ - --hash=sha256:47eba38322370347b1c47024defbd36374a211e8dd5b0dcbce7b34fdb6f8847b \ - --hash=sha256:48159d96962674eccdca9a3df280e864e8ac75e40a577cc97c5c42667ffabfc5 \ - --hash=sha256:49ee1a41738a23e98d98b937a0638357a2477bc99e61b0f768a8f654f45d9b7a \ - --hash=sha256:4a53aa7c16a60a6857e6b0b165261436396ef7293f8b5c9c828a3a203147ed4a \ - --hash=sha256:4b4cd51a57b698b2d06cb9993b7ac8dfe89a3b2878e96bc7948e9f19ff51dba6 \ - --hash=sha256:51c916108c56aa6e426ce611946f901badac950ee2ddaf302b7ed35d9958970d \ - --hash=sha256:550c114a8df0aafe9a05442a1162dfc8fec37e9af1d625ae6060fed6e756f303 \ - --hash=sha256:58cf54380c3884fb49fdd37dfb7a772632b6701d28edd3e2904743c5e1773602 \ - --hash=sha256:5b00be98ccd0fc123dcfad31d50030d25fcf31488cde9e61692029cd7394733b \ - --hash=sha256:5f93a5db13927a37d2d09637ccca4b2b6b48c19cd9eda7b17a2e9f77edee6a6f \ - --hash=sha256:64d0c62d88e67b98a1a5e71987b7aa7b5bcffc7dcee65b635823dbdd0a8dbbd0 \ - --hash=sha256:73f80bf4cd5d2ca7814da30a120de1f9408ee0619cc75da87d0641273d202a09 \ - --hash=sha256:752d76472d9804ac60f0078c79cdae8b956f293177acd2316cd1e15149aee132 \ - --hash=sha256:83c381767e2cc10e93e40281a04852facc4cd9334550e0f392f72d1c0a9c5753 \ - --hash=sha256:8fefe5868cd082db1186f2837d64cfbfa78b548ea0d0543e9b28935ccce81ce9 \ - --hash=sha256:9191d68659eda01e448188f60364c7763a7ca6653ed3f87ebb165822153a8547 \ - --hash=sha256:96f19de485a2925314f5020e85911fb447ff5fbef56e8c7c6927851b95533a1c \ - --hash=sha256:9a13dc409f2564de15dd68be65b462ba0dde01b19663720c68c1140c782d1d75 \ - --hash=sha256:a2c2a947fae7d1062ef08c4e369e0ba2086049a5e598fda41122535557012e9e \ - --hash=sha256:a2d4f1513d63c795e82948e1305f31a6d530626e5f9f2605408b300ae6095093 \ - --hash=sha256:a5bf613e96f49712073de86f20dbdd4014ca18efd4d34ed18c75bd808337851b \ - --hash=sha256:a6d2e0f9f7a69403e388894d4ca5ada5c47230716a03f2847cfc7bd1ecb589d6 \ - --hash=sha256:b492c3b040153e68b9fdaff5913305aaaba5bb433d8a7f73d5cf6a64ed3cc1dd \ - --hash=sha256:ba7c3a7e5f23e215cb271516197030c32aef2e754252c4c70a50aaff7031a2c8 \ - --hash=sha256:c0022283e9556e0f3643b7c3c03f05063ca72b3063291834cca43234f20c60bb \ - --hash=sha256:c174e322bb5783c099aaf87faeb240c8d210686b04bd61dfd05a8e5a83d88969 \ - --hash=sha256:c9c8e68d8b4a56fda8c025e538e639f8c5953f5073886b596c93ec9b620055e7 \ - --hash=sha256:cfcac6aa6f42397471e4943e0feb2244549db5c5d01efcd02725b96af417f3fe \ - --hash=sha256:d5e73fa573e273e4f2e5ff96f9043858a5e9311e94ffefd88a3186a910c70917 \ - --hash=sha256:def246fe9e180626731b26e89816e79aae2276f825420a07b4a647abaa84becc \ - --hash=sha256:dfbbc74ec68a0fd08cfec1f4b5e8cca3d3cd7de5501b01c4b5d209995033cde9 \ - --hash=sha256:e160bf9df356d841bb6c180eeeea1834085464626dc1b68fa4e1d59070affdc3 \ - --hash=sha256:e541c3ef484852ef36545f66209444c48b28661e864ccadb29daddb6a4b8e5f5 \ - --hash=sha256:f9b0b8a03ef474f56d1a842af1a2e01398b8f7654009823c6d9e0ecff4d5cfbf \ - --hash=sha256:f9ee9b52f5f857fbaf9d605a360884f034c92c1c23021fb90b2e39b8e64bede6 \ - --hash=sha256:fbdd522624141e40948ab3e8cdae6e04c748d78710e9f0f8d4dae2750831de19 \ - --hash=sha256:fd3d4602dc64914d462924a08c1a9816435a2155d74f325853c1f1ac3b2d9800 -linkify-it-py==2.2.0 \ - --hash=sha256:3adc40eb5af300b2605fcfdb968c24e1d780a90f1f2221af7c15e5111e94d443 \ - --hash=sha256:907acd2d17ac1fbb9ddb62c8957ccbd6158cac602231a15c3b0cd1e215f03cee -markdown-it-py==3.0.0 \ - --hash=sha256:355216845c60bd96232cd8d8c40e8f9765cc86f46880e43a8fd22dc1a1a8cab1 \ - --hash=sha256:e3f60a94fa066dc52ec76661e37c851cb232d92f9886b15cb560aaada2df8feb -mdit-py-plugins==0.6.1 \ - --hash=sha256:214c82fb2ac524472ab6a5bcab1de80f73b50443e187f401bfd77efbc7c6481d \ - --hash=sha256:a2bca0f039f39dbd35fb74ae1b5f998608c437463371f0ff7f49a19a17a114d0 -mdurl==0.1.2 \ - --hash=sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8 \ - --hash=sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba -platformdirs==4.12.4 ; python_full_version < '3.11' \ - --hash=sha256:63743c02414e755de4e31b8f68125c1407495b86c5a006e203c01ff8b9924250 \ - --hash=sha256:78bfb9db2a8471ed7eebe3c3c932da413911042994e699b384fbb4493fa872d7 -platformdirs==4.13.5 ; python_full_version >= '3.11' \ - --hash=sha256:05831adf87d392760f4b5c3a0a95579ab5a10b18b9b9bd43680dcb47544788a0 \ - --hash=sha256:53d3ec4280b815c1e0b02edad7f9ee8fda1ba9c0865dd9c2e6e63849ce82fc1e -pydub==0.25.1 \ - --hash=sha256:65617e33033874b59d87db603aa1ed450633288aefead953b30bded59cb599a6 \ - --hash=sha256:980a33ce9949cab2a569606b65674d748ecbca4f0796887fd6f46173a7b0d30f -pygments==2.20.0 \ - --hash=sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f \ - --hash=sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176 -rich==15.0.0 \ - --hash=sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb \ - --hash=sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36 -textual==8.2.8 \ - --hash=sha256:267375fd402dc8d981457212efa71f0e3365fd17bba144ba9bb3ed7563cb374a \ - --hash=sha256:3f106a9fbc73e39dd266c9712432087de78a6d644084c7c241d6a25c3169115b -typing-extensions==4.16.0 \ - --hash=sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8 \ - --hash=sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5 From 78946a119f6e57f70c75b0f78a12b3db76cf2280 Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 15:46:40 +0000 Subject: [PATCH 14/19] fix(realtime): truncate unheard audio discovered after interruption --- examples/realtime/push_to_talk_app.py | 32 ++++-- examples/realtime/test_push_to_talk_app.py | 108 ++++++++++++++++++++- 2 files changed, 129 insertions(+), 11 deletions(-) diff --git a/examples/realtime/push_to_talk_app.py b/examples/realtime/push_to_talk_app.py index 6156be977b..ab824f1751 100755 --- a/examples/realtime/push_to_talk_app.py +++ b/examples/realtime/push_to_talk_app.py @@ -143,7 +143,7 @@ def __init__(self) -> None: self.audio_player = AudioPlayerAsync() self._active_response_id: str | None = None self._displayed_item: tuple[str, str, int] | None = None - self._interrupted_response_ids: set[str] = set() + self._interrupted_responses: dict[str, set[tuple[str, int]]] = {} self.should_send_audio = asyncio.Event() self._mic_send_lock = asyncio.Lock() self._awaiting_new_input = False @@ -220,7 +220,7 @@ async def _receive_realtime_events(self) -> None: # interrupted playback, so this response belongs to an # earlier turn (even if K was already released). self._active_response_id = None - self._interrupted_response_ids.add(event.response.id) + self._interrupted_responses[event.response.id] = set() await conn.response.cancel(response_id=event.response.id) continue self._active_response_id = event.response.id @@ -230,7 +230,7 @@ async def _receive_realtime_events(self) -> None: if self._active_response_id == event.response.id: self._active_response_id = None if event.response.id is not None: - self._interrupted_response_ids.discard(event.response.id) + self._interrupted_responses.pop(event.response.id, None) self.audio_player.finish_response(event.response.id) if event.response.status in {"failed", "incomplete"}: details = event.response.status_details @@ -240,9 +240,22 @@ async def _receive_realtime_events(self) -> None: self.query_one(RichLog).write(f"{text}: {reason}" if reason else text) continue - if event.type == "response.output_audio.delta": - if event.response_id in self._interrupted_response_ids: + if event.type in {"response.output_audio.delta", "response.output_audio_transcript.delta"}: + handled = self._interrupted_responses.get(event.response_id) + if handled is not None: + part = (event.item_id, event.content_index) + if part not in handled: + # This item arrived after playback was stopped; none of it was heard. + # Reserve it before the worker runs so later chunks cannot truncate it twice. + handled.add(part) + self.run_worker( + self._interrupt_response( + None, [((event.response_id, event.item_id, event.content_index), 0)] + ) + ) continue + + if event.type == "response.output_audio.delta": bytes_data = base64.b64decode(event.delta) self.audio_player.add_data( bytes_data, @@ -253,8 +266,6 @@ async def _receive_realtime_events(self) -> None: continue if event.type == "response.output_audio_transcript.delta": - if event.response_id in self._interrupted_response_ids: - continue try: text = acc_items[event.item_id] except KeyError: @@ -360,7 +371,12 @@ async def on_key(self, event: events.Key) -> None: self._active_response_id = None unheard = self.audio_player.stop(interrupted_response_id=response_id) if response_id is not None: - self._interrupted_response_ids.add(response_id) + # Preserve the cutoff of partially played items, even while cancellation is pending. + self._interrupted_responses[response_id] = { + (item_id, content_index) + for (owner, item_id, content_index), _ in unheard + if owner == response_id + } if any(key == self._displayed_item for key, _ in unheard): bottom_pane = self.query_one("#bottom-pane", RichLog) bottom_pane.clear() diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py index 5083e1ac4e..c30941b539 100644 --- a/examples/realtime/test_push_to_talk_app.py +++ b/examples/realtime/test_push_to_talk_app.py @@ -562,7 +562,7 @@ async def __aiter__(self) -> Any: delta="first", ) assert display == ["first"] - app._interrupted_response_ids.add("old") + app._interrupted_responses["old"] = {("old", 0)} yield SimpleNamespace( type="response.output_audio_transcript.delta", response_id="old", @@ -625,10 +625,14 @@ async def cancel(**kwargs: Any) -> None: async def append(**_kwargs: Any) -> None: sent.append(("append", None)) + async def truncate(**kwargs: Any) -> None: + sent.append(("truncate", kwargs)) + class Connection: session = SimpleNamespace(update=update) response = SimpleNamespace(cancel=cancel) input_audio_buffer = SimpleNamespace(append=append) + conversation = SimpleNamespace(item=SimpleNamespace(truncate=truncate)) async def __aenter__(self) -> Any: return self @@ -667,6 +671,7 @@ async def __aiter__(self) -> Any: content_index=0, delta="a response", ) + await app.workers.wait_for_complete() if new_audio_sent: # Server VAD may legitimately answer a pause during the current recording. assert sent == [("append", None)] @@ -674,14 +679,17 @@ async def __aiter__(self) -> Any: assert display == ["a response"] else: # No new audio has been offered, so this can only be an older response. - assert sent == [("cancel", {"response_id": "response"})] + assert sent == [ + ("cancel", {"response_id": "response"}), + ("truncate", {"item_id": "item", "content_index": 0, "audio_end_ms": 0}), + ] assert not app.audio_player.playing assert display == [] yield SimpleNamespace( type="response.done", response=SimpleNamespace(id="response", status="completed", status_details=None), ) - assert not app._interrupted_response_ids + assert not app._interrupted_responses app.query_one = query_one original_client = app.client @@ -694,3 +702,97 @@ def connect(**_kwargs: Any) -> Connection: await app._receive_realtime_events() finally: await original_client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("response_before_k,played_frames", [(True, 0), (False, 0), (True, 480)]) +@pytest.mark.parametrize("first_delta", ["response.output_audio.delta", "response.output_audio_transcript.delta"]) +async def test_interruption_truncates_late_unheard_audio_once( + app_module: Any, response_before_k: bool, played_frames: int, first_delta: str +) -> None: + module, _ = app_module + app = module.RealtimeApp() + status = SimpleNamespace(is_recording=False) + display: list[str] = [] + sent: list[tuple[str, dict[str, Any]]] = [] + + class Log: + def clear(self) -> None: + display.clear() + + def write(self, message: str) -> None: + display.append(message) + + def query_one(*args: Any, **_kwargs: Any) -> Any: + return status if args[0] is module.AudioStatusIndicator else Log() + + async def update(**_kwargs: Any) -> None: + pass + + async def cancel(**kwargs: Any) -> None: + sent.append(("cancel", kwargs)) + + async def truncate(**kwargs: Any) -> None: + sent.append(("truncate", kwargs)) + + def delta(kind: str) -> SimpleNamespace: + return SimpleNamespace( + type=kind, + response_id="old-response", + item_id="assistant-item", + content_index=0, + delta="AAAAAAAAAAA=" if kind == "response.output_audio.delta" else "unheard text", + ) + + class Connection: + session = SimpleNamespace(update=update) + response = SimpleNamespace(cancel=cancel) + conversation = SimpleNamespace(item=SimpleNamespace(truncate=truncate)) + + async def __aenter__(self) -> Any: + return self + + async def __aexit__(self, *_args: Any) -> None: + pass + + async def __aiter__(self) -> Any: + created = SimpleNamespace(type="response.created", response=SimpleNamespace(id="old-response")) + if response_before_k: + yield created + if played_frames: + app.audio_player.add_data( + bytes(1920), response_id="old-response", item_id="assistant-item", content_index=0 + ) + app.audio_player.stream.callback( + np.empty((played_frames, 1), dtype=np.int16), played_frames, None, None + ) + await app.on_key(SimpleNamespace(key="k")) + if not response_before_k: + yield created + # Model multiple in-flight chunks, with either transcript or audio arriving first. + for kind in (first_delta, "response.output_audio.delta", "response.output_audio_transcript.delta"): + yield delta(kind) + await app.workers.wait_for_complete() + assert not app.audio_player.playing + assert display == [] + assert sent == [ + ("cancel", {"response_id": "old-response"}), + ("truncate", {"item_id": "assistant-item", "content_index": 0, "audio_end_ms": played_frames // 24}), + ] + yield SimpleNamespace( + type="response.done", + response=SimpleNamespace(id="old-response", status="cancelled", status_details=None), + ) + + app.query_one = query_one + original_client = app.client + + def connect(**_kwargs: Any) -> Connection: + return Connection() + + app.client = SimpleNamespace(realtime=SimpleNamespace(connect=connect)) + try: + await app._receive_realtime_events() + finally: + app._stop_mic_stream() + await original_client.close() From 69712bbafabd6996b135e65861791d69135e6c41 Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 16:05:57 +0000 Subject: [PATCH 15/19] fix(realtime): order delayed cancellation before mic sends --- examples/realtime/push_to_talk_app.py | 2 +- examples/realtime/test_push_to_talk_app.py | 85 ++++++++++++++++++++++ 2 files changed, 86 insertions(+), 1 deletion(-) diff --git a/examples/realtime/push_to_talk_app.py b/examples/realtime/push_to_talk_app.py index ab824f1751..f2eda46474 100755 --- a/examples/realtime/push_to_talk_app.py +++ b/examples/realtime/push_to_talk_app.py @@ -221,7 +221,7 @@ async def _receive_realtime_events(self) -> None: # earlier turn (even if K was already released). self._active_response_id = None self._interrupted_responses[event.response.id] = set() - await conn.response.cancel(response_id=event.response.id) + await self._interrupt_response(event.response.id, []) continue self._active_response_id = event.response.id continue diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py index c30941b539..8bcf72c700 100644 --- a/examples/realtime/test_push_to_talk_app.py +++ b/examples/realtime/test_push_to_talk_app.py @@ -796,3 +796,88 @@ def connect(**_kwargs: Any) -> Connection: finally: app._stop_mic_stream() await original_client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("manual_turn", [False, True]) +async def test_delayed_response_cancel_finishes_before_new_mic_audio(app_module: Any, manual_turn: bool) -> None: + module, mic_started = app_module + app = module.RealtimeApp() + status = SimpleNamespace(is_recording=False) + + def query_one(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + return status + + app.query_one = query_one + if manual_turn: + app.session = SimpleNamespace(audio=SimpleNamespace(input=SimpleNamespace(turn_detection=None))) + waiting = asyncio.Event() + release = asyncio.Event() + appended = asyncio.Event() + reception_done = asyncio.Event() + sent: list[str] = [] + + async def update(**_kwargs: Any) -> None: + pass + + async def cancel(**_kwargs: Any) -> None: + waiting.set() + await release.wait() + sent.append("cancel") + + async def append(**_kwargs: Any) -> None: + sent.append("append") + appended.set() + + async def commit() -> None: + sent.append("commit") + + async def create() -> None: + sent.append("create") + + class Connection: + session = SimpleNamespace(update=update) + response = SimpleNamespace(cancel=cancel, create=create) + input_audio_buffer = SimpleNamespace(append=append, commit=commit) + + async def __aenter__(self) -> Any: + return self + + async def __aexit__(self, *_args: Any) -> None: + pass + + async def __aiter__(self) -> Any: + await app.on_key(SimpleNamespace(key="k")) + yield SimpleNamespace(type="response.created", response=SimpleNamespace(id="old-response")) + await reception_done.wait() + + def connect(**_kwargs: Any) -> Connection: + return Connection() + + original_client = app.client + app.client = SimpleNamespace(realtime=SimpleNamespace(connect=connect)) + receiver = asyncio.create_task(app._receive_realtime_events()) + mic: asyncio.Task[None] | None = None + try: + await asyncio.wait_for(waiting.wait(), 2) + await asyncio.wait_for(mic_started.wait(), 2) + mic = asyncio.create_task(app.send_mic_audio()) + for _ in range(100): + if appended.is_set(): + break + await asyncio.sleep(0) + assert status.is_recording + assert sent == [] + release.set() + await asyncio.wait_for(appended.wait(), 2) + await app.on_key(SimpleNamespace(key="k")) + await app.workers.wait_for_complete() + assert sent == ["cancel", "append"] + (["commit", "create"] if manual_turn else []) + finally: + release.set() + reception_done.set() + if mic is not None: + mic.cancel() + receiver.cancel() + await asyncio.gather(receiver, *([mic] if mic is not None else []), return_exceptions=True) + await original_client.close() From fc69da10934788c200fe489b2de32960f7f207bb Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 16:21:29 +0000 Subject: [PATCH 16/19] fix(realtime): prioritize interruptions already queued during microphone sends --- examples/realtime/push_to_talk_app.py | 87 ++++++++++------ examples/realtime/test_push_to_talk_app.py | 110 +++++++++++++++++++-- 2 files changed, 160 insertions(+), 37 deletions(-) diff --git a/examples/realtime/push_to_talk_app.py b/examples/realtime/push_to_talk_app.py index f2eda46474..4209576568 100755 --- a/examples/realtime/push_to_talk_app.py +++ b/examples/realtime/push_to_talk_app.py @@ -31,6 +31,7 @@ import base64 import asyncio from typing import Any, cast +from functools import partial from dataclasses import dataclass from typing_extensions import override @@ -146,6 +147,7 @@ def __init__(self) -> None: self._interrupted_responses: dict[str, set[tuple[str, int]]] = {} self.should_send_audio = asyncio.Event() self._mic_send_lock = asyncio.Lock() + self._pending_interruptions = 0 self._awaiting_new_input = False self.connected = asyncio.Event() self._mic_stream: Any = None @@ -249,9 +251,12 @@ async def _receive_realtime_events(self) -> None: # Reserve it before the worker runs so later chunks cannot truncate it twice. handled.add(part) self.run_worker( - self._interrupt_response( - None, [((event.response_id, event.item_id, event.content_index), 0)] - ) + partial( + self._interrupt_response, + None, + [((event.response_id, event.item_id, event.content_index), 0)], + ), + group="realtime-interruption", ) continue @@ -307,18 +312,20 @@ async def send_mic_audio(self) -> None: async with self._mic_send_lock: if stream is not self._mic_stream or not self.should_send_audio.is_set(): continue - data, _ = stream.read(read_size) - self._awaiting_new_input = False - try: - await connection.input_audio_buffer.append( - audio=base64.b64encode(cast(Any, data)).decode("utf-8") - ) - recording.has_audio = True - except BaseException: - if recording is self._recording: - self._awaiting_new_input = True - raise + if not self._interruptions_pending(): + data, _ = stream.read(read_size) + self._awaiting_new_input = False + try: + await connection.input_audio_buffer.append( + audio=base64.b64encode(cast(Any, data)).decode("utf-8") + ) + recording.has_audio = True + except BaseException: + if recording is self._recording: + self._awaiting_new_input = True + raise + # Give any just-registered interruption worker time to start before retrying. await asyncio.sleep(0) except KeyboardInterrupt: pass @@ -385,31 +392,49 @@ async def on_key(self, event: events.Key) -> None: # Queue ordering work before waking the mic sender. Textual owns # worker cancellation on exit; no WebSocket writes hold the UI. if response_id is not None or unheard: - self.run_worker(self._interrupt_response(response_id, unheard)) + self.run_worker( + partial(self._interrupt_response, response_id, unheard), group="realtime-interruption" + ) self.should_send_audio.set() + def _interruptions_pending(self) -> bool: + # Inline receiver work reserves before taking the lock. Textual registers + # workers immediately and removes them even when cancelled before starting. + return self._pending_interruptions > 0 or any( + worker.group == "realtime-interruption" and not worker.is_finished for worker in self.workers + ) + async def _finish_manual_recording(self, recording: Recording) -> None: conn = await self._get_connection() - async with self._mic_send_lock: - if recording.has_audio: - await conn.input_audio_buffer.commit() - await conn.response.create() + while True: + async with self._mic_send_lock: + if not self._interruptions_pending(): + if recording.has_audio: + await conn.input_audio_buffer.commit() + await conn.response.create() + return + await asyncio.sleep(0) async def _interrupt_response( self, response_id: str | None, unheard: list[tuple[tuple[str, str, int], int]] ) -> None: - async with self._mic_send_lock: - if self.connection is None: - return - if response_id is not None: - await self.connection.response.cancel(response_id=response_id) - for (_, item_id, content_index), frames in unheard: - if content_index != 0: - self.query_one(RichLog).write("Cannot truncate audio: the API only supports content part 0.") - continue - await self.connection.conversation.item.truncate( - item_id=item_id, content_index=content_index, audio_end_ms=frames * 1000 // SAMPLE_RATE - ) + # Reserve before waiting: an older mic waiter must yield to new interruption work. + self._pending_interruptions += 1 + try: + async with self._mic_send_lock: + if self.connection is None: + return + if response_id is not None: + await self.connection.response.cancel(response_id=response_id) + for (_, item_id, content_index), frames in unheard: + if content_index != 0: + self.query_one(RichLog).write("Cannot truncate audio: the API only supports content part 0.") + continue + await self.connection.conversation.item.truncate( + item_id=item_id, content_index=content_index, audio_end_ms=frames * 1000 // SAMPLE_RATE + ) + finally: + self._pending_interruptions -= 1 if __name__ == "__main__": diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py index 8bcf72c700..77bbb33838 100644 --- a/examples/realtime/test_push_to_talk_app.py +++ b/examples/realtime/test_push_to_talk_app.py @@ -800,7 +800,10 @@ def connect(**_kwargs: Any) -> Connection: @pytest.mark.asyncio @pytest.mark.parametrize("manual_turn", [False, True]) -async def test_delayed_response_cancel_finishes_before_new_mic_audio(app_module: Any, manual_turn: bool) -> None: +@pytest.mark.parametrize("contended", [False, True]) +async def test_delayed_response_cancel_finishes_before_new_mic_audio( + app_module: Any, manual_turn: bool, contended: bool +) -> None: module, mic_started = app_module app = module.RealtimeApp() status = SimpleNamespace(is_recording=False) @@ -815,15 +818,18 @@ def query_one(*_args: Any, **_kwargs: Any) -> SimpleNamespace: release = asyncio.Event() appended = asyncio.Event() reception_done = asyncio.Event() + delayed_may_arrive = asyncio.Event() + delayed_delivered = asyncio.Event() sent: list[str] = [] async def update(**_kwargs: Any) -> None: pass - async def cancel(**_kwargs: Any) -> None: - waiting.set() - await release.wait() - sent.append("cancel") + async def cancel(*, response_id: str) -> None: + if not contended or response_id == "active-response": + waiting.set() + await release.wait() + sent.append(f"cancel:{response_id}") async def append(**_kwargs: Any) -> None: sent.append("append") @@ -847,7 +853,16 @@ async def __aexit__(self, *_args: Any) -> None: pass async def __aiter__(self) -> Any: + if contended: + yield SimpleNamespace(type="response.created", response=SimpleNamespace(id="active-response")) await app.on_key(SimpleNamespace(key="k")) + if contended: + await delayed_may_arrive.wait() + # The server has completed the interrupted response before starting another. + yield SimpleNamespace( + type="response.done", response=SimpleNamespace(id="active-response", status="completed") + ) + delayed_delivered.set() yield SimpleNamespace(type="response.created", response=SimpleNamespace(id="old-response")) await reception_done.wait() @@ -868,11 +883,19 @@ def connect(**_kwargs: Any) -> Connection: await asyncio.sleep(0) assert status.is_recording assert sent == [] + if contended: + # The microphone already waits behind A when the receiver learns of B. + delayed_may_arrive.set() + await asyncio.wait_for(delayed_delivered.wait(), 2) release.set() await asyncio.wait_for(appended.wait(), 2) await app.on_key(SimpleNamespace(key="k")) await app.workers.wait_for_complete() - assert sent == ["cancel", "append"] + (["commit", "create"] if manual_turn else []) + assert sent == ( + (["cancel:active-response"] if contended else []) + + ["cancel:old-response", "append"] + + (["commit", "create"] if manual_turn else []) + ) finally: release.set() reception_done.set() @@ -881,3 +904,78 @@ def connect(**_kwargs: Any) -> Connection: receiver.cancel() await asyncio.gather(receiver, *([mic] if mic is not None else []), return_exceptions=True) await original_client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("waiter", ["microphone", "manual"]) +@pytest.mark.parametrize("cancel_before_start", [False, True]) +async def test_queued_interruption_precedes_runnable_sender( + app_module: Any, waiter: str, cancel_before_start: bool +) -> None: + module, _ = app_module + app = module.RealtimeApp() + status = SimpleNamespace(is_recording=False) + sent: list[str] = [] + sent_first = asyncio.Event() + cancel_started = asyncio.Event() + release_cancel = asyncio.Event() + + def query_one(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + return status + + async def cancel(**_kwargs: Any) -> None: + cancel_started.set() + await release_cancel.wait() + sent.append("cancel") + + async def append(**_kwargs: Any) -> None: + sent.append("append") + sent_first.set() + + async def commit() -> None: + sent.append("commit") + sent_first.set() + + async def create() -> None: + sent.append("create") + + app.query_one = query_one + app.connection = SimpleNamespace( + response=SimpleNamespace(cancel=cancel, create=create), + input_audio_buffer=SimpleNamespace(append=append, commit=commit), + ) + app.connected.set() + await app.on_key(SimpleNamespace(key="k")) + await app._mic_send_lock.acquire() + if waiter == "microphone": + sender = asyncio.create_task(app.send_mic_audio()) + else: + # Audio from an earlier recording is already queued for manual completion. + sender = asyncio.create_task(app._finish_manual_recording(module.Recording(has_audio=True))) + try: + for _ in range(20): + await asyncio.sleep(0) + assert not sent + # Make the sender runnable, then register a K interruption before it gets CPU. + app._mic_send_lock.release() + await app.on_key(SimpleNamespace(key="k")) + app._active_response_id = "old-response" + await app.on_key(SimpleNamespace(key="k")) + if cancel_before_start: + app.workers.cancel_group(app, "realtime-interruption") + else: + await asyncio.wait_for(cancel_started.wait(), 2) + for _ in range(20): + await asyncio.sleep(0) + assert not sent + release_cancel.set() + await asyncio.wait_for(sent_first.wait(), 2) + expected = ([] if cancel_before_start else ["cancel"]) + ( + ["append"] if waiter == "microphone" else ["commit", "create"] + ) + assert sent == expected + finally: + release_cancel.set() + sender.cancel() + await asyncio.gather(sender, return_exceptions=True) + await app.client.close() From dd098453f4032812359d4a4821d44a09508db8a4 Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 17:33:36 +0000 Subject: [PATCH 17/19] fix(examples): finalize manual recording before sending the next --- examples/realtime/push_to_talk_app.py | 12 +- examples/realtime/test_push_to_talk_app.py | 138 +++++++++++++++++++++ 2 files changed, 147 insertions(+), 3 deletions(-) diff --git a/examples/realtime/push_to_talk_app.py b/examples/realtime/push_to_talk_app.py index 4209576568..84c6489b97 100755 --- a/examples/realtime/push_to_talk_app.py +++ b/examples/realtime/push_to_talk_app.py @@ -312,7 +312,9 @@ async def send_mic_audio(self) -> None: async with self._mic_send_lock: if stream is not self._mic_stream or not self.should_send_audio.is_set(): continue - if not self._interruptions_pending(): + if not self._interruptions_pending() and not any( + worker.group == "realtime-finalization" and not worker.is_finished for worker in self.workers + ): data, _ = stream.read(read_size) self._awaiting_new_input = False try: @@ -325,7 +327,7 @@ async def send_mic_audio(self) -> None: self._awaiting_new_input = True raise - # Give any just-registered interruption worker time to start before retrying. + # Let registered interruption or finalization work start before retrying. await asyncio.sleep(0) except KeyboardInterrupt: pass @@ -364,7 +366,11 @@ async def on_key(self, event: events.Key) -> None: # # However if we're in manual `turn_detection` mode then we need to # manually tell the model to commit the audio buffer and start responding. - self.run_worker(self._finish_manual_recording(self._recording)) + # Reserve the shared server buffer before the next K can send. + # Capture can start immediately; appends wait for this recording. + self.run_worker( + partial(self._finish_manual_recording, self._recording), group="realtime-finalization" + ) else: import sounddevice as sd # type: ignore diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py index 77bbb33838..81d3affee7 100644 --- a/examples/realtime/test_push_to_talk_app.py +++ b/examples/realtime/test_push_to_talk_app.py @@ -979,3 +979,141 @@ async def create() -> None: sender.cancel() await asyncio.gather(sender, return_exceptions=True) await app.client.close() + + +@pytest.mark.asyncio +async def test_manual_recordings_do_not_share_buffer_after_interruption( + app_module: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + import base64 + + module, _ = app_module + app = module.RealtimeApp() + status = SimpleNamespace(is_recording=False) + app.query_one = lambda *_args, **_kwargs: status + app.session = SimpleNamespace(audio=SimpleNamespace(input=SimpleNamespace(turn_detection=None))) + first_started, first_release = asyncio.Event(), asyncio.Event() + cancel_started, cancel_release = asyncio.Event(), asyncio.Event() + second_appended = asyncio.Event() + buffer: list[int] = [] + commits: list[list[int]] = [] + sent: list[str] = [] + + device = sys.modules["sounddevice"] + original_read = device.InputStream.read + + def read(stream: Any, frames: int) -> tuple[Any, bool]: + data, overflow = original_read(stream, frames) + data.fill(stream.marker) + return data, overflow + + monkeypatch.setattr(device.InputStream, "read", read) + + async def append(*, audio: str) -> None: + marker = int(np.frombuffer(base64.b64decode(audio), dtype=np.int16)[0]) + if marker == 1: + first_started.set() + await first_release.wait() + buffer.append(marker) + sent.append(f"append {marker}") + if marker == 2: + second_appended.set() + + async def cancel(**_kwargs: Any) -> None: + cancel_started.set() + await cancel_release.wait() + sent.append("cancel") + + async def commit() -> None: + commits.append(list(buffer)) + buffer.clear() + sent.append("commit") + + async def create() -> None: + sent.append("create") + + app.connection = SimpleNamespace( + response=SimpleNamespace(cancel=cancel, create=create), + input_audio_buffer=SimpleNamespace(append=append, commit=commit), + ) + app.connected.set() + mic = asyncio.create_task(app.send_mic_audio()) + try: + await app.on_key(SimpleNamespace(key="k")) + app._mic_stream.marker = 1 + await asyncio.wait_for(first_started.wait(), 2) + await app.on_key(SimpleNamespace(key="k")) + # The completed recording is waiting on its in-flight append; a new K + # begins capture and interrupts an earlier server response immediately. + app._active_response_id = "previous" + await app.on_key(SimpleNamespace(key="k")) + app._mic_stream.marker = 2 + assert status.is_recording + for _ in range(20): + await asyncio.sleep(0) + assert not sent + first_release.set() + await asyncio.wait_for(cancel_started.wait(), 2) + assert sent == ["append 1"] + assert status.is_recording + cancel_release.set() + await asyncio.wait_for(second_appended.wait(), 2) + await app.on_key(SimpleNamespace(key="k")) + await asyncio.wait_for(app.workers.wait_for_complete(), 2) + assert commits == [[1], [2]] + assert sent == ["append 1", "cancel", "commit", "create", "append 2", "commit", "create"] + finally: + first_release.set() + cancel_release.set() + mic.cancel() + await asyncio.gather(mic, return_exceptions=True) + await app.client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("cancel_before_start", [False, True]) +async def test_pending_manual_finalize_precedes_already_runnable_mic(app_module: Any, cancel_before_start: bool) -> None: + module, _ = app_module + app = module.RealtimeApp() + status = SimpleNamespace(is_recording=False) + app.query_one = lambda *_args, **_kwargs: status + app.session = SimpleNamespace(audio=SimpleNamespace(input=SimpleNamespace(turn_detection=None))) + sent: list[str] = [] + appended = asyncio.Event() + + async def append(**_kwargs: Any) -> None: + sent.append("append") + appended.set() + + async def commit() -> None: + sent.append("commit") + + async def create() -> None: + sent.append("create") + + app.connection = SimpleNamespace( + response=SimpleNamespace(create=create), + input_audio_buffer=SimpleNamespace(append=append, commit=commit), + ) + app.connected.set() + await app.on_key(SimpleNamespace(key="k")) + # Simulate audio already sent from the current recording. Hold the send lock + # until its waiter is runnable, then dispatch both keys without yielding. + app._recording.has_audio = True + await app._mic_send_lock.acquire() + mic = asyncio.create_task(app.send_mic_audio()) + try: + for _ in range(20): + await asyncio.sleep(0) + app._mic_send_lock.release() + await app.on_key(SimpleNamespace(key="k")) + await app.on_key(SimpleNamespace(key="k")) + if cancel_before_start: + app.workers.cancel_group(app, "realtime-finalization") + await asyncio.wait_for(appended.wait(), 2) + await asyncio.wait_for(app.workers.wait_for_complete(), 2) + assert sent == (["append"] if cancel_before_start else ["commit", "create", "append"]) + finally: + mic.cancel() + await asyncio.gather(mic, return_exceptions=True) + await app.client.close() From defb2228f8b4d93051e7dbc784d118e0b17c9c4d Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 17:37:03 +0000 Subject: [PATCH 18/19] fix(tests): type push-to-talk test UI lookups --- examples/realtime/test_push_to_talk_app.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py index 81d3affee7..41ea6be4d5 100644 --- a/examples/realtime/test_push_to_talk_app.py +++ b/examples/realtime/test_push_to_talk_app.py @@ -990,7 +990,10 @@ async def test_manual_recordings_do_not_share_buffer_after_interruption( module, _ = app_module app = module.RealtimeApp() status = SimpleNamespace(is_recording=False) - app.query_one = lambda *_args, **_kwargs: status + def query_one(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + return status + + app.query_one = query_one app.session = SimpleNamespace(audio=SimpleNamespace(input=SimpleNamespace(turn_detection=None))) first_started, first_release = asyncio.Event(), asyncio.Event() cancel_started, cancel_release = asyncio.Event(), asyncio.Event() @@ -1076,7 +1079,10 @@ async def test_pending_manual_finalize_precedes_already_runnable_mic(app_module: module, _ = app_module app = module.RealtimeApp() status = SimpleNamespace(is_recording=False) - app.query_one = lambda *_args, **_kwargs: status + def query_one(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + return status + + app.query_one = query_one app.session = SimpleNamespace(audio=SimpleNamespace(input=SimpleNamespace(turn_detection=None))) sent: list[str] = [] appended = asyncio.Event() From 5801a1abe797080c19fee5658df6df939abd4737 Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 17:43:46 +0000 Subject: [PATCH 19/19] fix(examples): stop playback before accounting for unheard audio --- examples/realtime/audio_util.py | 13 ++++-- examples/realtime/test_push_to_talk_app.py | 54 ++++++++++++++++++++++ 2 files changed, 62 insertions(+), 5 deletions(-) diff --git a/examples/realtime/audio_util.py b/examples/realtime/audio_util.py index 1c6e3db817..92219139dd 100644 --- a/examples/realtime/audio_util.py +++ b/examples/realtime/audio_util.py @@ -113,11 +113,16 @@ def start(self): def stop(self, *, interrupted_response_id: str | None = None): with self.lock: - # PortAudio supplies both times on the same clock. Exclude later callbacks - # before aborting; abort must not hold the lock a callback may be waiting on. - now = self.stream.time was_playing = self.playing self.playing = False + # Silence new callbacks before stopping the device; abort must not hold + # the lock a callback may need. Do accounting only after stopping sound. + # PortAudio's clock advances while stopped, so use the last known playing + # time immediately before abort, not the time after it returns. + now = self.stream.time + if was_playing: + self.stream.abort() + with self.lock: self._discard_finished_counts(now) pending = dict.fromkeys(key for _, key in self.queue) pending.update(dict.fromkeys(key for key, _, _, _ in self._scheduled)) @@ -131,8 +136,6 @@ def stop(self, *, interrupted_response_id: str | None = None): self._scheduled.clear() self._finished_responses.clear() self._frame_count = 0 - if was_playing: - self.stream.abort() return list(heard.items()) def terminate(self): diff --git a/examples/realtime/test_push_to_talk_app.py b/examples/realtime/test_push_to_talk_app.py index 41ea6be4d5..af092c8b30 100644 --- a/examples/realtime/test_push_to_talk_app.py +++ b/examples/realtime/test_push_to_talk_app.py @@ -1123,3 +1123,57 @@ async def create() -> None: mic.cancel() await asyncio.gather(mic, return_exceptions=True) await app.client.close() + + +@pytest.mark.parametrize("finish_response", [False, True]) +@pytest.mark.parametrize("abort_return_delay", [0.0, 0.08]) +def test_cutoff_accounts_after_stopping_without_counting_silence( + app_module: Any, monkeypatch: pytest.MonkeyPatch, finish_response: bool, abort_return_delay: float +) -> None: + import threading + + module, _ = app_module + player = module.AudioPlayerAsync() + key = ("response", "spoken", 0) + player.add_data(bytes(1920), response_id=key[0], item_id=key[1], content_index=key[2]) + player.stream.time = 0.98 + player.stream.callback( + np.empty((960, 1), dtype=np.int16), + 960, + SimpleNamespace(outputBufferDacTime=1.0, currentTime=0.98), + None, + ) + if finish_response: + player.finish_response(key[0]) + player.stream.time = 1.01 + stopped_at: float | None = None + original_abort = player.stream.abort + original_discard = player._discard_finished_counts + + def account(now: float) -> None: + # Accounting a large queue takes time. Until abort, queued DAC audio + # continues playing even though future callbacks have been silenced. + player.stream.time += 0.02 + original_discard(now) + + def abort() -> None: + nonlocal stopped_at + # The driver can stop playback before its abort method returns. The + # PortAudio stream clock continues afterward, even with no sound playing. + stopped_at = player.stream.time + out = np.full((480, 1), 99, dtype=np.int16) + callback = threading.Thread(target=player.stream.callback, args=(out, 480, None, None), daemon=True) + callback.start() + callback.join(1) + assert not callback.is_alive(), "abort held the lock required by the device callback" + assert not np.any(out) + player.stream.time += abort_return_delay + original_abort() + + monkeypatch.setattr(player, "_discard_finished_counts", account) + monkeypatch.setattr(player.stream, "abort", abort) + result = player.stop(interrupted_response_id=None if finish_response else key[0]) + assert stopped_at is not None + heard_frames = min(960, int((stopped_at - 1.0) * module.SAMPLE_RATE)) + expected = [] if finish_response and heard_frames == 960 else [(key, heard_frames)] + assert result == expected