Fix early wake-word chime timing

agent-sdk/xr-ai-voice/README.md

@@ -58,7 +58,7 @@ async with runtime:
 | `silence_duration` | `0.8` | Seconds of silence that finalize an utterance. |
 | `min_speech` | `0.15` | Minimum speech duration accepted as an utterance. |
 | `silero_threshold` | `0.5` | Silero VAD speech-probability threshold. |
-| `stop_probe_after_s` | `0.4` | Delay before an early wake/STOP transcription probe; set to `0` or less to disable probes. |
+| `stop_probe_after_s` | `0.25` | Cadence for up to three early wake/STOP transcription probes; set to `0` or less to disable probes. |
 
 `VoiceSession.text_topic` controls the completed-response echo sent through the
 hub data channel. Its default is `"agent.response"`; set it to `""` when the
@@ -118,9 +118,13 @@ does not execute application handlers. Typed-text ingress is also internal to
 health probes complete before the default hub transport opens its sockets.
 
 When wake phrases and the listening chime are enabled, the VAD/STT stage probes
-the opening audio while the user is still speaking. A recognized phrase emits
-the chime immediately, while only the final transcript enters the voice gate as
-a query. STOP commands use the same early-probe path for immediate interruption.
+the opening audio on a fixed cadence while the user is still speaking. Probe
+audio includes a short silent tail so offline STT can finalize the wake word. A
+recognized phrase emits the chime immediately, while only the final transcript
+enters the voice gate as a query. An in-flight probe gets a short grace period
+before final STT and is then cancelled, so a slow probe cannot stall the audio
+pipeline. A missed probe never inserts a late chime in front of response speech.
+STOP commands use the same early-probe path for immediate interruption.
 
 The final transcript accepts a wake phrase at its beginning or after
 sentence-final `.`, `?`, or `!` punctuation followed by whitespace or a closing

agent-sdk/xr-ai-voice/xr_ai_voice/_processors/vad_stt.py

@@ -48,6 +48,8 @@
 # True acknowledges and False requests another bounded probe.
 PartialTranscriptHandler = Callable[[str, str], Awaitable[bool]]
 _MAX_PARTIAL_PROBES = 3
+_PARTIAL_PROBE_TAIL_S = 0.12
+_PARTIAL_PROBE_FINISH_GRACE_S = 0.15
 
 
 @dataclass(frozen=True)
@@ -57,15 +59,15 @@ class VadConfig:
     Mirrors the constructor of :class:`xr_ai_vad.VadDetector`. Default
     values match the in-tree samples' current behavior.
 
-    ``stop_probe_after_s`` — seconds after ``on_speech_start`` to run an
-    extra STT pass on the partial audio buffer. This gives STOP commands a
-    fast interrupt path and lets a configured wake phrase be acknowledged
-    before the utterance ends. Set to ``0`` or negative to disable probes.
+    ``stop_probe_after_s`` — cadence in seconds for up to three STT probes of
+    the partial audio buffer. This gives STOP commands a fast interrupt path
+    and lets a configured wake phrase be acknowledged before the utterance
+    ends. Set to ``0`` or negative to disable probes.
     """
     silence_duration:   float = 0.8
     min_speech:         float = 0.15
     silero_threshold:   float = 0.5
-    stop_probe_after_s: float = 0.4
+    stop_probe_after_s: float = 0.25
 
 
 class VadSttProcessor(FrameProcessor):
@@ -107,6 +109,7 @@ def __init__(
         # One probe task per pid, so a fresh speech_start can cancel a
         # lingering task before scheduling the next.
         self._probe_task:     dict[str, asyncio.Task] = {}
+        self._probe_inflight: set[str] = set()
         # Pids whose probe has already pushed a STOP for the current
         # utterance — suppresses the duplicate that would fire when VAD
         # eventually finalizes the same speech run. Cleared on the next
@@ -190,14 +193,9 @@ async def on_speech_start() -> None:
             await self.push_frame(f)
 
         async def on_utterance(audio_bytes: bytes, sample_rate: int) -> None:
-            # Probe ran-or-not, the utterance has finalized. Close the
-            # probe buffer entry and cancel any pending probe (e.g.
-            # silence_duration < stop_probe_after_s). Await the
-            # cancellation so the probe task is fully torn down — see
-            # the comment in ``on_speech_start`` for why this matters.
-            await self._cancel_probe_task(pid)
             self._probe_buffer.pop(pid, None)
             self._probe_sr.pop(pid, None)
+            await self._finish_probe_for_utterance(pid)
 
             dur_s = (len(audio_bytes) // 2) / max(sample_rate, 1)
             logger.info("utterance finalize pid={!r} dur={:.2f}s", pid, dur_s)
@@ -297,9 +295,11 @@ async def _handle_audio(self, frame: InputAudioRawFrame) -> None:
     async def _run_partial_probes(self, pid: str) -> None:
         """Probe partial audio for STOP and an optional wake acknowledgement."""
         attempts = _MAX_PARTIAL_PROBES if self._on_partial_transcript else 1
+        started = time.monotonic()
         for attempt in range(1, attempts + 1):
             try:
-                await asyncio.sleep(self._vad_cfg.stop_probe_after_s)
+                due = started + attempt * self._vad_cfg.stop_probe_after_s
+                await asyncio.sleep(max(0.0, due - time.monotonic()))
             except asyncio.CancelledError:
                 return
 
@@ -308,41 +308,50 @@ async def _run_partial_probes(self, pid: str) -> None:
             if not buf or sr <= 0:
                 return
 
+            audio = bytes(buf) + bytes(round(sr * _PARTIAL_PROBE_TAIL_S) * 2)
+            before = time.monotonic()
+            self._probe_inflight.add(pid)
             try:
-                text = await self._transcribe(
-                    bytes(buf),
-                    sample_rate=sr,
-                    participant_id=pid,
-                    mode="partial-probe",
-                    attempt=attempt,
-                )
-            except asyncio.CancelledError:
-                return
-            except Exception:
-                logger.exception("partial-probe stt transcribe failed pid={!r}", pid)
-                return
-
-            stop_matched = bool(text and STOP_RE.match(text))
-            logger.info(
-                "early transcript probe fired pid={!r} attempt={} stop_matched={}",
-                pid, attempt, stop_matched,
-            )
-            if stop_matched:
-                await self._emit_early_stop(pid, text)
-                return
-
-            if text and self._on_partial_transcript is not None:
                 try:
-                    decision = await self._on_partial_transcript(pid, text)
+                    text = await self._transcribe(
+                        audio,
+                        sample_rate=sr,
+                        participant_id=pid,
+                        mode="partial-probe",
+                        attempt=attempt,
+                    )
                 except asyncio.CancelledError:
                     return
                 except Exception:
-                    logger.exception("partial-transcript handler failed pid={!r}", pid)
+                    logger.exception("partial-probe stt transcribe failed pid={!r}", pid)
                     return
-                if decision is True:
-                    logger.info("early wake phrase acknowledged pid={!r}", pid)
+
+                stop_matched = bool(text and STOP_RE.match(text))
+                logger.info(
+                    "early transcript probe fired pid={!r} attempt={} latency_ms={} stop_matched={}",
+                    pid, attempt, round((time.monotonic() - before) * 1000),
+                    stop_matched,
+                )
+                if stop_matched:
+                    await self._emit_early_stop(pid, text)
                     return
 
+                if text and self._on_partial_transcript is not None:
+                    try:
+                        decision = await self._on_partial_transcript(pid, text)
+                    except asyncio.CancelledError:
+                        return
+                    except Exception:
+                        logger.exception("partial-transcript handler failed pid={!r}", pid)
+                        return
+                    if decision is True:
+                        logger.info("early wake phrase acknowledged pid={!r}", pid)
+                        return
+            finally:
+                self._probe_inflight.discard(pid)
+            if pid not in self._probe_buffer:
+                return
+
     async def _transcribe(
         self,
         audio: bytes,
@@ -449,6 +458,22 @@ async def _evict_participant(self, pid: str) -> None:
             self._current_pid = None
         logger.info("evicted per-participant VAD state pid={!r}", pid)
 
+    async def _finish_probe_for_utterance(self, pid: str) -> None:
+        """Give active STT a short grace period; cancel all other probes."""
+        task = self._probe_task.get(pid)
+        if task is None or task.done() or pid not in self._probe_inflight:
+            await self._cancel_probe_task(pid)
+            return
+        self._probe_task.pop(pid, None)
+        try:
+            await asyncio.wait_for(task, timeout=_PARTIAL_PROBE_FINISH_GRACE_S)
+        except TimeoutError:
+            logger.info("partial probe grace expired pid={!r}", pid)
+        except asyncio.CancelledError:
+            logger.debug("partial probe cancelled during finalization pid={!r}", pid)
+        except Exception:
+            logger.exception("partial probe completion raised pid={!r}", pid)
+
     async def _cancel_probe_task(self, pid: str) -> None:
         """Cancel a pending probe task and await its teardown.
 

agent-sdk/xr-ai-voice/xr_ai_voice/_processors/voice_gate.py

@@ -75,6 +75,7 @@ def __init__(
             on_participant_joined = self._on_gate_participant_joined,
         )
         self._early_wake_ack: set[str] = set()
+        self._feeding_speech_transcript = False
         # Speech-onset timestamp (µs) of the transcript currently being fed to
         # the gate. ``VoiceGate.feed`` invokes ``_on_gate_query`` synchronously,
         # so the value is read back inside that callback.
@@ -131,10 +132,14 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
             # utterance (nanoseconds); the gate's query callback stamps it onto
             # the turn so downstream consumers anchor to when the user spoke.
             self._feeding_pts_us = frame.pts // 1_000 if frame.pts is not None else None
+            self._feeding_speech_transcript = bool(
+                frame.transport_source and frame.transport_source == frame.user_id
+            )
             try:
                 await self._gate.feed(frame.user_id, frame.text)
             finally:
                 self._feeding_pts_us = None
+                self._feeding_speech_transcript = False
                 self._early_wake_ack.discard(frame.user_id)
             return
 
@@ -162,7 +167,7 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
     # ── gate handlers ─────────────────────────────────────────────────────────
 
     async def _on_gate_query(self, pid: str, text: str, fresh_match: bool) -> None:
-        if fresh_match:
+        if fresh_match and not self._feeding_speech_transcript:
             await self._emit_chime(pid, early=False)
         await self.push_frame(GatedQueryFrame(
             participant_id = pid,