diff --git a/apps/desktop/src/app/chat/composer/hooks/use-composer-voice.ts b/apps/desktop/src/app/chat/composer/hooks/use-composer-voice.ts index 5ce92171d3f1..f8b75183f30c 100644 --- a/apps/desktop/src/app/chat/composer/hooks/use-composer-voice.ts +++ b/apps/desktop/src/app/chat/composer/hooks/use-composer-voice.ts @@ -1,7 +1,7 @@ import { useCallback, useEffect, useRef, useState } from 'react' import { useI18n } from '@/i18n' -import { chatMessageText } from '@/lib/chat-messages' +import { chatMessageText, collectUnspokenTurnSpeech } from '@/lib/chat-messages' import { triggerHaptic } from '@/lib/haptics' import { resetBrowseState } from '@/store/composer-input-history' import { notifyError } from '@/store/notifications' @@ -60,6 +60,7 @@ export function useComposerVoice({ onTranscribeAudio }) + /** Auto-speak selector: the latest unspoken reply only — a backlog collapses to the newest. */ const pendingResponse = () => { const messages = $messages.get() const last = messages.findLast(m => m.role === 'assistant' && !m.hidden) @@ -81,6 +82,13 @@ export function useComposerVoice({ } } + /** + * Voice-conversation selector: every unspoken assistant bubble of the turn, + * in order — narration interims AND the final answer, not just whichever + * bubble happens to be last. See `collectUnspokenTurnSpeech`. + */ + const pendingTurnResponse = () => collectUnspokenTurnSpeech($messages.get(), lastSpokenIdRef.current) + const consumePendingResponse = () => { const messages = $messages.get() const last = messages.findLast(m => m.role === 'assistant' && !m.hidden) @@ -108,7 +116,7 @@ export function useComposerVoice({ onFatalError: () => setVoiceConversationActive(false), onSubmit: submitVoiceTurn, onTranscribeAudio, - pendingResponse + pendingResponse: pendingTurnResponse }) // The `composer.voice` hotkey (Ctrl+B) toggles the conversation. Starting diff --git a/apps/desktop/src/lib/chat-messages.test.ts b/apps/desktop/src/lib/chat-messages.test.ts index 07234feac817..bd3253d41766 100644 --- a/apps/desktop/src/lib/chat-messages.test.ts +++ b/apps/desktop/src/lib/chat-messages.test.ts @@ -5,6 +5,7 @@ import { appendAssistantTextPart, appendReasoningPart, chatMessageText, + collectUnspokenTurnSpeech, mergeFinalAssistantText, preserveLocalAssistantErrors, reasoningPart, @@ -879,3 +880,82 @@ describe('mergeFinalAssistantText', () => { expect(result.filter(p => p.type === 'reasoning')).toHaveLength(1) }) }) + +describe('collectUnspokenTurnSpeech', () => { + const assistant = (id: string, text: string, extra: Partial = {}): ChatMessage => ({ + id, + role: 'assistant', + parts: text ? [{ type: 'text', text }] : [], + ...extra + }) + + const user = (id: string, text: string): ChatMessage => ({ + id, + role: 'user', + parts: [{ type: 'text', text }] + }) + + it('includes sealed interim narration AND the final answer of a tool-calling turn', () => { + const messages = [ + user('u1', 'what time is it?'), + assistant('a1', 'Let me check the clock.', { interim: true }), + assistant('a2', 'It is 9 PM.') + ] + + const speech = collectUnspokenTurnSpeech(messages, null) + + expect(speech).not.toBeNull() + expect(speech?.id).toBe('a1') + expect(speech?.text).toBe('Let me check the clock.\n\nIt is 9 PM.') + expect(speech?.pending).toBe(false) + }) + + it('keeps the binding id stable while later bubbles stream in', () => { + const turnStart = [user('u1', 'go'), assistant('a1', 'Let me check.', { interim: true })] + const first = collectUnspokenTurnSpeech(turnStart, null) + + const turnLater = [...turnStart, assistant('a2', 'Still work', { pending: true })] + const later = collectUnspokenTurnSpeech(turnLater, null) + + expect(first?.id).toBe('a1') + expect(later?.id).toBe('a1') + // The earlier snapshot's text is a prefix of the later one — the live + // session appends by length, so aggregation must be append-only. + expect(later?.text.startsWith(first?.text ?? '')).toBe(true) + expect(later?.pending).toBe(true) + }) + + it('starts after the last spoken message and skips hidden/empty bubbles', () => { + const messages = [ + assistant('a0', 'Spoken last turn.'), + user('u1', 'next'), + assistant('a1', '', { pending: false }), + assistant('a2', 'hidden note', { hidden: true }), + assistant('a3', 'The real reply.') + ] + + const speech = collectUnspokenTurnSpeech(messages, 'a0') + + expect(speech?.id).toBe('a3') + expect(speech?.text).toBe('The real reply.') + }) + + it('reports pending from the newest assistant bubble even when it has no text yet', () => { + const messages = [ + assistant('a1', 'Narration done.', { interim: true }), + assistant('a2', '', { pending: true }) + ] + + const speech = collectUnspokenTurnSpeech(messages, null) + + expect(speech?.id).toBe('a1') + expect(speech?.text).toBe('Narration done.') + expect(speech?.pending).toBe(true) + }) + + it('returns null when everything is spoken or there is no assistant text', () => { + expect(collectUnspokenTurnSpeech([], null)).toBeNull() + expect(collectUnspokenTurnSpeech([assistant('a1', 'Done.')], 'a1')).toBeNull() + expect(collectUnspokenTurnSpeech([user('u1', 'hello'), assistant('a1', '')], null)).toBeNull() + }) +}) diff --git a/apps/desktop/src/lib/chat-messages.ts b/apps/desktop/src/lib/chat-messages.ts index 4af96dce36bb..47b4ddf7d23c 100644 --- a/apps/desktop/src/lib/chat-messages.ts +++ b/apps/desktop/src/lib/chat-messages.ts @@ -150,6 +150,56 @@ export function chatMessageText(message: ChatMessage): string { .join('') } +export interface UnspokenTurnSpeech { + /** First unspoken assistant bubble — stable for the turn, the live speech session binds to it. */ + id: string + /** Whether the newest assistant bubble is still streaming. */ + pending: boolean + /** All unspoken assistant text in message order, bubbles joined on a blank line. */ + text: string +} + +/** + * Collect every unspoken assistant bubble after `lastSpokenId`, in order. + * + * A turn with tool calls produces several assistant bubbles — narration + * ("Let me check…") sealed as interims, then the final answer as a fresh + * bubble. Voice conversation speaks a turn through ONE live session bound to + * one response id, so it needs all of that text as a single growing string; + * selecting only one bubble silently drops everything after it. The blank-line + * join is a sentence boundary for the server's cutter, so a sealed bubble's + * tail is flushed as soon as the next bubble starts. + */ +export function collectUnspokenTurnSpeech(messages: ChatMessage[], lastSpokenId: string | null): UnspokenTurnSpeech | null { + const spokenIndex = lastSpokenId ? messages.findLastIndex(m => m.id === lastSpokenId) : -1 + + let id: string | null = null + let pending = false + const parts: string[] = [] + + for (const message of messages.slice(spokenIndex + 1)) { + if (message.role !== 'assistant' || message.hidden) { + continue + } + + pending = Boolean(message.pending) + const text = chatMessageText(message).trim() + + if (!text) { + continue + } + + id ??= message.id + parts.push(text) + } + + if (!id) { + return null + } + + return { id, pending, text: parts.join('\n\n') } +} + const normalizeWs = (value: string) => value.replace(/\s+/g, ' ').trim() /** diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 82f2e70ee472..f73a77bb655d 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -4566,9 +4566,31 @@ async def speak_stream_ws(ws: "WebSocket") -> None: chunker = SentenceChunker() + # The session stays open for a whole agent turn, and the client only + # sends `done` when the turn ends. During tool execution no text + # arrives, so without an idle flush a narration line with no trailing + # whitespace ("Let me check.") sits in the chunker until end-of-turn + # and is spoken long after the tool already finished. Mirror the CLI + # speaker pipeline: poll with a timeout and flush the buffer when the + # producer goes idle — immediately when the buffer ends on sentence + # punctuation, after a longer quiet spell otherwise. + idle_poll_seconds = 0.5 + idle_polls_before_force_flush = 4 # ~2s of silence + def _sentences(): + idle_polls = 0 while not stop.is_set(): - delta = text_q.get() + try: + delta = text_q.get(timeout=idle_poll_seconds) + except queue.Empty: + idle_polls += 1 + buffered = chunker.buf.strip() + if not buffered or ("" not in chunker.buf): + continue + if buffered.endswith((".", "!", "?", "…", ":")) or idle_polls >= idle_polls_before_force_flush: + yield from chunker.flush() + continue + idle_polls = 0 if delta is None: yield from chunker.flush() return diff --git a/tests/hermes_cli/test_web_server_speak_stream.py b/tests/hermes_cli/test_web_server_speak_stream.py index 50a36f35a90f..623402c50dfd 100644 --- a/tests/hermes_cli/test_web_server_speak_stream.py +++ b/tests/hermes_cli/test_web_server_speak_stream.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import time from urllib.parse import urlencode import pytest @@ -105,6 +106,60 @@ def test_incremental_deltas_are_cut_into_sentences(stream_client, monkeypatch): ] +def test_idle_flush_speaks_narration_before_done(stream_client, monkeypatch): + """A sentence-terminated buffer is spoken while the turn is still busy. + + The desktop keeps one session open for a whole agent turn and only sends + `done` at the end. Narration like "Let me check." has no trailing + whitespace, so the sentence cutter never fires on its own — the idle flush + must speak it during the tool-execution silence, not after the turn. + """ + streamer = _FakeStreamer([b"\x00\x00"]) + _patch_provider(monkeypatch, streamer) + + with stream_client.websocket_connect(_url()) as conn: + assert conn.receive_json()["type"] == "start" + conn.send_text(json.dumps({"text": "Let me check the config."})) + # No `done` — PCM must still arrive via the idle flush. + assert conn.receive_bytes() == b"\x00\x00" + conn.send_text(json.dumps({"done": True})) + assert conn.receive_json() == {"type": "end"} + + assert streamer.requests == ["Let me check the config."] + + +def test_idle_flush_eventually_speaks_unterminated_text(stream_client, monkeypatch): + """Text without sentence punctuation is force-flushed after a longer idle.""" + streamer = _FakeStreamer([b"\x00\x00"]) + _patch_provider(monkeypatch, streamer) + + with stream_client.websocket_connect(_url()) as conn: + assert conn.receive_json()["type"] == "start" + conn.send_text(json.dumps({"text": "Checking the wake-word branch now"})) + assert conn.receive_bytes() == b"\x00\x00" + conn.send_text(json.dumps({"done": True})) + assert conn.receive_json() == {"type": "end"} + + assert streamer.requests == ["Checking the wake-word branch now"] + + +def test_idle_flush_holds_open_think_block(stream_client, monkeypatch): + """An unterminated block is never flushed as speech.""" + streamer = _FakeStreamer([b"\x00\x00"]) + _patch_provider(monkeypatch, streamer) + + with stream_client.websocket_connect(_url()) as conn: + assert conn.receive_json()["type"] == "start" + conn.send_text(json.dumps({"text": "secret reasoning."})) + # Wait past the force-flush window; nothing may be synthesized. + time.sleep(2.5) + conn.send_text(json.dumps({"text": "Answer ready.", "done": True})) + assert conn.receive_bytes() == b"\x00\x00" + assert conn.receive_json() == {"type": "end"} + + assert streamer.requests == ["Answer ready."] + + def test_stop_frame_cuts_synthesis(stream_client, monkeypatch): streamer = _FakeStreamer([b"\x00\x00"]) _patch_provider(monkeypatch, streamer)