diff --git a/packages/app/e2e/browser/agent-message-submission.spec.ts b/packages/app/e2e/browser/agent-message-submission.spec.ts index 06a08c070c..ee97d87adb 100644 --- a/packages/app/e2e/browser/agent-message-submission.spec.ts +++ b/packages/app/e2e/browser/agent-message-submission.spec.ts @@ -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 { + 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()]); } diff --git a/packages/app/e2e/browser/agent-timeline-pagination.spec.ts b/packages/app/e2e/browser/agent-timeline-pagination.spec.ts index 07778d21a7..f2772b0b1b 100644 --- a/packages/app/e2e/browser/agent-timeline-pagination.spec.ts +++ b/packages/app/e2e/browser/agent-timeline-pagination.spec.ts @@ -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, }) => { diff --git a/packages/app/e2e/support/helpers/timeline-pagination.ts b/packages/app/e2e/support/helpers/timeline-pagination.ts index 7cc19461b2..7dd3dca8d7 100644 --- a/packages/app/e2e/support/helpers/timeline-pagination.ts +++ b/packages/app/e2e/support/helpers/timeline-pagination.ts @@ -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 { }), ); } + +function spokenTimelinePrompt(text: string): string { + return `\n${text}\n\nThis 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.`; +} + +export async function expectSpokenTimelinePrompt(page: Page, text: string): Promise { + await expectTimelinePromptVisible(page, text); + const row = page.getByTestId("user-message").filter({ hasText: text }); + await expect(row).not.toContainText(""); + await expect(row).not.toContainText(""); + await expect(row).not.toContainText("This message was spoken by the user."); +} + +export async function sendSpokenTimelinePrompt( + agent: MockAgentWorkspace, + text: string, +): Promise { + 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, +): Promise { + 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 { + await page.reload(); +} + +export async function captureSpokenTimeline( + page: Page, + testInfo: TestInfo, + size: "desktop" | "compact", +): Promise { + 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", + }); +} diff --git a/packages/app/src/agent-stream/presentation.test.ts b/packages/app/src/agent-stream/presentation.test.ts index 05df7180bf..c5b33b4c40 100644 --- a/packages/app/src/agent-stream/presentation.test.ts +++ b/packages/app/src/agent-stream/presentation.test.ts @@ -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 { + 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 = + "\nPlease fix the voice chat.\n\nThis 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."; + + 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: "\nFirst line.\nSecond line with XML.\n", + }; + expect(projectTimelineItems([item])).toEqual([ + { ...item, text: "First line.\nSecond line with XML." }, + ]); + }); + + 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: "unfinished" }, + { ...userMessage("quoted", 4), text: "Explain this example: " + envelope }, + { + ...userMessage("xml", 5), + text: "ExampleExplain this XML.", + }, + ]; + 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([]); + }); +}); diff --git a/packages/app/src/agent-stream/presentation.ts b/packages/app/src/agent-stream/presentation.ts index 9d05bdea5b..19808bdd43 100644 --- a/packages/app/src/agent-stream/presentation.ts +++ b/packages/app/src/agent-stream/presentation.ts @@ -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(); + + 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*([\s\S]*?)<\/spoken-input>(?:\s*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(); let liveSources = new Map(); 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; diff --git a/packages/server/src/server/session/voice/voice-session.test.ts b/packages/server/src/server/session/voice/voice-session.test.ts index d6232ff4d8..1511d6f2fd 100644 --- a/packages/server/src/server/session/voice/voice-session.test.ts +++ b/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,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[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 { + await vi.waitFor(() => expect(host.emitted.filter(isAudioOutput)).toHaveLength(1)); } async function settle(): Promise { @@ -110,6 +136,71 @@ async function settle(): Promise { } 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 () => { diff --git a/packages/server/src/server/session/voice/voice-session.ts b/packages/server/src/server/session/voice/voice-session.ts index b374e759ae..95537dd021 100644 --- a/packages/server/src/server/session/voice/voice-session.ts +++ b/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), diff --git a/packages/server/src/server/speech/providers/local/sherpa/sherpa-tts.test.ts b/packages/server/src/server/speech/providers/local/sherpa/sherpa-tts.test.ts index a7f60d2f29..06d4a5b41d 100644 --- a/packages/server/src/server/speech/providers/local/sherpa/sherpa-tts.test.ts +++ b/packages/server/src/server/speech/providers/local/sherpa/sherpa-tts.test.ts @@ -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) { + 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 { + await new Promise((resolve) => setImmediate(resolve)); +} - generate = generate; - free = free; - }, - }), -})); +function holdNativeSynthesis() { + const pending: Array<() => void> = []; + function wait() { + return new Promise((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"); + }, + ); }); diff --git a/packages/server/src/server/speech/providers/local/sherpa/sherpa-tts.ts b/packages/server/src/server/speech/providers/local/sherpa/sherpa-tts.ts index e212ba6afa..a078f81f51 100644 --- a/packages/server/src/server/speech/providers/local/sherpa/sherpa-tts.ts +++ b/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 = Promise.resolve(); + private closed = false; - constructor(config: SherpaTtsConfig, logger: pino.Logger) { + constructor( + config: SherpaTtsConfig, + logger: pino.Logger, + loadNative: () => Pick< + ReturnType, + "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 { + 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; + }); } } diff --git a/packages/server/src/server/speech/providers/local/worker-process.local.e2e.test.ts b/packages/server/src/server/speech/providers/local/worker-process.local.e2e.test.ts index 9634a7533e..278b8112da 100644 --- a/packages/server/src/server/speech/providers/local/worker-process.local.e2e.test.ts +++ b/packages/server/src/server/speech/providers/local/worker-process.local.e2e.test.ts @@ -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[] = []; + 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[] = []; - 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