Skip to content

Commit 35fc9f3

Browse files
fix(examples): handle interruptions and disconnects in push-to-talk (#4192)
Pressing Enter tried to use a button the app never mounted, and transcripts containing Rich tags could crash the session. This change displays transcripts and API errors as literal text, reports failed or incomplete responses, and closes the audio device and SDK client on exit. Q exits with code 0; an unexpected normal or abnormal disconnect exits with code 1. When K interrupts playback, the sample cancels the active response and truncates unheard audio, including queued items after response.done. Capture starts before awaiting cancellation and truncation; microphone sends wait until those writes finish. If audio or its transcript arrives after interruption, the sample truncates each previously unknown item part to zero once. It retains the original cutoff for partly played audio. Manual mode skips commit and response.create if no audio was sent before release. Manual finalization also reserves the server input buffer before the next recording can append, so successive recordings remain separate while capture and interruption can still proceed. A response.created event that arrives after K, before any new audio is offered, belongs to the earlier turn. The sample reserves interruption work before it waits for the send lock. Microphone appends and manual commits recheck that work after obtaining the lock and yield outside it when necessary. They cannot overtake a cancellation or truncation that was queued while they waited. It suppresses late audio and transcripts and truncates unheard parts, even after a quick release. Once new audio is offered, these events cannot identify whether a delayed response belongs to an earlier turn or to a current pause detected by server VAD. The sample preserves possible current VAD replies; it does not guarantee interruption of every old response in that interval. All 48 committed sample regressions pass locally on Python 3.10 and 3.13 against the public SDK, and on Python 3.10 against the internal SDK. They use fake audio devices and held writes. Contended delayed cancellations reproduced wrong send ordering in both VAD and manual modes. Separate tests reproduce a runnable microphone or manual completion overtaking an interruption worker that has not started. The new regressions also verify progress when that worker is cancelled before starting. Four late-audio cases reproduced missing truncations. The tests also cover frame counts, queued playback, recording while cancellation is blocked, manual and VAD modes, transcripts and connection closure. Scoped Ruff and pinned Pyright checks pass. Eight earlier Textual scenarios exercised SDK connections to a local WebSocket peer and virtual audio devices. Run the sample regressions explicitly after installing the sample dependencies: ```sh python -m pytest -o addopts= --noconftest -q examples/realtime/test_push_to_talk_app.py ``` The default SDK test run does not collect these example tests. The first microphone append restores recording state and immediately re-raises any failure, including asyncio.CancelledError. Narrowing that cleanup to Exception would skip restoring _awaiting_new_input when asyncio cancels an append. The committed canceled-append regression tests that no event is sent, the microphone closes, and cancellation propagates. Playback accounting uses PortAudio DAC timestamps to exclude audio still buffered on the device. Interrupting silences callbacks and aborts playback before processing the accounting queue. It measures the last known playing time immediately before abort; PortAudio's clock keeps advancing after playback stops and does not provide an exact hardware stop timestamp. No physical-device timing precision was tested; prolonged backpressure can fill the device buffer. Unsupported audio content parts are reported without sending an invalid truncate event. --------- Co-authored-by: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com>
1 parent 55fc500 commit 35fc9f3

3 files changed

Lines changed: 1424 additions & 59 deletions

File tree

‎examples/realtime/audio_util.py‎

Lines changed: 61 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -43,50 +43,100 @@ def __init__(self):
4343
)
4444
self.playing = False
4545
self._frame_count = 0
46+
self._played_by_item: dict[tuple[str, str, int], int] = {}
47+
# Item, scheduled DAC time, frames submitted before this span, span length.
48+
self._scheduled: list[tuple[tuple[str, str, int], float, int, int]] = []
49+
self._finished_responses: set[str] = set()
4650

4751
def callback(self, outdata, frames, time, status): # noqa
4852
with self.lock:
53+
if not self.playing:
54+
outdata.fill(0)
55+
return
4956
data = np.empty(0, dtype=np.int16)
5057

5158
# get next item from queue if there is still space in the buffer
5259
while len(data) < frames and len(self.queue) > 0:
53-
item = self.queue.pop(0)
60+
item, item_key = self.queue.pop(0)
5461
frames_needed = frames - len(data)
55-
data = np.concatenate((data, item[:frames_needed]))
62+
played = item[:frames_needed]
63+
before = self._played_by_item.get(item_key, 0)
64+
self._scheduled.append(
65+
(item_key, time.outputBufferDacTime + len(data) / SAMPLE_RATE, before, len(played))
66+
)
67+
data = np.concatenate((data, played))
68+
self._played_by_item[item_key] = before + len(played)
5669
if len(item) > frames_needed:
57-
self.queue.insert(0, item[frames_needed:])
70+
self.queue.insert(0, (item[frames_needed:], item_key))
5871

5972
self._frame_count += len(data)
73+
self._discard_finished_counts(time.currentTime)
6074

6175
# fill the rest of the frames with zeros if there is no more data
6276
if len(data) < frames:
6377
data = np.concatenate((data, np.zeros(frames - len(data), dtype=np.int16)))
6478

6579
outdata[:] = data.reshape(-1, 1)
6680

67-
def reset_frame_count(self):
68-
self._frame_count = 0
69-
7081
def get_frame_count(self):
7182
return self._frame_count
7283

73-
def add_data(self, data: bytes):
84+
def finish_response(self, response_id: str):
85+
with self.lock:
86+
self._finished_responses.add(response_id)
87+
self._discard_finished_counts(self.stream.time)
88+
89+
def _discard_finished_counts(self, now: float):
90+
self._scheduled = [span for span in self._scheduled if span[1] + span[3] / SAMPLE_RATE > now]
91+
pending = {key for _, key in self.queue}
92+
pending.update(key for key, _, _, _ in self._scheduled)
93+
for key in list(self._played_by_item):
94+
if key[0] in self._finished_responses and key not in pending:
95+
del self._played_by_item[key]
96+
self._finished_responses.intersection_update(key[0] for key in pending)
97+
98+
def note_transcript(self, response_id: str, item_id: str, content_index: int):
99+
with self.lock:
100+
self._played_by_item.setdefault((response_id, item_id, content_index), 0)
101+
102+
def add_data(self, data: bytes, *, response_id: str, item_id: str, content_index: int):
74103
with self.lock:
75104
# bytes is pcm16 single channel audio data, convert to numpy array
76105
np_data = np.frombuffer(data, dtype=np.int16)
77-
self.queue.append(np_data)
106+
self.queue.append((np_data, (response_id, item_id, content_index)))
78107
if not self.playing:
79108
self.start()
80109

81110
def start(self):
82111
self.playing = True
83112
self.stream.start()
84113

85-
def stop(self):
86-
self.playing = False
87-
self.stream.stop()
114+
def stop(self, *, interrupted_response_id: str | None = None):
115+
with self.lock:
116+
was_playing = self.playing
117+
self.playing = False
118+
# Silence new callbacks before stopping the device; abort must not hold
119+
# the lock a callback may need. Do accounting only after stopping sound.
120+
# PortAudio's clock advances while stopped, so use the last known playing
121+
# time immediately before abort, not the time after it returns.
122+
now = self.stream.time
123+
if was_playing:
124+
self.stream.abort()
88125
with self.lock:
126+
self._discard_finished_counts(now)
127+
pending = dict.fromkeys(key for _, key in self.queue)
128+
pending.update(dict.fromkeys(key for key, _, _, _ in self._scheduled))
129+
pending.update(dict.fromkeys(key for key in self._played_by_item if key[0] == interrupted_response_id))
130+
heard = {key: self._played_by_item.get(key, 0) for key in pending}
131+
for key, start, before, frames in self._scheduled:
132+
played = max(0, min(frames, int((now - start) * SAMPLE_RATE)))
133+
heard[key] = min(heard[key], before + played)
89134
self.queue = []
135+
self._played_by_item.clear()
136+
self._scheduled.clear()
137+
self._finished_responses.clear()
138+
self._frame_count = 0
139+
return list(heard.items())
90140

91141
def terminate(self):
92142
self.stream.close()

0 commit comments

Comments
 (0)