From 4aa813f0ea3bf230de5f3d3702349065bf488b45 Mon Sep 17 00:00:00 2001 From: "Andrex Ibiza, MBA" <84248988+andrexibiza@users.noreply.github.com> Date: Wed, 5 Aug 2026 17:08:32 -0500 Subject: [PATCH] refactor(cli): extract modal-prompts/voice mixins from cli.py (shard s4) --- cli.py | 1282 +---------------- hermes_cli/cli_modal_prompts_mixin.py | 579 ++++++++ hermes_cli/cli_voice_mixin.py | 770 ++++++++++ ...test_cli_modal_prompts_mixin_regression.py | 315 ++++ tests/cli/test_cli_voice_mixin_regression.py | 231 +++ 5 files changed, 1898 insertions(+), 1279 deletions(-) create mode 100644 hermes_cli/cli_modal_prompts_mixin.py create mode 100644 hermes_cli/cli_voice_mixin.py create mode 100644 tests/cli/test_cli_modal_prompts_mixin_regression.py create mode 100644 tests/cli/test_cli_voice_mixin_regression.py diff --git a/cli.py b/cli.py index aed3992922b4a..00e469c10d368 100644 --- a/cli.py +++ b/cli.py @@ -54,6 +54,8 @@ from hermes_cli.cli_agent_setup_mixin import CLIAgentSetupMixin from hermes_cli.cli_commands_mixin import CLICommandsMixin from hermes_cli.cli_billing_mixin import CLIBillingMixin +from hermes_cli.cli_modal_prompts_mixin import CLIModalPromptsMixin +from hermes_cli.cli_voice_mixin import CLIVoiceMixin from agent.interrupt_compat import request_hard_interrupt # prompt_toolkit for fixed input area TUI @@ -4202,7 +4204,7 @@ def __str__(self) -> str: return self.text -class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): +class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin, CLIModalPromptsMixin, CLIVoiceMixin): """ Interactive CLI for the Hermes Agent. @@ -12129,698 +12131,6 @@ def _on_tool_complete(self, tool_call_id: str, function_name: str, function_args except Exception: logger.debug("Edit diff preview failed for %s", function_name, exc_info=True) - # ==================================================================== - # Voice mode methods - # ==================================================================== - - def _voice_start_recording(self): - """Start capturing audio from the microphone.""" - if getattr(self, '_should_exit', False): - return - from tools.voice_mode import create_audio_recorder, check_voice_requirements - - reqs = check_voice_requirements() - if not reqs["audio_available"]: - if _is_termux_environment(): - details = reqs.get("details", "") - if "Termux:API Android app is not installed" in details: - raise RuntimeError( - "Termux:API command package detected, but the Android app is missing.\n" - "Install/update the Termux:API Android app, then retry /voice on.\n" - "Fallback: pkg install python-numpy portaudio && python -m pip install sounddevice" - ) - raise RuntimeError( - "Voice mode requires either Termux:API microphone access or Python audio libraries.\n" - "Option 1: pkg install termux-api and install the Termux:API Android app\n" - "Option 2: pkg install python-numpy portaudio && python -m pip install sounddevice" - ) - raise RuntimeError( - "Voice mode requires sounddevice and numpy.\n" - f"Install with: {sys.executable} -m pip install sounddevice numpy" - ) - if not reqs.get("stt_available", reqs.get("stt_key_set")): - raise RuntimeError( - "Voice mode requires an STT provider for transcription.\n" - "Option 1: uv pip install faster-whisper " - "(free, local; `pip install faster-whisper` also works if pip is on PATH)\n" - "Option 2: Set GROQ_API_KEY (free tier)\n" - "Option 3: Set VOICE_TOOLS_OPENAI_KEY (paid)" - ) - - # Prevent double-start from concurrent threads (atomic check-and-set) - with self._voice_lock: - if self._voice_recording: - return - self._voice_recording = True - - # Load silence detection params from config. Shape-safe: a - # hand-edited ``voice: true`` / ``voice: cmd+b`` leaves - # ``load_config()['voice']`` as a non-dict; coerce to {} so - # continuous recording falls back to the documented defaults - # instead of crashing on ``.get()``. - voice_cfg: dict = {} - try: - from hermes_cli.config import load_config - _cfg = load_config().get("voice") - voice_cfg = _cfg if isinstance(_cfg, dict) else {} - except Exception: - pass - - # Recorder creation can fail (no input device, PortAudio init error). - # Reset the flag on failure or _voice_recording stays True forever and - # every future voice start is silently skipped by the guard above. - if self._voice_recorder is None: - try: - self._voice_recorder = create_audio_recorder() - except Exception: - with self._voice_lock: - self._voice_recording = False - raise - - # Apply config-driven silence params (numeric-guarded so YAML - # scalar corruption doesn't break recording start-up). - # - # ``bool`` is explicitly excluded from the numeric check — in - # Python bool is a subclass of int, so a hand-edited - # ``silence_threshold: true`` would otherwise be forwarded as - # ``1`` instead of falling back to the 200 default (Copilot - # round-12 on #19835). - _threshold = voice_cfg.get("silence_threshold") - _duration = voice_cfg.get("silence_duration") - self._voice_recorder._silence_threshold = ( - _threshold if isinstance(_threshold, (int, float)) and not isinstance(_threshold, bool) else 200 - ) - self._voice_recorder._silence_duration = ( - _duration if isinstance(_duration, (int, float)) and not isinstance(_duration, bool) else 3.0 - ) - # voice.max_recording_seconds — hard cap on a single recording's length. - # Same numeric guard as the silence params (bool excluded: a hand-edited - # ``max_recording_seconds: true`` must not become ``1`` — it falls back - # to the documented 120 default, mirroring the silence-param handling). - # An explicit numeric value <= 0 disables the cap. Previously this - # documented key was never read (dead config); wiring it here makes it - # take effect. - _max_rec = voice_cfg.get("max_recording_seconds") - self._voice_recorder._max_recording_seconds = ( - (_max_rec if _max_rec > 0 else 0.0) - if isinstance(_max_rec, (int, float)) and not isinstance(_max_rec, bool) - else 120.0 - ) - - def _on_silence(): - """Called by AudioRecorder when silence is detected after speech.""" - with self._voice_lock: - if not self._voice_recording: - return - _cprint(f"\n{_DIM}Silence detected, auto-stopping...{_RST}") - if hasattr(self, '_app') and self._app: - self._app.invalidate() - self._voice_stop_and_transcribe() - - # Audio cue: single beep BEFORE starting stream (avoid CoreAudio conflict) - if self._voice_beeps_enabled(): - try: - from tools.voice_mode import play_beep - play_beep(frequency=880, count=1) - except Exception: - pass - - try: - self._voice_recorder.start(on_silence_stop=_on_silence) - except Exception: - with self._voice_lock: - self._voice_recording = False - raise - _label = self._voice_record_key_label() - if getattr(self._voice_recorder, "supports_silence_autostop", True): - _recording_hint = f"auto-stops on silence | {_label} to stop & exit continuous" - elif _is_termux_environment(): - _recording_hint = f"Termux:API capture | {_label} to stop" - else: - _recording_hint = f"{_label} to stop" - _cprint(f"\n{_ACCENT}● Recording...{_RST} {_DIM}({_recording_hint}){_RST}") - - # Periodically refresh prompt to update audio level indicator - def _refresh_level(): - while True: - with self._voice_lock: - still_recording = self._voice_recording - if not still_recording: - break - if hasattr(self, '_app') and self._app: - self._app.invalidate() - 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. - - For the local provider, prefer stt.local.model (default ``base``) so the - CLI passes a real model name into the local STT backend. - """ - try: - from hermes_cli.config import load_config - stt_config = load_config().get("stt", {}) - if not isinstance(stt_config, dict): - return None - provider = str(stt_config.get("provider") or "").strip().lower() - if provider == "local": - local_config = stt_config.get("local") or {} - if not isinstance(local_config, dict): - local_config = {} - return local_config.get("model") or "base" - return stt_config.get("model") - except Exception: - return None - - def _voice_stt_provider(self) -> str: - """Configured STT provider name (lowercased), or empty string.""" - try: - from hermes_cli.config import load_config - stt_config = load_config().get("stt", {}) - if not isinstance(stt_config, dict): - return "" - return str(stt_config.get("provider") or "").strip().lower() - except Exception: - return "" - - 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. - # Set _voice_processing immediately so concurrent Ctrl+B presses - # don't race into the START path while recorder.stop() holds its lock. - with self._voice_lock: - if not self._voice_recording: - return - self._voice_recording = False - self._voice_processing = True - - submitted = False - transcription_failed = False - wav_path = None - try: - if self._voice_recorder is None: - return - - wav_path = self._voice_recorder.stop() - - # Audio cue: double beep after stream stopped (no CoreAudio conflict) - if self._voice_beeps_enabled(): - try: - from tools.voice_mode import play_beep - play_beep(frequency=660, count=2) - except Exception: - pass - - if wav_path is None: - _cprint(f"{_DIM}No speech detected.{_RST}") - return - - # _voice_processing is already True (set atomically above) - if hasattr(self, '_app') and self._app: - self._app.invalidate() - - stt_model = self._voice_stt_model() - if self._voice_stt_provider() == "local": - _cprint( - f"{_DIM}Preparing local STT model '{stt_model}' " - f"(first use may download it from Hugging Face)...{_RST}" - ) - else: - _cprint(f"{_DIM}Transcribing...{_RST}") - - from tools.voice_mode import transcribe_recording - result = transcribe_recording(wav_path, model=stt_model) - - if result.get("success") and result.get("transcript", "").strip(): - transcript = result["transcript"].strip() - from tools.voice_mode import is_voice_stop_phrase - if is_voice_stop_phrase(transcript): - # Bare "stop" (or configured phrase) ends the voice chat - # instead of being sent to the agent. - _cprint(f"{_DIM}Stop phrase detected — ending voice chat.{_RST}") - self._disable_voice_mode() - return - self._attached_images.clear() - if hasattr(self, '_app') and self._app: - self._app.invalidate() - self._pending_input.put(_VoiceInputMessage(transcript)) - submitted = True - elif result.get("success"): - _cprint(f"{_DIM}No speech detected.{_RST}") - else: - error = result.get("error", "Unknown error") - _cprint(f"\n{_DIM}Transcription failed: {error}{_RST}") - transcription_failed = True - - except Exception as e: - _cprint(f"\n{_DIM}Voice processing error: {e}{_RST}") - transcription_failed = wav_path is not None - finally: - with self._voice_lock: - self._voice_processing = False - if hasattr(self, '_app') and self._app: - self._app.invalidate() - # Clean up temp file unless transcription failed. On failure, keep - # the source recording so long dictation is not lost. - try: - if wav_path and os.path.isfile(wav_path): - if transcription_failed: - _cprint(f"{_DIM}Recording preserved at: {wav_path}{_RST}") - else: - os.unlink(wav_path) - except Exception: - pass - - # Track consecutive no-speech cycles to avoid infinite restart loops. - # While the agent is mid-turn or TTS is speaking, the user is - # CORRECTLY silent (waiting/listening) — those cycles must not - # count, or a multi-minute tool run ends the voice chat under - # the user. The stop phrase and barge-in still work during the - # hold (they run on their own paths above). - stop_continuous_restart = False - _tts_done = getattr(self, "_voice_tts_done", None) - _activity_hold = bool( - getattr(self, "_agent_running", False) - or (_tts_done is not None and not _tts_done.is_set()) - ) - if not submitted: - if _activity_hold: - pass # held: keep listening without counting the cycle - else: - self._no_speech_count = getattr(self, '_no_speech_count', 0) + 1 - if self._no_speech_count >= 3: - self._voice_continuous = False - self._no_speech_count = 0 - _cprint(f"{_DIM}No speech detected 3 times, continuous mode stopped.{_RST}") - stop_continuous_restart = True - else: - self._no_speech_count = 0 - - # If no transcript was submitted but continuous mode is active, - # restart recording so the user can keep talking. - # (When transcript IS submitted, process_loop handles restart - # after chat() completes.) - if ( - self._voice_continuous - and not submitted - and not self._voice_recording - and not stop_continuous_restart - ): - 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.""" - if not self._voice_tts or not text: - return - self._voice_tts_done.clear() - threading.Thread( - target=self._voice_speak_response, - args=(text,), - daemon=True, - ).start() - # Spoken barge-in must work on the whole-file fallback path too. The - # full-duplex agent-turn listener normally already covers playback - # (armed at turn start in chat()); this arm is an idempotent safety - # net for speak calls outside a chat turn — the listener refuses to - # double-arm via _voice_fd_active. - if self._voice_continuous: - threading.Thread( - target=self._voice_full_duplex_listener, - daemon=True, - ).start() - - def _voice_speak_response(self, text: str): - """Speak the agent's response aloud using TTS (runs in background thread).""" - if not self._voice_tts: - return - self._voice_tts_done.clear() - try: - from tools.tts_tool import text_to_speech_tool - from tools.voice_mode import play_audio_file - - # Strip markdown and non-speech content for cleaner TTS via the - # shared cleaner (tools/tts_text_normalize): markdown, emoji, - # blocks, verifier footer, units, newline flattening. - try: - from tools.tts_text_normalize import prepare_spoken_text - tts_text = prepare_spoken_text(text, max_chars=4000) - except Exception: - # Legacy fallback pipeline — keep voice replies best-effort. - tts_text = text[:4000] if len(text) > 4000 else text - tts_text = re.sub(r'```[\s\S]*?```', ' ', tts_text) # fenced code blocks - tts_text = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', tts_text) # [text](url) -> text - tts_text = re.sub(r'https?://\S+', '', tts_text) # URLs - tts_text = re.sub(r'\*\*(.+?)\*\*', r'\1', tts_text) # bold - tts_text = re.sub(r'\*(.+?)\*', r'\1', tts_text) # italic - tts_text = re.sub(r'`(.+?)`', r'\1', tts_text) # inline code - tts_text = re.sub(r'^#+\s*', '', tts_text, flags=re.MULTILINE) # headers - tts_text = re.sub(r'^\s*[-*]\s+', '', tts_text, flags=re.MULTILINE) # list items - tts_text = re.sub(r'---+', '', tts_text) # horizontal rules - tts_text = re.sub(r'\n{3,}', '\n\n', tts_text) # excessive newlines - tts_text = tts_text.strip() - if not tts_text: - return - - # Use MP3 output for CLI playback (afplay doesn't handle OGG well). - # The TTS tool may auto-convert MP3->OGG, but the original MP3 remains. - os.makedirs(os.path.join(tempfile.gettempdir(), "hermes_voice"), exist_ok=True) - mp3_path = os.path.join( - tempfile.gettempdir(), "hermes_voice", - f"tts_{time.strftime('%Y%m%d_%H%M%S')}.mp3", - ) - - raw_result = text_to_speech_tool(text=tts_text, output_path=mp3_path) - try: - tts_result = json.loads(raw_result) if isinstance(raw_result, str) else {} - except Exception: - tts_result = {} - - # Prefer the requested MP3 when the provider produced it. This - # preserves reliable local playback while still supporting - # providers that write to and return a different path. - audio_path = mp3_path - if not os.path.isfile(mp3_path) or os.path.getsize(mp3_path) == 0: - audio_path = tts_result.get("file_path") or mp3_path - - if os.path.isfile(audio_path) and os.path.getsize(audio_path) > 0: - play_audio_file(audio_path) - # Clean up - try: - cleanup_paths = {audio_path, mp3_path} - for path in list(cleanup_paths): - ogg_path = path.rsplit(".", 1)[0] + ".ogg" - cleanup_paths.add(ogg_path) - for path in cleanup_paths: - if os.path.isfile(path): - os.unlink(path) - except OSError: - pass - except Exception as e: - logger.warning("Voice TTS playback failed: %s", e) - _cprint(f"{_DIM}TTS playback failed: {e}{_RST}") - finally: - self._voice_tts_done.set() - - - def _voice_full_duplex_listener(self) -> None: - """Full-duplex agent-turn listener: mic live for the WHOLE turn. - - Armed at utterance-submit (chat() start in continuous voice mode) and - disarmed when the turn is fully done (agent finished + TTS played). - Replaces the old per-playback ``_voice_barge_in_monitor``, which only - listened while TTS audio was playing — during LLM generation the mic - was dead, so the user could not interject by voice at all (and the - playback monitor calibrated against its own speaker bleed, making - the trigger unreachable; see tools.voice_mode.full_duplex_listen). - - Phase behaviour: - - * generation (no TTS audio yet): speech interrupts the in-flight - agent turn via ``self.agent.interrupt()`` — the same seam the - typed/Ctrl+C interrupt uses — and the captured utterance is - submitted as the next message. - * playback: speech cuts TTS (pipeline stop event + stop_playback) - and the interruption is captured with pre-roll and submitted. - - The stop phrase ends the voice chat in BOTH phases (a stop during - generation means "stop everything": the turn is already interrupted - at trip time, then ``_voice_submit_barge_utterance`` disables voice - mode). - """ - fd_active = getattr(self, "_voice_fd_active", None) - if fd_active is None: - fd_active = threading.Event() - self._voice_fd_active = fd_active - if fd_active.is_set(): - return # one listener owns the mic for this turn - fd_active.set() - 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 ( - full_duplex_listen, - is_audio_output_active, - stop_playback, - ) - - try: - _mult = float(voice_cfg.get("barge_in_threshold_multiplier", 0) or 0) - except (TypeError, ValueError): - _mult = 0.0 - try: - _grace_ms = int(float(voice_cfg.get("barge_in_grace_seconds", 0.5)) * 1000) - except (TypeError, ValueError): - _grace_ms = 500 - - tts_done = getattr(self, "_voice_tts_done", None) - - def _should_stop() -> bool: - if not (getattr(self, "_voice_mode", False) and getattr(self, "_voice_continuous", False)): - return True - if getattr(self, "_agent_running", False): - return False - # Agent finished — keep listening until TTS fully played. - if tts_done is not None and not tts_done.is_set(): - return False - return not is_audio_output_active() - - def _on_trigger(phase: str) -> None: - # Latch BEFORE cutting anything: suppresses process_loop's - # auto-restart until the capture is submitted. - self._voice_barge_capture.set() - if phase == "playback": - logger.debug( - "TTS CUT: full-duplex listener tripped during playback" - ) - from tools.tts_streaming import mark_speech_interrupted - mark_speech_interrupted() - _pipe_stop = getattr(self, "_voice_tts_stop", None) - if _pipe_stop is not None: - _pipe_stop.set() - stop_playback() - else: - # Generation phase: no audio to cut — interrupt the - # in-flight agent turn (same seam as typed interrupt). - logger.debug( - "full-duplex listener tripped during generation — " - "interrupting agent turn" - ) - _pipe_stop = getattr(self, "_voice_tts_stop", None) - if _pipe_stop is not None: - _pipe_stop.set() # never let the stale reply speak - try: - if self.agent is not None and getattr(self, "_agent_running", False): - _cprint(f"\n{_DIM}🎤 Voice interjection — interrupting…{_RST}") - self.agent.interrupt() - except Exception as e: - logger.debug("voice interjection interrupt failed: %s", e) - - wav_path = full_duplex_listen( - _should_stop, - is_playing=is_audio_output_active, - on_trigger=_on_trigger, - multiplier=_mult or None, - grace_ms=max(0, _grace_ms), - ) - 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 full-duplex listener failed: %s", e) - finally: - fd_active.clear() - - 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: - from tools.voice_mode import is_voice_stop_phrase - if is_voice_stop_phrase(transcript): - _cprint(f"\n{_DIM}Stop phrase detected — ending voice chat.{_RST}") - self._disable_voice_mode() - return - self._pending_input.put(_VoiceInputMessage(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: - from hermes_cli.config import load_config - from utils import is_truthy_value - voice_cfg = load_config().get("voice", {}) - if isinstance(voice_cfg, dict): - # is_truthy_value handles quoted YAML strings like "false" - # which bool() would misread as True (#49883). - return is_truthy_value(voice_cfg.get("beep_enabled", True), default=True) - except Exception: - pass - return True - - def _enable_voice_mode(self): - """Enable voice mode after checking requirements.""" - if self._voice_mode: - _cprint(f"{_DIM}Voice mode is already enabled.{_RST}") - return - - from tools.voice_mode import check_voice_requirements, detect_audio_environment - - # Environment detection -- warn and block in incompatible environments - env_check = detect_audio_environment() - if not env_check["available"]: - _cprint(f"\n{_ACCENT}Voice mode unavailable in this environment:{_RST}") - for warning in env_check["warnings"]: - _cprint(f" {_DIM}{warning}{_RST}") - return - - reqs = check_voice_requirements() - if not reqs["available"]: - _cprint(f"\n{_ACCENT}Voice mode requirements not met:{_RST}") - for line in reqs["details"].split("\n"): - _cprint(f" {_DIM}{line}{_RST}") - if reqs["missing_packages"]: - if _is_termux_environment(): - _cprint(f"\n {_BOLD}Option 1: pkg install termux-api{_RST}") - _cprint(f" {_DIM}Then install/update the Termux:API Android app for microphone capture{_RST}") - _cprint(f" {_BOLD}Option 2: pkg install python-numpy portaudio && python -m pip install sounddevice{_RST}") - else: - _cprint(f"\n {_BOLD}Install: {sys.executable} -m pip install {' '.join(reqs['missing_packages'])}{_RST}") - return - - with self._voice_lock: - self._voice_mode = True - - # Check config for auto_tts (shape-safe — malformed ``voice:`` YAML - # leaves ``voice_config`` as a non-dict, so guard before .get()). - try: - from hermes_cli.config import load_config - _raw_voice = load_config().get("voice") - voice_config = _raw_voice if isinstance(_raw_voice, dict) else {} - if voice_config.get("auto_tts", False): - with self._voice_lock: - self._voice_tts = True - except Exception: - pass - - # Voice mode instruction is injected as a user message prefix (not a - # system prompt change) to avoid invalidating the prompt cache. See - # _voice_message_prefix property and its usage in _process_message(). - - tts_status = " (TTS enabled)" if self._voice_tts else "" - # Use the startup-pinned cache so the advertised shortcut always - # matches the live prompt_toolkit binding — reading live config - # here would drift after a mid-session config edit (Copilot - # round-14 on #19835, same class as round-13). - _ptt_display = self._voice_record_key_label() - _cprint(f"\n{_ACCENT}Voice mode enabled{tts_status}{_RST}") - _cprint(f" {_DIM}{_ptt_display} to start/stop recording{_RST}") - # Spoken-stop hint sourced from voice.stop_phrases (first entry); the - # helper returns "" when stop phrases are disabled — show no hint then. - try: - from tools.voice_mode import voice_stop_hint - _stop_hint = voice_stop_hint() - except Exception: - _stop_hint = "" - if _stop_hint: - _cprint(f" {_DIM}{_stop_hint}{_RST}") - _cprint(f" {_DIM}/voice tts to toggle speech output{_RST}") - _cprint(f" {_DIM}/voice off to disable voice mode{_RST}") - - def _typed_voice_stop(self, user_input) -> bool: - """Typed bare stop phrase during an active voice chat ends the chat. - - Saying "stop" ends the voice chat (PR #73106); TYPING the same bare - stop phrase while voice mode is on must behave identically instead of - sending "stop" to the agent as a turn. Guarded on voice mode being ON - — typed "stop" outside voice chat passes through to the agent exactly - as before. Reuses ``is_voice_stop_phrase`` (same config - ``voice.stop_phrases``, same exact-match semantics), so longer typed - messages containing "stop" are never swallowed. - """ - if not isinstance(user_input, str): - return False - with self._voice_lock: - voice_on = self._voice_mode or self._voice_continuous - if not voice_on: - return False - try: - from tools.voice_mode import is_voice_stop_phrase - if not is_voice_stop_phrase(user_input): - return False - except Exception: - return False - _cprint(f"\n{_DIM}Stop phrase typed — ending voice chat.{_RST}") - self._disable_voice_mode() - return True - - def _disable_voice_mode(self): - """Disable voice mode, cancel any active recording, and stop TTS.""" - recorder = None - with self._voice_lock: - if self._voice_recording and self._voice_recorder: - self._voice_recorder.cancel() - self._voice_recording = False - recorder = self._voice_recorder - self._voice_mode = False - self._voice_tts = False - self._voice_continuous = False - - # Shut down the persistent audio stream in background - if recorder is not None: - def _bg_shutdown(rec=recorder): - try: - rec.shutdown() - except Exception: - pass - threading.Thread(target=_bg_shutdown, daemon=True).start() - self._voice_recorder = None - - # Stop any active TTS playback (file player + streaming pipeline) - try: - if self._voice_tts_stop is not None: - logger.info("TTS CUT: _disable_voice_mode setting stop event") - self._voice_tts_stop.set() - from tools.voice_mode import stop_playback - stop_playback() - except Exception: - pass - self._voice_tts_done.set() - - _cprint(f"\n{_DIM}Voice mode disabled.{_RST}") - # ── Wake word ("Hey Hermes") ───────────────────────────────────────── # # An always-on hotword listener (tools/wake_word.py) that, on detecting @@ -13045,592 +12355,6 @@ def _show_wake_word_status(self): if not owned: _cprint(f" {_DIM}Enable with /wake on{_RST}") - def _toggle_voice_tts(self): - """Toggle TTS output for voice mode.""" - if not self._voice_mode: - _cprint(f"{_DIM}Enable voice mode first: /voice on{_RST}") - return - - with self._voice_lock: - self._voice_tts = not self._voice_tts - status = "enabled" if self._voice_tts else "disabled" - - if self._voice_tts: - from tools.tts_tool import check_tts_requirements - if not check_tts_requirements(): - _cprint(f"{_DIM}Warning: No TTS provider available. Install edge-tts or set API keys.{_RST}") - - _cprint(f"{_ACCENT}Voice TTS {status}.{_RST}") - - def _show_voice_status(self): - """Show current voice mode status.""" - from tools.voice_mode import check_voice_requirements - - reqs = check_voice_requirements() - - _cprint(f"\n{_BOLD}Voice Mode Status{_RST}") - _cprint(f" Mode: {'ON' if self._voice_mode else 'OFF'}") - _cprint(f" TTS: {'ON' if self._voice_tts else 'OFF'}") - _cprint(f" Recording: {'YES' if self._voice_recording else 'no'}") - # Display the startup-pinned label so /voice status always - # matches the live prompt_toolkit binding (Copilot round-14 on - # #19835, same class as round-13). Reading live config here - # would drift after a mid-session config edit. - _cprint(f" Record key: {self._voice_record_key_label()}") - _cprint(f"\n {_BOLD}Requirements:{_RST}") - for line in reqs["details"].split("\n"): - _cprint(f" {line}") - - def _persist_prompt_summary(self, icon: str, label: str, detail: str, outcome: str) -> None: - """Print a one-line scrollback summary of a resolved modal prompt. - - Modal panels (approval / clarify) live in the prompt_toolkit layout and - vanish on the next repaint, so the question and the decision leave no - trace in the terminal scrollback. When display.persist_prompts is on - (default), emit a dim single line after the prompt resolves so the - decision survives in chat history. - """ - if not CLI_CONFIG.get("display", {}).get("persist_prompts", True): - return - detail = " ".join(detail.split()) - if len(detail) > 120: - detail = detail[:119] + "…" - outcome = " ".join(outcome.split()) - if len(outcome) > 120: - outcome = outcome[:119] + "…" - _cprint(f"\n{_DIM}{icon} {label}: {detail} → {outcome}{_RST}") - - def _clarify_callback(self, question, choices, multi_select=False): - """ - Platform callback for the clarify tool. Called from the agent thread. - - Sets up the interactive selection UI (or freetext prompt for open-ended - questions), then blocks until the user responds via the prompt_toolkit - key bindings. If no response arrives within the configured timeout the - question is dismissed and the agent is told to decide on its own. - - When ``multi_select`` is True, shows checkboxes and the user can - select multiple options with Space, confirming with Enter. - """ - import time as _time - - from tools.clarify_gateway import resolve_clarify_timeout - - # Canonical clarify timeout, shared with the gateway/TUI path. `<= 0` - # means unlimited (never auto-skip mid-think) → a null deadline. - timeout = resolve_clarify_timeout(CLI_CONFIG) - response_queue = queue.Queue() - is_open_ended = not choices - # multi-select support: only active when multi_select is True and choices exist - effective_multi = multi_select and not is_open_ended - - self._clarify_state = { - "question": question, - "choices": choices if not is_open_ended else [], - "selected": 0, - # multi-select support - "multi_select": effective_multi, - "selected_indices": set() if effective_multi else None, - "response_queue": response_queue, - } - self._clarify_deadline = None if timeout <= 0 else _time.monotonic() + timeout - # Open-ended questions skip straight to freetext input - self._clarify_freetext = is_open_ended - self._clarify_multi_base = None - - # Trigger an immediate prompt_toolkit repaint from this (non-main) - # thread. Modal prompts must paint at once and must not be gated by the - # _invalidate throttle / resize guard — see _paint_now / _invalidate (#41098). - self._paint_now() - - # Poll for the user's response. The countdown in the hint line updates - # on each repaint; refresh it once a second so the timer stays visible - # while we wait. Selection changes (↑/↓) trigger instant repaints via - # the key bindings. - _last_countdown_refresh = _time.monotonic() - while True: - try: - result = response_queue.get(timeout=1) - self._clarify_deadline = None - self._persist_prompt_summary("?", "Clarify", question, str(result)) - return result - except queue.Empty: - # None deadline = unlimited: never auto-skip, just keep polling. - if self._clarify_deadline is not None: - remaining = self._clarify_deadline - _time.monotonic() - if remaining <= 0: - break - now = _time.monotonic() - if now - _last_countdown_refresh >= 1.0: - _last_countdown_refresh = now - self._paint_now() - - # Timed out — tear down the UI and let the agent decide - self._clarify_state = None - self._clarify_freetext = False - self._clarify_deadline = None - self._clarify_multi_base = None - self._paint_now() - _cprint(f"\n{_DIM}(clarify timed out after {timeout}s — agent will decide){_RST}") - return ( - "The user did not provide a response within the time limit. " - "Use your best judgement to make the choice and proceed." - ) - - def _sudo_password_callback(self) -> str: - """ - Prompt for sudo password through the prompt_toolkit UI. - - Called from the agent thread when a sudo command is encountered. - Uses the same clarify-style mechanism: sets UI state, waits on a - queue for the user's response via the Enter key binding. - """ - import time as _time - - timeout = 45 - response_queue = queue.Queue() - - self._capture_modal_input_snapshot() - self._sudo_state = { - "response_queue": response_queue, - } - self._sudo_deadline = _time.monotonic() + timeout - - # Modal prompt — paint immediately, bypassing the throttle/resize guard - # so the prompt can't be dropped and time out unseen (#41098). - self._paint_now() - - while True: - try: - result = response_queue.get(timeout=1) - self._sudo_state = None - self._sudo_deadline = 0 - self._restore_modal_input_snapshot() - self._paint_now() - if result: - _cprint(f"\n{_DIM} ✓ Password received (cached for session){_RST}") - else: - _cprint(f"\n{_DIM} ⏭ Skipped{_RST}") - return result - except queue.Empty: - remaining = self._sudo_deadline - _time.monotonic() - if remaining <= 0: - break - self._paint_now() - - self._sudo_state = None - self._sudo_deadline = 0 - self._restore_modal_input_snapshot() - self._paint_now() - _cprint(f"\n{_DIM} ⏱ Timeout — continuing without sudo{_RST}") - return "" - - def _approval_callback(self, command: str, description: str, - *, allow_permanent: bool = True, - smart_denied: bool = False) -> str: - """ - Prompt for dangerous command approval through the prompt_toolkit UI. - - Called from the agent thread. Shows a selection UI similar to clarify - with choices: once / session / always / deny. Smart DENY owner - overrides show only once / deny. When allow_permanent is False for - another reason (for example tirith), only 'always' is hidden. - Long commands also get a 'view' option so the full command can be - expanded before deciding. - - Uses _approval_lock to serialize concurrent requests (e.g. from - parallel delegation subtasks) so each prompt gets its own turn - and the shared _approval_state / _approval_deadline aren't clobbered. - """ - import time as _time - - with self._approval_lock: - timeout = int(CLI_CONFIG.get("approvals", {}).get("timeout", 300)) - response_queue = queue.Queue() - - self._approval_state = { - "command": command, - "description": description, - "choices": self._approval_choices( - command, - allow_permanent=allow_permanent, - smart_denied=smart_denied, - ), - "selected": 0, - "response_queue": response_queue, - } - self._approval_deadline = _time.monotonic() + timeout - - # Modal prompt — paint immediately, bypassing the throttle/resize - # guard. A throttled paint here can be silently dropped (250ms - # window collision or in-flight resize), leaving the panel unseen so - # the command is denied on timeout without the user ever seeing it - # (#41098). The countdown refreshes below paint the same way. - self._paint_now() - - _last_countdown_refresh = _time.monotonic() - while True: - try: - result = response_queue.get(timeout=1) - self._approval_state = None - self._approval_deadline = 0 - self._paint_now() - _outcome_labels = { - "once": "allowed once", - "session": "allowed for session", - "always": "added to allowlist", - "deny": "denied", - } - self._persist_prompt_summary( - "⚠", "Approval", command, - _outcome_labels.get(result, str(result)), - ) - return result - except queue.Empty: - remaining = self._approval_deadline - _time.monotonic() - if remaining <= 0: - break - now = _time.monotonic() - if now - _last_countdown_refresh >= 1.0: - _last_countdown_refresh = now - self._paint_now() - - self._approval_state = None - self._approval_deadline = 0 - self._paint_now() - _cprint(f"\n{_DIM} ⏱ Timeout — denying command{_RST}") - self._persist_prompt_summary( - "⚠", "Approval", command, "timed out (no response)", - ) - return "timeout" - - def _approval_choices(self, command: str, *, allow_permanent: bool = True, - smart_denied: bool = False) -> list[str]: - """Return approval choices for a dangerous command prompt.""" - if smart_denied: - choices = ["once", "deny"] - else: - choices = ["once", "session", "always", "deny"] if allow_permanent else ["once", "session", "deny"] - if len(command) > 70: - choices.append("view") - return choices - - def _computer_use_approval_callback(self, action: str, args: dict, summary: str) -> str: - """Adapt the generic approval UI for the computer_use tool. - - The computer_use handler expects verdicts of the form - `approve_once` | `approve_session` | `always_approve` | `deny`. - The CLI's built-in approval UI returns `once` | `session` | `always` - | `deny`. Translate between the two. - """ - # Build a command-ish string so the existing UI renders something - # meaningful. `summary` is already a one-line human description. - verdict = self._approval_callback( - command=f"computer_use: {summary}", - description=f"Allow computer_use to perform `{action}`?", - ) - return { - "once": "approve_once", - "session": "approve_session", - "always": "always_approve", - "deny": "deny", - "timeout": "timeout", - }.get(verdict, "deny") - - def _handle_approval_selection(self) -> None: - """Process the currently selected dangerous-command approval choice.""" - state = self._approval_state - if not state: - return - - selected = state.get("selected", 0) - choices = state.get("choices") - if not isinstance(choices, list): - choices = [] - if not (0 <= selected < len(choices)): - return - - chosen = choices[selected] - if chosen == "view": - state["show_full"] = True - state["choices"] = [choice for choice in choices if choice != "view"] - if state["selected"] >= len(state["choices"]): - state["selected"] = max(0, len(state["choices"]) - 1) - self._invalidate() - return - - state["response_queue"].put(chosen) - self._approval_state = None - self._invalidate() - - def _get_approval_display_fragments(self): - """Render the dangerous-command approval panel for the prompt_toolkit UI. - - Layout priority: title + command + choices must always render, even if - the terminal is short or the description is long. Description is placed - at the bottom of the panel and gets truncated to fit the remaining row - budget. This prevents HSplit from clipping approve/deny off-screen when - tirith findings produce multi-paragraph descriptions or when the user - runs in a compact terminal pane. - """ - state = self._approval_state - if not state: - return [] - - def _panel_box_width(title_text: str, content_lines: list[str], min_width: int = 46, max_width: int = 76) -> int: - term_cols = shutil.get_terminal_size((100, 20)).columns - longest = max([len(title_text)] + [len(line) for line in content_lines] + [min_width - 4]) - inner = min(max(longest + 4, min_width - 2), max_width - 2, max(24, term_cols - 6)) - return inner + 2 - - def _wrap_panel_text(text: str, width: int, subsequent_indent: str = "") -> list[str]: - wrapped = textwrap.wrap( - text, - width=max(8, width), - replace_whitespace=False, - drop_whitespace=False, - subsequent_indent=subsequent_indent, - ) - return wrapped or [""] - - def _append_panel_line(lines, border_style: str, content_style: str, text: str, box_width: int) -> None: - inner_width = max(0, box_width - 2) - lines.append((border_style, "│ ")) - lines.append((content_style, text.ljust(inner_width))) - lines.append((border_style, " │\n")) - - def _append_blank_panel_line(lines, border_style: str, box_width: int) -> None: - lines.append((border_style, "│" + (" " * box_width) + "│\n")) - - command = state["command"] - description = state["description"] - choices = state["choices"] - selected = state.get("selected", 0) - show_full = state.get("show_full", False) - - title = "⚠️ Dangerous Command" - cmd_display = command - choice_labels = { - "once": "Allow once", - "session": "Allow for this session", - "always": "Add to permanent allowlist", - "deny": "Deny", - "view": "Show full command", - } - - preview_lines = _wrap_panel_text(description, 60) - preview_lines.extend(_wrap_panel_text(cmd_display, 60)) - for i, choice in enumerate(choices): - prefix = '❯ ' if i == selected else ' ' - preview_lines.extend(_wrap_panel_text( - f"{prefix}{choice_labels.get(choice, choice)}", - 60, - subsequent_indent=" ", - )) - - box_width = _panel_box_width(title, preview_lines) - inner_text_width = max(8, box_width - 2) - - # Pre-wrap the mandatory content — command + choices must always render. - cmd_wrapped = _wrap_panel_text(cmd_display, inner_text_width) - if not show_full and "view" in choices and len(cmd_wrapped) > 4: - cmd_wrapped = cmd_wrapped[:3] + _wrap_panel_text( - "… (choose Show full command)", - inner_text_width, - ) - - # (choice_index, wrapped_line) so we can re-apply selected styling below - choice_wrapped: list[tuple[int, str]] = [] - for i, choice in enumerate(choices): - label = choice_labels.get(choice, choice) - # Show number prefix for quick selection (1-9 for items 1-9, 0 for 10th item) - if i < 9: - num_prefix = str(i + 1) - elif i == 9: - num_prefix = '0' - else: - num_prefix = ' ' # No number for items beyond 10th - if i == selected: - prefix = f'❯ {num_prefix}. ' - else: - prefix = f' {num_prefix}. ' - for wrapped in _wrap_panel_text(f"{prefix}{label}", inner_text_width, subsequent_indent=" "): - choice_wrapped.append((i, wrapped)) - - # Budget vertical space so HSplit never clips the command or choices. - # Panel chrome (full layout with separators): - # top border + title + blank_after_title - # + blank_between_cmd_choices + bottom border = 5 rows. - # In tight terminals we collapse to: - # top border + title + bottom border = 3 rows (no blanks). - # - # reserved_below: rows consumed below the approval panel by the - # spinner/tool-progress line, status bar, input area, separators, and - # prompt symbol. Measured at ~6 rows during live PTY approval prompts; - # budget 6 so we don't overestimate the panel's room. - term_rows = shutil.get_terminal_size((100, 24)).lines - chrome_full = 5 - chrome_tight = 3 - reserved_below = 6 - - available = max(0, term_rows - reserved_below) - mandatory_full = chrome_full + len(cmd_wrapped) + len(choice_wrapped) - - # If the full-chrome panel doesn't fit, drop the separator blanks. - # This keeps the command and every choice on-screen in compact terminals. - use_compact_chrome = mandatory_full > available - chrome_rows = chrome_tight if use_compact_chrome else chrome_full - - # If the command itself is too long to leave room for choices (e.g. user - # hit "view" on a multi-hundred-character command), truncate it so the - # approve/deny buttons still render. Keep at least 1 row of command. - max_cmd_rows = max(1, available - chrome_rows - len(choice_wrapped)) - if len(cmd_wrapped) > max_cmd_rows: - keep = max(1, max_cmd_rows - 1) if max_cmd_rows > 1 else 1 - cmd_wrapped = cmd_wrapped[:keep] + _wrap_panel_text( - "… (command truncated — use /logs or /debug for full text)", - inner_text_width, - ) - - # Allocate any remaining rows to description. The extra -1 in full mode - # accounts for the blank separator between choices and description. - mandatory_no_desc = chrome_rows + len(cmd_wrapped) + len(choice_wrapped) - desc_sep_cost = 0 if use_compact_chrome else 1 - available_for_desc = available - mandatory_no_desc - desc_sep_cost - # Even on huge terminals, cap description height so the panel stays compact. - available_for_desc = max(0, min(available_for_desc, 10)) - - desc_wrapped = _wrap_panel_text(description, inner_text_width) if description else [] - if available_for_desc < 1 or not desc_wrapped: - desc_wrapped = [] - elif len(desc_wrapped) > available_for_desc: - keep = max(1, available_for_desc - 1) - desc_wrapped = desc_wrapped[:keep] + ["… (description truncated)"] - - # Render: title → command → choices → description (description last so - # any remaining overflow clips from the bottom of the least-critical - # content, never from the command or choices). Use compact chrome (no - # blank separators) when the terminal is tight. - lines = [] - lines.append(('class:approval-border', '╭' + ('─' * box_width) + '╮\n')) - _append_panel_line(lines, 'class:approval-border', 'class:approval-title', title, box_width) - if not use_compact_chrome: - _append_blank_panel_line(lines, 'class:approval-border', box_width) - - for wrapped in cmd_wrapped: - _append_panel_line(lines, 'class:approval-border', 'class:approval-cmd', wrapped, box_width) - if not use_compact_chrome: - _append_blank_panel_line(lines, 'class:approval-border', box_width) - - for i, wrapped in choice_wrapped: - style = 'class:approval-selected' if i == selected else 'class:approval-choice' - _append_panel_line(lines, 'class:approval-border', style, wrapped, box_width) - - if desc_wrapped: - if not use_compact_chrome: - _append_blank_panel_line(lines, 'class:approval-border', box_width) - for wrapped in desc_wrapped: - _append_panel_line(lines, 'class:approval-border', 'class:approval-desc', wrapped, box_width) - - lines.append(('class:approval-border', '╰' + ('─' * box_width) + '╯\n')) - return lines - - def _secret_capture_callback(self, var_name: str, prompt: str, metadata=None) -> dict: - return prompt_for_secret(self, var_name, prompt, metadata) - - def _capture_modal_input_snapshot(self) -> None: - """Temporarily clear the input buffer and save the user's in-progress draft.""" - if self._modal_input_snapshot is not None or not getattr(self, "_app", None): - return - try: - buf = self._app.current_buffer - self._modal_input_snapshot = { - "text": buf.text, - "cursor_position": buf.cursor_position, - } - buf.reset() - except Exception: - self._modal_input_snapshot = None - - def _restore_modal_input_snapshot(self) -> None: - """Restore any draft text that was present before a modal prompt opened.""" - snapshot = self._modal_input_snapshot - self._modal_input_snapshot = None - if not snapshot or not getattr(self, "_app", None): - return - try: - buf = self._app.current_buffer - buf.text = snapshot.get("text", "") - buf.cursor_position = min(snapshot.get("cursor_position", 0), len(buf.text)) - except Exception: - pass - - def _clear_active_overlays_for_interrupt(self) -> None: - """Drain and clear every input-blocking overlay left by an interrupted agent. - - approval/clarify/sudo/secret prompts each block a worker thread on a - ``response_queue.get()``. When the agent is interrupted the worker - thread is torn down, but the overlay's state dict stays set — leaving - the CLI input gated (``read_only`` condition + keypress filter) with no - thread servicing the prompt. The result is a frozen terminal until the - prompt's own timeout expires. Push a terminal value onto each queue so - any still-blocked thread unblocks cleanly, then nil the state out and - restore the user's pre-modal draft (#14026). - - Safe default per prompt: approval -> "deny", clarify/sudo/secret -> - cancel (None / empty). Each step is wrapped so a dead queue can't - prevent clearing the others. - """ - if self._approval_state: - try: - self._approval_state["response_queue"].put("deny") - except Exception: - pass - self._approval_state = None - if self._clarify_state: - try: - self._clarify_state["response_queue"].put( - "The user cancelled. Use your best judgement to proceed." - ) - except Exception: - pass - self._clarify_state = None - self._clarify_freetext = False - self._clarify_multi_base = None - if self._sudo_state: - try: - self._sudo_state["response_queue"].put("") - except Exception: - pass - self._sudo_state = None - self._sudo_deadline = 0 - self._restore_modal_input_snapshot() - if self._secret_state: - try: - self._cancel_secret_capture() - except Exception: - self._secret_state = None - - def _submit_secret_response(self, value: str) -> None: - if not self._secret_state: - return - self._secret_state["response_queue"].put(value) - self._secret_state = None - self._secret_deadline = 0 - # Modal teardown — paint directly so the secret panel clears at once and - # isn't held by the _invalidate throttle/resize guard (#41098). - self._paint_now() - - def _cancel_secret_capture(self) -> None: - self._submit_secret_response("") - - def _clear_secret_input_buffer(self) -> None: - if getattr(self, "_app", None): - try: - self._app.current_buffer.reset() - except Exception: - pass - def chat(self, message, images: list = None, voice_input: bool = False) -> Optional[str]: """ Send a message to the agent and get a response. diff --git a/hermes_cli/cli_modal_prompts_mixin.py b/hermes_cli/cli_modal_prompts_mixin.py new file mode 100644 index 0000000000000..b3410acb01add --- /dev/null +++ b/hermes_cli/cli_modal_prompts_mixin.py @@ -0,0 +1,579 @@ +"""Modal-prompt handlers (approval / clarify / sudo / secret) for the interactive CLI (god-file decomposition Wave 1). + +This module hosts the modal-prompt methods lifted out of ``cli.py``'s +``HermesCLI`` class (shard s4, cluster c8). ``HermesCLI`` inherits +``CLIModalPromptsMixin`` so every ``self.`` call resolves unchanged +via the MRO — behavior-neutral. + +Import discipline (mirrors ``hermes_cli/cli_commands_mixin.py``, the accepted +Phase-4 decomposition): + * Neutral, non-cyclic deps are imported at module top-level below. + * cli.py-internal symbols (``CLI_CONFIG``/``_cprint``/``_DIM``/``_RST``) + are imported LAZILY inside each method via ``from cli import ...`` — that + resolves at call time when ``cli`` is fully loaded, so this module never + imports ``cli`` at top level (no cycle). +""" + +from __future__ import annotations + +import queue +import shutil +import textwrap + +from hermes_cli.callbacks import prompt_for_secret + + +class CLIModalPromptsMixin: + def _persist_prompt_summary(self, icon: str, label: str, detail: str, outcome: str) -> None: + """Print a one-line scrollback summary of a resolved modal prompt. + + Modal panels (approval / clarify) live in the prompt_toolkit layout and + vanish on the next repaint, so the question and the decision leave no + trace in the terminal scrollback. When display.persist_prompts is on + (default), emit a dim single line after the prompt resolves so the + decision survives in chat history. + """ + from cli import CLI_CONFIG, _cprint, _DIM, _RST + if not CLI_CONFIG.get("display", {}).get("persist_prompts", True): + return + detail = " ".join(detail.split()) + if len(detail) > 120: + detail = detail[:119] + "…" + outcome = " ".join(outcome.split()) + if len(outcome) > 120: + outcome = outcome[:119] + "…" + _cprint(f"\n{_DIM}{icon} {label}: {detail} → {outcome}{_RST}") + + def _clarify_callback(self, question, choices, multi_select=False): + """ + Platform callback for the clarify tool. Called from the agent thread. + + Sets up the interactive selection UI (or freetext prompt for open-ended + questions), then blocks until the user responds via the prompt_toolkit + key bindings. If no response arrives within the configured timeout the + question is dismissed and the agent is told to decide on its own. + + When ``multi_select`` is True, shows checkboxes and the user can + select multiple options with Space, confirming with Enter. + """ + from cli import CLI_CONFIG, _cprint, _DIM, _RST + import time as _time + + from tools.clarify_gateway import resolve_clarify_timeout + + # Canonical clarify timeout, shared with the gateway/TUI path. `<= 0` + # means unlimited (never auto-skip mid-think) → a null deadline. + timeout = resolve_clarify_timeout(CLI_CONFIG) + response_queue = queue.Queue() + is_open_ended = not choices + # multi-select support: only active when multi_select is True and choices exist + effective_multi = multi_select and not is_open_ended + + self._clarify_state = { + "question": question, + "choices": choices if not is_open_ended else [], + "selected": 0, + # multi-select support + "multi_select": effective_multi, + "selected_indices": set() if effective_multi else None, + "response_queue": response_queue, + } + self._clarify_deadline = None if timeout <= 0 else _time.monotonic() + timeout + # Open-ended questions skip straight to freetext input + self._clarify_freetext = is_open_ended + self._clarify_multi_base = None + + # Trigger an immediate prompt_toolkit repaint from this (non-main) + # thread. Modal prompts must paint at once and must not be gated by the + # _invalidate throttle / resize guard — see _paint_now / _invalidate (#41098). + self._paint_now() + + # Poll for the user's response. The countdown in the hint line updates + # on each repaint; refresh it once a second so the timer stays visible + # while we wait. Selection changes (↑/↓) trigger instant repaints via + # the key bindings. + _last_countdown_refresh = _time.monotonic() + while True: + try: + result = response_queue.get(timeout=1) + self._clarify_deadline = None + self._persist_prompt_summary("?", "Clarify", question, str(result)) + return result + except queue.Empty: + # None deadline = unlimited: never auto-skip, just keep polling. + if self._clarify_deadline is not None: + remaining = self._clarify_deadline - _time.monotonic() + if remaining <= 0: + break + now = _time.monotonic() + if now - _last_countdown_refresh >= 1.0: + _last_countdown_refresh = now + self._paint_now() + + # Timed out — tear down the UI and let the agent decide + self._clarify_state = None + self._clarify_freetext = False + self._clarify_deadline = None + self._clarify_multi_base = None + self._paint_now() + _cprint(f"\n{_DIM}(clarify timed out after {timeout}s — agent will decide){_RST}") + return ( + "The user did not provide a response within the time limit. " + "Use your best judgement to make the choice and proceed." + ) + + def _sudo_password_callback(self) -> str: + """ + Prompt for sudo password through the prompt_toolkit UI. + + Called from the agent thread when a sudo command is encountered. + Uses the same clarify-style mechanism: sets UI state, waits on a + queue for the user's response via the Enter key binding. + """ + from cli import _cprint, _DIM, _RST + import time as _time + + timeout = 45 + response_queue = queue.Queue() + + self._capture_modal_input_snapshot() + self._sudo_state = { + "response_queue": response_queue, + } + self._sudo_deadline = _time.monotonic() + timeout + + # Modal prompt — paint immediately, bypassing the throttle/resize guard + # so the prompt can't be dropped and time out unseen (#41098). + self._paint_now() + + while True: + try: + result = response_queue.get(timeout=1) + self._sudo_state = None + self._sudo_deadline = 0 + self._restore_modal_input_snapshot() + self._paint_now() + if result: + _cprint(f"\n{_DIM} ✓ Password received (cached for session){_RST}") + else: + _cprint(f"\n{_DIM} ⏭ Skipped{_RST}") + return result + except queue.Empty: + remaining = self._sudo_deadline - _time.monotonic() + if remaining <= 0: + break + self._paint_now() + + self._sudo_state = None + self._sudo_deadline = 0 + self._restore_modal_input_snapshot() + self._paint_now() + _cprint(f"\n{_DIM} ⏱ Timeout — continuing without sudo{_RST}") + return "" + + def _approval_callback(self, command: str, description: str, + *, allow_permanent: bool = True, + smart_denied: bool = False) -> str: + """ + Prompt for dangerous command approval through the prompt_toolkit UI. + + Called from the agent thread. Shows a selection UI similar to clarify + with choices: once / session / always / deny. Smart DENY owner + overrides show only once / deny. When allow_permanent is False for + another reason (for example tirith), only 'always' is hidden. + Long commands also get a 'view' option so the full command can be + expanded before deciding. + + Uses _approval_lock to serialize concurrent requests (e.g. from + parallel delegation subtasks) so each prompt gets its own turn + and the shared _approval_state / _approval_deadline aren't clobbered. + """ + from cli import CLI_CONFIG, _cprint, _DIM, _RST + import time as _time + + with self._approval_lock: + timeout = int(CLI_CONFIG.get("approvals", {}).get("timeout", 300)) + response_queue = queue.Queue() + + self._approval_state = { + "command": command, + "description": description, + "choices": self._approval_choices( + command, + allow_permanent=allow_permanent, + smart_denied=smart_denied, + ), + "selected": 0, + "response_queue": response_queue, + } + self._approval_deadline = _time.monotonic() + timeout + + # Modal prompt — paint immediately, bypassing the throttle/resize + # guard. A throttled paint here can be silently dropped (250ms + # window collision or in-flight resize), leaving the panel unseen so + # the command is denied on timeout without the user ever seeing it + # (#41098). The countdown refreshes below paint the same way. + self._paint_now() + + _last_countdown_refresh = _time.monotonic() + while True: + try: + result = response_queue.get(timeout=1) + self._approval_state = None + self._approval_deadline = 0 + self._paint_now() + _outcome_labels = { + "once": "allowed once", + "session": "allowed for session", + "always": "added to allowlist", + "deny": "denied", + } + self._persist_prompt_summary( + "⚠", "Approval", command, + _outcome_labels.get(result, str(result)), + ) + return result + except queue.Empty: + remaining = self._approval_deadline - _time.monotonic() + if remaining <= 0: + break + now = _time.monotonic() + if now - _last_countdown_refresh >= 1.0: + _last_countdown_refresh = now + self._paint_now() + + self._approval_state = None + self._approval_deadline = 0 + self._paint_now() + _cprint(f"\n{_DIM} ⏱ Timeout — denying command{_RST}") + self._persist_prompt_summary( + "⚠", "Approval", command, "timed out (no response)", + ) + return "timeout" + + def _approval_choices(self, command: str, *, allow_permanent: bool = True, + smart_denied: bool = False) -> list[str]: + """Return approval choices for a dangerous command prompt.""" + if smart_denied: + choices = ["once", "deny"] + else: + choices = ["once", "session", "always", "deny"] if allow_permanent else ["once", "session", "deny"] + if len(command) > 70: + choices.append("view") + return choices + + def _computer_use_approval_callback(self, action: str, args: dict, summary: str) -> str: + """Adapt the generic approval UI for the computer_use tool. + + The computer_use handler expects verdicts of the form + `approve_once` | `approve_session` | `always_approve` | `deny`. + The CLI's built-in approval UI returns `once` | `session` | `always` + | `deny`. Translate between the two. + """ + # Build a command-ish string so the existing UI renders something + # meaningful. `summary` is already a one-line human description. + verdict = self._approval_callback( + command=f"computer_use: {summary}", + description=f"Allow computer_use to perform `{action}`?", + ) + return { + "once": "approve_once", + "session": "approve_session", + "always": "always_approve", + "deny": "deny", + "timeout": "timeout", + }.get(verdict, "deny") + + def _handle_approval_selection(self) -> None: + """Process the currently selected dangerous-command approval choice.""" + state = self._approval_state + if not state: + return + + selected = state.get("selected", 0) + choices = state.get("choices") + if not isinstance(choices, list): + choices = [] + if not (0 <= selected < len(choices)): + return + + chosen = choices[selected] + if chosen == "view": + state["show_full"] = True + state["choices"] = [choice for choice in choices if choice != "view"] + if state["selected"] >= len(state["choices"]): + state["selected"] = max(0, len(state["choices"]) - 1) + self._invalidate() + return + + state["response_queue"].put(chosen) + self._approval_state = None + self._invalidate() + + def _get_approval_display_fragments(self): + """Render the dangerous-command approval panel for the prompt_toolkit UI. + + Layout priority: title + command + choices must always render, even if + the terminal is short or the description is long. Description is placed + at the bottom of the panel and gets truncated to fit the remaining row + budget. This prevents HSplit from clipping approve/deny off-screen when + tirith findings produce multi-paragraph descriptions or when the user + runs in a compact terminal pane. + """ + state = self._approval_state + if not state: + return [] + + def _panel_box_width(title_text: str, content_lines: list[str], min_width: int = 46, max_width: int = 76) -> int: + term_cols = shutil.get_terminal_size((100, 20)).columns + longest = max([len(title_text)] + [len(line) for line in content_lines] + [min_width - 4]) + inner = min(max(longest + 4, min_width - 2), max_width - 2, max(24, term_cols - 6)) + return inner + 2 + + def _wrap_panel_text(text: str, width: int, subsequent_indent: str = "") -> list[str]: + wrapped = textwrap.wrap( + text, + width=max(8, width), + replace_whitespace=False, + drop_whitespace=False, + subsequent_indent=subsequent_indent, + ) + return wrapped or [""] + + def _append_panel_line(lines, border_style: str, content_style: str, text: str, box_width: int) -> None: + inner_width = max(0, box_width - 2) + lines.append((border_style, "│ ")) + lines.append((content_style, text.ljust(inner_width))) + lines.append((border_style, " │\n")) + + def _append_blank_panel_line(lines, border_style: str, box_width: int) -> None: + lines.append((border_style, "│" + (" " * box_width) + "│\n")) + + command = state["command"] + description = state["description"] + choices = state["choices"] + selected = state.get("selected", 0) + show_full = state.get("show_full", False) + + title = "⚠️ Dangerous Command" + cmd_display = command + choice_labels = { + "once": "Allow once", + "session": "Allow for this session", + "always": "Add to permanent allowlist", + "deny": "Deny", + "view": "Show full command", + } + + preview_lines = _wrap_panel_text(description, 60) + preview_lines.extend(_wrap_panel_text(cmd_display, 60)) + for i, choice in enumerate(choices): + prefix = '❯ ' if i == selected else ' ' + preview_lines.extend(_wrap_panel_text( + f"{prefix}{choice_labels.get(choice, choice)}", + 60, + subsequent_indent=" ", + )) + + box_width = _panel_box_width(title, preview_lines) + inner_text_width = max(8, box_width - 2) + + # Pre-wrap the mandatory content — command + choices must always render. + cmd_wrapped = _wrap_panel_text(cmd_display, inner_text_width) + if not show_full and "view" in choices and len(cmd_wrapped) > 4: + cmd_wrapped = cmd_wrapped[:3] + _wrap_panel_text( + "… (choose Show full command)", + inner_text_width, + ) + + # (choice_index, wrapped_line) so we can re-apply selected styling below + choice_wrapped: list[tuple[int, str]] = [] + for i, choice in enumerate(choices): + label = choice_labels.get(choice, choice) + # Show number prefix for quick selection (1-9 for items 1-9, 0 for 10th item) + if i < 9: + num_prefix = str(i + 1) + elif i == 9: + num_prefix = '0' + else: + num_prefix = ' ' # No number for items beyond 10th + if i == selected: + prefix = f'❯ {num_prefix}. ' + else: + prefix = f' {num_prefix}. ' + for wrapped in _wrap_panel_text(f"{prefix}{label}", inner_text_width, subsequent_indent=" "): + choice_wrapped.append((i, wrapped)) + + # Budget vertical space so HSplit never clips the command or choices. + # Panel chrome (full layout with separators): + # top border + title + blank_after_title + # + blank_between_cmd_choices + bottom border = 5 rows. + # In tight terminals we collapse to: + # top border + title + bottom border = 3 rows (no blanks). + # + # reserved_below: rows consumed below the approval panel by the + # spinner/tool-progress line, status bar, input area, separators, and + # prompt symbol. Measured at ~6 rows during live PTY approval prompts; + # budget 6 so we don't overestimate the panel's room. + term_rows = shutil.get_terminal_size((100, 24)).lines + chrome_full = 5 + chrome_tight = 3 + reserved_below = 6 + + available = max(0, term_rows - reserved_below) + mandatory_full = chrome_full + len(cmd_wrapped) + len(choice_wrapped) + + # If the full-chrome panel doesn't fit, drop the separator blanks. + # This keeps the command and every choice on-screen in compact terminals. + use_compact_chrome = mandatory_full > available + chrome_rows = chrome_tight if use_compact_chrome else chrome_full + + # If the command itself is too long to leave room for choices (e.g. user + # hit "view" on a multi-hundred-character command), truncate it so the + # approve/deny buttons still render. Keep at least 1 row of command. + max_cmd_rows = max(1, available - chrome_rows - len(choice_wrapped)) + if len(cmd_wrapped) > max_cmd_rows: + keep = max(1, max_cmd_rows - 1) if max_cmd_rows > 1 else 1 + cmd_wrapped = cmd_wrapped[:keep] + _wrap_panel_text( + "… (command truncated — use /logs or /debug for full text)", + inner_text_width, + ) + + # Allocate any remaining rows to description. The extra -1 in full mode + # accounts for the blank separator between choices and description. + mandatory_no_desc = chrome_rows + len(cmd_wrapped) + len(choice_wrapped) + desc_sep_cost = 0 if use_compact_chrome else 1 + available_for_desc = available - mandatory_no_desc - desc_sep_cost + # Even on huge terminals, cap description height so the panel stays compact. + available_for_desc = max(0, min(available_for_desc, 10)) + + desc_wrapped = _wrap_panel_text(description, inner_text_width) if description else [] + if available_for_desc < 1 or not desc_wrapped: + desc_wrapped = [] + elif len(desc_wrapped) > available_for_desc: + keep = max(1, available_for_desc - 1) + desc_wrapped = desc_wrapped[:keep] + ["… (description truncated)"] + + # Render: title → command → choices → description (description last so + # any remaining overflow clips from the bottom of the least-critical + # content, never from the command or choices). Use compact chrome (no + # blank separators) when the terminal is tight. + lines = [] + lines.append(('class:approval-border', '╭' + ('─' * box_width) + '╮\n')) + _append_panel_line(lines, 'class:approval-border', 'class:approval-title', title, box_width) + if not use_compact_chrome: + _append_blank_panel_line(lines, 'class:approval-border', box_width) + + for wrapped in cmd_wrapped: + _append_panel_line(lines, 'class:approval-border', 'class:approval-cmd', wrapped, box_width) + if not use_compact_chrome: + _append_blank_panel_line(lines, 'class:approval-border', box_width) + + for i, wrapped in choice_wrapped: + style = 'class:approval-selected' if i == selected else 'class:approval-choice' + _append_panel_line(lines, 'class:approval-border', style, wrapped, box_width) + + if desc_wrapped: + if not use_compact_chrome: + _append_blank_panel_line(lines, 'class:approval-border', box_width) + for wrapped in desc_wrapped: + _append_panel_line(lines, 'class:approval-border', 'class:approval-desc', wrapped, box_width) + + lines.append(('class:approval-border', '╰' + ('─' * box_width) + '╯\n')) + return lines + + def _secret_capture_callback(self, var_name: str, prompt: str, metadata=None) -> dict: + return prompt_for_secret(self, var_name, prompt, metadata) + + def _capture_modal_input_snapshot(self) -> None: + """Temporarily clear the input buffer and save the user's in-progress draft.""" + if self._modal_input_snapshot is not None or not getattr(self, "_app", None): + return + try: + buf = self._app.current_buffer + self._modal_input_snapshot = { + "text": buf.text, + "cursor_position": buf.cursor_position, + } + buf.reset() + except Exception: + self._modal_input_snapshot = None + + def _restore_modal_input_snapshot(self) -> None: + """Restore any draft text that was present before a modal prompt opened.""" + snapshot = self._modal_input_snapshot + self._modal_input_snapshot = None + if not snapshot or not getattr(self, "_app", None): + return + try: + buf = self._app.current_buffer + buf.text = snapshot.get("text", "") + buf.cursor_position = min(snapshot.get("cursor_position", 0), len(buf.text)) + except Exception: + pass + + def _clear_active_overlays_for_interrupt(self) -> None: + """Drain and clear every input-blocking overlay left by an interrupted agent. + + approval/clarify/sudo/secret prompts each block a worker thread on a + ``response_queue.get()``. When the agent is interrupted the worker + thread is torn down, but the overlay's state dict stays set — leaving + the CLI input gated (``read_only`` condition + keypress filter) with no + thread servicing the prompt. The result is a frozen terminal until the + prompt's own timeout expires. Push a terminal value onto each queue so + any still-blocked thread unblocks cleanly, then nil the state out and + restore the user's pre-modal draft (#14026). + + Safe default per prompt: approval -> "deny", clarify/sudo/secret -> + cancel (None / empty). Each step is wrapped so a dead queue can't + prevent clearing the others. + """ + if self._approval_state: + try: + self._approval_state["response_queue"].put("deny") + except Exception: + pass + self._approval_state = None + if self._clarify_state: + try: + self._clarify_state["response_queue"].put( + "The user cancelled. Use your best judgement to proceed." + ) + except Exception: + pass + self._clarify_state = None + self._clarify_freetext = False + self._clarify_multi_base = None + if self._sudo_state: + try: + self._sudo_state["response_queue"].put("") + except Exception: + pass + self._sudo_state = None + self._sudo_deadline = 0 + self._restore_modal_input_snapshot() + if self._secret_state: + try: + self._cancel_secret_capture() + except Exception: + self._secret_state = None + + def _submit_secret_response(self, value: str) -> None: + if not self._secret_state: + return + self._secret_state["response_queue"].put(value) + self._secret_state = None + self._secret_deadline = 0 + # Modal teardown — paint directly so the secret panel clears at once and + # isn't held by the _invalidate throttle/resize guard (#41098). + self._paint_now() + + def _cancel_secret_capture(self) -> None: + self._submit_secret_response("") + + def _clear_secret_input_buffer(self) -> None: + if getattr(self, "_app", None): + try: + self._app.current_buffer.reset() + except Exception: + pass diff --git a/hermes_cli/cli_voice_mixin.py b/hermes_cli/cli_voice_mixin.py new file mode 100644 index 0000000000000..4830f40f77452 --- /dev/null +++ b/hermes_cli/cli_voice_mixin.py @@ -0,0 +1,770 @@ +"""Voice-mode handlers for the interactive CLI (god-file decomposition Wave 1). + +This module hosts the ``_voice_*`` voice-mode methods lifted out of +``cli.py``'s ``HermesCLI`` class (shard s4, cluster c6). ``HermesCLI`` +inherits ``CLIVoiceMixin`` so every ``self.`` call resolves unchanged +via the MRO — behavior-neutral. + +Import discipline (mirrors ``hermes_cli/cli_commands_mixin.py``, the accepted +Phase-4 decomposition): + * Neutral, non-cyclic deps are imported at module top-level below. + * cli.py-internal symbols (``_cprint``/``_DIM``/``_RST``/``_ACCENT``/ + ``_BOLD``/``logger``/``_VoiceInputMessage``) are imported LAZILY inside + each method via ``from cli import ...`` — that resolves at call time when + ``cli`` is fully loaded, so this module never imports ``cli`` at top + level (no cycle). +""" + +from __future__ import annotations + +import json +import os +import re +import sys +import tempfile +import threading +import time +from typing import Optional + +from hermes_constants import is_termux as _is_termux_environment + + +class CLIVoiceMixin: + # ==================================================================== + # Voice mode methods + # ==================================================================== + + def _voice_start_recording(self): + """Start capturing audio from the microphone.""" + from cli import _ACCENT, _cprint, _DIM, _RST + if getattr(self, '_should_exit', False): + return + from tools.voice_mode import create_audio_recorder, check_voice_requirements + + reqs = check_voice_requirements() + if not reqs["audio_available"]: + if _is_termux_environment(): + details = reqs.get("details", "") + if "Termux:API Android app is not installed" in details: + raise RuntimeError( + "Termux:API command package detected, but the Android app is missing.\n" + "Install/update the Termux:API Android app, then retry /voice on.\n" + "Fallback: pkg install python-numpy portaudio && python -m pip install sounddevice" + ) + raise RuntimeError( + "Voice mode requires either Termux:API microphone access or Python audio libraries.\n" + "Option 1: pkg install termux-api and install the Termux:API Android app\n" + "Option 2: pkg install python-numpy portaudio && python -m pip install sounddevice" + ) + raise RuntimeError( + "Voice mode requires sounddevice and numpy.\n" + f"Install with: {sys.executable} -m pip install sounddevice numpy" + ) + if not reqs.get("stt_available", reqs.get("stt_key_set")): + raise RuntimeError( + "Voice mode requires an STT provider for transcription.\n" + "Option 1: uv pip install faster-whisper " + "(free, local; `pip install faster-whisper` also works if pip is on PATH)\n" + "Option 2: Set GROQ_API_KEY (free tier)\n" + "Option 3: Set VOICE_TOOLS_OPENAI_KEY (paid)" + ) + + # Prevent double-start from concurrent threads (atomic check-and-set) + with self._voice_lock: + if self._voice_recording: + return + self._voice_recording = True + + # Load silence detection params from config. Shape-safe: a + # hand-edited ``voice: true`` / ``voice: cmd+b`` leaves + # ``load_config()['voice']`` as a non-dict; coerce to {} so + # continuous recording falls back to the documented defaults + # instead of crashing on ``.get()``. + voice_cfg: dict = {} + try: + from hermes_cli.config import load_config + _cfg = load_config().get("voice") + voice_cfg = _cfg if isinstance(_cfg, dict) else {} + except Exception: + pass + + # Recorder creation can fail (no input device, PortAudio init error). + # Reset the flag on failure or _voice_recording stays True forever and + # every future voice start is silently skipped by the guard above. + if self._voice_recorder is None: + try: + self._voice_recorder = create_audio_recorder() + except Exception: + with self._voice_lock: + self._voice_recording = False + raise + + # Apply config-driven silence params (numeric-guarded so YAML + # scalar corruption doesn't break recording start-up). + # + # ``bool`` is explicitly excluded from the numeric check — in + # Python bool is a subclass of int, so a hand-edited + # ``silence_threshold: true`` would otherwise be forwarded as + # ``1`` instead of falling back to the 200 default (Copilot + # round-12 on #19835). + _threshold = voice_cfg.get("silence_threshold") + _duration = voice_cfg.get("silence_duration") + self._voice_recorder._silence_threshold = ( + _threshold if isinstance(_threshold, (int, float)) and not isinstance(_threshold, bool) else 200 + ) + self._voice_recorder._silence_duration = ( + _duration if isinstance(_duration, (int, float)) and not isinstance(_duration, bool) else 3.0 + ) + # voice.max_recording_seconds — hard cap on a single recording's length. + # Same numeric guard as the silence params (bool excluded: a hand-edited + # ``max_recording_seconds: true`` must not become ``1`` — it falls back + # to the documented 120 default, mirroring the silence-param handling). + # An explicit numeric value <= 0 disables the cap. Previously this + # documented key was never read (dead config); wiring it here makes it + # take effect. + _max_rec = voice_cfg.get("max_recording_seconds") + self._voice_recorder._max_recording_seconds = ( + (_max_rec if _max_rec > 0 else 0.0) + if isinstance(_max_rec, (int, float)) and not isinstance(_max_rec, bool) + else 120.0 + ) + + def _on_silence(): + """Called by AudioRecorder when silence is detected after speech.""" + with self._voice_lock: + if not self._voice_recording: + return + _cprint(f"\n{_DIM}Silence detected, auto-stopping...{_RST}") + if hasattr(self, '_app') and self._app: + self._app.invalidate() + self._voice_stop_and_transcribe() + + # Audio cue: single beep BEFORE starting stream (avoid CoreAudio conflict) + if self._voice_beeps_enabled(): + try: + from tools.voice_mode import play_beep + play_beep(frequency=880, count=1) + except Exception: + pass + + try: + self._voice_recorder.start(on_silence_stop=_on_silence) + except Exception: + with self._voice_lock: + self._voice_recording = False + raise + _label = self._voice_record_key_label() + if getattr(self._voice_recorder, "supports_silence_autostop", True): + _recording_hint = f"auto-stops on silence | {_label} to stop & exit continuous" + elif _is_termux_environment(): + _recording_hint = f"Termux:API capture | {_label} to stop" + else: + _recording_hint = f"{_label} to stop" + _cprint(f"\n{_ACCENT}● Recording...{_RST} {_DIM}({_recording_hint}){_RST}") + + # Periodically refresh prompt to update audio level indicator + def _refresh_level(): + while True: + with self._voice_lock: + still_recording = self._voice_recording + if not still_recording: + break + if hasattr(self, '_app') and self._app: + self._app.invalidate() + 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. + + For the local provider, prefer stt.local.model (default ``base``) so the + CLI passes a real model name into the local STT backend. + """ + try: + from hermes_cli.config import load_config + stt_config = load_config().get("stt", {}) + if not isinstance(stt_config, dict): + return None + provider = str(stt_config.get("provider") or "").strip().lower() + if provider == "local": + local_config = stt_config.get("local") or {} + if not isinstance(local_config, dict): + local_config = {} + return local_config.get("model") or "base" + return stt_config.get("model") + except Exception: + return None + + def _voice_stt_provider(self) -> str: + """Configured STT provider name (lowercased), or empty string.""" + try: + from hermes_cli.config import load_config + stt_config = load_config().get("stt", {}) + if not isinstance(stt_config, dict): + return "" + return str(stt_config.get("provider") or "").strip().lower() + except Exception: + return "" + + def _voice_restart_recording_async(self) -> None: + """Restart continuous-mode recording off-thread (start() can block).""" + from cli import _cprint, _DIM, _RST + 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.""" + from cli import _VoiceInputMessage, _cprint, _DIM, _RST + # Atomic guard: only one thread can enter stop-and-transcribe. + # Set _voice_processing immediately so concurrent Ctrl+B presses + # don't race into the START path while recorder.stop() holds its lock. + with self._voice_lock: + if not self._voice_recording: + return + self._voice_recording = False + self._voice_processing = True + + submitted = False + transcription_failed = False + wav_path = None + try: + if self._voice_recorder is None: + return + + wav_path = self._voice_recorder.stop() + + # Audio cue: double beep after stream stopped (no CoreAudio conflict) + if self._voice_beeps_enabled(): + try: + from tools.voice_mode import play_beep + play_beep(frequency=660, count=2) + except Exception: + pass + + if wav_path is None: + _cprint(f"{_DIM}No speech detected.{_RST}") + return + + # _voice_processing is already True (set atomically above) + if hasattr(self, '_app') and self._app: + self._app.invalidate() + + stt_model = self._voice_stt_model() + if self._voice_stt_provider() == "local": + _cprint( + f"{_DIM}Preparing local STT model '{stt_model}' " + f"(first use may download it from Hugging Face)...{_RST}" + ) + else: + _cprint(f"{_DIM}Transcribing...{_RST}") + + from tools.voice_mode import transcribe_recording + result = transcribe_recording(wav_path, model=stt_model) + + if result.get("success") and result.get("transcript", "").strip(): + transcript = result["transcript"].strip() + from tools.voice_mode import is_voice_stop_phrase + if is_voice_stop_phrase(transcript): + # Bare "stop" (or configured phrase) ends the voice chat + # instead of being sent to the agent. + _cprint(f"{_DIM}Stop phrase detected — ending voice chat.{_RST}") + self._disable_voice_mode() + return + self._attached_images.clear() + if hasattr(self, '_app') and self._app: + self._app.invalidate() + self._pending_input.put(_VoiceInputMessage(transcript)) + submitted = True + elif result.get("success"): + _cprint(f"{_DIM}No speech detected.{_RST}") + else: + error = result.get("error", "Unknown error") + _cprint(f"\n{_DIM}Transcription failed: {error}{_RST}") + transcription_failed = True + + except Exception as e: + _cprint(f"\n{_DIM}Voice processing error: {e}{_RST}") + transcription_failed = wav_path is not None + finally: + with self._voice_lock: + self._voice_processing = False + if hasattr(self, '_app') and self._app: + self._app.invalidate() + # Clean up temp file unless transcription failed. On failure, keep + # the source recording so long dictation is not lost. + try: + if wav_path and os.path.isfile(wav_path): + if transcription_failed: + _cprint(f"{_DIM}Recording preserved at: {wav_path}{_RST}") + else: + os.unlink(wav_path) + except Exception: + pass + + # Track consecutive no-speech cycles to avoid infinite restart loops. + # While the agent is mid-turn or TTS is speaking, the user is + # CORRECTLY silent (waiting/listening) — those cycles must not + # count, or a multi-minute tool run ends the voice chat under + # the user. The stop phrase and barge-in still work during the + # hold (they run on their own paths above). + stop_continuous_restart = False + _tts_done = getattr(self, "_voice_tts_done", None) + _activity_hold = bool( + getattr(self, "_agent_running", False) + or (_tts_done is not None and not _tts_done.is_set()) + ) + if not submitted: + if _activity_hold: + pass # held: keep listening without counting the cycle + else: + self._no_speech_count = getattr(self, '_no_speech_count', 0) + 1 + if self._no_speech_count >= 3: + self._voice_continuous = False + self._no_speech_count = 0 + _cprint(f"{_DIM}No speech detected 3 times, continuous mode stopped.{_RST}") + stop_continuous_restart = True + else: + self._no_speech_count = 0 + + # If no transcript was submitted but continuous mode is active, + # restart recording so the user can keep talking. + # (When transcript IS submitted, process_loop handles restart + # after chat() completes.) + if ( + self._voice_continuous + and not submitted + and not self._voice_recording + and not stop_continuous_restart + ): + 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.""" + if not self._voice_tts or not text: + return + self._voice_tts_done.clear() + threading.Thread( + target=self._voice_speak_response, + args=(text,), + daemon=True, + ).start() + # Spoken barge-in must work on the whole-file fallback path too. The + # full-duplex agent-turn listener normally already covers playback + # (armed at turn start in chat()); this arm is an idempotent safety + # net for speak calls outside a chat turn — the listener refuses to + # double-arm via _voice_fd_active. + if self._voice_continuous: + threading.Thread( + target=self._voice_full_duplex_listener, + daemon=True, + ).start() + + def _voice_speak_response(self, text: str): + """Speak the agent's response aloud using TTS (runs in background thread).""" + from cli import _cprint, _DIM, _RST, logger + if not self._voice_tts: + return + self._voice_tts_done.clear() + try: + from tools.tts_tool import text_to_speech_tool + from tools.voice_mode import play_audio_file + + # Strip markdown and non-speech content for cleaner TTS via the + # shared cleaner (tools/tts_text_normalize): markdown, emoji, + # blocks, verifier footer, units, newline flattening. + try: + from tools.tts_text_normalize import prepare_spoken_text + tts_text = prepare_spoken_text(text, max_chars=4000) + except Exception: + # Legacy fallback pipeline — keep voice replies best-effort. + tts_text = text[:4000] if len(text) > 4000 else text + tts_text = re.sub(r'```[\s\S]*?```', ' ', tts_text) # fenced code blocks + tts_text = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', tts_text) # [text](url) -> text + tts_text = re.sub(r'https?://\S+', '', tts_text) # URLs + tts_text = re.sub(r'\*\*(.+?)\*\*', r'\1', tts_text) # bold + tts_text = re.sub(r'\*(.+?)\*', r'\1', tts_text) # italic + tts_text = re.sub(r'`(.+?)`', r'\1', tts_text) # inline code + tts_text = re.sub(r'^#+\s*', '', tts_text, flags=re.MULTILINE) # headers + tts_text = re.sub(r'^\s*[-*]\s+', '', tts_text, flags=re.MULTILINE) # list items + tts_text = re.sub(r'---+', '', tts_text) # horizontal rules + tts_text = re.sub(r'\n{3,}', '\n\n', tts_text) # excessive newlines + tts_text = tts_text.strip() + if not tts_text: + return + + # Use MP3 output for CLI playback (afplay doesn't handle OGG well). + # The TTS tool may auto-convert MP3->OGG, but the original MP3 remains. + os.makedirs(os.path.join(tempfile.gettempdir(), "hermes_voice"), exist_ok=True) + mp3_path = os.path.join( + tempfile.gettempdir(), "hermes_voice", + f"tts_{time.strftime('%Y%m%d_%H%M%S')}.mp3", + ) + + raw_result = text_to_speech_tool(text=tts_text, output_path=mp3_path) + try: + tts_result = json.loads(raw_result) if isinstance(raw_result, str) else {} + except Exception: + tts_result = {} + + # Prefer the requested MP3 when the provider produced it. This + # preserves reliable local playback while still supporting + # providers that write to and return a different path. + audio_path = mp3_path + if not os.path.isfile(mp3_path) or os.path.getsize(mp3_path) == 0: + audio_path = tts_result.get("file_path") or mp3_path + + if os.path.isfile(audio_path) and os.path.getsize(audio_path) > 0: + play_audio_file(audio_path) + # Clean up + try: + cleanup_paths = {audio_path, mp3_path} + for path in list(cleanup_paths): + ogg_path = path.rsplit(".", 1)[0] + ".ogg" + cleanup_paths.add(ogg_path) + for path in cleanup_paths: + if os.path.isfile(path): + os.unlink(path) + except OSError: + pass + except Exception as e: + logger.warning("Voice TTS playback failed: %s", e) + _cprint(f"{_DIM}TTS playback failed: {e}{_RST}") + finally: + self._voice_tts_done.set() + + + def _voice_full_duplex_listener(self) -> None: + """Full-duplex agent-turn listener: mic live for the WHOLE turn. + + Armed at utterance-submit (chat() start in continuous voice mode) and + disarmed when the turn is fully done (agent finished + TTS played). + Replaces the old per-playback ``_voice_barge_in_monitor``, which only + listened while TTS audio was playing — during LLM generation the mic + was dead, so the user could not interject by voice at all (and the + playback monitor calibrated against its own speaker bleed, making + the trigger unreachable; see tools.voice_mode.full_duplex_listen). + + Phase behaviour: + + * generation (no TTS audio yet): speech interrupts the in-flight + agent turn via ``self.agent.interrupt()`` — the same seam the + typed/Ctrl+C interrupt uses — and the captured utterance is + submitted as the next message. + * playback: speech cuts TTS (pipeline stop event + stop_playback) + and the interruption is captured with pre-roll and submitted. + + The stop phrase ends the voice chat in BOTH phases (a stop during + generation means "stop everything": the turn is already interrupted + at trip time, then ``_voice_submit_barge_utterance`` disables voice + mode). + """ + from cli import _cprint, _DIM, _RST, logger + fd_active = getattr(self, "_voice_fd_active", None) + if fd_active is None: + fd_active = threading.Event() + self._voice_fd_active = fd_active + if fd_active.is_set(): + return # one listener owns the mic for this turn + fd_active.set() + 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 ( + full_duplex_listen, + is_audio_output_active, + stop_playback, + ) + + try: + _mult = float(voice_cfg.get("barge_in_threshold_multiplier", 0) or 0) + except (TypeError, ValueError): + _mult = 0.0 + try: + _grace_ms = int(float(voice_cfg.get("barge_in_grace_seconds", 0.5)) * 1000) + except (TypeError, ValueError): + _grace_ms = 500 + + tts_done = getattr(self, "_voice_tts_done", None) + + def _should_stop() -> bool: + if not (getattr(self, "_voice_mode", False) and getattr(self, "_voice_continuous", False)): + return True + if getattr(self, "_agent_running", False): + return False + # Agent finished — keep listening until TTS fully played. + if tts_done is not None and not tts_done.is_set(): + return False + return not is_audio_output_active() + + def _on_trigger(phase: str) -> None: + # Latch BEFORE cutting anything: suppresses process_loop's + # auto-restart until the capture is submitted. + self._voice_barge_capture.set() + if phase == "playback": + logger.debug( + "TTS CUT: full-duplex listener tripped during playback" + ) + from tools.tts_streaming import mark_speech_interrupted + mark_speech_interrupted() + _pipe_stop = getattr(self, "_voice_tts_stop", None) + if _pipe_stop is not None: + _pipe_stop.set() + stop_playback() + else: + # Generation phase: no audio to cut — interrupt the + # in-flight agent turn (same seam as typed interrupt). + logger.debug( + "full-duplex listener tripped during generation — " + "interrupting agent turn" + ) + _pipe_stop = getattr(self, "_voice_tts_stop", None) + if _pipe_stop is not None: + _pipe_stop.set() # never let the stale reply speak + try: + if self.agent is not None and getattr(self, "_agent_running", False): + _cprint(f"\n{_DIM}🎤 Voice interjection — interrupting…{_RST}") + self.agent.interrupt() + except Exception as e: + logger.debug("voice interjection interrupt failed: %s", e) + + wav_path = full_duplex_listen( + _should_stop, + is_playing=is_audio_output_active, + on_trigger=_on_trigger, + multiplier=_mult or None, + grace_ms=max(0, _grace_ms), + ) + 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 full-duplex listener failed: %s", e) + finally: + fd_active.clear() + + def _voice_submit_barge_utterance(self, wav_path: str) -> None: + """Transcribe a barge-captured interruption and queue it as the next turn.""" + from cli import _VoiceInputMessage, _cprint, _DIM, _RST + 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: + from tools.voice_mode import is_voice_stop_phrase + if is_voice_stop_phrase(transcript): + _cprint(f"\n{_DIM}Stop phrase detected — ending voice chat.{_RST}") + self._disable_voice_mode() + return + self._pending_input.put(_VoiceInputMessage(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: + from hermes_cli.config import load_config + from utils import is_truthy_value + voice_cfg = load_config().get("voice", {}) + if isinstance(voice_cfg, dict): + # is_truthy_value handles quoted YAML strings like "false" + # which bool() would misread as True (#49883). + return is_truthy_value(voice_cfg.get("beep_enabled", True), default=True) + except Exception: + pass + return True + + def _enable_voice_mode(self): + """Enable voice mode after checking requirements.""" + from cli import _ACCENT, _BOLD, _cprint, _DIM, _RST + if self._voice_mode: + _cprint(f"{_DIM}Voice mode is already enabled.{_RST}") + return + + from tools.voice_mode import check_voice_requirements, detect_audio_environment + + # Environment detection -- warn and block in incompatible environments + env_check = detect_audio_environment() + if not env_check["available"]: + _cprint(f"\n{_ACCENT}Voice mode unavailable in this environment:{_RST}") + for warning in env_check["warnings"]: + _cprint(f" {_DIM}{warning}{_RST}") + return + + reqs = check_voice_requirements() + if not reqs["available"]: + _cprint(f"\n{_ACCENT}Voice mode requirements not met:{_RST}") + for line in reqs["details"].split("\n"): + _cprint(f" {_DIM}{line}{_RST}") + if reqs["missing_packages"]: + if _is_termux_environment(): + _cprint(f"\n {_BOLD}Option 1: pkg install termux-api{_RST}") + _cprint(f" {_DIM}Then install/update the Termux:API Android app for microphone capture{_RST}") + _cprint(f" {_BOLD}Option 2: pkg install python-numpy portaudio && python -m pip install sounddevice{_RST}") + else: + _cprint(f"\n {_BOLD}Install: {sys.executable} -m pip install {' '.join(reqs['missing_packages'])}{_RST}") + return + + with self._voice_lock: + self._voice_mode = True + + # Check config for auto_tts (shape-safe — malformed ``voice:`` YAML + # leaves ``voice_config`` as a non-dict, so guard before .get()). + try: + from hermes_cli.config import load_config + _raw_voice = load_config().get("voice") + voice_config = _raw_voice if isinstance(_raw_voice, dict) else {} + if voice_config.get("auto_tts", False): + with self._voice_lock: + self._voice_tts = True + except Exception: + pass + + # Voice mode instruction is injected as a user message prefix (not a + # system prompt change) to avoid invalidating the prompt cache. See + # _voice_message_prefix property and its usage in _process_message(). + + tts_status = " (TTS enabled)" if self._voice_tts else "" + # Use the startup-pinned cache so the advertised shortcut always + # matches the live prompt_toolkit binding — reading live config + # here would drift after a mid-session config edit (Copilot + # round-14 on #19835, same class as round-13). + _ptt_display = self._voice_record_key_label() + _cprint(f"\n{_ACCENT}Voice mode enabled{tts_status}{_RST}") + _cprint(f" {_DIM}{_ptt_display} to start/stop recording{_RST}") + # Spoken-stop hint sourced from voice.stop_phrases (first entry); the + # helper returns "" when stop phrases are disabled — show no hint then. + try: + from tools.voice_mode import voice_stop_hint + _stop_hint = voice_stop_hint() + except Exception: + _stop_hint = "" + if _stop_hint: + _cprint(f" {_DIM}{_stop_hint}{_RST}") + _cprint(f" {_DIM}/voice tts to toggle speech output{_RST}") + _cprint(f" {_DIM}/voice off to disable voice mode{_RST}") + + def _typed_voice_stop(self, user_input) -> bool: + """Typed bare stop phrase during an active voice chat ends the chat. + + Saying "stop" ends the voice chat (PR #73106); TYPING the same bare + stop phrase while voice mode is on must behave identically instead of + sending "stop" to the agent as a turn. Guarded on voice mode being ON + — typed "stop" outside voice chat passes through to the agent exactly + as before. Reuses ``is_voice_stop_phrase`` (same config + ``voice.stop_phrases``, same exact-match semantics), so longer typed + messages containing "stop" are never swallowed. + """ + from cli import _cprint, _DIM, _RST + if not isinstance(user_input, str): + return False + with self._voice_lock: + voice_on = self._voice_mode or self._voice_continuous + if not voice_on: + return False + try: + from tools.voice_mode import is_voice_stop_phrase + if not is_voice_stop_phrase(user_input): + return False + except Exception: + return False + _cprint(f"\n{_DIM}Stop phrase typed — ending voice chat.{_RST}") + self._disable_voice_mode() + return True + + def _disable_voice_mode(self): + """Disable voice mode, cancel any active recording, and stop TTS.""" + from cli import _cprint, _DIM, _RST, logger + recorder = None + with self._voice_lock: + if self._voice_recording and self._voice_recorder: + self._voice_recorder.cancel() + self._voice_recording = False + recorder = self._voice_recorder + self._voice_mode = False + self._voice_tts = False + self._voice_continuous = False + + # Shut down the persistent audio stream in background + if recorder is not None: + def _bg_shutdown(rec=recorder): + try: + rec.shutdown() + except Exception: + pass + threading.Thread(target=_bg_shutdown, daemon=True).start() + self._voice_recorder = None + + # Stop any active TTS playback (file player + streaming pipeline) + try: + if self._voice_tts_stop is not None: + logger.info("TTS CUT: _disable_voice_mode setting stop event") + self._voice_tts_stop.set() + from tools.voice_mode import stop_playback + stop_playback() + except Exception: + pass + self._voice_tts_done.set() + + _cprint(f"\n{_DIM}Voice mode disabled.{_RST}") + + def _toggle_voice_tts(self): + """Toggle TTS output for voice mode.""" + from cli import _ACCENT, _cprint, _DIM, _RST + if not self._voice_mode: + _cprint(f"{_DIM}Enable voice mode first: /voice on{_RST}") + return + + with self._voice_lock: + self._voice_tts = not self._voice_tts + status = "enabled" if self._voice_tts else "disabled" + + if self._voice_tts: + from tools.tts_tool import check_tts_requirements + if not check_tts_requirements(): + _cprint(f"{_DIM}Warning: No TTS provider available. Install edge-tts or set API keys.{_RST}") + + _cprint(f"{_ACCENT}Voice TTS {status}.{_RST}") + + def _show_voice_status(self): + """Show current voice mode status.""" + from cli import _BOLD, _cprint, _RST + from tools.voice_mode import check_voice_requirements + + reqs = check_voice_requirements() + + _cprint(f"\n{_BOLD}Voice Mode Status{_RST}") + _cprint(f" Mode: {'ON' if self._voice_mode else 'OFF'}") + _cprint(f" TTS: {'ON' if self._voice_tts else 'OFF'}") + _cprint(f" Recording: {'YES' if self._voice_recording else 'no'}") + # Display the startup-pinned label so /voice status always + # matches the live prompt_toolkit binding (Copilot round-14 on + # #19835, same class as round-13). Reading live config here + # would drift after a mid-session config edit. + _cprint(f" Record key: {self._voice_record_key_label()}") + _cprint(f"\n {_BOLD}Requirements:{_RST}") + for line in reqs["details"].split("\n"): + _cprint(f" {line}") diff --git a/tests/cli/test_cli_modal_prompts_mixin_regression.py b/tests/cli/test_cli_modal_prompts_mixin_regression.py new file mode 100644 index 0000000000000..cf053c1035019 --- /dev/null +++ b/tests/cli/test_cli_modal_prompts_mixin_regression.py @@ -0,0 +1,315 @@ +"""Regression tests for the CLIModalPromptsMixin extraction. + +God-file decomposition Wave 1 (cli.py shard s4, cluster c8): the modal-prompt +methods (approval / clarify / sudo / secret) moved verbatim from +``cli.py``'s ``HermesCLI`` into ``hermes_cli/cli_modal_prompts_mixin.py``. +``HermesCLI`` now inherits ``CLIModalPromptsMixin``, so the behavior is +identical via the MRO. + +These tests exercise the mixin through a bare stub host (no HermesCLI, no +prompt_toolkit) and stub the ``cli`` module so the lazy ``from cli import ...`` +lines resolve without importing the full CLI (same isolation trick as +``tests/cli/test_cli_extension_hooks.py``). +""" + +from __future__ import annotations + +import sys +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from hermes_cli.cli_modal_prompts_mixin import CLIModalPromptsMixin + + +class _Stub(CLIModalPromptsMixin): + """Bare mixin host: each test sets only the attributes it exercises.""" + + +@pytest.fixture(autouse=True) +def _cli_stub(): + """Stub the cli module so lazy imports inside moved methods resolve.""" + cli = MagicMock() + cli._cprint = lambda *a, **k: None + cli._DIM = "" + cli._RST = "" + cli._ACCENT = "" + cli._BOLD = "" + cli.CLI_CONFIG = {} + with patch.dict(sys.modules, {"cli": cli}): + yield + + +# -------------------------------------------------------------------------- +# _approval_choices — pure choice-list construction +# -------------------------------------------------------------------------- + +def test_approval_choices_default(): + stub = _Stub() + assert stub._approval_choices("rm -rf /") == ["once", "session", "always", "deny"] + + +def test_approval_choices_no_permanent(): + stub = _Stub() + assert stub._approval_choices("rm -rf /", allow_permanent=False) == ["once", "session", "deny"] + + +def test_approval_choices_smart_denied(): + stub = _Stub() + assert stub._approval_choices("rm -rf /", smart_denied=True) == ["once", "deny"] + + +def test_approval_choices_long_command_adds_view(): + stub = _Stub() + long_cmd = "x" * 80 + assert stub._approval_choices(long_cmd) == ["once", "session", "always", "deny", "view"] + + +# -------------------------------------------------------------------------- +# _computer_use_approval_callback — verdict translation +# -------------------------------------------------------------------------- + +@pytest.mark.parametrize( + ("verdict", "expected"), + [ + ("once", "approve_once"), + ("session", "approve_session"), + ("always", "always_approve"), + ("deny", "deny"), + ("timeout", "timeout"), + ("unexpected", "deny"), + ], +) +def test_computer_use_verdict_mapping(verdict, expected): + stub = _Stub() + stub._approval_callback = MagicMock(return_value=verdict) + result = stub._computer_use_approval_callback("click", {"x": 1}, "click at 10,10") + assert result == expected + stub._approval_callback.assert_called_once() + call = stub._approval_callback.call_args + assert call.kwargs["command"].startswith("computer_use:") + assert "click at 10,10" in call.kwargs["command"] + + +# -------------------------------------------------------------------------- +# _get_approval_display_fragments — pure panel rendering +# -------------------------------------------------------------------------- + +def test_approval_fragments_empty_state(): + stub = _Stub() + stub._approval_state = None + assert stub._get_approval_display_fragments() == [] + + +def _state_with(command="rm -rf /", choices=None, selected=0): + return { + "command": command, + "description": "Delete everything", + "choices": choices if choices is not None else ["once", "session", "always", "deny"], + "selected": selected, + "response_queue": MagicMock(), + } + + +def _is_blank_separator(frag): + """A blank separator row is a single fragment with only borders/spaces.""" + return frag.count("│") >= 2 and frag.strip("│ \n") == "" + + +def test_approval_fragments_renders_full_panel(): + stub = _Stub() + stub._approval_state = _state_with() + with patch("shutil.get_terminal_size", return_value=SimpleNamespace(columns=120, lines=40)): + rendered = stub._get_approval_display_fragments() + text = "".join(frag for _, frag in rendered) + assert "╭" in text and "╰" in text + assert "Dangerous Command" in text + assert "Allow once" in text + assert "Add to permanent allowlist" in text + # full chrome: a blank separator row exists between title and choices + assert any(_is_blank_separator(frag) for _, frag in rendered) + + +def test_approval_fragments_compact_chrome_in_short_terminal(): + stub = _Stub() + stub._approval_state = _state_with() + with patch("shutil.get_terminal_size", return_value=SimpleNamespace(columns=120, lines=12)): + rendered = stub._get_approval_display_fragments() + text = "".join(frag for _, frag in rendered) + assert "╭" in text + assert "Allow once" in text + # compact chrome: no blank separator rows + assert not any(_is_blank_separator(frag) for _, frag in rendered) + + +def test_approval_fragments_truncates_overlong_command(): + stub = _Stub() + stub._approval_state = _state_with(command="z" * 400, choices=["once", "deny"]) + with patch("shutil.get_terminal_size", return_value=SimpleNamespace(columns=100, lines=12)): + rendered = stub._get_approval_display_fragments() + text = "".join(frag for _, frag in rendered) + assert "truncated" in text + assert "Deny" in text # choices still render + + +# -------------------------------------------------------------------------- +# _handle_approval_selection — state-machine transitions +# -------------------------------------------------------------------------- + +def test_handle_approval_selection_no_state_noop(): + stub = _Stub() + stub._approval_state = None + stub._invalidate = MagicMock() + stub._handle_approval_selection() # must not raise + + +def test_handle_approval_selection_submits_chosen(): + stub = _Stub() + queue_mock = MagicMock() + stub._approval_state = _state_with(choices=["once", "deny"], selected=0) + stub._approval_state["response_queue"] = queue_mock + stub._invalidate = MagicMock() + stub._handle_approval_selection() + queue_mock.put.assert_called_once_with("once") + assert stub._approval_state is None + stub._invalidate.assert_called_once() + + +def test_handle_approval_selection_view_expands_command(): + stub = _Stub() + stub._approval_state = _state_with(choices=["once", "session", "deny", "view"], selected=3) + stub._invalidate = MagicMock() + stub._handle_approval_selection() + assert stub._approval_state["show_full"] is True + assert "view" not in stub._approval_state["choices"] + assert stub._approval_state["response_queue"].put.call_count == 0 + stub._invalidate.assert_called_once() + + +def test_handle_approval_selection_out_of_range_noop(): + stub = _Stub() + stub._approval_state = _state_with(choices=["once", "deny"], selected=7) + stub._invalidate = MagicMock() + stub._handle_approval_selection() + assert stub._approval_state["response_queue"].put.call_count == 0 + assert stub._approval_state is not None + + +# -------------------------------------------------------------------------- +# modal input snapshot / secret capture helpers +# -------------------------------------------------------------------------- + +def test_capture_restore_modal_input_snapshot_roundtrip(): + stub = _Stub() + stub._modal_input_snapshot = None + buf = MagicMock() + buf.text = "half-typed draft" + buf.cursor_position = 7 + app = MagicMock() + app.current_buffer = buf + stub._app = app + + stub._capture_modal_input_snapshot() + assert stub._modal_input_snapshot == {"text": "half-typed draft", "cursor_position": 7} + buf.reset.assert_called_once() + + buf.text = "" + buf.cursor_position = 0 + stub._restore_modal_input_snapshot() + assert buf.text == "half-typed draft" + assert buf.cursor_position == 7 + + +def test_capture_modal_snapshot_skips_without_app(): + stub = _Stub() + stub._modal_input_snapshot = None + stub._app = None + stub._capture_modal_input_snapshot() + assert stub._modal_input_snapshot is None + + +def test_restore_modal_snapshot_clears_even_without_app(): + stub = _Stub() + stub._modal_input_snapshot = {"text": "x", "cursor_position": 0} + stub._app = None + stub._restore_modal_input_snapshot() # must not raise + assert stub._modal_input_snapshot is None + + +def test_clear_secret_input_buffer_resets_app_buffer(): + stub = _Stub() + buf = MagicMock() + app = MagicMock() + app.current_buffer = buf + stub._app = app + stub._clear_secret_input_buffer() + buf.reset.assert_called_once() + + +def test_clear_secret_input_buffer_no_app_noop(): + stub = _Stub() + stub._app = None + stub._clear_secret_input_buffer() # must not raise + + +def test_secret_capture_callback_forwards_to_prompt_for_secret(): + stub = _Stub() + with patch( + "hermes_cli.cli_modal_prompts_mixin.prompt_for_secret", + return_value={"ok": True}, + ) as pfs: + result = stub._secret_capture_callback("API_KEY", "Enter key", {"k": 1}) + assert result == {"ok": True} + pfs.assert_called_once_with(stub, "API_KEY", "Enter key", {"k": 1}) + + +# -------------------------------------------------------------------------- +# _clear_active_overlays_for_interrupt — drain every blocked prompt queue +# -------------------------------------------------------------------------- + +def test_clear_active_overlays_drains_queues_and_nils_state(): + stub = _Stub() + stub._modal_input_snapshot = None + stub._app = None + stub._paint_now = MagicMock() + approval_q = MagicMock() + clarify_q = MagicMock() + sudo_q = MagicMock() + stub._approval_state = {"response_queue": approval_q} + stub._clarify_state = {"response_queue": clarify_q} + stub._clarify_freetext = True + stub._clarify_multi_base = "x" + stub._sudo_state = {"response_queue": sudo_q} + stub._sudo_deadline = 123 + stub._secret_state = None + + stub._clear_active_overlays_for_interrupt() + + approval_q.put.assert_called_once_with("deny") + clarify_q.put.assert_called_once() + sudo_q.put.assert_called_once_with("") + assert stub._approval_state is None + assert stub._clarify_state is None + assert stub._clarify_freetext is False + assert stub._clarify_multi_base is None + assert stub._sudo_state is None + assert stub._sudo_deadline == 0 + + +def test_clear_active_overlays_cancels_secret_capture(): + stub = _Stub() + stub._modal_input_snapshot = None + stub._app = None + stub._paint_now = MagicMock() + secret_q = MagicMock() + stub._secret_state = {"response_queue": secret_q} + stub._approval_state = None + stub._clarify_state = None + stub._sudo_state = None + + stub._clear_active_overlays_for_interrupt() + + secret_q.put.assert_called_once_with("") + assert stub._secret_state is None + assert stub._secret_deadline == 0 diff --git a/tests/cli/test_cli_voice_mixin_regression.py b/tests/cli/test_cli_voice_mixin_regression.py new file mode 100644 index 0000000000000..cb593d0a61072 --- /dev/null +++ b/tests/cli/test_cli_voice_mixin_regression.py @@ -0,0 +1,231 @@ +"""Regression tests for the CLIVoiceMixin extraction. + +God-file decomposition Wave 1 (cli.py shard s4, cluster c6): the voice-mode +methods moved verbatim from ``cli.py``'s ``HermesCLI`` into +``hermes_cli/cli_voice_mixin.py``. ``HermesCLI`` now inherits +``CLIVoiceMixin``, so the behavior is identical via the MRO. + +These tests exercise the mixin through a bare stub host and stub the ``cli`` +module so the lazy ``from cli import ...`` lines resolve without importing the +full CLI (same isolation trick as ``tests/cli/test_cli_extension_hooks.py``). +""" + +from __future__ import annotations + +import sys +import threading +from unittest.mock import MagicMock, patch + +import pytest + +from hermes_cli.cli_voice_mixin import CLIVoiceMixin + + +class _Stub(CLIVoiceMixin): + """Bare mixin host: each test sets only the attributes it exercises.""" + + +class _VoiceInputMessage: + """Mirror of cli._VoiceInputMessage used by the stubbed cli module.""" + + def __init__(self, text: str): + self.text = text + + +@pytest.fixture(autouse=True) +def _cli_stub(): + """Stub the cli module so lazy imports inside moved methods resolve.""" + cli = MagicMock() + cli._cprint = lambda *a, **k: None + cli._DIM = "" + cli._RST = "" + cli._ACCENT = "" + cli._BOLD = "" + cli.logger = MagicMock() + cli._VoiceInputMessage = _VoiceInputMessage + with patch.dict(sys.modules, {"cli": cli}): + yield + + +def _lock() -> threading.Lock: + return threading.Lock() + + +# -------------------------------------------------------------------------- +# STT config resolution (pure, config-parsing) +# -------------------------------------------------------------------------- + +def test_stt_model_local_defaults_to_base(): + stub = _Stub() + with patch("hermes_cli.config.load_config", return_value={"stt": {"provider": "local"}}): + assert stub._voice_stt_model() == "base" + + +def test_stt_model_local_explicit_model(): + stub = _Stub() + with patch( + "hermes_cli.config.load_config", + return_value={"stt": {"provider": "local", "local": {"model": "tiny"}}}, + ): + assert stub._voice_stt_model() == "tiny" + + +def test_stt_model_remote_provider_model(): + stub = _Stub() + with patch( + "hermes_cli.config.load_config", + return_value={"stt": {"provider": "groq", "model": "whisper-large-v3"}}, + ): + assert stub._voice_stt_model() == "whisper-large-v3" + + +def test_stt_model_malformed_config_returns_none(): + stub = _Stub() + with patch("hermes_cli.config.load_config", return_value={"stt": "not-a-dict"}): + assert stub._voice_stt_model() is None + with patch("hermes_cli.config.load_config", side_effect=RuntimeError("boom")): + assert stub._voice_stt_model() is None + + +def test_stt_provider_lowercased(): + stub = _Stub() + with patch("hermes_cli.config.load_config", return_value={"stt": {"provider": "GROQ"}}): + assert stub._voice_stt_provider() == "groq" + + +def test_stt_provider_missing_or_malformed(): + stub = _Stub() + with patch("hermes_cli.config.load_config", return_value={}): + assert stub._voice_stt_provider() == "" + with patch("hermes_cli.config.load_config", return_value={"stt": []}): + assert stub._voice_stt_provider() == "" + + +# -------------------------------------------------------------------------- +# beep preference (config parsing with is_truthy_value semantics) +# -------------------------------------------------------------------------- + +def test_beeps_enabled_quoted_false_is_false(): + stub = _Stub() + with patch( + "hermes_cli.config.load_config", + return_value={"voice": {"beep_enabled": "false"}}, + ): + assert stub._voice_beeps_enabled() is False + + +def test_beeps_enabled_default_true(): + stub = _Stub() + with patch("hermes_cli.config.load_config", return_value={}): + assert stub._voice_beeps_enabled() is True + with patch("hermes_cli.config.load_config", side_effect=RuntimeError("boom")): + assert stub._voice_beeps_enabled() is True + + +# -------------------------------------------------------------------------- +# _typed_voice_stop — typed stop-phrase gating +# -------------------------------------------------------------------------- + +def test_typed_voice_stop_non_string_passthrough(): + stub = _Stub() + stub._voice_lock = _lock() + stub._voice_mode = True + stub._voice_continuous = False + assert stub._typed_voice_stop(42) is False + + +def test_typed_voice_stop_requires_voice_mode(): + stub = _Stub() + stub._voice_lock = _lock() + stub._voice_mode = False + stub._voice_continuous = False + assert stub._typed_voice_stop("stop") is False + + +def test_typed_voice_stop_accepts_stop_phrase(): + stub = _Stub() + stub._voice_lock = _lock() + stub._voice_mode = True + stub._voice_continuous = False + stub._disable_voice_mode = MagicMock() + with patch("tools.voice_mode.is_voice_stop_phrase", return_value=True): + assert stub._typed_voice_stop("stop") is True + stub._disable_voice_mode.assert_called_once() + + +def test_typed_voice_stop_non_stop_phrase_passthrough(): + stub = _Stub() + stub._voice_lock = _lock() + stub._voice_mode = True + stub._voice_continuous = False + stub._disable_voice_mode = MagicMock() + with patch("tools.voice_mode.is_voice_stop_phrase", return_value=False): + assert stub._typed_voice_stop("hello") is False + stub._disable_voice_mode.assert_not_called() + + +# -------------------------------------------------------------------------- +# _voice_submit_barge_utterance — transcript queueing / restart fallback +# -------------------------------------------------------------------------- + +def _barge_stub(tmp_path): + stub = _Stub() + stub._voice_lock = _lock() + stub._voice_barge_capture = MagicMock() + stub._pending_input = MagicMock() + stub._voice_restart_recording_async = MagicMock() + stub._disable_voice_mode = MagicMock() + stub._voice_mode = True + stub._voice_continuous = True + stub._voice_recording = False + wav = tmp_path / "barge.wav" + wav.write_bytes(b"RIFF") + return stub, wav + + +def test_barge_utterance_submits_transcript_and_cleans_wav(tmp_path): + stub, wav = _barge_stub(tmp_path) + with patch("hermes_cli.config.load_config", return_value={}), patch( + "tools.voice_mode.transcribe_recording", + return_value={"success": True, "transcript": " hello world "}, + ), patch("tools.voice_mode.is_voice_stop_phrase", return_value=False): + stub._voice_submit_barge_utterance(str(wav)) + assert stub._pending_input.put.call_count == 1 + msg = stub._pending_input.put.call_args[0][0] + assert isinstance(msg, _VoiceInputMessage) + assert msg.text == "hello world" + assert not wav.exists() # cleaned up after successful transcription + stub._voice_restart_recording_async.assert_not_called() + + +def test_barge_utterance_stop_phrase_disables_voice(tmp_path): + stub, wav = _barge_stub(tmp_path) + with patch("hermes_cli.config.load_config", return_value={}), patch( + "tools.voice_mode.transcribe_recording", + return_value={"success": True, "transcript": "stop"}, + ), patch("tools.voice_mode.is_voice_stop_phrase", return_value=True): + stub._voice_submit_barge_utterance(str(wav)) + stub._disable_voice_mode.assert_called_once() + assert stub._pending_input.put.call_count == 0 + + +def test_barge_utterance_failure_restarts_recording(tmp_path): + stub, wav = _barge_stub(tmp_path) + with patch("hermes_cli.config.load_config", return_value={}), patch( + "tools.voice_mode.transcribe_recording", + return_value={"success": False, "error": "no speech"}, + ): + stub._voice_submit_barge_utterance(str(wav)) + stub._voice_restart_recording_async.assert_called_once() + assert stub._pending_input.put.call_count == 0 + + +def test_barge_utterance_no_restart_when_voice_turned_off(tmp_path): + stub, wav = _barge_stub(tmp_path) + stub._voice_mode = False + with patch("hermes_cli.config.load_config", return_value={}), patch( + "tools.voice_mode.transcribe_recording", + return_value={"success": False, "error": "no speech"}, + ): + stub._voice_submit_barge_utterance(str(wav)) + stub._voice_restart_recording_async.assert_not_called()