Fix voice reply latency and hide internal prompt wrappers (#4927)

* 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. * test(voice): clarify playback and timeline regression journeys * test(voice): centralize native worker cleanup and repair fixture path * test(app): exercise workspace eviction without reloading the app * test(app): preserve streamed output during workspace eviction Hold the turn until the workspace is evicted instead of dropping all stream events during navigation. Slow CI runners can finish the entire response before navigation ends; waiting for producer completion before release makes that ordering deterministic. * test(app): stabilize subscriptions during workspace eviction Opening another chat replaces the timeline subscription, making buffered frames from earlier generations unroutable. Preopen all chats before holding the stream, finish production before eviction, and assert both early and final response content with no subscription replacement.
2026-10-28 04:03:05 +00:00 · 2026-09-16 13:36:44 +02:00
parent 0eac75be7d
commit 64b1a62ed3
10 changed files with 688 additions and 157 deletions
@@ -36,14 +36,17 @@ import { delayBrowserAgentCreatedStatus } from "../support/helpers/new-workspace
 import { installDaemonWebSocketGate } from "../support/helpers/daemon-websocket-gate";
 import { gotoAppShell, openSettings, selectModel } from "../support/helpers/app";
 import { observeTimelineSubscriptions } from "../support/helpers/timeline-delivery";
+import { rememberTimelineRequestCounts } from "../support/helpers/timeline-resume";
 import {
-  expectOneResumeCheckWithoutTail,
-  rememberTimelineRequestCounts,
-} from "../support/helpers/timeline-resume";
-import {
+  switchWorkspaceViaSidebar,
  waitForWorkspaceInSidebar,
  workspaceDeckEntryLocator,
 } from "../support/helpers/workspace-ui";
+import {
+  openSessions,
+  clickSessionRow,
+  expectWorkspaceTabVisible,
+} from "../support/helpers/archive-tab";
 import { expectInFlightForkAvailable } from "../support/helpers/assistant-fork";
 import {
  scrollTimelineToNewestLoadedEdge,
@@ -428,6 +431,18 @@ async function expectInterruptedTurnOrderAfterReconnect(
  }
 }

+async function visitEvictionWorkspaces(
+  page: Page,
+  agents: readonly { agentId: string }[],
+): Promise<void> {
+  for (const [index, agent] of agents.entries()) {
+    await openSessions(page);
+    await clickSessionRow(page, `Workspace eviction ${index + 1}.`);
+    await expectWorkspaceTabVisible(page, agent.agentId);
+    await expectComposerVisible(page);
+  }
+}
+
 async function expectHiddenStreamingSubmissionOrderAfterWorkspaceEviction(
  page: Page,
  testInfo: { workerIndex: number },
@@ -443,54 +458,76 @@ async function expectHiddenStreamingSubmissionOrderAfterWorkspaceEviction(
    Array.from({ length: WORKSPACE_DECK_MAX_MOUNTED_WORKSPACES }, (_unused, index) =>
      seedMockAgentWorkspace({
        repoPrefix: `submission-workspace-eviction-${testInfo.workerIndex}-${index}-`,
-        title: `Workspace eviction ${index + 1}`,
+        title: `Workspace eviction ${index + 1}.`,
      }),
    ),
  );
  const prompt = "Keep this hidden image prompt before its streaming output.";
  const targetDeckEntry = workspaceDeckEntryLocator(page, getServerId(), target.workspaceId);
+  const openAgentIds = [target.agentId, ...evictionAgents.map((agent) => agent.agentId)];

  try {
    await openAgentRoute(page, target);
    await expectComposerVisible(page);
-    await subscriptions.waitForSubscribedAgents([target.agentId]);
+    // Open every chat before holding output. Adding chats replaces the timeline
+    // subscription and retires the IDs stamped on already-buffered frames.
+    await visitEvictionWorkspaces(page, evictionAgents);
+    await subscriptions.waitForSubscribedAgents(openAgentIds);
+    await switchWorkspaceViaSidebar({
+      page,
+      serverId: getServerId(),
+      workspaceId: target.workspaceId,
+    });
+    await expectWorkspaceTabVisible(page, target.agentId);
+    await expectComposerVisible(page);
+    const subscriptionRequests = gate.getClientRequestCount(
+      "agent.timeline.set_subscription.request",
+    );

    const userMessageCount = gate.getAgentStreamItemCount("user_message");
-    gate.setAgentStreamSuppressed(true);
+    gate.setAgentStreamItemSuppressed("user_message", true);
+    gate.holdNextAgentStreamEvent("turn_started");
    const promptRow = await submitMessageWithImage(page, prompt);
    await gate.waitForAgentStreamItem("user_message", userMessageCount + 1);
-
-    for (const evictionAgent of evictionAgents) {
-      await openAgentRoute(page, evictionAgent);
-      await expectComposerVisible(page);
-    }
-    await expect(targetDeckEntry).toHaveCount(0);
-    await subscriptions.waitForSubscribedAgents([
-      target.agentId,
-      ...evictionAgents.map((agent) => agent.agentId),
-    ]);
-    gate.setAgentStreamSuppressed(false);
-
+    await gate.waitForHeldAgentStreamEvent("turn_started");
+    // Finish production before navigation so passing cannot depend on late
+    // chunks arriving after the final workspace switch.
    await target.client.waitForFinish(target.agentId, 30_000);
+
+    // Navigate inside the app: a document reload discards the retained deck and
+    // adds startup history fetches, so it cannot prove eviction/resume behavior.
+    await visitEvictionWorkspaces(page, evictionAgents);
+    await expect(targetDeckEntry).toHaveCount(0);
+    await subscriptions.waitForSubscribedAgents(openAgentIds);
+    expect(gate.getClientRequestCount("agent.timeline.set_subscription.request")).toBe(
+      subscriptionRequests,
+    );
+    gate.releaseHeldAgentStreamEvent("turn_started");
    const requestsBeforeReturn = rememberTimelineRequestCounts(gate, target.agentId);
    await waitForWorkspaceInSidebar(page, {
      serverId: getServerId(),
      workspaceId: target.workspaceId,
    });
-    await openAgentRoute(page, target);
+    await switchWorkspaceViaSidebar({
+      page,
+      serverId: getServerId(),
+      workspaceId: target.workspaceId,
+    });
    await expectComposerVisible(page);
-    await subscriptions.waitForSubscribedAgents([
-      target.agentId,
-      ...evictionAgents.map((agent) => agent.agentId),
-    ]);
+    await subscriptions.waitForSubscribedAgents(openAgentIds);

+    const responseStart = page.getByText("Cycle 1", { exact: true });
    const response = page.getByText("(end of synthetic stream)", { exact: true }).last();
    await expect(promptRow).toBeVisible();
+    await expect(responseStart).toBeVisible();
    await expect(response).toBeVisible();
+    await expectRenderedBefore(promptRow, responseStart);
    await expectRenderedBefore(promptRow, response);
-    expectOneResumeCheckWithoutTail(gate, requestsBeforeReturn);
+    // Open chats stay subscribed when their workspace view is evicted. Returning
+    // uses that live timeline without another resume check or startup tail fetch.
+    expect(rememberTimelineRequestCounts(gate, target.agentId)).toEqual(requestsBeforeReturn);
  } finally {
-    gate.setAgentStreamSuppressed(false);
+    gate.setAgentStreamItemSuppressed("user_message", false);
    gate.restore();
    await Promise.all([...evictionAgents.map((agent) => agent.cleanup()), target.cleanup()]);
  }
@@ -1,5 +1,10 @@
 import { expect, test } from "../support/fixtures";
 import {
+  withSpokenTimeline,
+  captureSpokenTimeline,
+  reloadSpokenTimeline,
+  expectSpokenTimelinePrompt,
+  sendSpokenTimelinePrompt,
  expectStableHistoryStartGutter,
  expectTimelineAtHistoryStart,
  expectTimelinePromptCentered,
@@ -27,6 +32,24 @@ import {
 } from "../support/helpers/timeline-pagination";

 test.describe("Agent timeline pagination", () => {
+  test("shows spoken words without voice instructions in history, live updates and after reload", async ({
+    page,
+  }, testInfo) => {
+    const historyText = "Please check the voice history.";
+    const liveText = "Now check the live voice message.";
+    await withSpokenTimeline(historyText, async (agent) => {
+      await openAgentTimeline(page, agent);
+      await expectSpokenTimelinePrompt(page, historyText);
+      await sendSpokenTimelinePrompt(agent, liveText);
+      await expectSpokenTimelinePrompt(page, liveText);
+      await captureSpokenTimeline(page, testInfo, "desktop");
+      await reloadSpokenTimeline(page);
+      await expectSpokenTimelinePrompt(page, liveText);
+      await captureSpokenTimeline(page, testInfo, "compact");
+      await expectSpokenTimelinePrompt(page, liveText);
+    });
+  });
+
  test("keeps the history-start gutter and visible position stable through the final page", async ({
    page,
  }) => {
@@ -1,4 +1,4 @@
-import { expect, type Page } from "@playwright/test";
+import { expect, type TestInfo, type Page } from "@playwright/test";
 import { buildAgentRoute, seedMockAgentWorkspace, type MockAgentWorkspace } from "./mock-agent";
 import { readReplicaCache } from "./replica-cache-storage";
 import {
@@ -574,3 +574,58 @@ async function waitForTimelineGeometryToSettle(page: Page): Promise<void> {
      }),
  );
 }
+
+function spokenTimelinePrompt(text: string): string {
+  return `<spoken-input>\n${text}\n</spoken-input>\n<instruction>This message was spoken by the user. Respond using the speak tool only, not normal messages, because the user may not be looking at the chat.</instruction>`;
+}
+
+export async function expectSpokenTimelinePrompt(page: Page, text: string): Promise<void> {
+  await expectTimelinePromptVisible(page, text);
+  const row = page.getByTestId("user-message").filter({ hasText: text });
+  await expect(row).not.toContainText("<spoken-input>");
+  await expect(row).not.toContainText("<instruction>");
+  await expect(row).not.toContainText("This message was spoken by the user.");
+}
+
+export async function sendSpokenTimelinePrompt(
+  agent: MockAgentWorkspace,
+  text: string,
+): Promise<void> {
+  await agent.client.sendAgentMessage(agent.agentId, spokenTimelinePrompt(text));
+  await agent.client.waitForFinish(agent.agentId, 15_000);
+}
+
+export async function withSpokenTimeline(
+  historyText: string,
+  run: (agent: MockAgentWorkspace) => Promise<void>,
+): Promise<void> {
+  const agent = await seedMockAgentWorkspace({
+    repoPrefix: "spoken-timeline-",
+    title: "Spoken timeline presentation",
+    initialPrompt: spokenTimelinePrompt(historyText),
+  });
+  try {
+    await agent.client.waitForFinish(agent.agentId, 15_000);
+    await run(agent);
+  } finally {
+    await agent.cleanup();
+  }
+}
+
+export async function reloadSpokenTimeline(page: Page): Promise<void> {
+  await page.reload();
+}
+
+export async function captureSpokenTimeline(
+  page: Page,
+  testInfo: TestInfo,
+  size: "desktop" | "compact",
+): Promise<void> {
+  await page.setViewportSize(
+    size === "compact" ? { width: 390, height: 844 } : { width: 1280, height: 900 },
+  );
+  await testInfo.attach(`spoken-${size}-web`, {
+    body: await page.screenshot({ path: testInfo.outputPath(`spoken-${size}-web.png`) }),
+    contentType: "image/png",
+  });
+}
@@ -8,9 +8,11 @@ import {
  hydrateStreamState,
  type StreamItem,
  type ToolCallItem,
+  type UserMessageItem,
 } from "@/types/stream";
 import { transformTimelineItem, type TimelineItemTransform } from "@/plugins/timeline/model";
 import { createStreamPresentation } from "./presentation";
+import { buildAgentStreamRenderModel } from "./model";

 const runtime = {
  paseo: {},
@@ -349,3 +351,121 @@ describe("stream presentation through installed plugins", () => {
    ]);
  });
 });
+
+function createTimestamp(seed: number): Date {
+  return new Date(`2026-01-01T00:00:${seed.toString().padStart(2, "0")}.000Z`);
+}
+
+function userMessage(id: string, seed: number): UserMessageItem {
+  return {
+    kind: "user_message",
+    id,
+    text: id,
+    timestamp: createTimestamp(seed),
+  };
+}
+
+function assistantMessage(
+  id: string,
+  seed: number,
+): Extract<StreamItem, { kind: "assistant_message" }> {
+  return {
+    kind: "assistant_message",
+    id,
+    text: id,
+    timestamp: createTimestamp(seed),
+  };
+}
+
+describe("timeline presentation", () => {
+  const present = createStreamPresentation();
+  function projectTimelineItems(items: StreamItem[], transform?: TimelineItemTransform) {
+    return present({ ...presentationOptions, tail: items, head: [], transform }).tail;
+  }
+  const envelope =
+    "<spoken-input>\nPlease fix the voice chat.\n</spoken-input>\n<instruction>This message was spoken by the user. Respond using the speak tool only, not normal messages, because the user may not be looking at the chat.</instruction>";
+
+  it.each(["live", "history"])(
+    "shows only spoken words from a %s user message without mutating its source",
+    (source) => {
+      const item: UserMessageItem = Object.freeze({
+        ...userMessage(source, 1),
+        text: envelope,
+        messageId: "provider-message",
+        clientMessageId: "client-message",
+        turnId: "turn-1",
+        timelineCursor: { epoch: "epoch", seq: 12 },
+      });
+      const presentMessage = () =>
+        present({
+          ...presentationOptions,
+          tail: source === "history" ? [item] : [],
+          head: source === "live" ? [item] : [],
+          transform: undefined,
+        });
+      const projected = rows(presentMessage());
+      expect(projected).toEqual([{ ...item, text: "Please fix the voice chat." }]);
+      const model = buildAgentStreamRenderModel({
+        tail: source === "history" ? projected : [],
+        head: source === "live" ? projected : [],
+        isTurnActive: source === "live",
+        activeTurnStartedAt: item.timestamp,
+        platform: "native",
+        isMobileBreakpoint: true,
+      });
+      const rendered = [...model.history, ...model.segments.liveHead];
+      expect(rendered).toContainEqual({ ...item, text: "Please fix the voice chat." });
+      expect(item.text).toBe(envelope);
+      expect(rows(presentMessage())[0]).toBe(projected[0]);
+    },
+  );
+
+  it("supports older envelopes and preserves multiline spoken content", () => {
+    const item = {
+      ...userMessage("legacy", 1),
+      text: "<spoken-input>\nFirst line.\nSecond line with <example>XML</example>.\n</spoken-input>",
+    };
+    expect(projectTimelineItems([item])).toEqual([
+      { ...item, text: "First line.\nSecond line with <example>XML</example>." },
+    ]);
+  });
+
+  it("leaves ordinary messages, assistant examples and incomplete wrappers unchanged", () => {
+    const items: StreamItem[] = [
+      userMessage("ordinary", 1),
+      { ...assistantMessage("example", 2), text: envelope },
+      { ...userMessage("incomplete", 3), text: "<spoken-input>unfinished" },
+      { ...userMessage("quoted", 4), text: "Explain this example: " + envelope },
+      {
+        ...userMessage("xml", 5),
+        text: "<spoken-input>Example</spoken-input><instruction>Explain this XML.</instruction>",
+      },
+    ];
+    expect(projectTimelineItems(items)).toEqual(items);
+    expect(projectTimelineItems(items)[0]).toBe(items[0]);
+  });
+
+  it("keeps plugin transforms on the original source and respects replacements", () => {
+    const source: StreamItem = { ...userMessage("spoken", 1), text: envelope };
+    const inputs: string[] = [];
+    const transformed = projectTimelineItems([source], ({ item, sourceId }) => {
+      if (item.type === "user_message") inputs.push(item.text);
+      return [
+        {
+          type: "plugin",
+          pluginId: "test",
+          id: sourceId,
+          kind: "voice",
+          version: 1,
+          data: { text: "Custom voice row" },
+        },
+      ];
+    });
+    expect(inputs).toEqual([envelope]);
+    expect(transformed).toMatchObject([{ kind: "plugin", data: { text: "Custom voice row" } }]);
+    expect(projectTimelineItems([source], () => undefined)).toEqual([
+      { ...source, text: "Please fix the voice chat." },
+    ]);
+    expect(projectTimelineItems([source], () => [])).toEqual([]);
+  });
+});
@@ -1,4 +1,4 @@
-import type { AssistantMessageItem, StreamItem } from "@/types/stream";
+import type { AssistantMessageItem, StreamItem, UserMessageItem } from "@/types/stream";
 import type { TimelineItemTransform } from "@/plugins/timeline/model";
 import { projectPluginTimelineItems } from "@/plugins/timeline/projection";
 import { splitMarkdownBlocks } from "@/utils/split-markdown-blocks";
@@ -25,6 +25,23 @@ function retainItems(previous: StreamItem[], next: StreamItem[]): StreamItem[] {

 /** Source messages reach plugins before any Markdown splitting or Overview grouping. */
 export function createStreamPresentation() {
+  const userMessageCache = new WeakMap<UserMessageItem, UserMessageItem>();
+
+  function presentUserMessage(item: UserMessageItem): UserMessageItem {
+    const cached = userMessageCache.get(item);
+    if (cached) return cached;
+
+    // Recognize the complete voice envelope, including older messages without the
+    // instruction suffix. Leave quoted examples and unrelated XML untouched.
+    const spokenInput =
+      /^\s*<spoken-input>([\s\S]*?)<\/spoken-input>(?:\s*<instruction>This message was spoken by the user\.[\s\S]*?<\/instruction>)?\s*$/.exec(
+        item.text,
+      );
+    const projected = spokenInput ? { ...item, text: spokenInput[1].trim() } : item;
+    userMessageCache.set(item, projected);
+    return projected;
+  }
+
  const blocksBySource = new WeakMap<AssistantMessageItem, AssistantMessageItem[]>();
  let liveSources = new Map<string, AssistantMessageItem>();
  let historySource: StreamItem[] | undefined;
@@ -38,6 +55,7 @@ export function createStreamPresentation() {
  let preparedHistory: PreparedToolCallHistory | null = null;

  function nativeBlocks(item: StreamItem): StreamItem[] {
+    if (item.kind === "user_message") return [presentUserMessage(item)];
    if (item.kind !== "assistant_message") return [item];
    const cached = blocksBySource.get(item);
    if (cached) return cached;
@@ -95,7 +113,7 @@ export function createStreamPresentation() {
          // Preserve live block identities at completion; fetched native Markdown
          // stays whole so links and other cross-block constructs keep their context.
          if (item.kind === "assistant_message") return blocksBySource.get(item) ?? [item];
-          return [item];
+          return [item.kind === "user_message" ? presentUserMessage(item) : item];
        },
      );
      historySource = input.tail;
@@ -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,33 @@ 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 waitForAudioOutput(host: FakeVoiceHost): Promise<void> {
+  await vi.waitFor(() => expect(host.emitted.filter(isAudioOutput)).toHaveLength(1));
 }

 async function settle(): Promise<void> {
@@ -110,6 +136,71 @@ 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[] = [];
+    async function recordInterruption(agentId: string) {
+      interrupted.push(agentId);
+    }
+    host.interruptAgentIfRunning = recordInterruption;
+    await voiceSession.handleSetVoiceMode(true, VOICE_AGENT_ID);
+    const playback = speak({ text: "A spoken response." });
+    try {
+      await waitForAudioOutput(host);
+      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;
+    function acknowledgeLaterAudio(message: SessionOutboundMessage) {
+      emit(message);
+      if (message.type === "audio_output" && abortRequested)
+        voiceSession.handleAudioPlayed(message.payload.id);
+    }
+    host.emit = acknowledgeLaterAudio;
+    await voiceSession.handleSetVoiceMode(true, VOICE_AGENT_ID);
+    const playback = speak({
+      text: "First sentence. Second sentence. Third sentence.",
+      signal: external.signal,
+    });
+    try {
+      await waitForAudioOutput(host);
+      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 () => {
@@ -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),
@@ -1,53 +1,141 @@
-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),
-}));
+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 };
+}

-vi.mock("./sherpa-onnx-node-loader.js", () => ({
-  loadSherpaOnnxNode: () => ({
-    OfflineTts: class {
-      public readonly sampleRate = 24000;
+async function yieldNativeQueue(): Promise<void> {
+  await new Promise<void>((resolve) => setImmediate(resolve));
+}

-      generate = generate;
-      free = free;
-    },
-  }),
-}));
+function holdNativeSynthesis() {
+  const pending: Array<() => void> = [];
+  function wait() {
+    return new Promise<void>((resolve) => pending.push(resolve));
+  }
+  return { pending, wait };
+}

 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, wait } = holdNativeSynthesis();
+    const { tts, requests } = createTts(undefined, wait);
+    const first = tts.synthesizeSpeech("first");
+    const second = tts.synthesizeSpeech("second");
+    try {
+      await yieldNativeQueue();
+      expect(requests).toEqual([]);
+      expect(pending).toHaveLength(1);
+      pending.shift()!();
+      const audio = await first;
+      audio.stream.destroy();
+      await yieldNativeQueue();
+      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 () => {
+    const { pending, wait } = holdNativeSynthesis();
+    const { tts, requests, released } = createTts(undefined, wait);
+    const first = tts.synthesizeSpeech("first");
+    const second = tts.synthesizeSpeech("second");
+    const rejected = expect(second).rejects.toThrow("TTS provider closed");
+    await yieldNativeQueue();
+    tts.free();
+    expect(released).toEqual([]);
+    pending.shift()!();
+    (await first).stream.destroy();
+    await rejected;
+    await yieldNativeQueue();
+    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");
+    },
+  );
 });
@@ -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;
+    });
  }
 }
@@ -5,7 +5,7 @@ import { randomUUID } from "node:crypto";
 import { homedir } from "node:os";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
-import { expect, test } from "vitest";
+import { expect, onTestFinished, test } from "vitest";

 import { parsePcm16MonoWav, wordSimilarity } from "../../../test-utils/dictation-e2e.js";
 import { ensureSherpaOnnxModels } from "./sherpa/model-downloader.js";
@@ -30,7 +30,7 @@ function hasParakeetModel(dir: string): boolean {
 }

 function fixturePath(fileName: string): string {
-  return path.resolve(process.cwd(), "..", "app", "e2e", "fixtures", fileName);
+  return path.resolve(process.cwd(), "..", "app", "e2e", "support", "fixtures", fileName);
 }

 function resolveWorkerUrl(): URL {
@@ -68,6 +68,10 @@ function forkWorker(): ChildProcess {
    stdio: ["ignore", "ignore", "pipe", "ipc"],
  });
  worker.setMaxListeners(100);
+  onTestFinished(() => {
+    if (worker.connected) worker.disconnect();
+    if (!worker.killed) worker.kill();
+  });
  return worker;
 }

@@ -92,75 +96,133 @@ workerSpeechTest(
      messages.push(message);
    });

-    try {
-      const config: LocalSpeechWorkerConfig = {
-        modelsDir,
-        voiceSttModel: "parakeet-tdt-0.6b-v2-int8",
-        dictationSttModel: "parakeet-tdt-0.6b-v2-int8",
-        voiceTtsModel: "kokoro-en-v0_19",
-      };
-      const sessionId = randomUUID();
+    const config: LocalSpeechWorkerConfig = {
+      modelsDir,
+      voiceSttModel: "parakeet-tdt-0.6b-v2-int8",
+      dictationSttModel: "parakeet-tdt-0.6b-v2-int8",
+      voiceTtsModel: "kokoro-en-v0_19",
+    };
+    const sessionId = randomUUID();

-      await sendRequest(
-        worker,
-        messages,
-        {
-          type: "session.create",
-          config,
-          sessionId,
-          kind: "dictationStt",
-        },
-        () => stderr,
+    await sendRequest(
+      worker,
+      messages,
+      {
+        type: "session.create",
+        config,
+        sessionId,
+        kind: "dictationStt",
+      },
+      () => stderr,
+    );
+
+    const wav = await readFile(fixturePath("recording.wav"));
+    const { sampleRate, pcm16 } = parsePcm16MonoWav(wav);
+    expect(sampleRate).toBe(16000);
+
+    const chunkBytes = 3200;
+    const appendPromises: Promise<unknown>[] = [];
+    for (let offset = 0; offset < pcm16.length; offset += chunkBytes) {
+      const audio = pcm16.subarray(offset, Math.min(pcm16.length, offset + chunkBytes));
+      appendPromises.push(
+        sendRequest(
+          worker,
+          messages,
+          {
+            type: "session.append",
+            sessionId,
+            audio: bufferToWorkerBytes(audio),
+          },
+          () => stderr,
+        ),
      );
-
-      const wav = await readFile(fixturePath("recording.wav"));
-      const { sampleRate, pcm16 } = parsePcm16MonoWav(wav);
-      expect(sampleRate).toBe(16000);
-
-      const chunkBytes = 3200;
-      const appendPromises: Promise<unknown>[] = [];
-      for (let offset = 0; offset < pcm16.length; offset += chunkBytes) {
-        const audio = pcm16.subarray(offset, Math.min(pcm16.length, offset + chunkBytes));
-        appendPromises.push(
-          sendRequest(
-            worker,
-            messages,
-            {
-              type: "session.append",
-              sessionId,
-              audio: bufferToWorkerBytes(audio),
-            },
-            () => stderr,
-          ),
-        );
-      }
-      await Promise.all(appendPromises);
-
-      await sendRequest(
-        worker,
-        messages,
-        {
-          type: "session.commit",
-          sessionId,
-        },
-        () => stderr,
-      );
-
-      const finalTranscript = await waitForFinalTranscript(worker, messages, sessionId, stderr);
-      const baseline = await readFile(fixturePath("recording.baseline.txt"), "utf8");
-      expect(wordSimilarity(finalTranscript, baseline)).toBeGreaterThan(0.45);
-    } finally {
-      if (worker.connected) {
-        worker.disconnect();
-      }
-      if (!worker.killed) {
-        worker.kill();
-      }
    }
+    await Promise.all(appendPromises);
+
+    await sendRequest(
+      worker,
+      messages,
+      {
+        type: "session.commit",
+        sessionId,
+      },
+      () => stderr,
+    );
+
+    const finalTranscript = await waitForFinalTranscript(worker, messages, sessionId, stderr);
+    const baseline = await readFile(fixturePath("recording.baseline.txt"), "utf8");
+    expect(wordSimilarity(finalTranscript, baseline)).toBeGreaterThan(0.45);
  },
  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",
+  };
+  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"]);
+}, 120_000);
+
 type RequestInput = LocalSpeechWorkerRequest extends infer Request
  ? Request extends LocalSpeechWorkerRequest
    ? Omit<Request, "requestId">