diff --git a/tests/tools/test_voice_mode.py b/tests/tools/test_voice_mode.py index 94fe7478d3..be9f24f6de 100644 --- a/tests/tools/test_voice_mode.py +++ b/tests/tools/test_voice_mode.py @@ -1069,7 +1069,7 @@ class TestStreamLeakOnStartFailure: # ============================================================================ -# full_duplex_listen — VAD barge-in listener +# listen_for_speech — VAD barge-in monitor # ============================================================================ class _FakeInputStream: @@ -1092,6 +1092,118 @@ class _FakeInputStream: return self._np.full((frames, 1), level, dtype=self._np.int16), False +class TestListenForSpeech: + """listen_for_speech: calibration → sustained-speech trigger → barge-in.""" + + CALIB_BLOCKS = 14 # 400ms / 30ms + TRIP_BLOCKS = 10 # 300ms / 30ms + + def _run(self, mock_sd, levels, should_stop=None, **kwargs): + np = pytest.importorskip("numpy") + stream = _FakeInputStream(np, levels) + mock_sd.InputStream.return_value = stream + from tools.voice_mode import listen_for_speech + stops = iter([False] * 200 + [True] * 10_000) + return listen_for_speech(should_stop or (lambda: next(stops)), **kwargs), stream + + def test_sustained_speech_triggers(self, mock_sd): + levels = [0] * self.CALIB_BLOCKS + [5000] * 50 + heard, _ = self._run(mock_sd, levels) + assert heard is True + + + def test_returns_false_when_audio_unavailable(self, monkeypatch): + monkeypatch.setattr("tools.voice_mode._import_audio", MagicMock(side_effect=OSError("no audio"))) + from tools.voice_mode import listen_for_speech + assert listen_for_speech(lambda: False) is False + + def test_quiet_then_loud_playback_does_not_trip(self, mock_sd): + """TTS that starts quiet and gets louder must NOT trip barge-in. + + This is the core regression: a one-shot calibration freezes the + floor from the quiet opening, then louder TTS exceeds the stale + floor and false-triggers. The rolling window keeps the floor + current so the louder passage is absorbed into the floor. + """ + levels = [100] * self.CALIB_BLOCKS + [200] * 30 + [500] * 30 + [1000] * 30 + heard, _ = self._run(mock_sd, levels) + assert heard is False + + def test_silence_calibration_does_not_false_trip_on_tts(self, mock_sd): + """Calibration during an inter-sentence gap must NOT false-trip. + + If the grace period ends during a pause between TTS sentences, the + calibration window samples near-silence. Without the min_floor clamp, + min_floor locks near zero, the trigger drops to 400 RMS (SILENCE_RMS_THRESHOLD + * 2), and the next TTS sentence at 800 RMS exceeds it — those blocks are + excluded from the rolling window (rms >= trigger), the floor freezes, and + after sustained_ms the VAD false-triggers and cuts playback mid-sentence. + + With the clamp, min_floor stays at SILENCE_RMS_THRESHOLD * 2 = 400, the + trigger is max(400, 400 * 8.0) = 3200, and 800-RMS TTS stays below it and + feeds the rolling floor. No false trip. + """ + # calibration_ms=800 → CALIB_BLOCKS = 800/30 ≈ 26 blocks of silence + # Then TTS resumes at 800 RMS — must NOT trip (below 3200 trigger). + calib = 800 // 30 + levels = [0] * calib + [800] * 100 + heard, _ = self._run( + mock_sd, levels, + sustained_ms=1000, + calibration_ms=800, + ) + assert heard is False + + +class TestListenForSpeechCapture: + """capture=True: the barge monitor records the interruption with pre-roll, + so the utterance is complete from its first syllable — nothing is lost + between detection and a recorder restart.""" + + CALIB_BLOCKS = 14 # 400ms / 30ms + LOUD_BLOCKS = 30 # speech: trips after 10, keeps talking + BLOCK = 480 # 16000 * 0.03 + + def _run(self, mock_sd, monkeypatch, levels, should_stop=None, **kwargs): + np = pytest.importorskip("numpy") + stream = _FakeInputStream(np, levels) + mock_sd.InputStream.return_value = stream + written = {} + monkeypatch.setattr( + "tools.voice_mode.AudioRecorder._write_wav", + staticmethod(lambda audio: written.update(audio=audio) or "/tmp/barge.wav"), + ) + from tools.voice_mode import listen_for_speech + stops = iter([False] * 200 + [True] * 10_000) + path = listen_for_speech( + should_stop or (lambda: next(stops)), capture=True, **kwargs + ) + return path, written.get("audio"), stream + + def test_captured_utterance_includes_speech_onset(self, mock_sd, monkeypatch): + """Every loud block — including the ones BEFORE detection tripped — + must land in the WAV. That pre-roll is the whole point.""" + triggered = [] + levels = [0] * self.CALIB_BLOCKS + [5000] * self.LOUD_BLOCKS + [0] * 500 + path, audio, _ = self._run( + mock_sd, monkeypatch, levels, + should_stop=lambda: False, + on_trigger=lambda: triggered.append(True), + ) + assert path == "/tmp/barge.wav" + assert triggered == [True] + assert int((audio == 5000).sum()) == self.LOUD_BLOCKS * self.BLOCK + + def test_no_trip_returns_none(self, mock_sd, monkeypatch): + triggered = [] + path, audio, _ = self._run( + mock_sd, monkeypatch, [0] * 500, + on_trigger=lambda: triggered.append(True), + ) + assert path is None + assert audio is None + assert triggered == [] + class TestFullDuplexListen: """full_duplex_listen: one agent-turn listener spanning generation and playback — pre-playback calibration, phase-aware trigger, grace window.""" diff --git a/tools/voice_mode.py b/tools/voice_mode.py index 8e8b8fd352..a7293da8a0 100644 --- a/tools/voice_mode.py +++ b/tools/voice_mode.py @@ -1080,6 +1080,167 @@ def _play_audio_file_impl(file_path: str) -> bool: return False +# ── Barge-in — detect the user speaking over TTS playback ── +# Public API kept from main: the per-playback barge monitor (rolling-floor VAD). No in-tree +# caller since full_duplex_listen took over the voice path, but plugins may import it. +def listen_for_speech( + should_stop: Callable[[], bool], + threshold: Optional[int] = None, + sustained_ms: int = 300, + calibration_ms: int = 400, + capture: bool = False, + on_trigger: Optional[Callable[[], None]] = None, + pre_roll_ms: int = 1200, + endpoint_silence_ms: int = 1250, + max_utterance_ms: int = 30_000, +): + """Block until sustained speech is heard on the mic, or *should_stop*. + + Barge-in monitor: run in a side thread while TTS is playing. Without + *capture* it returns ``True`` when the user started talking (cut playback). + With ``capture=True`` it ALSO records the interruption — a rolling + *pre_roll_ms* buffer means the utterance is kept from its first syllable, + not from the moment detection tripped — and keeps rolling until the user + goes quiet for *endpoint_silence_ms*, then returns the WAV path (or + ``None`` if speech never tripped). *on_trigger* fires at the moment of + detection so the caller can stop playback while capture continues. + + The noise floor is calibrated from the first *calibration_ms* of input — + playback is already audible then, so speaker bleed is baked into the + floor and only louder-than-playback speech trips the trigger. Requiring + *sustained_ms* of consecutive above-threshold blocks filters out coughs, + keyboard thumps, and playback transients. + """ + try: + sd, np = _import_audio() + except (ImportError, OSError): + return None if capture else False + + from collections import deque + + block = int(SAMPLE_RATE * 0.03) # 30ms blocks + calib_blocks = max(1, calibration_ms // 30) + trip_blocks = max(1, sustained_ms // 30) + endpoint_blocks = max(1, endpoint_silence_ms // 30) + max_blocks = max(1, max_utterance_ms // 30) + + # Rolling floor window: continuously tracks TTS speaker-bleed volume + # throughout playback, not just the first calibration_ms. This is the + # key fix for false barge-in — a one-shot calibration freezes a floor + # from the opening TTS passage, but later louder passages exceed the + # stale floor and false-trigger. The rolling window keeps the floor + # current so only genuinely louder-than-playback speech trips the VAD. + floor_window: "deque[float]" = deque(maxlen=max(calib_blocks, 100)) # ~3s rolling + pre_roll: deque = deque(maxlen=max(1, pre_roll_ms // 30)) + consecutive = 0 + min_floor = 0.0 # baseline from initial calibration; floor never drops below this + block_idx = 0 # block counter for diagnostic logging + + try: + with sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="int16", blocksize=block) as stream: + while not should_stop(): + data, _ = stream.read(block) + rms = float(np.sqrt(np.mean(data.astype(np.float64) ** 2))) + if capture: + pre_roll.append(data.copy()) + block_idx += 1 + + # Wait for at least calib_blocks before evaluating. During + # the initial warmup we always feed the window so calibration + # has data to work with. + if len(floor_window) < calib_blocks: + floor_window.append(rms) + continue + + # Lock a minimum floor from the initial calibration samples. + # During inter-sentence pauses the rolling window can flush + # with near-silence, collapsing the 90th-percentile floor + # toward zero and false-triggering on the next rising + # sentence. min_floor keeps the trigger from ever dropping + # below the baseline TTS playback level established during + # the initial calibration_ms window. + # + # If the grace period ended during an inter-sentence gap the + # calibration samples near-silence. Locking a near-zero + # floor sets the trigger so low that TTS blocks exceed it, + # are excluded from the rolling window (rms >= trigger), and + # the floor freezes — guaranteeing a false trigger the moment + # TTS resumes. Clamp min_floor to SILENCE_RMS_THRESHOLD * 2 + # (400 RMS) so the 8x multiplier yields a trigger of at least + # (500-2000 RMS) stays below it and feeds the rolling window, + # while genuine speech (3000-8000 RMS) can still trip it. + if min_floor == 0.0 and len(floor_window) >= calib_blocks: + _pct90 = float(np.percentile(list(floor_window), 90)) + min_floor = max(_pct90, SILENCE_RMS_THRESHOLD * 2) + else: + _pct90 = float(np.percentile(list(floor_window), 90)) + + # Use the 90th percentile of the ROLLING window for the + # noise floor so the trigger reflects the loudest parts of + # recent playback — not a frozen snapshot from TTS onset. + _floor = max(_pct90, min_floor) + # 8.0x multiplier: TTS speaker bleed has wide + # volume variation between sentences and within sentences. + # At 5x, louder TTS passages exceed the trigger, get + # excluded from the floor window, and create a low-stale + # floor that false-triggers on the next loud passage. + # 8x gives enough headroom for TTS dynamics to stay below + # the trigger and get absorbed into the rolling floor. + trigger = max(float(threshold or SILENCE_RMS_THRESHOLD * 2), _floor * 8.0) + # Ceiling: never let the trigger exceed 4000 RMS, otherwise + # a very loud TTS passage would push the trigger so high + # that genuine speech (which is typically 3000–8000 RMS) + # couldn't trip it. + trigger = min(trigger, 4000.0) + + # Only feed the floor window with blocks that are NOT above + # the current trigger — speech blocks would inflate the floor + # and make the trigger unreachable. + if rms < trigger: + floor_window.append(rms) + + consecutive = consecutive + 1 if rms >= trigger else 0 + if consecutive > 0: + logger.debug( + "VAD above-trigger: block=%d rms=%.0f floor=%.0f trigger=%.0f " + "consec=%d/%d min_floor=%.0f window_len=%d", + block_idx, rms, _floor, trigger, consecutive, + trip_blocks, min_floor, len(floor_window), + ) + if consecutive < trip_blocks: + continue + + # Tripped — the user is talking over playback. + logger.info( + "VAD TRIPPED: block=%d rms=%.0f floor=%.0f trigger=%.0f " + "consec=%d min_floor=%.0f — cutting TTS playback", + block_idx, rms, _floor, trigger, consecutive, min_floor, + ) + if on_trigger: + try: + on_trigger() + except Exception as e: + logger.debug("Barge-in trigger callback failed: %s", e) + if not capture: + return True + + # Keep rolling until the user goes quiet. Playback is stopped + # now, so plain silence endpointing (recorder threshold) works. + frames: List[Any] = list(pre_roll) + quiet = 0 + for _ in range(max_blocks): + data, _ = stream.read(block) + frames.append(data.copy()) + rms = float(np.sqrt(np.mean(data.astype(np.float64) ** 2))) + quiet = quiet + 1 if rms < SILENCE_RMS_THRESHOLD else 0 + if quiet >= endpoint_blocks: + break + return AudioRecorder._write_wav(np.concatenate(frames, axis=0)) + except Exception as e: + logger.debug("Barge-in listener failed: %s", e) + return None if capture else False + + # ── Full-duplex agent-turn listener ── # One listener for the WHOLE agent turn (armed at utterance submit, disarmed when # response + TTS are done): calibrates against the QUIET room at turn start,