From 05575abf29e4ebdb8eebed13417c6e9e415422d1 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Wed, 19 Aug 2026 17:00:30 -0400 Subject: [PATCH 01/11] feat(voice): speak assistant text while it streams --- .../lib/nativeAssistantSpeech.test.ts | 79 ++++++++++++++++--- .../lib/nativeAssistantSpeech.ts | 65 ++++++++++----- 2 files changed, 117 insertions(+), 27 deletions(-) diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts index a51f8a92b..16102e1f9 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts @@ -104,13 +104,77 @@ describe("native assistant speech queue", () => { }); }); + it("starts speaking complete sentences while the response is streaming", async () => { + mocks.speakPocketVoice.mockResolvedValue(); + startNativeAssistantSpeech("session-1", vi.fn()); + + useChatStore + .getState() + .setMessages("session-1", [ + assistant([{ type: "text", text: "The first sentence is ready." }]), + ]); + await vi.waitFor(() => { + expect(mocks.speakPocketVoice).toHaveBeenCalledWith( + "The first sentence is ready.", + ); + }); + + useChatStore.getState().setMessages("session-1", [ + assistant([ + { + type: "text", + text: "The first sentence is ready. The second is still streaming", + }, + ]), + ]); + await Promise.resolve(); + expect(mocks.speakPocketVoice).toHaveBeenCalledTimes(1); + + useChatStore.getState().setMessages("session-1", [ + assistant([ + { + type: "text", + text: "The first sentence is ready. The second is still streaming!", + }, + ]), + ]); + await vi.waitFor(() => { + expect(mocks.speakPocketVoice).toHaveBeenNthCalledWith( + 2, + "The second is still streaming!", + ); + }); + }); + + it("flushes an unfinished streamed sentence when the response completes", async () => { + mocks.speakPocketVoice.mockResolvedValue(); + startNativeAssistantSpeech("session-1", vi.fn()); + + useChatStore + .getState() + .setMessages("session-1", [ + assistant([{ type: "text", text: "An unfinished tail" }]), + ]); + await Promise.resolve(); + expect(mocks.speakPocketVoice).not.toHaveBeenCalled(); + + useChatStore + .getState() + .setMessages("session-1", [ + assistant([{ type: "text", text: "An unfinished tail" }], "completed"), + ]); + await vi.waitFor(() => { + expect(mocks.speakPocketVoice).toHaveBeenCalledWith("An unfinished tail"); + }); + }); + it("speaks text at tool boundaries even when tools normalize before text", async () => { mocks.speakPocketVoice.mockResolvedValue(); startNativeAssistantSpeech("session-1", vi.fn()); useChatStore .getState() .setMessages("session-1", [ - assistant([{ type: "text", text: "First block." }]), + assistant([{ type: "text", text: "First block" }]), ]); await Promise.resolve(); expect(mocks.speakPocketVoice).not.toHaveBeenCalled(); @@ -124,11 +188,11 @@ describe("native assistant speech queue", () => { arguments: {}, status: "completed", }, - { type: "text", text: "First block." }, + { type: "text", text: "First block" }, ]), ]); await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenCalledWith("First block."); + expect(mocks.speakPocketVoice).toHaveBeenCalledWith("First block"); }); useChatStore.getState().setMessages("session-1", [ @@ -140,7 +204,7 @@ describe("native assistant speech queue", () => { arguments: {}, status: "completed", }, - { type: "text", text: "First block. Second block." }, + { type: "text", text: "First block Second block" }, ]), ]); await Promise.resolve(); @@ -162,14 +226,11 @@ describe("native assistant speech queue", () => { arguments: {}, status: "completed", }, - { type: "text", text: "First block. Second block." }, + { type: "text", text: "First block Second block" }, ]), ]); await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenNthCalledWith( - 2, - "Second block.", - ); + expect(mocks.speakPocketVoice).toHaveBeenNthCalledWith(2, "Second block"); }); }); diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts index 125f4c281..cae75e216 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts @@ -11,6 +11,28 @@ type SpeakableSegment = { sourceText: string; }; +const STREAMING_SPEECH_BOUNDARY = /(?:[.!?](?:["')\]}]+)?(?=\s|$)|\n+)/g; + +function splitSpeakableText( + text: string, + flushRemainder: boolean, +): { segments: string[]; consumedLength: number } { + const segments: string[] = []; + let start = 0; + for (const match of text.matchAll(STREAMING_SPEECH_BOUNDARY)) { + const end = (match.index ?? 0) + match[0].length; + const segment = text.slice(start, end).trim(); + if (segment) segments.push(segment); + start = end; + } + if (flushRemainder) { + const remainder = text.slice(start).trim(); + if (remainder) segments.push(remainder); + return { segments, consumedLength: text.length }; + } + return { segments, consumedLength: start }; +} + let stopSubscription: (() => void) | null = null; let stopVoiceSubscription: (() => void) | null = null; let playbackQueue = Promise.resolve(); @@ -66,7 +88,7 @@ function setSegmentStatus( content: message.content.map((content, index) => index === segment.contentIndex && content.type === "text" && - content.text.trim() === segment.sourceText + content.text.trim().startsWith(segment.sourceText) ? { ...content, speech: { status } } : content, ), @@ -105,7 +127,7 @@ export function startNativeAssistantSpeech( const initialMessages = useChatStore.getState().messagesBySession[sessionId] ?? []; const toolCountByMessage = new Map(); - const spokenTextBySlot = new Map(); + const consumedTextBySlot = new Map(); const completedMessages = new Set(); for (const message of initialMessages) { toolCountByMessage.set( @@ -119,7 +141,7 @@ export function startNativeAssistantSpeech( let textOrdinal = 0; message.content.forEach((content) => { if (content.type === "text") { - spokenTextBySlot.set( + consumedTextBySlot.set( `${message.id}\0text:${textOrdinal}`, content.text.trim(), ); @@ -156,8 +178,6 @@ export function startNativeAssistantSpeech( const crossedToolBoundary = toolCount > priorToolCount; toolCountByMessage.set(message.id, toolCount); if (completed) completedMessages.add(message.id); - if (!crossedToolBoundary && !completed) continue; - let textOrdinal = 0; message.content.forEach((content, contentIndex) => { if (content.type !== "text") return; @@ -165,19 +185,28 @@ export function startNativeAssistantSpeech( const slot = `${message.id}\0text:${textOrdinal}`; textOrdinal += 1; if (!sourceText) return; - const spokenText = spokenTextBySlot.get(slot) ?? ""; - if (sourceText === spokenText) return; - const text = sourceText.startsWith(spokenText) - ? sourceText.slice(spokenText.length).trim() - : sourceText; - spokenTextBySlot.set(slot, sourceText); - if (!text) return; - segments.push({ - key: `${slot}\0${spokenText.length}\0${sourceText.length}`, - messageId: message.id, - contentIndex, - text, - sourceText, + const consumedText = consumedTextBySlot.get(slot) ?? ""; + const consumedPrefix = sourceText.startsWith(consumedText) + ? consumedText + : ""; + const pendingText = sourceText.slice(consumedPrefix.length); + const split = splitSpeakableText( + pendingText, + crossedToolBoundary || completed, + ); + if (split.consumedLength === 0) return; + consumedTextBySlot.set( + slot, + sourceText.slice(0, consumedPrefix.length + split.consumedLength), + ); + split.segments.forEach((text, segmentIndex) => { + segments.push({ + key: `${slot}\0${consumedPrefix.length}\0${sourceText.length}\0${segmentIndex}`, + messageId: message.id, + contentIndex, + text, + sourceText, + }); }); }); } From 722b5763fa6fcee9c7533fe34284ad94731dbf72 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Wed, 19 Aug 2026 18:04:23 -0400 Subject: [PATCH 02/11] fix(voice): keep streaming speech state exact --- src/features/chat/stores/chatStore.ts | 4 +- .../lib/nativeAssistantSpeech.test.ts | 46 ++++++++++++++----- .../lib/nativeAssistantSpeech.ts | 28 ++++++++++- 3 files changed, 64 insertions(+), 14 deletions(-) diff --git a/src/features/chat/stores/chatStore.ts b/src/features/chat/stores/chatStore.ts index ccc3498be..7331fb0a1 100644 --- a/src/features/chat/stores/chatStore.ts +++ b/src/features/chat/stores/chatStore.ts @@ -271,10 +271,12 @@ function appendTextToMessage(message: Message, text: string): Message { const lastContent = message.content[message.content.length - 1]; if (lastContent?.type === "text") { const nextContent = [...message.content]; - nextContent[nextContent.length - 1] = { + const nextTextContent = { ...lastContent, text: lastContent.text + text, }; + delete nextTextContent.speech; + nextContent[nextContent.length - 1] = nextTextContent; return { ...message, content: nextContent }; } diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts index 16102e1f9..567ab07a9 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts @@ -119,29 +119,51 @@ describe("native assistant speech queue", () => { ); }); - useChatStore.getState().setMessages("session-1", [ - assistant([ - { - type: "text", - text: "The first sentence is ready. The second is still streaming", - }, - ]), - ]); + useChatStore + .getState() + .appendStreamingText( + "session-1", + "assistant-1", + " The second is still streaming", + ); await Promise.resolve(); expect(mocks.speakPocketVoice).toHaveBeenCalledTimes(1); + expect( + useChatStore.getState().messagesBySession["session-1"]?.[0]?.content[0], + ).not.toHaveProperty("speech"); + + useChatStore + .getState() + .appendStreamingText("session-1", "assistant-1", "!"); + await vi.waitFor(() => { + expect(mocks.speakPocketVoice).toHaveBeenNthCalledWith( + 2, + "The second is still streaming!", + ); + }); + }); + + it("waits through abbreviations before speaking a streamed sentence", async () => { + mocks.speakPocketVoice.mockResolvedValue(); + startNativeAssistantSpeech("session-1", vi.fn()); useChatStore.getState().setMessages("session-1", [ assistant([ { type: "text", - text: "The first sentence is ready. The second is still streaming!", + text: "Dr. Smith is reviewing the U.S.", }, ]), ]); + await Promise.resolve(); + expect(mocks.speakPocketVoice).not.toHaveBeenCalled(); + + useChatStore + .getState() + .appendStreamingText("session-1", "assistant-1", " economy."); await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenNthCalledWith( - 2, - "The second is still streaming!", + expect(mocks.speakPocketVoice).toHaveBeenCalledWith( + "Dr. Smith is reviewing the U.S. economy.", ); }); }); diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts index cae75e216..259b9aeca 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts @@ -12,6 +12,31 @@ type SpeakableSegment = { }; const STREAMING_SPEECH_BOUNDARY = /(?:[.!?](?:["')\]}]+)?(?=\s|$)|\n+)/g; +const NON_TERMINAL_ABBREVIATIONS = new Set([ + "dr.", + "etc.", + "fig.", + "jr.", + "mr.", + "mrs.", + "ms.", + "no.", + "prof.", + "sr.", + "st.", + "vs.", +]); + +function isSpeakableBoundary(text: string, match: RegExpMatchArray): boolean { + if (match[0].startsWith("\n") || !match[0].startsWith(".")) return true; + const periodIndex = match.index ?? 0; + const prefix = text.slice(0, periodIndex + 1); + const token = prefix.match(/[\p{L}.]+$/u)?.[0].toLocaleLowerCase(); + if (!token) return true; + return ( + !NON_TERMINAL_ABBREVIATIONS.has(token) && !/^(?:\p{L}\.){2,}$/u.test(token) + ); +} function splitSpeakableText( text: string, @@ -20,6 +45,7 @@ function splitSpeakableText( const segments: string[] = []; let start = 0; for (const match of text.matchAll(STREAMING_SPEECH_BOUNDARY)) { + if (!isSpeakableBoundary(text, match)) continue; const end = (match.index ?? 0) + match[0].length; const segment = text.slice(start, end).trim(); if (segment) segments.push(segment); @@ -88,7 +114,7 @@ function setSegmentStatus( content: message.content.map((content, index) => index === segment.contentIndex && content.type === "text" && - content.text.trim().startsWith(segment.sourceText) + content.text.trim() === segment.sourceText ? { ...content, speech: { status } } : content, ), From 05e349604117f14637e4bedb9ed3cd2b16bbcb32 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Wed, 19 Aug 2026 18:09:10 -0400 Subject: [PATCH 03/11] fix(voice): wait through streamed initials --- .../voice-conversation/lib/nativeAssistantSpeech.test.ts | 4 ++-- src/features/voice-conversation/lib/nativeAssistantSpeech.ts | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts index 567ab07a9..c3b862cfd 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts @@ -151,7 +151,7 @@ describe("native assistant speech queue", () => { assistant([ { type: "text", - text: "Dr. Smith is reviewing the U.S.", + text: "Dr. J. Smith is reviewing the U.S.", }, ]), ]); @@ -163,7 +163,7 @@ describe("native assistant speech queue", () => { .appendStreamingText("session-1", "assistant-1", " economy."); await vi.waitFor(() => { expect(mocks.speakPocketVoice).toHaveBeenCalledWith( - "Dr. Smith is reviewing the U.S. economy.", + "Dr. J. Smith is reviewing the U.S. economy.", ); }); }); diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts index 259b9aeca..c3fc8fefe 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts @@ -34,7 +34,7 @@ function isSpeakableBoundary(text: string, match: RegExpMatchArray): boolean { const token = prefix.match(/[\p{L}.]+$/u)?.[0].toLocaleLowerCase(); if (!token) return true; return ( - !NON_TERMINAL_ABBREVIATIONS.has(token) && !/^(?:\p{L}\.){2,}$/u.test(token) + !NON_TERMINAL_ABBREVIATIONS.has(token) && !/^(?:\p{L}\.)+$/u.test(token) ); } From b441005d6eeb764e63db15ac73faaaabf409552f Mon Sep 17 00:00:00 2001 From: John Tennant Date: Wed, 19 Aug 2026 18:18:16 -0400 Subject: [PATCH 04/11] fix(voice): complete block speech after its final segment --- .../lib/nativeAssistantSpeech.test.ts | 45 +++++++++++++++++++ .../lib/nativeAssistantSpeech.ts | 11 ++++- 2 files changed, 55 insertions(+), 1 deletion(-) diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts index c3b862cfd..db394da3c 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts @@ -168,6 +168,51 @@ describe("native assistant speech queue", () => { }); }); + it("marks a multi-sentence text block spoken only after its final segment", async () => { + let finishFirst: (() => void) | undefined; + let finishSecond: (() => void) | undefined; + mocks.speakPocketVoice + .mockImplementationOnce( + () => + new Promise((resolve) => { + finishFirst = resolve; + }), + ) + .mockImplementationOnce( + () => + new Promise((resolve) => { + finishSecond = resolve; + }), + ); + startNativeAssistantSpeech("session-1", vi.fn()); + + useChatStore + .getState() + .setMessages("session-1", [ + assistant( + [{ type: "text", text: "First sentence. Second sentence." }], + "completed", + ), + ]); + await vi.waitFor(() => { + expect(mocks.speakPocketVoice).toHaveBeenCalledTimes(1); + }); + finishFirst?.(); + await vi.waitFor(() => { + expect(mocks.speakPocketVoice).toHaveBeenCalledTimes(2); + expect( + useChatStore.getState().messagesBySession["session-1"]?.[0]?.content[0], + ).toMatchObject({ speech: { status: "speaking" } }); + }); + + finishSecond?.(); + await vi.waitFor(() => { + expect( + useChatStore.getState().messagesBySession["session-1"]?.[0]?.content[0], + ).toMatchObject({ speech: { status: "spoken" } }); + }); + }); + it("flushes an unfinished streamed sentence when the response completes", async () => { mocks.speakPocketVoice.mockResolvedValue(); startNativeAssistantSpeech("session-1", vi.fn()); diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts index c3fc8fefe..296e5f4fa 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts @@ -9,6 +9,7 @@ type SpeakableSegment = { contentIndex: number; text: string; sourceText: string; + completesSourceText: boolean; }; const STREAMING_SPEECH_BOUNDARY = /(?:[.!?](?:["')\]}]+)?(?=\s|$)|\n+)/g; @@ -225,6 +226,8 @@ export function startNativeAssistantSpeech( slot, sourceText.slice(0, consumedPrefix.length + split.consumedLength), ); + const completesSourceText = + consumedPrefix.length + split.consumedLength === sourceText.length; split.segments.forEach((text, segmentIndex) => { segments.push({ key: `${slot}\0${consumedPrefix.length}\0${sourceText.length}\0${segmentIndex}`, @@ -232,6 +235,8 @@ export function startNativeAssistantSpeech( contentIndex, text, sourceText, + completesSourceText: + completesSourceText && segmentIndex === split.segments.length - 1, }); }); }); @@ -263,7 +268,11 @@ export function startNativeAssistantSpeech( current.setUiState("agent-speaking"); try { await speakPocketVoice(segment.text); - if (activeGeneration === generation && segmentEpoch === speechEpoch) { + if ( + activeGeneration === generation && + segmentEpoch === speechEpoch && + segment.completesSourceText + ) { setSegmentStatus(sessionId, segment, "spoken"); } } catch (error) { From 1300022449148cbe170d3ad02d8ab5fa02d080e9 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Wed, 19 Aug 2026 22:25:04 -0400 Subject: [PATCH 05/11] fix(voice): overlap streamed text synthesis and playback --- src-tauri/src/commands/pocket_voice.rs | 37 ++ src-tauri/src/lib.rs | 1 + src/features/chat/stores/chatStore.ts | 4 +- .../api/pocketVoice.test.ts | 51 +++ .../voice-conversation/api/pocketVoice.ts | 33 ++ .../lib/nativeAssistantSpeech.test.ts | 345 ++++---------- .../lib/nativeAssistantSpeech.ts | 431 ++++++++++-------- 7 files changed, 443 insertions(+), 459 deletions(-) diff --git a/src-tauri/src/commands/pocket_voice.rs b/src-tauri/src/commands/pocket_voice.rs index 8e622015b..bd285becd 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -148,6 +148,7 @@ struct ActivePocketStream { #[derive(Debug)] enum PocketStreamCommand { Append(String), + Flush, Finish, Stop, } @@ -811,6 +812,23 @@ pub fn finish_pocket_voice_stream( } } +#[tauri::command] +pub fn flush_pocket_voice_stream( + state: State<'_, PocketVoiceState>, + stream_id: String, +) -> Result<(), String> { + #[cfg(not(target_os = "macos"))] + { + let _ = (state, stream_id); + return Err("Pocket voice playback is currently supported on macOS only".to_string()); + } + + #[cfg(target_os = "macos")] + { + send_pocket_stream_command(&state, &stream_id, PocketStreamCommand::Flush) + } +} + #[cfg(target_os = "macos")] fn send_pocket_stream_command( state: &PocketVoiceState, @@ -1889,6 +1907,25 @@ fn run_pocket_voice_stream( return Ok(PocketStreamEventState::Interrupted); } } + Ok(PocketStreamCommand::Flush) => { + if !synthesize_pocket_stream_ready( + app, + stream_id, + &engine, + &style, + &active, + &player, + channels, + rate, + &mut speed_processor, + &mut pending, + &mut first_chunk_pending, + &mut playback_started, + true, + )? { + return Ok(PocketStreamEventState::Interrupted); + } + } Ok(PocketStreamCommand::Finish) => { if !synthesize_pocket_stream_ready( app, diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index ce157f5e7..9a8cd3c4c 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -623,6 +623,7 @@ pub fn run() { commands::pocket_voice::speak_pocket_voice, commands::pocket_voice::start_pocket_voice_stream, commands::pocket_voice::append_pocket_voice_stream, + commands::pocket_voice::flush_pocket_voice_stream, commands::pocket_voice::finish_pocket_voice_stream, commands::pocket_voice::stop_pocket_voice, commands::pocket_voice::remove_voice_model, diff --git a/src/features/chat/stores/chatStore.ts b/src/features/chat/stores/chatStore.ts index 7331fb0a1..ccc3498be 100644 --- a/src/features/chat/stores/chatStore.ts +++ b/src/features/chat/stores/chatStore.ts @@ -271,12 +271,10 @@ function appendTextToMessage(message: Message, text: string): Message { const lastContent = message.content[message.content.length - 1]; if (lastContent?.type === "text") { const nextContent = [...message.content]; - const nextTextContent = { + nextContent[nextContent.length - 1] = { ...lastContent, text: lastContent.text + text, }; - delete nextTextContent.speech; - nextContent[nextContent.length - 1] = nextTextContent; return { ...message, content: nextContent }; } diff --git a/src/features/voice-conversation/api/pocketVoice.test.ts b/src/features/voice-conversation/api/pocketVoice.test.ts index ed4254413..2d66dd261 100644 --- a/src/features/voice-conversation/api/pocketVoice.test.ts +++ b/src/features/voice-conversation/api/pocketVoice.test.ts @@ -9,14 +9,19 @@ vi.mock("@tauri-apps/api/core", () => ({ invoke: mocks.invoke })); vi.mock("@tauri-apps/api/event", () => ({ listen: mocks.listen })); import { + appendPocketVoiceStream, + finishPocketVoiceStream, + flushPocketVoiceStream, getPocketVoiceStatus, installVoiceModel, + listenToPocketVoiceStream, listenToPocketVoiceStatus, previewPocketVoice, removeVoiceModel, selectPocketVoice, setPocketPlaybackSpeed, speakPocketVoice, + startPocketVoiceStream, stopPocketVoice, } from "./pocketVoice"; @@ -79,4 +84,50 @@ describe("Pocket voice API", () => { await listenToPocketVoiceStatus(callback); expect(callback).toHaveBeenCalledWith(status); }); + + it("uses the streaming utterance commands", async () => { + mocks.invoke.mockResolvedValue(undefined); + + await startPocketVoiceStream("stream-1"); + await appendPocketVoiceStream("stream-1", "Hello"); + await flushPocketVoiceStream("stream-1"); + await finishPocketVoiceStream("stream-1"); + + expect(mocks.invoke).toHaveBeenNthCalledWith( + 1, + "start_pocket_voice_stream", + { streamId: "stream-1" }, + ); + expect(mocks.invoke).toHaveBeenNthCalledWith( + 2, + "append_pocket_voice_stream", + { streamId: "stream-1", text: "Hello" }, + ); + expect(mocks.invoke).toHaveBeenNthCalledWith( + 3, + "flush_pocket_voice_stream", + { streamId: "stream-1" }, + ); + expect(mocks.invoke).toHaveBeenNthCalledWith( + 4, + "finish_pocket_voice_stream", + { streamId: "stream-1" }, + ); + }); + + it("unwraps playback stream events", async () => { + const callback = vi.fn(); + const streamEvent = { + streamId: "stream-1", + state: "started", + error: null, + }; + mocks.listen.mockImplementation(async (_event, handler) => { + handler({ payload: streamEvent }); + return vi.fn(); + }); + + await listenToPocketVoiceStream(callback); + expect(callback).toHaveBeenCalledWith(streamEvent); + }); }); diff --git a/src/features/voice-conversation/api/pocketVoice.ts b/src/features/voice-conversation/api/pocketVoice.ts index 394cf91a1..b8152a8b5 100644 --- a/src/features/voice-conversation/api/pocketVoice.ts +++ b/src/features/voice-conversation/api/pocketVoice.ts @@ -34,6 +34,12 @@ export interface PocketVoiceStatus { voices: PocketVoice[]; } +export interface PocketVoiceStreamEvent { + streamId: string; + state: "started" | "completed" | "interrupted" | "failed"; + error: string | null; +} + export type VoiceModelKind = "pocket" | "parakeet"; export type VoiceModelDownloadPhase = | "queued" @@ -81,6 +87,25 @@ export function speakPocketVoice(text: string): Promise { return invoke("speak_pocket_voice", { text }); } +export function startPocketVoiceStream(streamId: string): Promise { + return invoke("start_pocket_voice_stream", { streamId }); +} + +export function appendPocketVoiceStream( + streamId: string, + text: string, +): Promise { + return invoke("append_pocket_voice_stream", { streamId, text }); +} + +export function flushPocketVoiceStream(streamId: string): Promise { + return invoke("flush_pocket_voice_stream", { streamId }); +} + +export function finishPocketVoiceStream(streamId: string): Promise { + return invoke("finish_pocket_voice_stream", { streamId }); +} + export function stopPocketVoice(): Promise { return invoke("stop_pocket_voice"); } @@ -98,3 +123,11 @@ export function listenToPocketVoiceStatus( onStatus(event.payload), ); } + +export function listenToPocketVoiceStream( + onEvent: (event: PocketVoiceStreamEvent) => void, +): Promise { + return listen("pocket-voice:stream-event", (event) => + onEvent(event.payload), + ); +} diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts index db394da3c..aa917d4e0 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts @@ -2,13 +2,31 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { Message } from "@/shared/types/messages"; import { useChatStore } from "@/features/chat/stores/chatStore"; import { useVoiceConversationStore } from "../stores/voiceConversationStore"; +import type { PocketVoiceStreamEvent } from "../api/pocketVoice"; const mocks = vi.hoisted(() => ({ - speakPocketVoice: vi.fn<(text: string) => Promise>(), - stopPocketVoice: vi.fn<() => Promise>(), + start: vi.fn<(streamId: string) => Promise>(), + append: vi.fn<(streamId: string, text: string) => Promise>(), + flush: vi.fn<(streamId: string) => Promise>(), + finish: vi.fn<(streamId: string) => Promise>(), + stop: vi.fn<() => Promise>(), + streamHandler: null as ((event: PocketVoiceStreamEvent) => void) | null, })); -vi.mock("../api/pocketVoice", () => mocks); +vi.mock("../api/pocketVoice", () => ({ + startPocketVoiceStream: (streamId: string) => mocks.start(streamId), + appendPocketVoiceStream: (streamId: string, text: string) => + mocks.append(streamId, text), + flushPocketVoiceStream: (streamId: string) => mocks.flush(streamId), + finishPocketVoiceStream: (streamId: string) => mocks.finish(streamId), + stopPocketVoice: () => mocks.stop(), + listenToPocketVoiceStream: async ( + handler: (event: PocketVoiceStreamEvent) => void, + ) => { + mocks.streamHandler = handler; + return vi.fn(); + }, +})); import { startNativeAssistantSpeech, @@ -31,10 +49,22 @@ function assistant( }; } -describe("native assistant speech queue", () => { +function emit( + state: PocketVoiceStreamEvent["state"], + error: string | null = null, +) { + const streamId = mocks.start.mock.calls[0]?.[0] as string; + mocks.streamHandler?.({ streamId, state, error }); +} + +describe("native assistant speech stream", () => { beforeEach(() => { - mocks.speakPocketVoice.mockReset(); - mocks.stopPocketVoice.mockReset().mockResolvedValue(true); + mocks.start.mockReset().mockResolvedValue(); + mocks.append.mockReset().mockResolvedValue(); + mocks.flush.mockReset().mockResolvedValue(); + mocks.finish.mockReset().mockResolvedValue(); + mocks.stop.mockReset().mockResolvedValue(true); + mocks.streamHandler = null; useChatStore.setState({ messagesBySession: {}, sessionStateById: {}, @@ -58,209 +88,81 @@ describe("native assistant speech queue", () => { stopNativeAssistantSpeech(); }); - it("plays rapid assistant blocks exactly once in transcript order", async () => { - let finishFirst: (() => void) | undefined; - mocks.speakPocketVoice - .mockImplementationOnce( - () => - new Promise((resolve) => { - finishFirst = resolve; - }), - ) - .mockResolvedValueOnce(); - startNativeAssistantSpeech("session-1", vi.fn()); - - useChatStore.getState().setMessages("session-1", [ - assistant( - [ - { type: "text", text: "First block." }, - { - type: "toolRequest", - id: "tool-1", - name: "Read", - arguments: {}, - status: "completed", - }, - { type: "text", text: "Second block." }, - ], - "completed", - ), - ]); - - await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenCalledTimes(1); - }); - expect(mocks.speakPocketVoice).toHaveBeenNthCalledWith(1, "First block."); - finishFirst?.(); - await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenCalledTimes(2); - }); - expect(mocks.speakPocketVoice).toHaveBeenNthCalledWith(2, "Second block."); - await vi.waitFor(() => { - const content = - useChatStore.getState().messagesBySession["session-1"]?.[0]?.content; - expect(content?.[0]).toMatchObject({ speech: { status: "spoken" } }); - expect(content?.[2]).toMatchObject({ speech: { status: "spoken" } }); - }); - }); - - it("starts speaking complete sentences while the response is streaming", async () => { - mocks.speakPocketVoice.mockResolvedValue(); + it("pushes raw assistant deltas without frontend sentence segmentation", async () => { startNativeAssistantSpeech("session-1", vi.fn()); - useChatStore .getState() .setMessages("session-1", [ - assistant([{ type: "text", text: "The first sentence is ready." }]), + assistant([{ type: "text", text: "First sentence. Later" }]), ]); + await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenCalledWith( - "The first sentence is ready.", + expect(mocks.start).toHaveBeenCalledTimes(1); + expect(mocks.append).toHaveBeenCalledWith( + mocks.start.mock.calls[0]?.[0], + "First sentence. Later", ); }); - - useChatStore - .getState() - .appendStreamingText( - "session-1", - "assistant-1", - " The second is still streaming", - ); - await Promise.resolve(); - expect(mocks.speakPocketVoice).toHaveBeenCalledTimes(1); expect( useChatStore.getState().messagesBySession["session-1"]?.[0]?.content[0], ).not.toHaveProperty("speech"); useChatStore .getState() - .appendStreamingText("session-1", "assistant-1", "!"); + .appendStreamingText("session-1", "assistant-1", " text."); await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenNthCalledWith( + expect(mocks.append).toHaveBeenNthCalledWith( 2, - "The second is still streaming!", + mocks.start.mock.calls[0]?.[0], + " text.", ); }); }); - it("waits through abbreviations before speaking a streamed sentence", async () => { - mocks.speakPocketVoice.mockResolvedValue(); + it("derives speaking and completion state from backend playback events", async () => { startNativeAssistantSpeech("session-1", vi.fn()); - - useChatStore.getState().setMessages("session-1", [ - assistant([ - { - type: "text", - text: "Dr. J. Smith is reviewing the U.S.", - }, - ]), - ]); - await Promise.resolve(); - expect(mocks.speakPocketVoice).not.toHaveBeenCalled(); - - useChatStore - .getState() - .appendStreamingText("session-1", "assistant-1", " economy."); - await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenCalledWith( - "Dr. J. Smith is reviewing the U.S. economy.", - ); - }); - }); - - it("marks a multi-sentence text block spoken only after its final segment", async () => { - let finishFirst: (() => void) | undefined; - let finishSecond: (() => void) | undefined; - mocks.speakPocketVoice - .mockImplementationOnce( - () => - new Promise((resolve) => { - finishFirst = resolve; - }), - ) - .mockImplementationOnce( - () => - new Promise((resolve) => { - finishSecond = resolve; - }), - ); - startNativeAssistantSpeech("session-1", vi.fn()); - useChatStore .getState() .setMessages("session-1", [ - assistant( - [{ type: "text", text: "First sentence. Second sentence." }], - "completed", - ), + assistant([{ type: "text", text: "First sentence." }]), ]); - await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenCalledTimes(1); - }); - finishFirst?.(); - await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenCalledTimes(2); - expect( - useChatStore.getState().messagesBySession["session-1"]?.[0]?.content[0], - ).toMatchObject({ speech: { status: "speaking" } }); - }); - - finishSecond?.(); - await vi.waitFor(() => { - expect( - useChatStore.getState().messagesBySession["session-1"]?.[0]?.content[0], - ).toMatchObject({ speech: { status: "spoken" } }); - }); - }); + await vi.waitFor(() => expect(mocks.append).toHaveBeenCalled()); - it("flushes an unfinished streamed sentence when the response completes", async () => { - mocks.speakPocketVoice.mockResolvedValue(); - startNativeAssistantSpeech("session-1", vi.fn()); + emit("started"); + expect( + useChatStore.getState().messagesBySession["session-1"]?.[0]?.content[0], + ).toMatchObject({ speech: { status: "speaking" } }); useChatStore .getState() - .setMessages("session-1", [ - assistant([{ type: "text", text: "An unfinished tail" }]), - ]); - await Promise.resolve(); - expect(mocks.speakPocketVoice).not.toHaveBeenCalled(); + .appendStreamingText("session-1", "assistant-1", " Second sentence."); + expect( + useChatStore.getState().messagesBySession["session-1"]?.[0]?.content[0], + ).toMatchObject({ speech: { status: "speaking" } }); useChatStore .getState() .setMessages("session-1", [ - assistant([{ type: "text", text: "An unfinished tail" }], "completed"), + assistant( + [{ type: "text", text: "First sentence. Second sentence." }], + "completed", + ), ]); - await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenCalledWith("An unfinished tail"); - }); + await vi.waitFor(() => expect(mocks.finish).toHaveBeenCalledTimes(1)); + emit("completed"); + expect( + useChatStore.getState().messagesBySession["session-1"]?.[0]?.content[0], + ).toMatchObject({ speech: { status: "spoken" } }); }); - it("speaks text at tool boundaries even when tools normalize before text", async () => { - mocks.speakPocketVoice.mockResolvedValue(); + it("flushes buffered text at a tool boundary without ending the stream", async () => { startNativeAssistantSpeech("session-1", vi.fn()); useChatStore .getState() .setMessages("session-1", [ - assistant([{ type: "text", text: "First block" }]), + assistant([{ type: "text", text: "Before the tool" }]), ]); - await Promise.resolve(); - expect(mocks.speakPocketVoice).not.toHaveBeenCalled(); - - useChatStore.getState().setMessages("session-1", [ - assistant([ - { - type: "toolRequest", - id: "tool-1", - name: "Read", - arguments: {}, - status: "completed", - }, - { type: "text", text: "First block" }, - ]), - ]); - await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenCalledWith("First block"); - }); + await vi.waitFor(() => expect(mocks.append).toHaveBeenCalledTimes(1)); useChatStore.getState().setMessages("session-1", [ assistant([ @@ -271,115 +173,32 @@ describe("native assistant speech queue", () => { arguments: {}, status: "completed", }, - { type: "text", text: "First block Second block" }, + { type: "text", text: "Before the tool" }, ]), ]); - await Promise.resolve(); - expect(mocks.speakPocketVoice).toHaveBeenCalledTimes(1); - useChatStore.getState().setMessages("session-1", [ - assistant([ - { - type: "toolRequest", - id: "tool-1", - name: "Read", - arguments: {}, - status: "completed", - }, - { - type: "toolRequest", - id: "tool-2", - name: "Read again", - arguments: {}, - status: "completed", - }, - { type: "text", text: "First block Second block" }, - ]), - ]); - await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenNthCalledWith(2, "Second block"); - }); + await vi.waitFor(() => expect(mocks.flush).toHaveBeenCalledTimes(1)); + expect(mocks.finish).not.toHaveBeenCalled(); }); - it("marks the active and queued blocks interrupted on barge-in", async () => { - let finishFirst: (() => void) | undefined; - mocks.speakPocketVoice.mockImplementation( - () => - new Promise((resolve) => { - finishFirst = resolve; - }), - ); - startNativeAssistantSpeech("session-1", vi.fn()); - useChatStore.getState().setMessages("session-1", [ - assistant( - [ - { type: "text", text: "First block." }, - { - type: "toolRequest", - id: "tool-1", - name: "Read", - arguments: {}, - status: "completed", - }, - { type: "text", text: "Second block." }, - ], - "completed", - ), - ]); - await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenCalledTimes(1); - }); - - useVoiceConversationStore.setState({ userSpeaking: true }); - - await vi.waitFor(() => { - expect(mocks.stopPocketVoice).toHaveBeenCalled(); - const content = - useChatStore.getState().messagesBySession["session-1"]?.[0]?.content; - expect(content?.[0]).toMatchObject({ - speech: { status: "interrupted" }, - }); - expect(content?.[2]).toMatchObject({ - speech: { status: "interrupted" }, - }); - }); - finishFirst?.(); - await Promise.resolve(); - expect(mocks.speakPocketVoice).toHaveBeenCalledTimes(1); - expect(takeVoicePlaybackNotices("session-1")).toBe( - "[voice: tts-delivery-failed]\n" + - "TTS delivery was interrupted because the user started speaking; the assistant reply was not fully spoken.\n" + - "Original text: First block.\n" + - "This is TTS delivery state, not live user voice input. Do not respond to this control message or repeat the reply unless re-delivery is still appropriate.\n" + - "[voice: tts-delivery-failed]\n" + - "TTS delivery was interrupted because the user started speaking; the assistant reply was not fully spoken.\n" + - "Original text: Second block.\n" + - "This is TTS delivery state, not live user voice input. Do not respond to this control message or repeat the reply unless re-delivery is still appropriate.", - ); - }); - - it("marks active speech interrupted when voice conversation stops", async () => { - mocks.speakPocketVoice.mockImplementation(() => new Promise(() => {})); + it("interrupts one utterance status even when many deltas are queued", async () => { startNativeAssistantSpeech("session-1", vi.fn()); useChatStore .getState() .setMessages("session-1", [ - assistant([{ type: "text", text: "Still speaking." }], "completed"), + assistant([{ type: "text", text: "One. Two. Three." }]), ]); - await vi.waitFor(() => { - expect(mocks.speakPocketVoice).toHaveBeenCalledWith("Still speaking."); - }); - - stopNativeAssistantSpeech(); + await vi.waitFor(() => expect(mocks.append).toHaveBeenCalled()); + emit("started"); + useVoiceConversationStore.setState({ userSpeaking: true }); + await vi.waitFor(() => expect(mocks.stop).toHaveBeenCalled()); expect( useChatStore.getState().messagesBySession["session-1"]?.[0]?.content[0], ).toMatchObject({ speech: { status: "interrupted" } }); - expect(takeVoicePlaybackNotices("session-1")).toBe( - "[voice: tts-delivery-failed]\n" + - "TTS delivery was interrupted because the user started speaking; the assistant reply was not fully spoken.\n" + - "Original text: Still speaking.\n" + - "This is TTS delivery state, not live user voice input. Do not respond to this control message or repeat the reply unless re-delivery is still appropriate.", + expect(takeVoicePlaybackNotices("session-1")).toContain( + "Original text: One. Two. Three.", ); + expect(takeVoicePlaybackNotices("session-1")).toBeNull(); }); }); diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts index 296e5f4fa..0a1c67fbe 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts @@ -1,87 +1,54 @@ import { useChatStore } from "@/features/chat/stores/chatStore"; -import { speakPocketVoice, stopPocketVoice } from "../api/pocketVoice"; +import { + appendPocketVoiceStream, + finishPocketVoiceStream, + flushPocketVoiceStream, + listenToPocketVoiceStream, + startPocketVoiceStream, + stopPocketVoice, + type PocketVoiceStreamEvent, +} from "../api/pocketVoice"; import { useVoiceConversationStore } from "../stores/voiceConversationStore"; type SpeechFailureHandler = (text: string, error: unknown) => void; -type SpeakableSegment = { - key: string; - messageId: string; - contentIndex: number; +type SpeechTarget = { messageId: string; textOrdinal: number }; +type ActiveUtterance = { + id: string; + sessionId: string; + target: SpeechTarget; text: string; - sourceText: string; - completesSourceText: boolean; + finishing: boolean; + onFailure: SpeechFailureHandler; }; - -const STREAMING_SPEECH_BOUNDARY = /(?:[.!?](?:["')\]}]+)?(?=\s|$)|\n+)/g; -const NON_TERMINAL_ABBREVIATIONS = new Set([ - "dr.", - "etc.", - "fig.", - "jr.", - "mr.", - "mrs.", - "ms.", - "no.", - "prof.", - "sr.", - "st.", - "vs.", -]); - -function isSpeakableBoundary(text: string, match: RegExpMatchArray): boolean { - if (match[0].startsWith("\n") || !match[0].startsWith(".")) return true; - const periodIndex = match.index ?? 0; - const prefix = text.slice(0, periodIndex + 1); - const token = prefix.match(/[\p{L}.]+$/u)?.[0].toLocaleLowerCase(); - if (!token) return true; - return ( - !NON_TERMINAL_ABBREVIATIONS.has(token) && !/^(?:\p{L}\.)+$/u.test(token) - ); -} - -function splitSpeakableText( - text: string, - flushRemainder: boolean, -): { segments: string[]; consumedLength: number } { - const segments: string[] = []; - let start = 0; - for (const match of text.matchAll(STREAMING_SPEECH_BOUNDARY)) { - if (!isSpeakableBoundary(text, match)) continue; - const end = (match.index ?? 0) + match[0].length; - const segment = text.slice(start, end).trim(); - if (segment) segments.push(segment); - start = end; - } - if (flushRemainder) { - const remainder = text.slice(start).trim(); - if (remainder) segments.push(remainder); - return { segments, consumedLength: text.length }; - } - return { segments, consumedLength: start }; -} +type SpeechStatus = + | "speaking" + | "spoken" + | "interrupted" + | "notSpoken" + | "failed"; let stopSubscription: (() => void) | null = null; let stopVoiceSubscription: (() => void) | null = null; -let playbackQueue = Promise.resolve(); +let stopStreamSubscription: (() => void) | null = null; +let streamListenerReady: Promise = Promise.resolve(); +let commandQueue = Promise.resolve(); let generation = 0; -let speechEpoch = 0; +let commandEpoch = 0; let activeSpeechSessionId: string | null = null; -const scheduledSegments = new Map(); +let activeUtterance: ActiveUtterance | null = null; const pendingNotices = new Map(); const recordedNoticeKeys = new Set(); function recordPlaybackNotice( sessionId: string, - segment: SpeakableSegment, + key: string, + text: string, status: "interrupted" | "notSpoken" | "failed", ) { - const key = `${sessionId}\0${segment.key}\0${status}`; - if (recordedNoticeKeys.has(key)) return; - recordedNoticeKeys.add(key); - const text = - segment.text.length > 500 - ? `${segment.text.slice(0, 497).trimEnd()}…` - : segment.text; + const noticeKey = `${sessionId}\0${key}\0${status}`; + if (recordedNoticeKeys.has(noticeKey)) return; + recordedNoticeKeys.add(noticeKey); + const excerpt = text.length > 500 ? `${text.slice(0, 497).trimEnd()}…` : text; const outcome = status === "interrupted" ? "TTS delivery was interrupted because the user started speaking; the assistant reply was not fully spoken." @@ -89,7 +56,7 @@ function recordPlaybackNotice( ? "TTS delivery was blocked because the user was speaking; the assistant reply was not spoken." : "Native TTS could not deliver the assistant reply."; const notice = - `[voice: tts-delivery-failed]\n${outcome}\nOriginal text: ${text}\n` + + `[voice: tts-delivery-failed]\n${outcome}\nOriginal text: ${excerpt}\n` + "This is TTS delivery state, not live user voice input. Do not respond to this control message or repeat the reply unless re-delivery is still appropriate."; pendingNotices.set(sessionId, [ ...(pendingNotices.get(sessionId) ?? []), @@ -103,44 +70,139 @@ export function takeVoicePlaybackNotices(sessionId: string): string | null { return notices?.join("\n") ?? null; } -function setSegmentStatus( +function targetKey(target: SpeechTarget): string { + return `${target.messageId}\0text:${target.textOrdinal}`; +} + +function setTargetStatus( sessionId: string, - segment: SpeakableSegment, - status: "speaking" | "spoken" | "interrupted" | "notSpoken" | "failed", + target: SpeechTarget, + status: SpeechStatus, ) { useChatStore .getState() - .updateMessage(sessionId, segment.messageId, (message) => ({ - ...message, - content: message.content.map((content, index) => - index === segment.contentIndex && - content.type === "text" && - content.text.trim() === segment.sourceText - ? { ...content, speech: { status } } - : content, - ), - })); + .updateMessage(sessionId, target.messageId, (message) => { + let textOrdinal = 0; + return { + ...message, + content: message.content.map((content) => { + if (content.type !== "text") return content; + const matches = textOrdinal === target.textOrdinal; + textOrdinal += 1; + return matches ? { ...content, speech: { status } } : content; + }), + }; + }); } -export function stopNativeAssistantSpeech(): void { - if (activeSpeechSessionId) { - for (const segment of scheduledSegments.values()) { - setSegmentStatus(activeSpeechSessionId, segment, "interrupted"); - recordPlaybackNotice(activeSpeechSessionId, segment, "interrupted"); +function failActiveUtterance( + utteranceId: string, + error: unknown, + onFailure: SpeechFailureHandler, +) { + const utterance = activeUtterance; + if (!utterance || utterance.id !== utteranceId) return; + setTargetStatus(utterance.sessionId, utterance.target, "failed"); + recordPlaybackNotice( + utterance.sessionId, + utterance.id, + utterance.text, + "failed", + ); + useVoiceConversationStore.getState().setUiState("listening"); + activeUtterance = null; + onFailure(utterance.text, error); +} + +function queueStreamCommand( + utterance: ActiveUtterance, + operation: () => Promise, + onFailure: SpeechFailureHandler, +) { + const queuedEpoch = commandEpoch; + commandQueue = commandQueue.then(async () => { + if (queuedEpoch !== commandEpoch) return; + try { + await operation(); + } catch (error) { + failActiveUtterance(utterance.id, error, onFailure); } + }); +} + +function handleStreamEvent(event: PocketVoiceStreamEvent) { + const utterance = activeUtterance; + if (!utterance || utterance.id !== event.streamId) return; + const voice = useVoiceConversationStore.getState(); + + switch (event.state) { + case "started": + setTargetStatus(utterance.sessionId, utterance.target, "speaking"); + voice.setUiState("agent-speaking"); + break; + case "completed": + setTargetStatus(utterance.sessionId, utterance.target, "spoken"); + voice.setUiState("listening"); + activeUtterance = null; + break; + case "interrupted": + setTargetStatus(utterance.sessionId, utterance.target, "interrupted"); + recordPlaybackNotice( + utterance.sessionId, + utterance.id, + utterance.text, + "interrupted", + ); + voice.setUiState("listening"); + activeUtterance = null; + break; + case "failed": + setTargetStatus(utterance.sessionId, utterance.target, "failed"); + recordPlaybackNotice( + utterance.sessionId, + utterance.id, + utterance.text, + "failed", + ); + voice.setUiState("listening"); + activeUtterance = null; + utterance.onFailure( + utterance.text, + event.error ?? new Error("Pocket voice stream failed"), + ); + break; + } +} + +function interruptActiveUtterance() { + const utterance = activeUtterance; + commandEpoch += 1; + activeUtterance = null; + if (utterance) { + setTargetStatus(utterance.sessionId, utterance.target, "interrupted"); + recordPlaybackNotice( + utterance.sessionId, + utterance.id, + utterance.text, + "interrupted", + ); } + void stopPocketVoice().catch(() => undefined); + commandQueue = commandQueue.then(async () => { + await stopPocketVoice().catch(() => undefined); + }); +} + +export function stopNativeAssistantSpeech(): void { generation += 1; - speechEpoch += 1; + interruptActiveUtterance(); stopSubscription?.(); stopSubscription = null; stopVoiceSubscription?.(); stopVoiceSubscription = null; - scheduledSegments.clear(); + stopStreamSubscription?.(); + stopStreamSubscription = null; activeSpeechSessionId = null; - void stopPocketVoice().catch(() => { - // Stopping an inactive or unavailable native player is best-effort during - // lifecycle teardown. - }); } export function startNativeAssistantSpeech( @@ -151,6 +213,16 @@ export function startNativeAssistantSpeech( stopNativeAssistantSpeech(); activeSpeechSessionId = sessionId; const activeGeneration = generation; + streamListenerReady = listenToPocketVoiceStream(handleStreamEvent).then( + (unlisten) => { + if (activeGeneration !== generation) { + unlisten(); + return; + } + stopStreamSubscription = unlisten; + }, + ); + const initialMessages = useChatStore.getState().messagesBySession[sessionId] ?? []; const toolCountByMessage = new Map(); @@ -166,17 +238,38 @@ export function startNativeAssistantSpeech( completedMessages.add(message.id); } let textOrdinal = 0; - message.content.forEach((content) => { - if (content.type === "text") { - consumedTextBySlot.set( - `${message.id}\0text:${textOrdinal}`, - content.text.trim(), - ); - textOrdinal += 1; - } - }); + for (const content of message.content) { + if (content.type !== "text") continue; + consumedTextBySlot.set( + `${message.id}\0text:${textOrdinal}`, + content.text, + ); + textOrdinal += 1; + } } + const ensureUtterance = (target: SpeechTarget): ActiveUtterance => { + if (activeUtterance) return activeUtterance; + const utterance: ActiveUtterance = { + id: crypto.randomUUID(), + sessionId, + target, + text: "", + finishing: false, + onFailure, + }; + activeUtterance = utterance; + queueStreamCommand( + utterance, + async () => { + await streamListenerReady; + await startPocketVoiceStream(utterance.id); + }, + onFailure, + ); + return utterance; + }; + const inspectNow = () => { if (activeGeneration !== generation) return; const voice = useVoiceConversationStore.getState(); @@ -186,7 +279,7 @@ export function startNativeAssistantSpeech( ) { return; } - const segments: SpeakableSegment[] = []; + const messages = useChatStore.getState().messagesBySession[sessionId] ?? []; for (const message of messages) { if ( @@ -199,95 +292,59 @@ export function startNativeAssistantSpeech( (content) => content.type === "toolRequest", ).length; const priorToolCount = toolCountByMessage.get(message.id) ?? 0; + const crossedToolBoundary = toolCount > priorToolCount; const completed = message.metadata?.completionStatus === "completed" && !completedMessages.has(message.id); - const crossedToolBoundary = toolCount > priorToolCount; toolCountByMessage.set(message.id, toolCount); if (completed) completedMessages.add(message.id); + let textOrdinal = 0; - message.content.forEach((content, contentIndex) => { - if (content.type !== "text") return; - const sourceText = content.text.trim(); - const slot = `${message.id}\0text:${textOrdinal}`; + for (const content of message.content) { + if (content.type !== "text") continue; + const target = { messageId: message.id, textOrdinal }; + const slot = targetKey(target); textOrdinal += 1; - if (!sourceText) return; - const consumedText = consumedTextBySlot.get(slot) ?? ""; - const consumedPrefix = sourceText.startsWith(consumedText) - ? consumedText - : ""; - const pendingText = sourceText.slice(consumedPrefix.length); - const split = splitSpeakableText( - pendingText, - crossedToolBoundary || completed, + const previous = consumedTextBySlot.get(slot) ?? ""; + if (content.text === previous) continue; + const delta = content.text.startsWith(previous) + ? content.text.slice(previous.length) + : content.text; + consumedTextBySlot.set(slot, content.text); + if (!delta) continue; + + if (voice.userSpeaking) { + setTargetStatus(sessionId, target, "notSpoken"); + recordPlaybackNotice(sessionId, slot, content.text, "notSpoken"); + continue; + } + + const utterance = ensureUtterance(target); + if (utterance.finishing) continue; + utterance.text += delta; + queueStreamCommand( + utterance, + () => appendPocketVoiceStream(utterance.id, delta), + onFailure, ); - if (split.consumedLength === 0) return; - consumedTextBySlot.set( - slot, - sourceText.slice(0, consumedPrefix.length + split.consumedLength), + } + + const utterance = activeUtterance; + if (crossedToolBoundary && utterance && !utterance.finishing) { + queueStreamCommand( + utterance, + () => flushPocketVoiceStream(utterance.id), + onFailure, + ); + } + if (completed && utterance && !utterance.finishing) { + utterance.finishing = true; + queueStreamCommand( + utterance, + () => finishPocketVoiceStream(utterance.id), + onFailure, ); - const completesSourceText = - consumedPrefix.length + split.consumedLength === sourceText.length; - split.segments.forEach((text, segmentIndex) => { - segments.push({ - key: `${slot}\0${consumedPrefix.length}\0${sourceText.length}\0${segmentIndex}`, - messageId: message.id, - contentIndex, - text, - sourceText, - completesSourceText: - completesSourceText && segmentIndex === split.segments.length - 1, - }); - }); - }); - } - for (const segment of segments) { - if (voice.userSpeaking) { - setSegmentStatus(sessionId, segment, "notSpoken"); - recordPlaybackNotice(sessionId, segment, "notSpoken"); - continue; } - const segmentEpoch = speechEpoch; - scheduledSegments.set(segment.key, segment); - playbackQueue = playbackQueue.then(async () => { - if ( - activeGeneration !== generation || - segmentEpoch !== speechEpoch || - !scheduledSegments.has(segment.key) - ) { - return; - } - const current = useVoiceConversationStore.getState(); - if (current.userSpeaking) { - setSegmentStatus(sessionId, segment, "notSpoken"); - recordPlaybackNotice(sessionId, segment, "notSpoken"); - scheduledSegments.delete(segment.key); - return; - } - setSegmentStatus(sessionId, segment, "speaking"); - current.setUiState("agent-speaking"); - try { - await speakPocketVoice(segment.text); - if ( - activeGeneration === generation && - segmentEpoch === speechEpoch && - segment.completesSourceText - ) { - setSegmentStatus(sessionId, segment, "spoken"); - } - } catch (error) { - if (activeGeneration === generation && segmentEpoch === speechEpoch) { - setSegmentStatus(sessionId, segment, "failed"); - recordPlaybackNotice(sessionId, segment, "failed"); - onFailure(segment.text, error); - } - } finally { - scheduledSegments.delete(segment.key); - if (activeGeneration === generation) { - useVoiceConversationStore.getState().setUiState("listening"); - } - } - }); } }; @@ -315,19 +372,7 @@ export function startNativeAssistantSpeech( const becameUserSpeaking = voice.userSpeaking && !wasUserSpeaking; wasUserSpeaking = voice.userSpeaking; if (!becameUserSpeaking || activeGeneration !== generation) return; - - speechEpoch += 1; - for (const segment of scheduledSegments.values()) { - setSegmentStatus(sessionId, segment, "interrupted"); - recordPlaybackNotice(sessionId, segment, "interrupted"); - } - scheduledSegments.clear(); - void stopPocketVoice().catch((error) => { - console.error("Native Pocket barge-in stop failed", { - sessionId, - error, - }); - }); + interruptActiveUtterance(); }); queueMicrotask(inspect); } From 66a5bdcce853e8e09817326722bf9e7173096c30 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Wed, 19 Aug 2026 22:40:20 -0400 Subject: [PATCH 06/11] fix(voice): keep stream flushing lint-clean cross-platform --- src-tauri/src/commands/pocket_voice.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src-tauri/src/commands/pocket_voice.rs b/src-tauri/src/commands/pocket_voice.rs index bd285becd..e7aa50674 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -820,7 +820,7 @@ pub fn flush_pocket_voice_stream( #[cfg(not(target_os = "macos"))] { let _ = (state, stream_id); - return Err("Pocket voice playback is currently supported on macOS only".to_string()); + Err("Pocket voice playback is currently supported on macOS only".to_string()) } #[cfg(target_os = "macos")] From 58b397e90aa9527103f9ddba659225a5dcc8b0b1 Mon Sep 17 00:00:00 2001 From: jtennant Date: Thu, 20 Aug 2026 14:11:55 +0000 Subject: [PATCH 07/11] fix(voice): preserve first streamed reply --- .../hooks/useVoiceConversationController.ts | 18 +++++-- .../lib/nativeAssistantSpeech.test.ts | 51 ++++++++++++++++++- .../lib/nativeAssistantSpeech.ts | 25 ++++++--- 3 files changed, 83 insertions(+), 11 deletions(-) diff --git a/src/features/voice-conversation/hooks/useVoiceConversationController.ts b/src/features/voice-conversation/hooks/useVoiceConversationController.ts index 3b1d2dbea..df364dd87 100644 --- a/src/features/voice-conversation/hooks/useVoiceConversationController.ts +++ b/src/features/voice-conversation/hooks/useVoiceConversationController.ts @@ -13,6 +13,7 @@ import { } from "../stores/voiceConversationStore"; import { startNativeAssistantSpeech, + stopNativeAssistantSpeech, takeVoicePlaybackNotices, } from "../lib/nativeAssistantSpeech"; import type { VoiceConversationStatus } from "../api/voiceConversation"; @@ -623,9 +624,7 @@ export function useVoiceConversationController({ status.sessionId, ]); - useEffect(() => { - if (status.lifecycle !== "running" || status.sessionId !== sessionId) - return; + const startAssistantSpeech = useCallback(() => { startNativeAssistantSpeech(sessionId, (text, playbackError) => { addErrorNotification( sessionId, @@ -639,7 +638,13 @@ export function useVoiceConversationController({ error: playbackError, }); }); - }, [sessionId, status.lifecycle, status.sessionId]); + }, [sessionId]); + + useEffect(() => { + if (status.lifecycle !== "running" || status.sessionId !== sessionId) + return; + startAssistantSpeech(); + }, [sessionId, startAssistantSpeech, status.lifecycle, status.sessionId]); const isActive = status.sessionId !== null && status.lifecycle !== "stopped"; const controlEnabled = enabled && isGooseSession && !readOnly && !disabled; @@ -681,12 +686,16 @@ export function useVoiceConversationController({ // subscriber must exist before the microphone lifecycle starts. ensureVoiceEventDeliveryInitialized(); activeSendRoute = { sessionId, send: onSend }; + // Capture the history boundary before native startup can admit a + // transcript and produce the first assistant response. + startAssistantSpeech(); try { await start(sessionId); } catch (startError) { const backendStatus = useVoiceConversationStore.getState().status; if (backendStatus.sessionId !== sessionId) { activeSendRoute = null; + stopNativeAssistantSpeech(); } addErrorNotification(sessionId, errorText(startError)); } @@ -700,6 +709,7 @@ export function useVoiceConversationController({ pocketReady, sessionId, start, + startAssistantSpeech, stop, ]); diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts index aa917d4e0..33cd2fd28 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts @@ -39,9 +39,10 @@ function assistant( completionStatus: NonNullable< Message["metadata"] >["completionStatus"] = "inProgress", + id = "assistant-1", ): Message { return { - id: "assistant-1", + id, role: "assistant", created: 1, content, @@ -119,6 +120,54 @@ describe("native assistant speech stream", () => { }); }); + it("preserves the first live reply while speech is arming", async () => { + const history = assistant( + [{ type: "text", text: "Historical response." }], + "completed", + "assistant-history", + ); + useChatStore.getState().setMessages("session-1", [history]); + useVoiceConversationStore.setState((voice) => ({ + status: { + ...voice.status, + lifecycle: "stopped", + sessionId: null, + ownerWindowLabel: null, + revision: 0, + }, + uiState: "off", + })); + + startNativeAssistantSpeech("session-1", vi.fn()); + useChatStore + .getState() + .setMessages("session-1", [ + history, + assistant([{ type: "text", text: "First live reply." }]), + ]); + useVoiceConversationStore.setState((voice) => ({ + status: { + ...voice.status, + lifecycle: "running", + sessionId: "session-1", + ownerWindowLabel: "main", + revision: 1, + }, + uiState: "listening", + })); + + await vi.waitFor(() => + expect(mocks.append).toHaveBeenCalledWith( + mocks.start.mock.calls[0]?.[0], + "First live reply.", + ), + ); + expect(mocks.append).not.toHaveBeenCalledWith( + expect.any(String), + "Historical response.", + ); + }); + it("derives speaking and completion state from backend playback events", async () => { startNativeAssistantSpeech("session-1", vi.fn()); useChatStore diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts index 0a1c67fbe..95605ab42 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts @@ -360,15 +360,28 @@ export function startNativeAssistantSpeech( }; stopSubscription = useChatStore.subscribe(inspect); - let wasUserSpeaking = useVoiceConversationStore.getState().userSpeaking; + const initialVoice = useVoiceConversationStore.getState(); + let reachedRunning = + initialVoice.status.lifecycle === "running" && + initialVoice.status.sessionId === sessionId; + let wasUserSpeaking = initialVoice.userSpeaking; stopVoiceSubscription = useVoiceConversationStore.subscribe((voice) => { - if ( - voice.status.lifecycle !== "running" || - voice.status.sessionId !== sessionId - ) { - stopNativeAssistantSpeech(); + const runningForSession = + voice.status.lifecycle === "running" && + voice.status.sessionId === sessionId; + if (!runningForSession) { + if ( + reachedRunning || + voice.status.lifecycle === "unavailable" || + (voice.status.sessionId !== null && + voice.status.sessionId !== sessionId) + ) { + stopNativeAssistantSpeech(); + } return; } + reachedRunning = true; + inspect(); const becameUserSpeaking = voice.userSpeaking && !wasUserSpeaking; wasUserSpeaking = voice.userSpeaking; if (!becameUserSpeaking || activeGeneration !== generation) return; From 655ba1fea1335345c8b4b8d469d676abcc9c24e5 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Thu, 20 Aug 2026 10:59:25 -0400 Subject: [PATCH 08/11] fix(voice): flush sped-up audio at tool boundaries --- .../src/commands/pocket_playback_speed_dsp.rs | 53 +++++++++++++++++++ src-tauri/src/commands/pocket_voice.rs | 17 ++++++ 2 files changed, 70 insertions(+) diff --git a/src-tauri/src/commands/pocket_playback_speed_dsp.rs b/src-tauri/src/commands/pocket_playback_speed_dsp.rs index 346568f09..221768342 100644 --- a/src-tauri/src/commands/pocket_playback_speed_dsp.rs +++ b/src-tauri/src/commands/pocket_playback_speed_dsp.rs @@ -11,6 +11,7 @@ const UNITY_EPSILON: f32 = 0.000_1; pub(super) struct StreamingSpeedProcessor { speed: f32, + sample_rate: u32, stretch: Option, input_latency: usize, output_latency: usize, @@ -27,6 +28,7 @@ impl StreamingSpeedProcessor { if (speed - DEFAULT_PLAYBACK_SPEED).abs() <= UNITY_EPSILON { return Ok(Self { speed, + sample_rate, stretch: None, input_latency: 0, output_latency: 0, @@ -46,6 +48,7 @@ impl StreamingSpeedProcessor { Ok(Self { speed, + sample_rate, stretch: Some(stretch), input_latency, output_latency, @@ -123,6 +126,12 @@ impl StreamingSpeedProcessor { Ok(output) } + pub(super) fn drain_and_reset(&mut self) -> Result, String> { + let output = self.finish()?; + *self = Self::new(self.speed, self.sample_rate)?; + Ok(output) + } + fn trim_and_count(&mut self, samples: &[f32]) -> Vec { let trim = self.trim_remaining.min(samples.len()); self.trim_remaining -= trim; @@ -230,6 +239,50 @@ mod tests { assert_frequency(&output[2_000..], frequency, 4.0); } + #[test] + fn two_x_boundary_drain_emits_complete_stretched_length() { + let input: Vec = (0..SAMPLE_RATE) + .map(|sample| { + (2.0 * std::f32::consts::PI * 220.0 * sample as f32 / SAMPLE_RATE as f32).sin() + }) + .collect(); + let mut processor = StreamingSpeedProcessor::new(2.0, SAMPLE_RATE).expect("processor"); + let mut output = Vec::new(); + for block in input.chunks(1_920) { + output.extend(processor.process(block).expect("stream block")); + } + + let tail = processor.drain_and_reset().expect("boundary drain"); + assert!(!tail.is_empty(), "boundary drain did not emit a tail"); + output.extend(tail); + + assert_eq!(output.len(), stretched_len(input.len(), 2.0)); + } + + #[test] + fn two_x_processing_continues_after_boundary_reset() { + let first = vec![0.25; SAMPLE_RATE as usize]; + let second: Vec = (0..SAMPLE_RATE) + .map(|sample| { + (2.0 * std::f32::consts::PI * 440.0 * sample as f32 / SAMPLE_RATE as f32).sin() + }) + .collect(); + let mut processor = StreamingSpeedProcessor::new(2.0, SAMPLE_RATE).expect("processor"); + for block in first.chunks(1_920) { + processor.process(block).expect("first stream block"); + } + processor.drain_and_reset().expect("boundary drain"); + + let mut output = Vec::new(); + for block in second.chunks(1_920) { + output.extend(processor.process(block).expect("second stream block")); + } + output.extend(processor.finish().expect("finish second segment")); + + assert_eq!(output.len(), stretched_len(second.len(), 2.0)); + assert_frequency(&output[2_000..], 440.0, 8.0); + } + fn assert_frequency(samples: &[f32], expected: f32, tolerance: f32) { let measured = zero_crossing_frequency(samples); assert!( diff --git a/src-tauri/src/commands/pocket_voice.rs b/src-tauri/src/commands/pocket_voice.rs index e7aa50674..ebcd90331 100644 --- a/src-tauri/src/commands/pocket_voice.rs +++ b/src-tauri/src/commands/pocket_voice.rs @@ -1925,6 +1925,23 @@ fn run_pocket_voice_stream( )? { return Ok(PocketStreamEventState::Interrupted); } + let tail = speed_processor.drain_and_reset()?; + if !tail.is_empty() { + player.append(SamplesBuffer::new(channels, rate, tail)); + if !playback_started { + playback_started = true; + emit_pocket_stream_event( + app, + stream_id, + PocketStreamEventState::Started, + None, + ); + println!("VOICE_CONVERSATION_PLAYBACK_STARTED"); + std::io::stdout() + .flush() + .map_err(|error| format!("signal Pocket playback start: {error}"))?; + } + } } Ok(PocketStreamCommand::Finish) => { if !synthesize_pocket_stream_ready( From 92fc2b20a6b80ffb232c8a8b683dfd0adccb1d16 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Thu, 20 Aug 2026 12:38:16 -0400 Subject: [PATCH 09/11] fix(voice): preserve speech targets across tools --- .../lib/nativeAssistantSpeech.test.ts | 60 +++++++++++++++++++ .../lib/nativeAssistantSpeech.ts | 39 +++++++++--- 2 files changed, 90 insertions(+), 9 deletions(-) diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts index 33cd2fd28..01353bf8f 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts @@ -230,6 +230,66 @@ describe("native assistant speech stream", () => { expect(mocks.finish).not.toHaveBeenCalled(); }); + it("updates every text block around a tool with the utterance status", async () => { + startNativeAssistantSpeech("session-1", vi.fn()); + useChatStore + .getState() + .setMessages("session-1", [ + assistant([{ type: "text", text: "Before the tool." }]), + ]); + await vi.waitFor(() => expect(mocks.append).toHaveBeenCalledTimes(1)); + emit("started"); + + useChatStore.getState().setMessages("session-1", [ + assistant([ + { type: "text", text: "Before the tool." }, + { + type: "toolRequest", + id: "tool-1", + name: "Read", + arguments: {}, + status: "completed", + }, + { type: "text", text: "After the tool." }, + ]), + ]); + + await vi.waitFor(() => { + expect(mocks.flush).toHaveBeenCalledTimes(1); + expect(mocks.append).toHaveBeenLastCalledWith( + mocks.start.mock.calls[0]?.[0], + "After the tool.", + ); + }); + expect( + useChatStore.getState().messagesBySession["session-1"]?.[0]?.content[2], + ).toMatchObject({ speech: { status: "speaking" } }); + + useChatStore.getState().setMessages("session-1", [ + assistant( + [ + { type: "text", text: "Before the tool." }, + { + type: "toolRequest", + id: "tool-1", + name: "Read", + arguments: {}, + status: "completed", + }, + { type: "text", text: "After the tool." }, + ], + "completed", + ), + ]); + await vi.waitFor(() => expect(mocks.finish).toHaveBeenCalledTimes(1)); + emit("completed"); + + const content = + useChatStore.getState().messagesBySession["session-1"]?.[0]?.content; + expect(content?.[0]).toMatchObject({ speech: { status: "spoken" } }); + expect(content?.[2]).toMatchObject({ speech: { status: "spoken" } }); + }); + it("interrupts one utterance status even when many deltas are queued", async () => { startNativeAssistantSpeech("session-1", vi.fn()); useChatStore diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts index 95605ab42..4ed0e8973 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts @@ -15,9 +15,10 @@ type SpeechTarget = { messageId: string; textOrdinal: number }; type ActiveUtterance = { id: string; sessionId: string; - target: SpeechTarget; + targets: SpeechTarget[]; text: string; finishing: boolean; + status: SpeechStatus | null; onFailure: SpeechFailureHandler; }; type SpeechStatus = @@ -95,6 +96,13 @@ function setTargetStatus( }); } +function setUtteranceStatus(utterance: ActiveUtterance, status: SpeechStatus) { + utterance.status = status; + for (const target of utterance.targets) { + setTargetStatus(utterance.sessionId, target, status); + } +} + function failActiveUtterance( utteranceId: string, error: unknown, @@ -102,7 +110,7 @@ function failActiveUtterance( ) { const utterance = activeUtterance; if (!utterance || utterance.id !== utteranceId) return; - setTargetStatus(utterance.sessionId, utterance.target, "failed"); + setUtteranceStatus(utterance, "failed"); recordPlaybackNotice( utterance.sessionId, utterance.id, @@ -137,16 +145,16 @@ function handleStreamEvent(event: PocketVoiceStreamEvent) { switch (event.state) { case "started": - setTargetStatus(utterance.sessionId, utterance.target, "speaking"); + setUtteranceStatus(utterance, "speaking"); voice.setUiState("agent-speaking"); break; case "completed": - setTargetStatus(utterance.sessionId, utterance.target, "spoken"); + setUtteranceStatus(utterance, "spoken"); voice.setUiState("listening"); activeUtterance = null; break; case "interrupted": - setTargetStatus(utterance.sessionId, utterance.target, "interrupted"); + setUtteranceStatus(utterance, "interrupted"); recordPlaybackNotice( utterance.sessionId, utterance.id, @@ -157,7 +165,7 @@ function handleStreamEvent(event: PocketVoiceStreamEvent) { activeUtterance = null; break; case "failed": - setTargetStatus(utterance.sessionId, utterance.target, "failed"); + setUtteranceStatus(utterance, "failed"); recordPlaybackNotice( utterance.sessionId, utterance.id, @@ -179,7 +187,7 @@ function interruptActiveUtterance() { commandEpoch += 1; activeUtterance = null; if (utterance) { - setTargetStatus(utterance.sessionId, utterance.target, "interrupted"); + setUtteranceStatus(utterance, "interrupted"); recordPlaybackNotice( utterance.sessionId, utterance.id, @@ -249,13 +257,26 @@ export function startNativeAssistantSpeech( } const ensureUtterance = (target: SpeechTarget): ActiveUtterance => { - if (activeUtterance) return activeUtterance; + if (activeUtterance) { + if ( + !activeUtterance.targets.some( + (candidate) => targetKey(candidate) === targetKey(target), + ) + ) { + activeUtterance.targets.push(target); + if (activeUtterance.status) { + setTargetStatus(sessionId, target, activeUtterance.status); + } + } + return activeUtterance; + } const utterance: ActiveUtterance = { id: crypto.randomUUID(), sessionId, - target, + targets: [target], text: "", finishing: false, + status: null, onFailure, }; activeUtterance = utterance; From 04f48746ae0a6b849bed870bc05084925ff7bf6d Mon Sep 17 00:00:00 2001 From: John Tennant Date: Thu, 20 Aug 2026 21:18:37 -0400 Subject: [PATCH 10/11] fix(voice): queue replies behind finishing playback --- .../useVoiceConversationController.test.ts | 15 ++++++ .../lib/nativeAssistantSpeech.test.ts | 47 +++++++++++++++++++ .../lib/nativeAssistantSpeech.ts | 11 +++++ 3 files changed, 73 insertions(+) diff --git a/src/features/voice-conversation/hooks/useVoiceConversationController.test.ts b/src/features/voice-conversation/hooks/useVoiceConversationController.test.ts index c4834088c..68e06b5c6 100644 --- a/src/features/voice-conversation/hooks/useVoiceConversationController.test.ts +++ b/src/features/voice-conversation/hooks/useVoiceConversationController.test.ts @@ -3,6 +3,18 @@ import { act, renderHook, waitFor } from "@testing-library/react"; import { useChatStore } from "@/features/chat/stores/chatStore"; import { useVoiceConversationStore } from "../stores/voiceConversationStore"; +const nativeAssistantSpeechMocks = vi.hoisted(() => ({ + start: vi.fn(), + stop: vi.fn(), + takeNotices: vi.fn<() => string | null>(() => null), +})); + +vi.mock("../lib/nativeAssistantSpeech", () => ({ + startNativeAssistantSpeech: nativeAssistantSpeechMocks.start, + stopNativeAssistantSpeech: nativeAssistantSpeechMocks.stop, + takeVoicePlaybackNotices: nativeAssistantSpeechMocks.takeNotices, +})); + import { canBindVoiceSendRoute, canClaimVoiceSendRoute, @@ -73,6 +85,9 @@ describe("voice transcript delivery coordination", () => { }); beforeEach(() => { + nativeAssistantSpeechMocks.start.mockClear(); + nativeAssistantSpeechMocks.stop.mockClear(); + nativeAssistantSpeechMocks.takeNotices.mockClear(); useChatStore.setState({ messagesBySession: {}, sessionStateById: {} }); }); it("serializes deliveries for the same session and re-evaluates in order", async () => { diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts index 01353bf8f..ff188c60d 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts @@ -290,6 +290,53 @@ describe("native assistant speech stream", () => { expect(content?.[2]).toMatchObject({ speech: { status: "spoken" } }); }); + it("queues the next reply until the finishing stream completes", async () => { + startNativeAssistantSpeech("session-1", vi.fn()); + useChatStore.getState().setMessages("session-1", [ + assistant( + [{ type: "text", text: "First reply." }], + "completed", + "assistant-1", + ), + ]); + await vi.waitFor(() => expect(mocks.finish).toHaveBeenCalledTimes(1)); + const firstStreamId = mocks.start.mock.calls[0]?.[0] as string; + + useChatStore.getState().setMessages("session-1", [ + assistant( + [{ type: "text", text: "First reply." }], + "completed", + "assistant-1", + ), + assistant( + [{ type: "text", text: "Second reply." }], + "completed", + "assistant-2", + ), + ]); + await Promise.resolve(); + expect(mocks.start).toHaveBeenCalledTimes(1); + expect(mocks.append).not.toHaveBeenCalledWith( + expect.any(String), + "Second reply.", + ); + + mocks.streamHandler?.({ + streamId: firstStreamId, + state: "completed", + error: null, + }); + + await vi.waitFor(() => { + expect(mocks.start).toHaveBeenCalledTimes(2); + expect(mocks.append).toHaveBeenCalledWith( + mocks.start.mock.calls[1]?.[0], + "Second reply.", + ); + expect(mocks.finish).toHaveBeenCalledTimes(2); + }); + }); + it("interrupts one utterance status even when many deltas are queued", async () => { startNativeAssistantSpeech("session-1", vi.fn()); useChatStore diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts index 4ed0e8973..0b3d1c6dc 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.ts @@ -20,6 +20,7 @@ type ActiveUtterance = { finishing: boolean; status: SpeechStatus | null; onFailure: SpeechFailureHandler; + onTerminal: () => void; }; type SpeechStatus = | "speaking" @@ -120,6 +121,7 @@ function failActiveUtterance( useVoiceConversationStore.getState().setUiState("listening"); activeUtterance = null; onFailure(utterance.text, error); + utterance.onTerminal(); } function queueStreamCommand( @@ -152,6 +154,7 @@ function handleStreamEvent(event: PocketVoiceStreamEvent) { setUtteranceStatus(utterance, "spoken"); voice.setUiState("listening"); activeUtterance = null; + utterance.onTerminal(); break; case "interrupted": setUtteranceStatus(utterance, "interrupted"); @@ -163,6 +166,7 @@ function handleStreamEvent(event: PocketVoiceStreamEvent) { ); voice.setUiState("listening"); activeUtterance = null; + utterance.onTerminal(); break; case "failed": setUtteranceStatus(utterance, "failed"); @@ -178,6 +182,7 @@ function handleStreamEvent(event: PocketVoiceStreamEvent) { utterance.text, event.error ?? new Error("Pocket voice stream failed"), ); + utterance.onTerminal(); break; } } @@ -194,6 +199,7 @@ function interruptActiveUtterance() { utterance.text, "interrupted", ); + utterance.onTerminal(); } void stopPocketVoice().catch(() => undefined); commandQueue = commandQueue.then(async () => { @@ -278,6 +284,7 @@ export function startNativeAssistantSpeech( finishing: false, status: null, onFailure, + onTerminal: () => queueMicrotask(inspect), }; activeUtterance = utterance; queueStreamCommand( @@ -300,6 +307,10 @@ export function startNativeAssistantSpeech( ) { return; } + // The backend owns the current stream until its terminal playback event. + // Leave later transcript changes entirely unconsumed so that terminal + // handling can inspect them into a distinct utterance. + if (activeUtterance?.finishing) return; const messages = useChatStore.getState().messagesBySession[sessionId] ?? []; for (const message of messages) { From 168968a5e9396fb4aa296889431e967136f0ea59 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Thu, 20 Aug 2026 22:06:53 -0400 Subject: [PATCH 11/11] style(voice): format queued reply test --- .../lib/nativeAssistantSpeech.test.ts | 42 ++++++++++--------- 1 file changed, 23 insertions(+), 19 deletions(-) diff --git a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts index ff188c60d..3d0c56120 100644 --- a/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts +++ b/src/features/voice-conversation/lib/nativeAssistantSpeech.test.ts @@ -292,28 +292,32 @@ describe("native assistant speech stream", () => { it("queues the next reply until the finishing stream completes", async () => { startNativeAssistantSpeech("session-1", vi.fn()); - useChatStore.getState().setMessages("session-1", [ - assistant( - [{ type: "text", text: "First reply." }], - "completed", - "assistant-1", - ), - ]); + useChatStore + .getState() + .setMessages("session-1", [ + assistant( + [{ type: "text", text: "First reply." }], + "completed", + "assistant-1", + ), + ]); await vi.waitFor(() => expect(mocks.finish).toHaveBeenCalledTimes(1)); const firstStreamId = mocks.start.mock.calls[0]?.[0] as string; - useChatStore.getState().setMessages("session-1", [ - assistant( - [{ type: "text", text: "First reply." }], - "completed", - "assistant-1", - ), - assistant( - [{ type: "text", text: "Second reply." }], - "completed", - "assistant-2", - ), - ]); + useChatStore + .getState() + .setMessages("session-1", [ + assistant( + [{ type: "text", text: "First reply." }], + "completed", + "assistant-1", + ), + assistant( + [{ type: "text", text: "Second reply." }], + "completed", + "assistant-2", + ), + ]); await Promise.resolve(); expect(mocks.start).toHaveBeenCalledTimes(1); expect(mocks.append).not.toHaveBeenCalledWith(