feat(voice): speak assistant text while it streams

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<ssstretch::Stretch>,
     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<Vec<f32>, String> {
+        let output = self.finish()?;
+        *self = Self::new(self.speed, self.sample_rate)?;
+        Ok(output)
+    }
+
     fn trim_and_count(&mut self, samples: &[f32]) -> Vec<f32> {
         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<f32> = (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<f32> = (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!(

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);
+        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,42 @@ 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);
+                }
+                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(
                     app,

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,

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);
+  });
 });

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<void> {
   return invoke("speak_pocket_voice", { text });
 }
 
+export function startPocketVoiceStream(streamId: string): Promise<void> {
+  return invoke("start_pocket_voice_stream", { streamId });
+}
+
+export function appendPocketVoiceStream(
+  streamId: string,
+  text: string,
+): Promise<void> {
+  return invoke("append_pocket_voice_stream", { streamId, text });
+}
+
+export function flushPocketVoiceStream(streamId: string): Promise<void> {
+  return invoke("flush_pocket_voice_stream", { streamId });
+}
+
+export function finishPocketVoiceStream(streamId: string): Promise<void> {
+  return invoke("finish_pocket_voice_stream", { streamId });
+}
+
 export function stopPocketVoice(): Promise<boolean> {
   return invoke<boolean>("stop_pocket_voice");
 }
@@ -98,3 +123,11 @@ export function listenToPocketVoiceStatus(
     onStatus(event.payload),
   );
 }
+
+export function listenToPocketVoiceStream(
+  onEvent: (event: PocketVoiceStreamEvent) => void,
+): Promise<UnlistenFn> {
+  return listen<PocketVoiceStreamEvent>("pocket-voice:stream-event", (event) =>
+    onEvent(event.payload),
+  );
+}

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

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,
   ]);