From 705a2ab633fa363850c1e23ccc91c51c3263aba2 Mon Sep 17 00:00:00 2001 From: fl0rianr <226492742+fl0rianr@users.noreply.github.com> Date: Thu, 2 Jul 2026 19:30:07 +0200 Subject: [PATCH] Fix flaky Whisper realtime transcription test Accumulate realtime transcription deltas instead of relying only on the completed event transcript field. Some backends/SDK event shapes can deliver text via delta events while completed marks the final state with an empty transcript, causing a false negative despite successful transcription. Also make realtime event field access robust for SDK objects and dict-like events. --- test/server_whisper.py | 181 ++++++++++++++++++++++++++++++++--------- 1 file changed, 143 insertions(+), 38 deletions(-) diff --git a/test/server_whisper.py b/test/server_whisper.py index dab87e8b5..d4a3a435a 100644 --- a/test/server_whisper.py +++ b/test/server_whisper.py @@ -381,13 +381,76 @@ def _get_pcm16_chunks(self): print(f"[INFO] Split into {len(chunks)} chunks") return chunks + def _event_payload(self, event): + """Return a best-effort dict representation of an SDK realtime event.""" + if isinstance(event, dict): + return event + + for dump_method in ("model_dump", "dict"): + method = getattr(event, dump_method, None) + if method is None: + continue + try: + payload = method() + except Exception: + continue + if isinstance(payload, dict): + return payload + + return {} + + def _event_value(self, event, *field_names): + """Read a field from either an SDK event object or a plain dict.""" + for field_name in field_names: + if isinstance(event, dict): + value = event.get(field_name) + else: + value = getattr(event, field_name, None) + if value is not None: + return value + + payload = self._event_payload(event) + for field_name in field_names: + value = payload.get(field_name) + if value is not None: + return value + + return None + + def _event_type(self, event): + """Return the realtime event type for SDK event objects or dicts.""" + return self._event_value(event, "type") or "" + + def _event_text(self, event, *field_names): + """Extract the first non-empty text field from a realtime event.""" + payload = self._event_payload(event) + + for field_name in field_names: + values = [] + value = self._event_value(event, field_name) + if value is not None: + values.append(value) + if field_name in payload: + values.append(payload[field_name]) + + for value in values: + if isinstance(value, str): + if value.strip(): + return value + elif value is not None: + value = str(value) + if value.strip(): + return value + + return "" + async def _drain_realtime_events(self, conn, timeout_s=0.25): """Collect any pending realtime events until a short timeout expires.""" events = [] while True: try: event = await asyncio.wait_for(conn.recv(), timeout=timeout_s) - print(f"[INFO] Drained message: {event.type}") + print(f"[INFO] Drained message: {self._event_type(event)}") events.append(event) except asyncio.TimeoutError: break @@ -466,28 +529,64 @@ async def _test_007_realtime_websocket_transcription(self): print("[INFO] Committing audio buffer...") await conn.input_audio_buffer.commit() - # Collect messages until we get the transcription result - transcript = None - deadline = time.time() + TIMEOUT_MODEL_OPERATION - while time.time() < deadline: + # Collect messages until the server reports that transcription completed. + # Some SDK/server combinations expose interim text on delta events, while + # completed events may only signal completion and carry an empty transcript. + transcript_parts = [] + completed_transcript = "" + saw_completed = False + received_event_types = [] + deadline = time.monotonic() + TIMEOUT_MODEL_OPERATION + + while time.monotonic() < deadline: + timeout_s = max(0.1, min(30, deadline - time.monotonic())) try: - event = await asyncio.wait_for(conn.recv(), timeout=30) - print(f"[INFO] Received message: {event.type}") - - if ( - event.type - == "conversation.item.input_audio_transcription.completed" - ): - transcript = getattr(event, "transcript", "") - break + event = await asyncio.wait_for(conn.recv(), timeout=timeout_s) except asyncio.TimeoutError: break - self.assertIsNotNone(transcript, "Should receive a transcription result") + event_type = self._event_type(event) + received_event_types.append(event_type) + print(f"[INFO] Received message: {event_type}") + + if event_type == "error": + self.fail(f"Realtime transcription returned error: {event}") + + if ( + event_type + == "conversation.item.input_audio_transcription.delta" + ): + delta = self._event_text(event, "delta", "transcript", "text") + if delta: + transcript_parts.append(delta) + continue + + if ( + event_type + == "conversation.item.input_audio_transcription.completed" + ): + saw_completed = True + completed_transcript = self._event_text( + event, "transcript", "text", "delta" + ) + break + + transcript = (completed_transcript or "".join(transcript_parts)).strip() + + self.assertTrue( + saw_completed, + ( + "Should receive a transcription completion event; " + f"received events: {received_event_types}" + ), + ) self.assertGreater( - len(transcript.strip()), + len(transcript), 0, - "Transcription should not be empty", + ( + "Transcription should not be empty; " + f"received events: {received_event_types}" + ), ) print(f"[OK] WebSocket transcription result: {transcript}") @@ -527,31 +626,37 @@ async def _test_008_realtime_manual_commit(self): await conn.input_audio_buffer.commit() transcript = None - deadline = time.time() + TIMEOUT_MODEL_OPERATION - while time.time() < deadline: + deadline = time.monotonic() + TIMEOUT_MODEL_OPERATION + while time.monotonic() < deadline: + timeout_s = max(0.1, min(30, deadline - time.monotonic())) try: - event = await asyncio.wait_for(conn.recv(), timeout=30) - print(f"[INFO] Received message: {event.type}") - - self.assertNotIn( - event.type, - { - "input_audio_buffer.speech_started", - "input_audio_buffer.speech_stopped", - "conversation.item.input_audio_transcription.delta", - }, - "Manual commit mode should not emit VAD or interim events", - ) - - if ( - event.type - == "conversation.item.input_audio_transcription.completed" - ): - transcript = getattr(event, "transcript", "") - break + event = await asyncio.wait_for(conn.recv(), timeout=timeout_s) except asyncio.TimeoutError: break + event_type = self._event_type(event) + print(f"[INFO] Received message: {event_type}") + + if event_type == "error": + self.fail(f"Realtime transcription returned error: {event}") + + self.assertNotIn( + event_type, + { + "input_audio_buffer.speech_started", + "input_audio_buffer.speech_stopped", + "conversation.item.input_audio_transcription.delta", + }, + "Manual commit mode should not emit VAD or interim events", + ) + + if ( + event_type + == "conversation.item.input_audio_transcription.completed" + ): + transcript = self._event_text(event, "transcript", "text", "delta") + break + self.assertIsNotNone(transcript, "Should receive a transcription result") self.assertGreater( len(transcript.strip()),