Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 30 additions & 9 deletions cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -13707,6 +13707,33 @@ def _on_wake_word(self):
_cprint(f"{_DIM}Wake capture failed: {e}{_RST}")
# Leave _wake_suspended set; the watchdog resumes once idle.

def _manual_ptt_start_recording(self):
"""Start manual push-to-talk recording (Ctrl+B), on a daemon thread.

Manual push-to-talk claims the same physical microphone as an active
wake-word listener. Unlike ``_on_wake_word`` (which pauses before
capturing because the detector already owns the mic when it fires),
this path can run while the detector's own stream is still open —
hand the mic off first or ``AudioRecorder`` opens a second,
independent input stream on the same device (PortAudio init error on
single-owner-capture platforms). Setting ``_wake_suspended`` lets the
existing wake watchdog (``_start_wake_watchdog``) resume the listener
once recording goes idle, exactly as it does for the wake-triggered
capture.
"""
try:
from tools.wake_word import pause_listening
if pause_listening(owner=self):
self._wake_suspended = True
except Exception as e:
logger.debug("wake word pause failed: %s", e)
try:
self._voice_start_recording()
if hasattr(self, '_app') and self._app:
self._app.invalidate()
except Exception as e:
_cprint(f"\n{_DIM}Voice recording failed: {e}{_RST}")

def _start_wake_watchdog(self):
"""Resume the paused detector when the CLI returns to a stable idle."""
if getattr(self, "_wake_watchdog_started", False):
Expand Down Expand Up @@ -16900,15 +16927,9 @@ def handle_voice_record(event):
# Dispatch to a daemon thread so play_beep(sd.wait),
# AudioRecorder.start(lock acquire), and config I/O
# never block the prompt_toolkit event loop.
def _start_recording():
try:
cli_ref._voice_start_recording()
if hasattr(cli_ref, '_app') and cli_ref._app:
cli_ref._app.invalidate()
except Exception as e:
_cprint(f"\n{_DIM}Voice recording failed: {e}{_RST}")

threading.Thread(target=_start_recording, daemon=True).start()
threading.Thread(
target=cli_ref._manual_ptt_start_recording, daemon=True
).start()
event.app.invalidate()
from prompt_toolkit.keys import Keys

Expand Down
61 changes: 61 additions & 0 deletions tests/tools/test_voice_cli_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -375,6 +375,67 @@ def test_non_local_stt_keeps_generic_transcribing_status(self):
mock_transcribe.assert_called_once_with("/tmp/test.wav", model="whisper-1")


class TestManualPTTStartRecording:
"""HermesCLI._manual_ptt_start_recording: the daemon-thread body
handle_voice_record's Ctrl+B path dispatches to, with real CLI instance.

Manual push-to-talk claims the same physical microphone as an active
wake-word listener. _on_wake_word already hands the mic off before
capturing (it owns the mic when it fires); this covers the manual
push-to-talk path, which can otherwise run while the wake detector's
own input stream is still open and open a second, independent
sd.InputStream on the same device (a PortAudio error on
single-owner-capture platforms, e.g. Windows).
"""

def test_pauses_wake_word_before_starting_recording(self):
cli = _make_voice_cli(_wake_suspended=False)
calls: list[str] = []

def _fake_pause_listening(*, owner):
assert owner is cli
calls.append("pause_listening")
return True

cli._voice_start_recording = MagicMock(
side_effect=lambda: calls.append("voice_start_recording")
)
with patch("tools.wake_word.pause_listening", side_effect=_fake_pause_listening):
cli._manual_ptt_start_recording()

assert calls == ["pause_listening", "voice_start_recording"], (
"pause_listening must run BEFORE _voice_start_recording opens "
"the AudioRecorder's own input stream on the same microphone"
)
assert cli._wake_suspended is True

def test_skips_wake_suspended_when_pause_declines(self):
"""pause_listening() returning False means there was nothing to
pause (or it declined) — _wake_suspended must stay False, since the
watchdog only needs to resume a listener that was actually paused."""
cli = _make_voice_cli(_wake_suspended=False)
cli._voice_start_recording = MagicMock()

with patch("tools.wake_word.pause_listening", return_value=False):
cli._manual_ptt_start_recording()

assert cli._wake_suspended is False
cli._voice_start_recording.assert_called_once()

def test_proceeds_when_pause_raises(self):
"""A pause_listening() failure (e.g. wake_word module unavailable)
must not block manual push-to-talk from recording — the mic
hand-off is best-effort, not a hard prerequisite."""
cli = _make_voice_cli(_wake_suspended=False)
cli._voice_start_recording = MagicMock()

with patch("tools.wake_word.pause_listening", side_effect=RuntimeError("boom")):
cli._manual_ptt_start_recording()

assert cli._wake_suspended is False
cli._voice_start_recording.assert_called_once()


# ---------------------------------------------------------------------------
# Barge-in capture — the interruption is transcribed and queued directly
# ---------------------------------------------------------------------------
Expand Down
Loading