From 0694614ab73b69dbc0c1dac77e658c9fd305cbee Mon Sep 17 00:00:00 2001 From: 273-B_L0 <273-B_L0@proton.me> Date: Sun, 5 Apr 2026 21:08:44 -0700 Subject: [PATCH] fix(cli): route voice reply behavior by input origin --- cli.py | 98 ++++++++++++++++--- hermes_cli/config.py | 1 + tests/cli/test_cli_init.py | 89 +++++++++++++++++ tests/tools/test_clipboard.py | 36 +++++-- tests/tools/test_voice_cli_integration.py | 39 ++++++-- website/docs/user-guide/configuration.md | 3 +- .../docs/user-guide/features/voice-mode.md | 6 ++ 7 files changed, 241 insertions(+), 31 deletions(-) diff --git a/cli.py b/cli.py index 560b5d9a2e376..ea7eb0c84ae83 100644 --- a/cli.py +++ b/cli.py @@ -38,6 +38,8 @@ import uuid import textwrap from collections import deque +from dataclasses import dataclass +from enum import Enum from urllib.parse import unquote, urlparse from contextlib import contextmanager from pathlib import Path @@ -90,6 +92,30 @@ import threading import queue + +class _CLIInputOrigin(str, Enum): + TEXT = "text" + VOICE = "voice" + + +@dataclass(frozen=True) +class _CLIQueuedInput: + """Structured queue payload for turns that carry origin metadata.""" + + text: str + images: tuple[Path, ...] = () + origin: _CLIInputOrigin = _CLIInputOrigin.TEXT + + +def _normalize_cli_queued_input(payload): + """Normalize structured and legacy CLI queue payloads at one boundary.""" + if isinstance(payload, _CLIQueuedInput): + return payload.text, list(payload.images), payload.origin + if isinstance(payload, tuple): + text, images = payload + return text, images, _CLIInputOrigin.TEXT + return payload, [], _CLIInputOrigin.TEXT + def CanonicalUsage(*args, **kwargs): from agent.usage_pricing import CanonicalUsage as _CanonicalUsage @@ -11298,7 +11324,9 @@ def _voice_stop_and_transcribe(self): self._attached_images.clear() if hasattr(self, '_app') and self._app: self._app.invalidate() - self._pending_input.put(transcript) + self._pending_input.put( + _CLIQueuedInput(transcript, origin=_CLIInputOrigin.VOICE) + ) submitted = True elif result.get("success"): _cprint(f"{_DIM}No speech detected.{_RST}") @@ -11426,6 +11454,30 @@ def _voice_beeps_enabled(self) -> bool: pass return True + def _get_voice_message_reply_mode(self) -> str: + """Return normalized CLI voice reply mode from config.""" + try: + from hermes_cli.config import load_config + + voice_config = load_config().get("voice", {}) + reply_mode = str(voice_config.get("message_reply_mode", "all")).strip().lower() + if reply_mode in ("voice_only", "all"): + return reply_mode + except Exception: + pass + return "all" + + def _should_speak_voice_response(self, input_origin: _CLIInputOrigin | str) -> bool: + """Return whether TTS output is enabled for this input origin.""" + try: + origin = _CLIInputOrigin(input_origin) + except (TypeError, ValueError): + origin = _CLIInputOrigin.TEXT + return self._voice_tts and ( + origin == _CLIInputOrigin.VOICE + or self._get_voice_message_reply_mode() == "all" + ) + def _enable_voice_mode(self): """Enable voice mode after checking requirements.""" if self._voice_mode: @@ -12073,7 +12125,12 @@ def _clear_secret_input_buffer(self) -> None: except Exception: pass - def chat(self, message, images: list = None) -> Optional[str]: + def chat( + self, + message, + images: list = None, + input_origin: _CLIInputOrigin | str = _CLIInputOrigin.TEXT, + ) -> Optional[str]: """ Send a message to the agent and get a response. @@ -12088,6 +12145,7 @@ def chat(self, message, images: list = None) -> Optional[str]: Args: message: The user's message (str or multimodal content list) images: Optional list of Path objects for attached images + input_origin: Where the turn came from ("text" or "voice") Returns: The agent's response, or None on error @@ -12096,6 +12154,11 @@ def chat(self, message, images: list = None) -> Optional[str]: # register secure secret capture here as well. set_secret_capture_callback(self._secret_capture_callback) + try: + input_origin = _CLIInputOrigin(input_origin) + except (TypeError, ValueError): + input_origin = _CLIInputOrigin.TEXT + # Reset the per-turn interrupt flag. Any subsequent path that # discovers an interrupt (below, after run_conversation) will flip # this to True. Early returns (credential refresh failure, etc.) @@ -12235,7 +12298,9 @@ def chat(self, message, images: list = None) -> Optional[str]: stream_callback = None stop_event = None - if self._voice_tts: + should_speak_response = self._should_speak_voice_response(input_origin) + + if should_speak_response: try: from tools.tts_tool import ( _load_tts_config as _load_tts_cfg, @@ -12288,7 +12353,11 @@ def stream_callback(delta: str): # model responds concisely. The prefix is API-call-local only — # run_conversation persists the original clean user message. _voice_prefix = "" - if self._voice_mode and isinstance(message, str): + if ( + self._voice_mode + and input_origin == _CLIInputOrigin.VOICE + and isinstance(message, str) + ): _voice_prefix = ( "[Voice input — respond concisely and conversationally, " "2-3 sentences max. No code blocks or markdown.] " @@ -12710,9 +12779,9 @@ def run_agent(): f"response may be incomplete{_RST}" ) - # Speak response aloud if voice TTS is enabled - # Skip batch TTS when streaming TTS already handled it - if self._voice_tts and response and not use_streaming_tts: + # Speak response aloud only when this turn qualifies for voice output. + # Skip batch TTS when streaming TTS already handled it. + if should_speak_response and response and not use_streaming_tts: self._voice_speak_response_async(response) @@ -15191,10 +15260,11 @@ def process_loop(): # post-resize transient suppression should end here. self._status_bar_suppressed_after_resize = False - # Unpack image payload: (text, [Path, ...]) or plain str - submit_images = [] - if isinstance(user_input, tuple): - user_input, submit_images = user_input + # Normalize plain strings, legacy (text, images) tuples, and + # structured turns carrying input-origin metadata. + user_input, submit_images, input_origin = _normalize_cli_queued_input( + user_input + ) if isinstance(user_input, str): user_input = _strip_leaked_bracketed_paste_wrappers(user_input) @@ -15277,7 +15347,11 @@ def process_loop(): app.invalidate() # Refresh status line try: - self.chat(user_input, images=submit_images or None) + self.chat( + user_input, + images=submit_images or None, + input_origin=input_origin, + ) finally: self._agent_running = False self._spinner_text = "" diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 03570bceba498..d4db7dee13cf2 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -2152,6 +2152,7 @@ def _ensure_hermes_home_managed(home: Path): "max_recording_seconds": 120, "auto_tts": False, "beep_enabled": True, # Play record start/stop beeps in CLI voice mode + "message_reply_mode": "all", "silence_threshold": 200, # RMS below this = silence (0-32767) "silence_duration": 3.0, # Seconds of silence before auto-stop }, diff --git a/tests/cli/test_cli_init.py b/tests/cli/test_cli_init.py index a990f6bf34277..070cb57fad80d 100644 --- a/tests/cli/test_cli_init.py +++ b/tests/cli/test_cli_init.py @@ -325,6 +325,95 @@ def test_voice_and_interrupt_state_initialized_before_run(self): assert hasattr(cli, "_interrupt_queue") assert hasattr(cli, "_pending_input") + def test_voice_reply_policy_covers_text_voice_and_tts_disabled(self): + from cli import _CLIInputOrigin + + cli = _make_cli() + cli._voice_tts = True + cli._get_voice_message_reply_mode = lambda: "voice_only" + assert cli._should_speak_voice_response(_CLIInputOrigin.TEXT) is False + assert cli._should_speak_voice_response(_CLIInputOrigin.VOICE) is True + + cli._get_voice_message_reply_mode = lambda: "all" + assert cli._should_speak_voice_response(_CLIInputOrigin.TEXT) is True + + cli._voice_tts = False + assert cli._should_speak_voice_response(_CLIInputOrigin.VOICE) is False + + def test_invalid_voice_reply_mode_falls_back_to_all(self): + cli = _make_cli() + + with patch( + "hermes_cli.config.load_config", + return_value={"voice": {"message_reply_mode": "invalid"}}, + ): + assert cli._get_voice_message_reply_mode() == "all" + + @staticmethod + def _make_voice_routing_cli(): + cli = _make_cli() + captured = {} + + def fake_run_conversation(**kwargs): + captured.update(kwargs) + return { + "final_response": "Spoken reply", + "messages": [{"role": "assistant", "content": "Spoken reply"}], + "completed": True, + } + + cli._voice_tts = True + cli._voice_mode = True + cli._get_voice_message_reply_mode = lambda: "voice_only" + cli._ensure_runtime_credentials = lambda: True + cli._resolve_turn_agent_config = lambda message: { + "signature": "sig", + "model": None, + "runtime": None, + "label": "test", + } + cli._active_agent_route_signature = "sig" + cli._reset_stream_state = lambda: None + cli._flush_stream = lambda: None + cli._invalidate = lambda *args, **kwargs: None + cli._voice_speak_response_async = MagicMock() + cli.agent = SimpleNamespace( + run_conversation=fake_run_conversation, + interrupt=lambda _msg=None: None, + _active_children=[], + _interrupt_requested=False, + ) + + return cli, captured + + def test_voice_origin_gets_voice_prompt_and_spoken_reply(self): + cli, captured = self._make_voice_routing_cli() + + with patch("tools.tts_tool._load_tts_config", return_value={"provider": "openai"}), patch( + "tools.tts_tool._get_provider", return_value="openai" + ): + response = cli.chat("Hello there", input_origin="voice") + + assert response == "Spoken reply" + assert captured["user_message"].startswith("[Voice input — respond concisely") + assert captured["persist_user_message"] == "Hello there" + cli._voice_speak_response_async.assert_called_once_with("Spoken reply") + + def test_typed_origin_stays_plain_and_silent_in_voice_only_mode(self): + cli, captured = self._make_voice_routing_cli() + + with patch("tools.tts_tool._load_tts_config") as load_tts_config, patch( + "tools.tts_tool._get_provider" + ) as get_tts_provider: + response = cli.chat("Hello there", input_origin="text") + + assert response == "Spoken reply" + assert captured["user_message"] == "Hello there" + assert captured["persist_user_message"] is None + load_tts_config.assert_not_called() + get_tts_provider.assert_not_called() + cli._voice_speak_response_async.assert_not_called() + class TestHistoryDisplay: def test_history_numbers_only_visible_messages_and_summarizes_tools(self, capsys): diff --git a/tests/tools/test_clipboard.py b/tests/tools/test_clipboard.py index 4a3b31ee56a52..124fbdc8a5c7c 100644 --- a/tests/tools/test_clipboard.py +++ b/tests/tools/test_clipboard.py @@ -1079,7 +1079,11 @@ def test_voice_transcript_clears_stale_attached_images(self, cli): cli._voice_stop_and_transcribe() assert cli._attached_images == [] - assert cli._pending_input.get_nowait() == "hello" + from cli import _CLIInputOrigin, _CLIQueuedInput + + assert cli._pending_input.get_nowait() == _CLIQueuedInput( + "hello", origin=_CLIInputOrigin.VOICE + ) # ═════════════════════════════════════════════════════════════════════════ @@ -1087,29 +1091,41 @@ def test_voice_transcript_clears_stale_attached_images(self, cli): # ═════════════════════════════════════════════════════════════════════════ class TestQueueRouting: - """Test that (text, images) tuples are correctly unpacked and routed.""" + """Test production normalization of structured and legacy queue payloads.""" def test_plain_string_stays_string(self): """Regular text input has no images.""" - user_input = "hello world" - submit_images = [] - if isinstance(user_input, tuple): - user_input, submit_images = user_input + from cli import _CLIInputOrigin, _normalize_cli_queued_input + + user_input, submit_images, origin = _normalize_cli_queued_input("hello world") assert user_input == "hello world" assert submit_images == [] + assert origin == _CLIInputOrigin.TEXT def test_tuple_unpacks_text_and_images(self, tmp_path): """(text, images) tuple is correctly split.""" img = tmp_path / "test.png" img.write_bytes(FAKE_PNG) - user_input = ("describe this", [img]) + from cli import _CLIInputOrigin, _normalize_cli_queued_input - submit_images = [] - if isinstance(user_input, tuple): - user_input, submit_images = user_input + user_input, submit_images, origin = _normalize_cli_queued_input( + ("describe this", [img]) + ) assert user_input == "describe this" assert len(submit_images) == 1 assert submit_images[0] == img + assert origin == _CLIInputOrigin.TEXT + + def test_structured_voice_input_preserves_origin(self): + from cli import _CLIInputOrigin, _CLIQueuedInput, _normalize_cli_queued_input + + user_input, submit_images, origin = _normalize_cli_queued_input( + _CLIQueuedInput("hello", origin=_CLIInputOrigin.VOICE) + ) + + assert user_input == "hello" + assert submit_images == [] + assert origin == _CLIInputOrigin.VOICE def test_empty_text_with_images(self, tmp_path): """Images without text — text should be empty string.""" diff --git a/tests/tools/test_voice_cli_integration.py b/tests/tools/test_voice_cli_integration.py index dc6f2061f2c78..53a1506bfb344 100644 --- a/tests/tools/test_voice_cli_integration.py +++ b/tests/tools/test_voice_cli_integration.py @@ -302,14 +302,14 @@ class TestVoiceMessagePrefix: """Voice mode should inject instruction via user message prefix, not by modifying the system prompt (which breaks prompt cache).""" - def test_prefix_added_when_voice_mode_active(self): - """When voice mode is active and message is str, agent_message - should have the voice instruction prefix.""" + def test_prefix_added_only_for_voice_origin(self): + """Only STT-originated prompts should get the voice instruction prefix.""" voice_mode = True + input_origin = "voice" message = "What's the weather like?" agent_message = message - if voice_mode and isinstance(message, str): + if voice_mode and input_origin == "voice" and isinstance(message, str): agent_message = ( "[Voice input — respond concisely and conversationally, " "2-3 sentences max. No code blocks or markdown.] " @@ -319,13 +319,30 @@ def test_prefix_added_when_voice_mode_active(self): assert agent_message.startswith("[Voice input") assert "What's the weather like?" in agent_message + def test_no_prefix_for_typed_input_while_voice_mode_active(self): + """Typed prompts should remain plain text even when voice mode is on.""" + voice_mode = True + input_origin = "text" + message = "What's the weather like?" + + agent_message = message + if voice_mode and input_origin == "voice" and isinstance(message, str): + agent_message = ( + "[Voice input — respond concisely and conversationally, " + "2-3 sentences max. No code blocks or markdown.] " + + message + ) + + assert agent_message == message + def test_no_prefix_when_voice_mode_inactive(self): """When voice mode is off, message passes through unchanged.""" voice_mode = False + input_origin = "voice" message = "What's the weather like?" agent_message = message - if voice_mode and isinstance(message, str): + if voice_mode and input_origin == "voice" and isinstance(message, str): agent_message = ( "[Voice input — respond concisely and conversationally, " "2-3 sentences max. No code blocks or markdown.] " @@ -337,10 +354,11 @@ def test_no_prefix_when_voice_mode_inactive(self): def test_no_prefix_for_multimodal_content(self): """When message is a list (multimodal), no prefix is added.""" voice_mode = True + input_origin = "voice" message = [{"type": "text", "text": "describe this"}, {"type": "image_url"}] agent_message = message - if voice_mode and isinstance(message, str): + if voice_mode and input_origin == "voice" and isinstance(message, str): agent_message = ( "[Voice input — respond concisely and conversationally, " "2-3 sentences max. No code blocks or markdown.] " @@ -353,13 +371,14 @@ def test_history_stays_clean(self): """conversation_history should contain the original message, not the prefixed version.""" voice_mode = True + input_origin = "voice" message = "Hello there" conversation_history = [] conversation_history.append({"role": "user", "content": message}) agent_message = message - if voice_mode and isinstance(message, str): + if voice_mode and input_origin == "voice" and isinstance(message, str): agent_message = ( "[Voice input — respond concisely and conversationally, " "2-3 sentences max. No code blocks or markdown.] " @@ -1184,7 +1203,11 @@ def test_successful_transcription_queues_input( recorder.stop.return_value = "/tmp/test.wav" cli = _make_voice_cli(_voice_recording=True, _voice_recorder=recorder) cli._voice_stop_and_transcribe() - assert cli._pending_input.get_nowait() == "hello world" + from cli import _CLIInputOrigin, _CLIQueuedInput + + assert cli._pending_input.get_nowait() == _CLIQueuedInput( + "hello world", origin=_CLIInputOrigin.VOICE + ) @patch("cli._cprint") @patch("cli.os.unlink") diff --git a/website/docs/user-guide/configuration.md b/website/docs/user-guide/configuration.md index e65399a67d853..6a77d37a4796c 100644 --- a/website/docs/user-guide/configuration.md +++ b/website/docs/user-guide/configuration.md @@ -1576,11 +1576,12 @@ voice: max_recording_seconds: 120 # Hard stop for long recordings auto_tts: false # Enable spoken replies automatically when /voice on beep_enabled: true # Play record start/stop beeps in CLI voice mode + message_reply_mode: "all" # "all" speaks typed + voice turns; "voice_only" speaks voice turns only silence_threshold: 200 # RMS threshold for speech detection silence_duration: 3.0 # Seconds of silence before auto-stop ``` -Use `/voice on` in the CLI to enable microphone mode, `record_key` to start/stop recording, and `/voice tts` to toggle spoken replies. See [Voice Mode](/user-guide/features/voice-mode) for end-to-end setup and platform-specific behavior. +Use `/voice on` in the CLI to enable microphone mode, `record_key` to start/stop recording, and `/voice tts` to toggle spoken replies. With `message_reply_mode: "voice_only"`, typed turns remain text-only while replies to microphone input are spoken; the default `"all"` preserves spoken replies for both input types. See [Voice Mode](/user-guide/features/voice-mode) for end-to-end setup and platform-specific behavior. ## Streaming diff --git a/website/docs/user-guide/features/voice-mode.md b/website/docs/user-guide/features/voice-mode.md index 14a4235e3e7cd..dbbc58ff9957a 100644 --- a/website/docs/user-guide/features/voice-mode.md +++ b/website/docs/user-guide/features/voice-mode.md @@ -142,6 +142,12 @@ Then use these commands inside the CLI: This loop continues until you press **Ctrl+B** during recording (exits continuous mode) or 3 consecutive recordings detect no speech. +By default, enabling TTS speaks replies to both microphone and typed input. Set +`voice.message_reply_mode: "voice_only"` in `~/.hermes/config.yaml` to keep +typed turns text-only while still speaking replies to microphone input. Accepted +values are `"all"` (the default) and `"voice_only"`; the setting has no effect +while TTS is disabled. + :::tip The record key is configurable via `voice.record_key` in `~/.hermes/config.yaml` (default: `ctrl+b`). :::