Merge pull request #69936 from NousResearch/bb/voice-speak-whole-turn

fix(voice): speak the whole desktop turn and idle-flush held narration

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

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> = {}): 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()
+  })
+})

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()
 
 /**

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 ("<think" in chunker.buf and "</think>" 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

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 <think> 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": "<think>secret reasoning."}))
+        # Wait past the force-flush window; nothing may be synthesized.
+        time.sleep(2.5)
+        conn.send_text(json.dumps({"text": "</think>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)