feat(voice): CLI + TUI tell the model when its spoken reply was cut

CLI: the VAD monitor's playback cut and the record-key interrupt both
mark the latch; the chat path prepends the note via the existing
_prepend_note_to_message channel. TUI: _tts_stream_stop() grows a
user_barge flag (False for /voice off — a mode change isn't an
interruption; no-op when speech already finished), the VAD monitor
marks on cut, prompt.submit accepts interrupted:true from clients, and
_run_prompt_submit pops the latch into the run message.

cli.py

@@ -11296,6 +11296,8 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
 
             def _cut_playback():
                 if not self._voice_tts_done.is_set():
+                    from tools.tts_streaming import mark_speech_interrupted
+                    mark_speech_interrupted()
                     self._voice_barge_capture.set()
                     stop_event.set()
                     stop_playback()
@@ -12302,6 +12304,11 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
                 if _srn:
                     agent_message = _prepend_note_to_message(agent_message, _srn)
                     self._pending_skills_reload_note = None
+                # Barged mid-speech (VAD or record key)? Tell the model it was
+                # cut off — same one-shot, API-local note channel as above.
+                from tools.tts_streaming import SPEECH_INTERRUPTED_NOTE, take_speech_interrupted
+                if take_speech_interrupted():
+                    agent_message = _prepend_note_to_message(agent_message, SPEECH_INTERRUPTED_NOTE)
                 _moa_cfg = getattr(self, "_pending_moa_config", None)
                 self._pending_moa_config = None
                 if _moa_cfg is None:
@@ -14179,6 +14186,8 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
                 # the stop event drains the streaming pipeline if one is live.
                 if not cli_ref._voice_tts_done.is_set():
                     try:
+                        from tools.tts_streaming import mark_speech_interrupted
+                        mark_speech_interrupted()
                         if cli_ref._voice_tts_stop is not None:
                             cli_ref._voice_tts_stop.set()
                         from tools.voice_mode import stop_playback

tests/test_tui_gateway_server.py

@@ -11821,11 +11821,50 @@ def test_tts_stream_begin_barges_in_on_previous_pipeline(monkeypatch):
     server._tts_stream_stop()
 
 
+def test_tts_stream_stop_latches_interruption_for_next_turn(monkeypatch):
+    """Cutting live speech (interrupt / typing barge) marks the latch the next
+    turn's model note consumes; a mode change (user_barge=False) does not."""
+    import tools.tts_streaming as ts
+
+    ts._interrupted_at = None
+    monkeypatch.setenv("HERMES_VOICE_TTS", "1")
+    monkeypatch.setenv("HERMES_VOICE", "0")
+    _fake_tts_modules(monkeypatch)
+
+    server._tts_stream_begin()
+    server._tts_stream_stop()  # default: user barge
+    assert ts.take_speech_interrupted() is True
+
+    server._tts_stream_begin()
+    server._tts_stream_stop(user_barge=False)  # /voice off
+    assert ts.take_speech_interrupted() is False
+
+
+def test_tts_stream_stop_after_natural_finish_does_not_latch(monkeypatch):
+    """Speech that already finished (done set) isn't an interruption."""
+    import tools.tts_streaming as ts
+
+    ts._interrupted_at = None
+    monkeypatch.setenv("HERMES_VOICE_TTS", "1")
+    monkeypatch.setenv("HERMES_VOICE", "0")
+    _fake_tts_modules(monkeypatch)
+
+    server._tts_stream_begin()
+    with server._tts_stream_lock:
+        server._tts_stream_state["done"].set()
+    server._tts_stream_stop()
+    assert ts.take_speech_interrupted() is False
+
+
 def test_tts_stream_vad_barge_in_cuts_pipeline_and_submits_capture(monkeypatch, tmp_path):
     """User speech during playback cuts TTS at the moment of detection
     (voice.interrupted), then the captured interruption is transcribed and
     emitted as voice.transcript so the TUI submits it — complete from its
-    first syllable, no re-record round trip."""
+    first syllable, no re-record round trip. The cut also latches the
+    speech-interrupted note for the next turn."""
+    import tools.tts_streaming as ts
+
+    ts._interrupted_at = None
     monkeypatch.setenv("HERMES_VOICE_TTS", "1")
     monkeypatch.setenv("HERMES_VOICE", "1")
     monkeypatch.setattr(server, "_load_cfg", lambda: {"voice": {"barge_in": True}})
@@ -11859,4 +11898,5 @@ def test_tts_stream_vad_barge_in_cuts_pipeline_and_submits_capture(monkeypatch,
     assert ("voice.interrupted", None) in events
     assert ("voice.transcript", {"text": "stop, actually—"}) in events
     assert not wav.exists()  # capture temp file cleaned up
+    assert ts.take_speech_interrupted() is True  # VAD cut latches the model note
     server._tts_stream_stop()

tui_gateway/server.py

@@ -9745,6 +9745,12 @@ def _(rid, params: dict) -> dict:
     raw_text = params.get("text", "")
     text = sanitize_user_prompt_text(raw_text) if isinstance(raw_text, str) else raw_text
     truncate_user_ordinal = params.get("truncate_before_user_ordinal")
+    if params.get("interrupted"):
+        # Client-side barge-in (desktop VAD / typing over playback) — latch it
+        # so this turn's model message carries the interruption note.
+        from tools.tts_streaming import mark_speech_interrupted
+
+        mark_speech_interrupted()
     session, err = _sess_nowait(params, rid)
     if err:
         return err
@@ -10430,8 +10436,22 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
 
             # Streaming TTS: voice-mode replies are spoken sentence-by-sentence
             # as tokens arrive (CLI parity) instead of after the full turn.
+            # begin() first — it cuts any still-speaking previous turn, and
+            # that cut IS this turn's barge-in, so it must latch before we
+            # consume the latch below.
             tts_queue = _tts_stream_begin()
 
+            # Barged mid-speech? Tell the model (API-message note, same
+            # enrichment channel as attached images) so it can react
+            # ("rude!") instead of being oblivious to its own interruption.
+            from tools.tts_streaming import SPEECH_INTERRUPTED_NOTE, take_speech_interrupted
+
+            if take_speech_interrupted():
+                if isinstance(run_message, str):
+                    run_message = f"{SPEECH_INTERRUPTED_NOTE}\n\n{run_message}"
+                elif isinstance(run_message, list):
+                    run_message = [{"type": "text", "text": SPEECH_INTERRUPTED_NOTE}, *run_message]
+
             def _stream(delta):
                 with session["history_lock"]:
                     _append_inflight_delta(session, delta)
@@ -15568,13 +15588,22 @@ def _tts_stream_begin() -> Optional[queue.Queue]:
     return text_queue
 
 
-def _tts_stream_stop() -> None:
-    """Barge-in: cut any in-flight streaming TTS (new turn, interrupt, /voice off)."""
+def _tts_stream_stop(user_barge: bool = True) -> None:
+    """Cut any in-flight streaming TTS (new turn, interrupt, /voice off).
+
+    *user_barge* latches the interruption for the next turn's model note
+    (``mark_speech_interrupted``) — pass ``False`` for mode changes like
+    ``/voice off`` where the user isn't talking over the reply.
+    """
     global _tts_stream_state
     with _tts_stream_lock:
         state, _tts_stream_state = _tts_stream_state, None
     if state is None:
         return
+    if user_barge and not state["done"].is_set():
+        from tools.tts_streaming import mark_speech_interrupted
+
+        mark_speech_interrupted()
     state["stop"].set()
     try:
         from tools.voice_mode import stop_playback
@@ -15594,6 +15623,7 @@ def _tts_stream_barge_in_monitor(stop: threading.Event, done: threading.Event) -
     lost between detection and the next recording start.
     """
     try:
+        from tools.tts_streaming import mark_speech_interrupted
         from tools.voice_mode import listen_for_speech, stop_playback, transcribe_recording
 
         barged = threading.Event()
@@ -15601,6 +15631,7 @@ def _tts_stream_barge_in_monitor(stop: threading.Event, done: threading.Event) -
         def _cut_playback():
             if not done.is_set():
                 barged.set()
+                mark_speech_interrupted()
                 stop.set()
                 stop_playback()
                 _voice_emit("voice.interrupted")
@@ -15716,7 +15747,7 @@ def _(rid, params: dict) -> dict:
             # Clear TTS so it can be toggled independently after voice is off,
             # and silence any in-flight streaming speech.
             os.environ["HERMES_VOICE_TTS"] = "0"
-            _tts_stream_stop()
+            _tts_stream_stop(user_barge=False)
 
         return _ok(
             rid,
@@ -15734,7 +15765,7 @@ def _(rid, params: dict) -> dict:
         # Runtime-only flag (CLI parity) — see voice.toggle on/off above.
         os.environ["HERMES_VOICE_TTS"] = "1" if new_value else "0"
         if not new_value:
-            _tts_stream_stop()
+            _tts_stream_stop(user_barge=False)
         # Include ``record_key`` on every branch so a /voice tts toggle
         # doesn't reset the TUI's cached shortcut to the default when a
         # user has a custom binding configured (Copilot review, round 2