fix(wake): stabilize exclusive and duplex audio handoffs

cli.py

@@ -11870,10 +11870,14 @@ def _voice_start_recording(self):
         # continuous recording falls back to the documented defaults
         # instead of crashing on ``.get()``.
         voice_cfg: dict = {}
+        wake_cfg: dict = {}
         try:
             from hermes_cli.config import load_config
-            _cfg = load_config().get("voice")
+            _full_cfg = load_config()
+            _cfg = _full_cfg.get("voice")
             voice_cfg = _cfg if isinstance(_cfg, dict) else {}
+            _wake_cfg = _full_cfg.get("wake_word")
+            wake_cfg = _wake_cfg if isinstance(_wake_cfg, dict) else {}
         except Exception:
             pass
 
@@ -11932,7 +11936,16 @@ def _on_silence():
         if self._voice_beeps_enabled():
             try:
                 from tools.voice_mode import play_beep
-                play_beep(frequency=880, count=1)
+                if (
+                    getattr(self, "_wake_suspended", False)
+                    and wake_cfg.get("duplex_output_device") is not None
+                ):
+                    # A newly reopened HFP sink can discard the first short
+                    # buffer while SCO starts. Warm it with silence so the
+                    # actual wake cue remains audible.
+                    play_beep(frequency=880, count=1, pre_roll=0.25)
+                else:
+                    play_beep(frequency=880, count=1)
             except Exception:
                 pass
 
@@ -12024,7 +12037,25 @@ def _voice_stop_and_transcribe(self):
             if self._voice_recorder is None:
                 return
 
-            wav_path = self._voice_recorder.stop()
+            recorder = self._voice_recorder
+            release_for_wake = getattr(self, "_wake_suspended", False)
+            try:
+                wav_path = recorder.stop()
+            finally:
+                # A wake-triggered capture hands the device to another owner:
+                # the wake listener after a single-turn command, or a
+                # full-duplex/follow-up recorder in continuous conversation.
+                # AudioRecorder.stop() deliberately keeps its stream alive for
+                # ordinary voice mode, but no wake hand-off can reopen an
+                # exclusive device while that stream still owns it. Release it
+                # before transcription, including when stop() itself fails.
+                if release_for_wake:
+                    try:
+                        recorder.shutdown()
+                    except Exception:
+                        pass
+                    if self._voice_recorder is recorder:
+                        self._voice_recorder = None
 
             # Audio cue: double beep after stream stopped (no CoreAudio conflict)
             if self._voice_beeps_enabled():

hermes_cli/config_defaults.py

@@ -1481,6 +1481,7 @@
 
     "voice": {
         "record_key": "ctrl+b",
+        "output_device": None,         # PortAudio output index/name for beeps; null uses platform default
         "max_recording_seconds": 120,
         "auto_tts": False,
         "beep_enabled": True,         # Play record start/stop beeps in CLI voice mode
@@ -1504,6 +1505,7 @@
         "enabled": False,
         "surface": "auto",            # eligible surface: "auto" (first claimant) | "cli" | "tui" | "gui"
         "input_device": None,          # PortAudio input device index/name; null uses the process default
+        "duplex_output_device": None,  # optional output kept silently open for HFP-style full-duplex capture
         "provider": "openwakeword",   # "openwakeword" (free, local) | "sherpa" (free, ANY phrase, no training) | "porcupine" (premium; needs PORCUPINE_ACCESS_KEY)
         "phrase": "hey hermes",       # for "sherpa" this IS the detected phrase (any text works); for other engines it's a cosmetic label — detection is keyed by the model/keyword below
         "sensitivity": 0.6,           # 0.0-1.0 detection threshold, consistent across engines (higher = stricter, fewer false triggers)

tests/tools/test_voice_cli_integration.py

@@ -180,6 +180,39 @@ def test_start_recording_skips_beep_when_disabled(
         recorder.start.assert_called_once()
         mock_beep.assert_not_called()
 
+    @patch("cli._cprint")
+    @patch("cli.threading.Thread")
+    @patch("tools.voice_mode.play_beep")
+    @patch("tools.voice_mode.create_audio_recorder")
+    @patch(
+        "tools.voice_mode.check_voice_requirements",
+        return_value={
+            "available": True,
+            "audio_available": True,
+            "stt_available": True,
+            "details": "OK",
+            "missing_packages": [],
+        },
+    )
+    @patch(
+        "hermes_cli.config.load_config",
+        return_value={
+            "voice": {"beep_enabled": True},
+            "wake_word": {"duplex_output_device": "jabra_bluetooth"},
+        },
+    )
+    def test_wake_recording_warms_duplex_output_before_first_beep(
+        self, _cfg, _req, mock_create, mock_beep, mock_thread, _cp
+    ):
+        recorder = MagicMock(supports_silence_autostop=True)
+        mock_create.return_value = recorder
+        mock_thread.return_value = MagicMock(start=MagicMock())
+
+        cli = _make_voice_cli(_wake_suspended=True)
+        cli._voice_start_recording()
+
+        mock_beep.assert_called_once_with(frequency=880, count=1, pre_roll=0.25)
+
 
 class TestMaxRecordingSecondsConfigReal:
     """voice.max_recording_seconds must reach the recorder from config.
@@ -321,6 +354,68 @@ def test_no_speech_detected(self, _beep, _cp):
         cli._voice_stop_and_transcribe()
         assert cli._pending_input.empty()
 
+    @patch("cli._cprint")
+    @patch("tools.voice_mode.play_beep")
+    def test_wake_turn_without_speech_releases_capture_device(self, _beep, _cp):
+        recorder = MagicMock()
+        recorder.stop.return_value = None
+        cli = _make_voice_cli(
+            _voice_recording=True,
+            _voice_recorder=recorder,
+            _wake_suspended=True,
+        )
+
+        cli._voice_stop_and_transcribe()
+
+        recorder.shutdown.assert_called_once_with()
+        assert cli._voice_recorder is None
+        assert cli._pending_input.empty()
+
+    @pytest.mark.parametrize("continuous", [False, True])
+    def test_wake_turn_releases_capture_device_before_transcription(self, continuous):
+        events = []
+        recorder = MagicMock()
+        recorder.stop.return_value = "/tmp/test.wav"
+        recorder.shutdown.side_effect = lambda: events.append("shutdown")
+        cli = _make_voice_cli(
+            _voice_recording=True,
+            _voice_recorder=recorder,
+            _voice_continuous=continuous,
+            _wake_suspended=True,
+        )
+
+        def transcribe(*_args, **_kwargs):
+            events.append("transcribe")
+            return {"success": True, "transcript": "hello world"}
+
+        with patch("cli._cprint"), \
+             patch("cli.os.path.isfile", return_value=False), \
+             patch("hermes_cli.config.load_config", return_value={"stt": {}}), \
+             patch("tools.voice_mode.play_beep"), \
+             patch("tools.voice_mode.transcribe_recording",
+                   side_effect=transcribe) as mock_transcribe:
+            cli._voice_stop_and_transcribe()
+
+        assert events == ["shutdown", "transcribe"]
+        assert cli._voice_recorder is None
+        mock_transcribe.assert_called_once_with("/tmp/test.wav", model=None)
+
+    @patch("cli._cprint")
+    def test_wake_turn_releases_capture_device_when_stop_fails(self, _cp):
+        recorder = MagicMock()
+        recorder.stop.side_effect = RuntimeError("wav write failed")
+        cli = _make_voice_cli(
+            _voice_recording=True,
+            _voice_recorder=recorder,
+            _wake_suspended=True,
+        )
+
+        cli._voice_stop_and_transcribe()
+
+        recorder.shutdown.assert_called_once_with()
+        assert cli._voice_recorder is None
+        assert cli._voice_processing is False
+
     @patch("cli._cprint")
     @patch("cli.os.unlink")
     @patch("cli.os.path.isfile", return_value=True)
@@ -341,6 +436,8 @@ def test_successful_transcription_queues_input(
         from cli import _VoiceInputMessage
         assert isinstance(queued, _VoiceInputMessage)
         assert str(queued) == "hello world"
+        recorder.shutdown.assert_not_called()
+        assert cli._voice_recorder is recorder
 
 
     def test_non_local_stt_keeps_generic_transcribing_status(self):

tests/tools/test_voice_mode.py

@@ -708,6 +708,35 @@ def test_beep_calls_sounddevice_play(self, mock_sd):
         assert audio_arg.dtype == np.int16
         assert len(audio_arg) > 0
 
+    def test_beep_can_prepend_silent_device_warmup(self, mock_sd):
+        np = pytest.importorskip("numpy")
+        from tools import voice_mode as vm
+
+        mock_stream = MagicMock(active=False)
+        mock_sd.get_stream.return_value = mock_stream
+
+        vm.play_beep(frequency=880, duration=0.1, count=1, pre_roll=0.25)
+
+        audio = mock_sd.play.call_args[0][0]
+        warmup_samples = int(vm.SAMPLE_RATE * 0.25)
+        assert len(audio) == warmup_samples + int(vm.SAMPLE_RATE * 0.1)
+        assert np.all(audio[:warmup_samples] == 0)
+        assert np.any(audio[warmup_samples:] != 0)
+
+    def test_beep_uses_configured_output_device(self, mock_sd):
+        pytest.importorskip("numpy")
+        from tools.voice_mode import play_beep
+
+        mock_stream = MagicMock(active=False)
+        mock_sd.get_stream.return_value = mock_stream
+        with patch(
+            "tools.voice_mode.configured_output_device",
+            return_value="Jabra Speak2 55 UC",
+        ):
+            play_beep()
+
+        assert mock_sd.play.call_args.kwargs["device"] == "Jabra Speak2 55 UC"
+
 # ============================================================================
 # Silence detection
 # ============================================================================

tests/tools/test_wake_word.py

@@ -23,11 +23,22 @@
 
 
 def test_config_defaults_and_clamping():
+    from hermes_cli.config_defaults import DEFAULT_CONFIG
+
+    assert ww._DEFAULTS["duplex_output_device"] is None
+    assert DEFAULT_CONFIG["wake_word"]["duplex_output_device"] is None
     assert ww._provider({}) == "openwakeword"
     assert ww._provider({"provider": "Porcupine"}) == "porcupine"
     assert ww._input_device({}) is None
     assert ww._input_device({"input_device": 7}) == 7
     assert ww._input_device({"input_device": " Microphone Array "}) == "Microphone Array"
+    assert ww._duplex_output_device({}) is None
+    assert ww._duplex_output_device({"duplex_output_device": 8}) == 8
+    assert (
+        ww._duplex_output_device({"duplex_output_device": " Jabra Bluetooth "})
+        == "Jabra Bluetooth"
+    )
+    assert ww._duplex_output_device({"duplex_output_device": ""}) is None
     assert ww._input_device({"input_device": ""}) is None
     assert ww._input_device({"input_device": False}) is None
     assert ww._sensitivity({"sensitivity": 5}) == 1.0
@@ -472,6 +483,79 @@ def _stream(**kwargs):
         det.stop()
 
 
+def test_detector_keeps_optional_duplex_output_open_and_silent(monkeypatch):
+    opened = []
+    input_streams = []
+    output_streams = []
+
+    class _OutputStream(_FakeStream):
+        def __init__(self, **kwargs):
+            super().__init__(**kwargs)
+            output_streams.append(self)
+
+    class _OutputBuffer:
+        def __init__(self):
+            self.value = None
+
+        def fill(self, value):
+            self.value = value
+
+    class _CallbackInputStream(_FakeStream):
+        def __init__(self, **kwargs):
+            super().__init__(**kwargs)
+            input_streams.append(self)
+
+        def read(self, _n):
+            raise AssertionError("duplex wake capture must not use blocking read()")
+
+    def _input_stream(**kwargs):
+        opened.append(("input", kwargs))
+        return _CallbackInputStream(**kwargs)
+
+    def _output_stream(**kwargs):
+        opened.append(("output", kwargs))
+        return _OutputStream(**kwargs)
+
+    fake_sd = types.SimpleNamespace(
+        InputStream=_input_stream,
+        OutputStream=_output_stream,
+        query_devices=lambda selector, kind: {
+            "name": str(selector),
+            "hostapi": 0,
+            "max_input_channels": 1,
+            "default_samplerate": 16000.0,
+        },
+        query_hostapis=lambda index: {"name": "ALSA"},
+    )
+    monkeypatch.setattr(ww, "_import_audio", lambda: (fake_sd, None))
+
+    det = ww.WakeWordDetector(
+        _FakeEngine(fire=False),
+        lambda: None,
+        input_device="Bluetooth Mic",
+        duplex_output_device="Bluetooth Speaker",
+    )
+    det.start()
+    try:
+        assert [kind for kind, _kwargs in opened] == ["output", "input"]
+        output_kwargs = opened[0][1]
+        assert output_kwargs["device"] == "Bluetooth Speaker"
+        assert output_kwargs["samplerate"] == ww.SAMPLE_RATE
+        assert output_kwargs["channels"] == 1
+        assert output_kwargs["dtype"] == "int16"
+        buffer = _OutputBuffer()
+        output_kwargs["callback"](buffer, 4, None, None)
+        assert buffer.value == 0
+        input_kwargs = opened[1][1]
+        assert callable(input_kwargs["callback"])
+        input_kwargs["callback"](_Frame([500] * 4), 4, None, None)
+    finally:
+        det.stop()
+
+    assert input_streams[0].closed is True
+    assert output_streams[0].closed is True
+
+
 def test_windows_silent_hint_names_selected_device(monkeypatch):
     monkeypatch.setattr(ww.sys, "platform", "win32")
     hint = ww.silent_audio_hint(
@@ -523,6 +607,7 @@ def test_detector_flags_silent_stream_and_recovers(monkeypatch):
 
 def test_detection_callback_can_pause_and_close_stream(monkeypatch, tmp_path):
     streams = []
+    callback_saw_closed_stream = []
 
     def _stream(**kw):
         stream = _FakeStream(**kw)
@@ -537,13 +622,15 @@ def _stream(**kw):
     paused = threading.Event()
 
     def _on_wake():
+        callback_saw_closed_stream.append(streams[0].closed)
         if ww.pause_listening(owner=owner):
             paused.set()
 
     ww.start_listening(_on_wake, owner=owner, config={})
     assert paused.wait(2)
     assert ww.is_listening() is False
     assert streams[0].closed is True
+    assert callback_saw_closed_stream == [True]
     assert ww.stop_listening(owner=owner) is True