From b135a8badd79d022f734a66ac965d27142b3b5e1 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Wed, 22 Jul 2026 17:47:15 -0500 Subject: [PATCH] feat(voice): stream any provider in the CLI, with barge-in capture The streaming gate broadens from ElevenLabs-only to any provider that passes check_tts_requirements(). In continuous voice mode a mic monitor runs during playback: talking over the agent cuts TTS at detection while the monitor keeps recording, then the captured interruption is transcribed and queued as the next turn (process_loop's auto-restart stands down while the capture owns the mic). New voice.barge_in config key, default true. --- cli.py | 146 +++++++++++++----- hermes_cli/config.py | 1 + tests/tools/test_voice_cli_integration.py | 178 +++++++++------------- 3 files changed, 189 insertions(+), 136 deletions(-) diff --git a/cli.py b/cli.py index b9641e53ee206..18e167c009716 100644 --- a/cli.py +++ b/cli.py @@ -4205,6 +4205,8 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): self._voice_continuous = False self._voice_tts_done = threading.Event() self._voice_tts_done.set() + self._voice_tts_stop = None # active streaming pipeline's stop event + self._voice_barge_capture = threading.Event() # barge monitor is capturing the interruption # Status bar visibility (toggled via /statusbar) self._status_bar_visible = True @@ -11099,6 +11101,26 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): time.sleep(0.15) threading.Thread(target=_refresh_level, daemon=True).start() + def _voice_stt_model(self) -> Optional[str]: + """STT model override from config, or None for the provider default.""" + try: + from hermes_cli.config import load_config + stt_config = load_config().get("stt", {}) + return stt_config.get("model") if isinstance(stt_config, dict) else None + except Exception: + return None + + def _voice_restart_recording_async(self) -> None: + """Restart continuous-mode recording off-thread (start() can block).""" + def _restart_recording(): + try: + self._voice_start_recording() + if hasattr(self, '_app') and self._app: + self._app.invalidate() + except Exception as e: + _cprint(f"{_DIM}Voice auto-restart failed: {e}{_RST}") + threading.Thread(target=_restart_recording, daemon=True).start() + def _voice_stop_and_transcribe(self): """Stop recording, transcribe via STT, and queue the transcript as input.""" # Atomic guard: only one thread can enter stop-and-transcribe. @@ -11136,17 +11158,8 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): self._app.invalidate() _cprint(f"{_DIM}Transcribing...{_RST}") - # Get STT model from config - stt_model = None - try: - from hermes_cli.config import load_config - stt_config = load_config().get("stt", {}) - stt_model = stt_config.get("model") - except Exception: - pass - from tools.voice_mode import transcribe_recording - result = transcribe_recording(wav_path, model=stt_model) + result = transcribe_recording(wav_path, model=self._voice_stt_model()) if result.get("success") and result.get("transcript", "").strip(): transcript = result["transcript"].strip() @@ -11197,14 +11210,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): # (When transcript IS submitted, process_loop handles restart # after chat() completes.) if self._voice_continuous and not submitted and not self._voice_recording: - def _restart_recording(): - try: - self._voice_start_recording() - if hasattr(self, '_app') and self._app: - self._app.invalidate() - except Exception as e: - _cprint(f"{_DIM}Voice auto-restart failed: {e}{_RST}") - threading.Thread(target=_restart_recording, daemon=True).start() + self._voice_restart_recording_async() def _voice_speak_response_async(self, text: str) -> None: """Schedule TTS and mark it pending before continuous recording can restart.""" @@ -11270,6 +11276,68 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): self._voice_tts_done.set() + def _voice_barge_in_monitor(self, stop_event: threading.Event) -> None: + """VAD barge-in: cut streaming TTS the moment the user starts talking. + + Runs for one turn alongside the streaming pipeline (continuous voice + mode only — the mic is otherwise idle during playback). On speech, + playback is cut immediately while the monitor KEEPS capturing (with + pre-roll, so the interruption is transcribed from its first syllable + — restarting the recorder after detection would lose the opening + words). ``_voice_barge_capture`` suppresses process_loop's auto- + restart until the captured utterance has been submitted. + """ + try: + from hermes_cli.config import load_config + voice_cfg = load_config().get("voice") or {} + if not (isinstance(voice_cfg, dict) and voice_cfg.get("barge_in", True)): + return + from tools.voice_mode import listen_for_speech, stop_playback + + def _cut_playback(): + if not self._voice_tts_done.is_set(): + self._voice_barge_capture.set() + stop_event.set() + stop_playback() + + wav_path = listen_for_speech( + lambda: stop_event.is_set() or self._voice_tts_done.is_set(), + capture=True, + on_trigger=_cut_playback, + ) + if wav_path and self._voice_barge_capture.is_set(): + self._voice_submit_barge_utterance(wav_path) + else: + self._voice_barge_capture.clear() + except Exception as e: + self._voice_barge_capture.clear() + logger.debug("Voice barge-in monitor failed: %s", e) + + def _voice_submit_barge_utterance(self, wav_path: str) -> None: + """Transcribe a barge-captured interruption and queue it as the next turn.""" + submitted = False + try: + from tools.voice_mode import transcribe_recording + result = transcribe_recording(wav_path, model=self._voice_stt_model()) + transcript = (result.get("transcript") or "").strip() if result.get("success") else "" + if transcript: + self._pending_input.put(transcript) + submitted = True + elif not result.get("success"): + _cprint(f"\n{_DIM}Transcription failed: {result.get('error', 'Unknown error')}{_RST}") + except Exception as e: + _cprint(f"\n{_DIM}Voice processing error: {e}{_RST}") + finally: + try: + if os.path.isfile(wav_path): + os.unlink(wav_path) + except OSError: + pass + self._voice_barge_capture.clear() + # No usable transcript: hand the mic back to the normal loop. + if not submitted and self._voice_mode and self._voice_continuous and not self._voice_recording: + self._voice_restart_recording_async() + def _voice_beeps_enabled(self) -> bool: """Return whether CLI voice mode should play record start/stop beeps.""" try: @@ -11363,8 +11431,10 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): threading.Thread(target=_bg_shutdown, daemon=True).start() self._voice_recorder = None - # Stop any active TTS playback + # Stop any active TTS playback (file player + streaming pipeline) try: + if self._voice_tts_stop is not None: + self._voice_tts_stop.set() from tools.voice_mode import stop_playback stop_playback() except Exception: @@ -12117,9 +12187,9 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): self._reasoning_shown_this_turn = False # --- Streaming TTS setup --- - # When ElevenLabs is the TTS provider and sounddevice is available, - # we stream audio sentence-by-sentence as the agent generates tokens - # instead of waiting for the full response. + # Any working TTS provider streams sentence-by-sentence as the agent + # generates tokens: PCM-streaming providers (ElevenLabs, OpenAI) play + # chunks as they arrive, everything else synthesizes per sentence. use_streaming_tts = False _streaming_box_opened = False text_queue = None @@ -12130,20 +12200,12 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): if self._voice_tts: try: from tools.tts_tool import ( - _load_tts_config as _load_tts_cfg, - _get_provider as _get_prov, - _import_elevenlabs, _import_sounddevice, + check_tts_requirements, stream_tts_to_speaker, ) - _tts_cfg = _load_tts_cfg() - if _get_prov(_tts_cfg) == "elevenlabs": - # Verify both ElevenLabs SDK and audio output are available - _import_elevenlabs() - _import_sounddevice() - use_streaming_tts = True - except (ImportError, OSError): - pass + _import_sounddevice() + use_streaming_tts = check_tts_requirements() except Exception: pass @@ -12171,6 +12233,13 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): daemon=True, ) tts_thread.start() + # Expose the pipeline's stop event so barge-in paths (voice + # key, VAD monitor) can cut playback from outside this turn. + self._voice_tts_stop = stop_event + if self._voice_continuous: + threading.Thread( + target=self._voice_barge_in_monitor, args=(stop_event,), daemon=True + ).start() def stream_callback(delta: str): if text_queue is not None: @@ -13316,6 +13385,8 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): self._voice_continuous = False # Whether to auto-restart after agent responds self._voice_tts_done = threading.Event() # Signals TTS playback finished self._voice_tts_done.set() # Initially "done" (no TTS pending) + self._voice_tts_stop = None # active streaming pipeline's stop event + self._voice_barge_capture = threading.Event() # barge monitor is capturing the interruption if os.environ.get("HERMES_DEFER_AGENT_STARTUP") != "1": self._install_tool_callbacks() @@ -14104,9 +14175,12 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): return # Interrupt TTS if playing, so user can start talking. - # stop_playback() is fast (just terminates a subprocess). + # stop_playback() is fast (just terminates a subprocess); + # the stop event drains the streaming pipeline if one is live. if not cli_ref._voice_tts_done.is_set(): try: + if cli_ref._voice_tts_stop is not None: + cli_ref._voice_tts_stop.set() from tools.voice_mode import stop_playback stop_playback() cli_ref._voice_tts_done.set() @@ -15328,6 +15402,10 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): if self._voice_tts: self._voice_tts_done.wait(timeout=60) time.sleep(0.3) + # A barge-in capture already owns the mic and + # will submit the interruption itself. + if self._voice_barge_capture.is_set(): + return self._voice_start_recording() app.invalidate() except Exception as e: diff --git a/hermes_cli/config.py b/hermes_cli/config.py index fd6de3a58dbb3..fb538fcf3059b 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -2320,6 +2320,7 @@ DEFAULT_CONFIG = { "beep_enabled": True, # Play record start/stop beeps in CLI voice mode "silence_threshold": 200, # RMS below this = silence (0-32767) "silence_duration": 3.0, # Seconds of silence before auto-stop + "barge_in": True, # Stop TTS playback when the user starts talking }, "human_delay": { diff --git a/tests/tools/test_voice_cli_integration.py b/tests/tools/test_voice_cli_integration.py index bf96ff501281e..1e566f74806fb 100644 --- a/tests/tools/test_voice_cli_integration.py +++ b/tests/tools/test_voice_cli_integration.py @@ -29,6 +29,8 @@ def _make_voice_cli(**overrides): cli._voice_continuous = False cli._voice_tts_done = threading.Event() cli._voice_tts_done.set() + cli._voice_tts_stop = None + cli._voice_barge_capture = threading.Event() cli._pending_input = queue.Queue() cli._app = None cli._attached_images = [] @@ -176,114 +178,40 @@ class TestVoiceStateLock: # ============================================================================ class TestStreamingTTSActivation: - """Verify streaming TTS uses lazy imports to check availability.""" + """The CLI streaming gate: sounddevice + a working provider, ANY provider. - def test_activates_when_elevenlabs_and_sounddevice_available(self): - """use_streaming_tts should be True when provider is elevenlabs - and both lazy imports succeed.""" - use_streaming_tts = False + Mirrors cli.py's gate exactly — streaming engages whenever audio output + exists and check_tts_requirements() passes, regardless of which provider + is configured (non-streamers get the per-sentence sync path downstream). + """ + + @staticmethod + def _gate() -> bool: + """The cli.py streaming-TTS gate, verbatim.""" try: - from tools.tts_tool import ( - _load_tts_config as _load_tts_cfg, - _get_provider as _get_prov, - _import_elevenlabs, - _import_sounddevice, - ) - assert callable(_import_elevenlabs) - assert callable(_import_sounddevice) - except ImportError: - pytest.skip("tools.tts_tool not available") + from tools.tts_tool import _import_sounddevice, check_tts_requirements + _import_sounddevice() + return check_tts_requirements() + except Exception: + return False - with patch("tools.tts_tool._load_tts_config") as mock_cfg, \ - patch("tools.tts_tool._get_provider", return_value="elevenlabs"), \ - patch("tools.tts_tool._import_elevenlabs") as mock_el, \ - patch("tools.tts_tool._import_sounddevice") as mock_sd: - mock_cfg.return_value = {"provider": "elevenlabs"} - mock_el.return_value = MagicMock() - mock_sd.return_value = MagicMock() + def test_activates_for_any_working_provider(self): + """Any provider that passes check_tts_requirements engages streaming.""" + with patch("tools.tts_tool._import_sounddevice", return_value=MagicMock()), \ + patch("tools.tts_tool.check_tts_requirements", return_value=True): + assert self._gate() is True - from tools.tts_tool import ( - _load_tts_config as load_cfg, - _get_provider as get_prov, - _import_elevenlabs as import_el, - _import_sounddevice as import_sd, - ) - cfg = load_cfg() - if get_prov(cfg) == "elevenlabs": - import_el() - import_sd() - use_streaming_tts = True - - assert use_streaming_tts is True - - def test_does_not_activate_when_elevenlabs_missing(self): - """use_streaming_tts stays False when elevenlabs import fails.""" - use_streaming_tts = False - with patch("tools.tts_tool._load_tts_config", return_value={"provider": "elevenlabs"}), \ - patch("tools.tts_tool._get_provider", return_value="elevenlabs"), \ - patch("tools.tts_tool._import_elevenlabs", side_effect=ImportError("no elevenlabs")): - try: - from tools.tts_tool import ( - _load_tts_config as load_cfg, - _get_provider as get_prov, - _import_elevenlabs as import_el, - _import_sounddevice as import_sd, - ) - cfg = load_cfg() - if get_prov(cfg) == "elevenlabs": - import_el() - import_sd() - use_streaming_tts = True - except (ImportError, OSError): - pass - - assert use_streaming_tts is False + def test_does_not_activate_when_provider_unavailable(self): + """No working TTS provider → no streaming pipeline.""" + with patch("tools.tts_tool._import_sounddevice", return_value=MagicMock()), \ + patch("tools.tts_tool.check_tts_requirements", return_value=False): + assert self._gate() is False def test_does_not_activate_when_sounddevice_missing(self): - """use_streaming_tts stays False when sounddevice import fails.""" - use_streaming_tts = False - with patch("tools.tts_tool._load_tts_config", return_value={"provider": "elevenlabs"}), \ - patch("tools.tts_tool._get_provider", return_value="elevenlabs"), \ - patch("tools.tts_tool._import_elevenlabs", return_value=MagicMock()), \ - patch("tools.tts_tool._import_sounddevice", side_effect=OSError("no PortAudio")): - try: - from tools.tts_tool import ( - _load_tts_config as load_cfg, - _get_provider as get_prov, - _import_elevenlabs as import_el, - _import_sounddevice as import_sd, - ) - cfg = load_cfg() - if get_prov(cfg) == "elevenlabs": - import_el() - import_sd() - use_streaming_tts = True - except (ImportError, OSError): - pass - - assert use_streaming_tts is False - - def test_does_not_activate_for_non_elevenlabs_provider(self): - """use_streaming_tts stays False when provider is not elevenlabs.""" - use_streaming_tts = False - with patch("tools.tts_tool._load_tts_config", return_value={"provider": "edge"}), \ - patch("tools.tts_tool._get_provider", return_value="edge"): - try: - from tools.tts_tool import ( - _load_tts_config as load_cfg, - _get_provider as get_prov, - _import_elevenlabs as import_el, - _import_sounddevice as import_sd, - ) - cfg = load_cfg() - if get_prov(cfg) == "elevenlabs": - import_el() - import_sd() - use_streaming_tts = True - except (ImportError, OSError): - pass - - assert use_streaming_tts is False + """No audio output device → no streaming pipeline, even with a provider.""" + with patch("tools.tts_tool._import_sounddevice", side_effect=OSError("no PortAudio")), \ + patch("tools.tts_tool.check_tts_requirements", return_value=True): + assert self._gate() is False def test_stale_boolean_imports_no_longer_exist(self): """Confirm _HAS_ELEVENLABS and _HAS_AUDIO are not in tts_tool module.""" @@ -1326,3 +1254,49 @@ class TestRefreshLevelLock: assert not t.is_alive() assert not t.is_alive(), "Refresh thread did not stop" assert iterations > 0, "Refresh thread never ran" + + +# --------------------------------------------------------------------------- +# Barge-in capture — the interruption is transcribed and queued directly +# --------------------------------------------------------------------------- + + +class TestVoiceBargeCaptureSubmit: + """_voice_submit_barge_utterance: the barge monitor's captured WAV becomes + the next turn without a re-record round trip.""" + + def test_transcript_is_queued_and_wav_removed(self, tmp_path, monkeypatch): + cli = _make_voice_cli() + cli._voice_barge_capture.set() + wav = tmp_path / "barge.wav" + wav.write_bytes(b"RIFF") + + monkeypatch.setattr( + "tools.voice_mode.transcribe_recording", + lambda path, model=None: {"success": True, "transcript": "stop, do it differently"}, + ) + + cli._voice_submit_barge_utterance(str(wav)) + + assert cli._pending_input.get_nowait() == "stop, do it differently" + assert not cli._voice_barge_capture.is_set() + assert not wav.exists() + + def test_no_speech_hands_mic_back_without_queueing(self, tmp_path, monkeypatch): + cli = _make_voice_cli(_voice_mode=True, _voice_continuous=True) + cli._voice_barge_capture.set() + wav = tmp_path / "barge.wav" + wav.write_bytes(b"RIFF") + restarted = threading.Event() + cli._voice_start_recording = lambda: restarted.set() + + monkeypatch.setattr( + "tools.voice_mode.transcribe_recording", + lambda path, model=None: {"success": True, "transcript": "", "no_speech": True}, + ) + + cli._voice_submit_barge_utterance(str(wav)) + + assert cli._pending_input.empty() + assert not cli._voice_barge_capture.is_set() + assert restarted.wait(2.0) # continuous mode resumes listening