diff --git a/apps/desktop/src/app/chat/composer/hooks/use-voice-conversation.ts b/apps/desktop/src/app/chat/composer/hooks/use-voice-conversation.ts index dc8e7dcdf138..2ec43c83c0b8 100644 --- a/apps/desktop/src/app/chat/composer/hooks/use-voice-conversation.ts +++ b/apps/desktop/src/app/chat/composer/hooks/use-voice-conversation.ts @@ -3,6 +3,7 @@ import { useCallback, useEffect, useRef, useState } from 'react' import { useI18n } from '@/i18n' import { monitorSpeechDuringPlayback } from '@/lib/voice-barge-in' import { + markVoicePlaybackInterrupted, playSpeechText, type SpeechStreamSession, startSpeechStream, @@ -267,6 +268,7 @@ export function useVoiceConversation({ onSpeech: () => { bargeCapturePendingRef.current = true onBarge() + markVoicePlaybackInterrupted() stopVoicePlayback() }, onUtterance: audio => { diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx b/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx index f83aa1235725..030ed8751fbf 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx @@ -536,6 +536,44 @@ describe('usePromptActions submit / queue drain semantics', () => { ) }) + it('flags prompt.submit with interrupted:true after a voice-playback barge', async () => { + const { markVoicePlaybackInterrupted } = await import('@/lib/voice-playback') + const requestGateway = vi.fn(async () => ({}) as never) + + let handle: HarnessHandle | null = null + await actRender( + (handle = h)} + refreshSessions={async () => undefined} + requestGateway={requestGateway} + /> + ) + + markVoicePlaybackInterrupted() + await handle!.submitText('stop! rude interruption') + + // The latch is one-shot: the flag rides this submit, the next is clean. + expect(requestGateway).toHaveBeenCalledWith( + 'prompt.submit', + { + session_id: RUNTIME_SESSION_ID, + text: 'stop! rude interruption', + interrupted: true + }, + 1_800_000 + ) + + await handle!.submitText('follow-up without a barge') + expect(requestGateway).toHaveBeenLastCalledWith( + 'prompt.submit', + { + session_id: RUNTIME_SESSION_ID, + text: 'follow-up without a barge' + }, + 1_800_000 + ) + }) + it('a fromQueue drain sends even when busyRef is still true on the settle edge', async () => { // busyRef lags $busy by one effect tick on the busy→false settle edge, so a // drained queue send would otherwise hit the busy guard and silently no-op. diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts index 9493227b9da9..d50647a68942 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts @@ -6,7 +6,12 @@ import { type ChatMessage, textPart } from '@/lib/chat-messages' import { optimisticAttachmentRef } from '@/lib/chat-runtime' import { sanitizeComposerInput } from '@/lib/composer-input-sanitize' import { setMutableRef } from '@/lib/mutable-ref' -import { isVoicePlaybackActive, stopVoicePlayback } from '@/lib/voice-playback' +import { + isVoicePlaybackActive, + markVoicePlaybackInterrupted, + stopVoicePlayback, + takeVoicePlaybackInterrupted +} from '@/lib/voice-playback' import { $composerAttachments, clearComposerAttachments, @@ -144,9 +149,14 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { // Typing barge-in: a new send silences any in-flight spoken reply. if (isVoicePlaybackActive()) { + markVoicePlaybackInterrupted() stopVoicePlayback() } + // Barged mid-speech (here or via the voice loop's VAD)? Flag the submit + // so the backend notes the interruption to the model. + const interrupted = takeVoicePlaybackInterrupted() + // Queue drains carry their source session explicitly. A background drain // must never inherit the currently selected session after the user moves // to another chat. @@ -456,6 +466,12 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { rewriteOptimistic(sessionId) const text = buildContextText(syncedAttachments) + const submitParams = (targetId: string) => ({ + session_id: targetId, + text, + ...(interrupted && { interrupted }) + }) + // On sleep/wake the gateway's in-memory session may have been cleared // while the desktop app still holds the old session ID. Detect this, // resume the stored session to re-register it, and retry once. @@ -463,7 +479,7 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { try { await withSessionBusyRetry(() => - requestGateway('prompt.submit', { session_id: sessionId, text }, PROMPT_SUBMIT_REQUEST_TIMEOUT_MS) + requestGateway('prompt.submit', submitParams(sessionId), PROMPT_SUBMIT_REQUEST_TIMEOUT_MS) ) } catch (firstErr) { const recoverStoredSessionId = targetStoredSessionId ?? selectedStoredSessionIdRef.current @@ -491,7 +507,7 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) { } await withSessionBusyRetry(() => - requestGateway('prompt.submit', { session_id: recoveredId, text }, PROMPT_SUBMIT_REQUEST_TIMEOUT_MS) + requestGateway('prompt.submit', submitParams(recoveredId), PROMPT_SUBMIT_REQUEST_TIMEOUT_MS) ) } else { submitErr = firstErr diff --git a/apps/desktop/src/lib/voice-playback.ts b/apps/desktop/src/lib/voice-playback.ts index 142423d5d5bd..80d9a3f22a2e 100644 --- a/apps/desktop/src/lib/voice-playback.ts +++ b/apps/desktop/src/lib/voice-playback.ts @@ -433,3 +433,24 @@ export async function playSpeechText(text: string, options: VoicePlaybackOptions export function isVoicePlaybackActive() { return $voicePlayback.get().status !== 'idle' } + +// --------------------------------------------------------------------------- +// Interruption latch — the next prompt.submit carries `interrupted: true` so +// the model knows its spoken reply was cut off (it can react: "rude!"). +// Marked by the barge-in paths (VAD, typing over playback); TTL'd so a stale +// barge never annotates an unrelated message minutes later. +// --------------------------------------------------------------------------- + +const INTERRUPT_TTL_MS = 120_000 +let interruptedAt: null | number = null + +export function markVoicePlaybackInterrupted() { + interruptedAt = Date.now() +} + +export function takeVoicePlaybackInterrupted(): boolean { + const at = interruptedAt + interruptedAt = null + + return at !== null && Date.now() - at < INTERRUPT_TTL_MS +} diff --git a/cli.py b/cli.py index 18e167c00971..870fd468b3aa 100644 --- a/cli.py +++ b/cli.py @@ -11296,6 +11296,8 @@ def _voice_barge_in_monitor(self, stop_event: threading.Event) -> None: 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 @@ def run_agent(): 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 @@ def handle_voice_record(event): # 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 diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index e5169339c8c2..03d96b45966e 100644 --- a/tests/test_tui_gateway_server.py +++ b/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 fake_listen(should_stop, capture=False, on_trigger=None, **_kw): 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() diff --git a/tests/tools/test_tts_streaming.py b/tests/tools/test_tts_streaming.py index 77241ea485b7..dd5b921d1695 100644 --- a/tests/tools/test_tts_streaming.py +++ b/tests/tools/test_tts_streaming.py @@ -53,6 +53,26 @@ def test_paragraph_break_is_a_boundary(self): ] +# ── Interruption latch ─────────────────────────────────────────────────── + + +class TestSpeechInterruptedLatch: + def test_take_pops_and_reports_recent_barge(self): + ts.mark_speech_interrupted() + assert ts.take_speech_interrupted() is True + assert ts.take_speech_interrupted() is False # one-shot + + def test_untouched_latch_is_false(self): + ts._interrupted_at = None + assert ts.take_speech_interrupted() is False + + def test_stale_barge_expires(self, monkeypatch): + ts.mark_speech_interrupted() + at = ts._interrupted_at + monkeypatch.setattr(ts.time, "monotonic", lambda: at + ts._INTERRUPT_TTL_S + 1) + assert ts.take_speech_interrupted() is False + + # ── Registry + resolver ────────────────────────────────────────────────── diff --git a/tools/tts_streaming.py b/tools/tts_streaming.py index 8d773d954f59..746886ccce3f 100644 --- a/tools/tts_streaming.py +++ b/tools/tts_streaming.py @@ -23,6 +23,7 @@ import logging import re +import time from abc import ABC, abstractmethod from typing import Callable, Dict, Iterator, List, Optional @@ -30,6 +31,34 @@ logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Interruption latch — lets the model know it was cut off mid-speech +# --------------------------------------------------------------------------- +# When the user barges in on a spoken reply (talks over it, types, hits the +# record key), the surface marks the latch; the next turn's submit path takes +# it and prepends SPEECH_INTERRUPTED_NOTE to the model-bound message (API-call +# local — never persisted, same as the CLI's model-switch notes). The TTL +# keeps a stale barge from annotating an unrelated message minutes later. + +SPEECH_INTERRUPTED_NOTE = ( + "[Note: the user interrupted your previous spoken reply before it finished.]" +) +_INTERRUPT_TTL_S = 120.0 +_interrupted_at: Optional[float] = None + + +def mark_speech_interrupted() -> None: + global _interrupted_at + _interrupted_at = time.monotonic() + + +def take_speech_interrupted() -> bool: + """Pop the latch; True when a barge happened within the TTL.""" + global _interrupted_at + at, _interrupted_at = _interrupted_at, None + return at is not None and time.monotonic() - at < _INTERRUPT_TTL_S + # Sentence boundary: after .!? followed by whitespace, or a blank line. SENTENCE_BOUNDARY_RE = re.compile(r"(?<=[.!?])(?:\s|\n)|(?:\n\n)") _THINK_BLOCK_RE = re.compile(r"].*?", flags=re.DOTALL) diff --git a/tui_gateway/server.py b/tui_gateway/server.py index ba7372673835..4212f7d0957f 100644 --- a/tui_gateway/server.py +++ b/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(): # 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 diff --git a/website/docs/user-guide/features/voice-mode.md b/website/docs/user-guide/features/voice-mode.md index d0945478dac6..7c3222b6438a 100644 --- a/website/docs/user-guide/features/voice-mode.md +++ b/website/docs/user-guide/features/voice-mode.md @@ -174,6 +174,8 @@ You can interrupt the agent mid-speech: - **Talk over it** — in continuous voice mode, a voice-activity monitor listens while the agent speaks and cuts playback the moment you start talking, then goes straight back to recording. The detector calibrates its noise floor against the playback itself, so speaker bleed doesn't self-trigger. Disable with `voice.barge_in: false` in `config.yaml`. - **Type or press the record key** — sending a new message or hitting the push-to-talk key stops playback instantly on every surface. +The agent **knows** it was interrupted: the next message carries a short note telling the model its spoken reply was cut off, so it can react naturally ("rude!") or pick up where it left off instead of being oblivious. + ### Hallucination Filter Whisper sometimes generates phantom text from silence or background noise ("Thank you for watching", "Subscribe", etc.). The agent filters these out using a set of 26 known hallucination phrases across multiple languages, plus a regex pattern that catches repetitive variations.