fix(cli): route voice reply behavior by input origin

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 = ""

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
     },

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):

tests/tools/test_clipboard.py

@@ -1079,37 +1079,53 @@ 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
+        )
 
 
 # ═════════════════════════════════════════════════════════════════════════
 # Level 4: Queue routing — tuple unpacking in process_loop
 # ═════════════════════════════════════════════════════════════════════════
 
 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."""