fix(voice): speed up replies and hide prompt wrappers

Keep voice prompt cleanup in the client presentation path so provider history and live messages render consistently without rewriting canonical input. Correct native TTS scheduling and threading, and reconnect interruption to speech detection and session cancellation.

packages/server/src/server/session/voice/voice-session.test.ts

@@ -1,3 +1,5 @@
+import { Readable } from "node:stream";
+import type { VoiceSpeakHandler } from "../../voice-types.js";
 import { EventEmitter } from "node:events";
 import pino from "pino";
 import { describe, expect, test, vi } from "vitest";
@@ -7,6 +9,7 @@ import type { ManagedAgent } from "../../agent/agent-manager.js";
 import type { SessionOutboundMessage } from "../../messages.js";
 import type {
   SpeechToTextProvider,
+  TextToSpeechProvider,
   StreamingTranscriptionCommittedEvent,
   StreamingTranscriptionEvent,
   StreamingTranscriptionSession,
@@ -79,7 +82,8 @@ function createFakeHost(): FakeVoiceHost {
   };
 }
 
-function createVoiceSession() {
+function createVoiceSession(tts: TextToSpeechProvider | null = null) {
+  let speakHandler: VoiceSpeakHandler | undefined;
   const detector = new FakeVoiceTurnDetectionSession();
   const sttSession = new FakeVoiceSttSession();
   const stt: SpeechToTextProvider = {
@@ -96,11 +100,29 @@ function createVoiceSession() {
     logger: pino({ level: "silent" }),
     sessionId: "voice-session-test",
     sttLanguage: "en",
-    tts: null,
+    tts,
+    voiceBridge: {
+      registerVoiceSpeakHandler: (_id, handler) => {
+        speakHandler = handler;
+      },
+    },
     stt,
     voice: { turnDetection },
   });
-  return { voiceSession, detector, sttSession, host };
+  return {
+    voiceSession,
+    detector,
+    sttSession,
+    host,
+    speak: (args: Parameters<VoiceSpeakHandler>[0]) => {
+      if (!speakHandler) throw new Error("Voice speak handler not registered");
+      return speakHandler(args);
+    },
+  };
+}
+
+function isAudioOutput(message: SessionOutboundMessage): boolean {
+  return message.type === "audio_output";
 }
 
 async function settle(): Promise<void> {
@@ -110,6 +132,69 @@ async function settle(): Promise<void> {
 }
 
 describe("VoiceSession streaming transcription", () => {
+  test("interrupts playback and the agent on speech detection without partial transcripts", async () => {
+    const tts: TextToSpeechProvider = {
+      async synthesizeSpeech() {
+        return { stream: Readable.from([Buffer.from("audio")]), format: "pcm;rate=24000" };
+      },
+    };
+    const { voiceSession, detector, host, speak } = createVoiceSession(tts);
+    const interrupted: string[] = [];
+    host.interruptAgentIfRunning = async (agentId) => {
+      interrupted.push(agentId);
+    };
+    await voiceSession.handleSetVoiceMode(true, VOICE_AGENT_ID);
+    const playback = speak({ text: "A spoken response." });
+    try {
+      await vi.waitFor(() => expect(host.emitted.filter(isAudioOutput)).toHaveLength(1));
+      detector.emit("speech_started");
+      await settle();
+      expect(host.emitted).toContainEqual({
+        type: "voice_input_state",
+        payload: { isSpeaking: true },
+      });
+      expect(interrupted).toEqual([VOICE_AGENT_ID]);
+      await playback;
+    } finally {
+      await voiceSession.cleanup();
+      await playback.catch(() => {});
+    }
+  });
+
+  test("session abort stops later audio even when the speak tool supplies its own signal", async () => {
+    const tts: TextToSpeechProvider = {
+      async synthesizeSpeech() {
+        return { stream: Readable.from([Buffer.from("audio")]), format: "pcm;rate=24000" };
+      },
+    };
+    const { voiceSession, host, speak } = createVoiceSession(tts);
+    const external = new AbortController();
+    let abortRequested = false;
+    const emit = host.emit;
+    host.emit = (message) => {
+      emit(message);
+      if (message.type === "audio_output" && abortRequested)
+        voiceSession.handleAudioPlayed(message.payload.id);
+    };
+    await voiceSession.handleSetVoiceMode(true, VOICE_AGENT_ID);
+    const playback = speak({
+      text: "First sentence. Second sentence. Third sentence.",
+      signal: external.signal,
+    });
+    try {
+      await vi.waitFor(() => expect(host.emitted.filter(isAudioOutput)).toHaveLength(1));
+      abortRequested = true;
+      await voiceSession.handleAbort();
+      await playback;
+      expect(host.emitted.filter(isAudioOutput)).toHaveLength(1);
+      expect(external.signal.aborted).toBe(false);
+    } finally {
+      external.abort();
+      await voiceSession.cleanup();
+      await playback.catch(() => {});
+    }
+  });
+
   test("surfaces a refused voice-mode agent interruption", async () => {
     const { voiceSession, host } = createVoiceSession();
     host.interruptAgentIfRunning = vi.fn(async () => {

packages/server/src/server/session/voice/voice-session.ts

@@ -587,13 +587,9 @@ export class VoiceSession {
       sttLanguage: this.sttLanguage,
       callbacks: {
         onSpeechStarted: async () => {
+          // Voice STT providers return final transcripts only. Use the detector's
+          // confirmed speech event so interruption does not wait for transcription.
           this.sessionLogger.debug("Voice VAD speech_started");
-        },
-        onPartialTranscript: async ({ segmentId, transcript }) => {
-          this.sessionLogger.info(
-            { segmentId, transcriptLength: transcript.trim().length },
-            "voice_input_state emitting isSpeaking=true",
-          );
           this.emit({
             type: "voice_input_state",
             payload: {
@@ -602,6 +598,12 @@ export class VoiceSession {
           });
           await this.handleVoiceSpeechStart();
         },
+        onPartialTranscript: async ({ segmentId, transcript }) => {
+          this.sessionLogger.debug(
+            { segmentId, transcriptLength: transcript.trim().length },
+            "Voice partial transcript",
+          );
+        },
         onSpeechStopped: async () => {
           this.handleVoiceSpeechStopped();
           this.setPhase("transcribing");
@@ -1058,7 +1060,9 @@ export class VoiceSession {
         },
         "Voice speak tool call received by session handler",
       );
-      const abortSignal = signal ?? this.abortController.signal;
+      const abortSignal = signal
+        ? AbortSignal.any([signal, this.abortController.signal])
+        : this.abortController.signal;
       await this.ttsManager.generateAndWaitForPlayback(
         text,
         (msg) => this.emit(msg),

packages/server/src/server/speech/providers/local/sherpa/sherpa-tts.test.ts

@@ -1,53 +1,135 @@
-import { describe, expect, it, vi, beforeEach } from "vitest";
+import { afterEach, describe, expect, it } from "vitest";
+import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
 import pino from "pino";
+import { SherpaOnnxTTS } from "./sherpa-tts.js";
 
-const generate = vi.fn();
-const free = vi.fn();
+const directories: string[] = [];
+afterEach(() => {
+  for (const directory of directories.splice(0))
+    rmSync(directory, { recursive: true, force: true });
+});
 
-vi.mock("node:fs", () => ({
-  existsSync: vi.fn(() => true),
-}));
-
-vi.mock("./sherpa-onnx-node-loader.js", () => ({
-  loadSherpaOnnxNode: () => ({
-    OfflineTts: class {
-      public readonly sampleRate = 24000;
-
-      generate = generate;
-      free = free;
-    },
-  }),
-}));
+function createTts(numThreads?: number, beforeGenerate?: () => Promise<void>) {
+  const modelDir = mkdtempSync(join(tmpdir(), "paseo-tts-"));
+  directories.push(modelDir);
+  for (const name of ["model.onnx", "voices.bin", "tokens.txt"])
+    writeFileSync(join(modelDir, name), "");
+  mkdirSync(join(modelDir, "espeak-ng-data"));
+  let nativeConfig: unknown;
+  const requests: unknown[] = [];
+  const released: boolean[] = [];
+  const tts = new SherpaOnnxTTS(
+    { preset: "kokoro-en-v0_19", modelDir, numThreads },
+    pino({ level: "silent" }),
+    () => ({
+      OfflineTts: class {
+        sampleRate = 24000;
+        free() {
+          released.push(true);
+        }
+        constructor(config: unknown) {
+          nativeConfig = config;
+        }
+        async generateAsync(request: unknown) {
+          await beforeGenerate?.();
+          return this.generate(request);
+        }
+        generate(request: unknown) {
+          requests.push(request);
+          return { samples: Float32Array.from([0, 0.5, -0.5, 0.25]), sampleRate: 24000 };
+        }
+      },
+    }),
+  );
+  return { tts, requests, nativeConfig, released };
+}
 
 describe("SherpaOnnxTTS", () => {
-  beforeEach(() => {
-    generate.mockReset();
-    free.mockReset();
-  });
-
-  it("disables external buffers when calling sherpa generate", async () => {
-    generate.mockReturnValue({
-      samples: Float32Array.from([0, 0.5, -0.5, 0.25]),
-      sampleRate: 24000,
-    });
-
-    const { SherpaOnnxTTS } = await import("./sherpa-tts.js");
-    const tts = new SherpaOnnxTTS(
-      {
-        preset: "kokoro-en-v0_19",
-        modelDir: "/tmp/fake-model",
-      },
-      pino({ level: "silent" }),
-    );
-
+  it("requests Electron-compatible copied samples and returns PCM audio", async () => {
+    const { tts, requests } = createTts();
     const result = await tts.synthesizeSpeech("hello");
-
-    expect(generate).toHaveBeenCalledWith({
-      text: "hello",
-      sid: 0,
-      speed: 1,
-      enableExternalBuffer: false,
-    });
+    expect(requests).toEqual([{ text: "hello", sid: 0, speed: 1, enableExternalBuffer: false }]);
     expect(result.format).toBe("pcm;rate=24000");
+    const chunks: Buffer[] = [];
+    for await (const chunk of result.stream) chunks.push(chunk);
+    expect(Buffer.concat(chunks).length).toBe(8);
   });
+
+  it("keeps the event loop available and serializes native synthesis", async () => {
+    const pending: Array<() => void> = [];
+    const { tts, requests } = createTts(
+      undefined,
+      () => new Promise<void>((resolve) => pending.push(resolve)),
+    );
+    const first = tts.synthesizeSpeech("first");
+    const second = tts.synthesizeSpeech("second");
+    try {
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      expect(requests).toEqual([]);
+      expect(pending).toHaveLength(1);
+      pending.shift()!();
+      const audio = await first;
+      audio.stream.destroy();
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      expect(requests).toHaveLength(1);
+      expect(pending).toHaveLength(1);
+      pending.shift()!();
+      (await second).stream.destroy();
+      expect(requests.map((request) => (request as { text: string }).text)).toEqual([
+        "first",
+        "second",
+      ]);
+    } finally {
+      for (const release of pending) release();
+      await Promise.allSettled([first, second]);
+      tts.free();
+    }
+  });
+
+  it("continues queued synthesis after a native failure", async () => {
+    let attempts = 0;
+    const { tts, requests } = createTts(undefined, async () => {
+      attempts += 1;
+      if (attempts === 1) throw new Error("generation failed");
+    });
+    const first = tts.synthesizeSpeech("first");
+    const second = tts.synthesizeSpeech("second");
+    await expect(first).rejects.toThrow("generation failed");
+    (await second).stream.destroy();
+    expect(requests).toEqual([{ text: "second", sid: 0, speed: 1, enableExternalBuffer: false }]);
+    tts.free();
+  });
+
+  it("waits for active native synthesis before freeing and drops queued synthesis", async () => {
+    let finish!: () => void;
+    const gate = new Promise<void>((resolve) => {
+      finish = resolve;
+    });
+    const { tts, requests, released } = createTts(undefined, () => gate);
+    const first = tts.synthesizeSpeech("first");
+    const second = tts.synthesizeSpeech("second");
+    const rejected = expect(second).rejects.toThrow("TTS provider closed");
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    tts.free();
+    expect(released).toEqual([]);
+    finish();
+    (await first).stream.destroy();
+    await rejected;
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    expect(requests).toHaveLength(1);
+    expect(released).toEqual([true]);
+    await expect(tts.synthesizeSpeech("third")).rejects.toThrow("TTS provider closed");
+  });
+
+  it.each([undefined, 4])(
+    "passes the requested thread count at the native model boundary (%s)",
+    (threads) => {
+      const { nativeConfig } = createTts(threads);
+      expect(nativeConfig).toMatchObject({ model: { numThreads: threads ?? 2, provider: "cpu" } });
+      expect(nativeConfig).not.toHaveProperty("numThreads");
+      expect(nativeConfig).not.toHaveProperty("provider");
+    },
+  );
 });

packages/server/src/server/speech/providers/local/sherpa/sherpa-tts.ts

@@ -25,12 +25,12 @@ function assertFileExists(filePath: string, label: string): void {
 
 interface SherpaOfflineTtsNative {
   sampleRate?: number;
-  generate: (args: {
+  generateAsync: (args: {
     text: string;
     sid: number;
     speed: number;
     enableExternalBuffer: boolean;
-  }) => { samples?: Float32Array | number[]; sampleRate?: number } | undefined;
+  }) => Promise<{ samples?: Float32Array | number[]; sampleRate?: number } | undefined>;
   free?: () => void;
 }
 
@@ -39,13 +39,22 @@ export class SherpaOnnxTTS implements TextToSpeechProvider {
   private readonly speakerId: number;
   private readonly speed: number;
   private readonly logger: pino.Logger;
+  private synthesisQueue: Promise<void> = Promise.resolve();
+  private closed = false;
 
-  constructor(config: SherpaTtsConfig, logger: pino.Logger) {
+  constructor(
+    config: SherpaTtsConfig,
+    logger: pino.Logger,
+    loadNative: () => Pick<
+      ReturnType<typeof loadSherpaOnnxNode>,
+      "OfflineTts"
+    > = loadSherpaOnnxNode,
+  ) {
     this.logger = logger.child({ module: "speech", provider: "local", component: "tts" });
     this.speakerId = config.speakerId ?? 0;
     this.speed = config.speed ?? 1.0;
 
-    const sherpa = loadSherpaOnnxNode();
+    const sherpa = loadNative();
     if (typeof sherpa.OfflineTts !== "function") {
       throw new Error("sherpa-onnx-node OfflineTts is unavailable");
     }
@@ -61,6 +70,9 @@ export class SherpaOnnxTTS implements TextToSpeechProvider {
     assertFileExists(dataDir, "TTS espeak-ng dataDir");
 
     const modelConfig = {
+      // The native parser reads these under model, despite the upstream JS typedef.
+      numThreads: config.numThreads ?? 2,
+      provider: "cpu",
       kokoro: {
         model: modelPath,
         voices: voicesPath,
@@ -72,8 +84,6 @@ export class SherpaOnnxTTS implements TextToSpeechProvider {
 
     const offlineTtsConfig = {
       model: modelConfig,
-      numThreads: config.numThreads ?? 2,
-      provider: "cpu",
       maxNumSentences: 1,
     };
 
@@ -92,8 +102,22 @@ export class SherpaOnnxTTS implements TextToSpeechProvider {
       throw new Error("Cannot synthesize empty text");
     }
 
-    const audio = this.tts.generate({
-      text: trimmed,
+    // The native handle is not reentrant. Keep prefetch off the worker event loop
+    // while allowing only one generation at a time on this model.
+    const result = this.synthesisQueue.then(() => this.generateSpeech(trimmed));
+    this.synthesisQueue = result.then(
+      () => undefined,
+      () => undefined,
+    );
+    return result;
+  }
+
+  private async generateSpeech(text: string): Promise<SpeechStreamResult> {
+    if (this.closed) {
+      throw new Error("TTS provider closed");
+    }
+    const audio = await this.tts.generateAsync({
+      text,
       sid: this.speakerId,
       speed: this.speed,
       // Electron rejects native external-backed typed arrays. Request a copied buffer
@@ -138,10 +162,19 @@ export class SherpaOnnxTTS implements TextToSpeechProvider {
   }
 
   free(): void {
-    try {
-      this.tts?.free?.();
-    } catch {
-      // ignore
+    if (this.closed) {
+      return;
     }
+    this.closed = true;
+    // Let the active native call finish before releasing its handle. Queued
+    // calls observe closed and never enter native code.
+    void this.synthesisQueue.then(() => {
+      try {
+        this.tts.free?.();
+      } catch {
+        // ignore
+      }
+      return undefined;
+    });
   }
 }

packages/server/src/server/speech/providers/local/worker-process.local.e2e.test.ts

@@ -161,6 +161,78 @@ workerSpeechTest(
   120_000,
 );
 
+test("delivers audio and serves control requests while later TTS is synthesizing", async () => {
+  const worker = forkWorker();
+  const messages: LocalSpeechWorkerToParentMessage[] = [];
+  let stderr = "";
+  worker.stderr?.on("data", (chunk) => {
+    stderr += String(chunk);
+  });
+  worker.on("message", (message: LocalSpeechWorkerToParentMessage) => messages.push(message));
+  const config: LocalSpeechWorkerConfig = {
+    modelsDir,
+    voiceSttModel: "parakeet-tdt-0.6b-v2-int8",
+    dictationSttModel: "parakeet-tdt-0.6b-v2-int8",
+    voiceTtsModel: "kokoro-en-v0_19",
+  };
+  try {
+    await sendRequest(
+      worker,
+      messages,
+      { type: "tts.synthesize", config, text: "Ready." },
+      () => stderr,
+    );
+    const completed: string[] = [];
+    const first = sendRequest(
+      worker,
+      messages,
+      {
+        type: "tts.synthesize",
+        config,
+        text: "I found the issue.",
+      },
+      () => stderr,
+    ).then(() => completed.push("first"));
+    const second = sendRequest(
+      worker,
+      messages,
+      {
+        type: "tts.synthesize",
+        config,
+        text: "The voice feature is slow because it waits for the whole sentence to finish generating before sending any audio to your phone, so I am checking whether we can start playback earlier while the rest is still being synthesized.",
+      },
+      () => stderr,
+    ).then(() => completed.push("second"));
+    await sendRequest(
+      worker,
+      messages,
+      {
+        type: "session.close",
+        sessionId: "responsiveness-probe",
+      },
+      () => stderr,
+    );
+    expect(completed).toEqual([]);
+    await first;
+    expect(completed).toEqual(["first"]);
+    await sendRequest(
+      worker,
+      messages,
+      {
+        type: "session.close",
+        sessionId: "responsiveness-probe",
+      },
+      () => stderr,
+    );
+    expect(completed).toEqual(["first"]);
+    await second;
+    expect(completed).toEqual(["first", "second"]);
+  } finally {
+    if (worker.connected) worker.disconnect();
+    if (!worker.killed) worker.kill();
+  }
+}, 120_000);
+
 type RequestInput = LocalSpeechWorkerRequest extends infer Request
   ? Request extends LocalSpeechWorkerRequest
     ? Omit<Request, "requestId">