diff --git a/packages/server/src/server/speech/providers/local/worker-bytes.test.ts b/packages/server/src/server/speech/providers/local/worker-bytes.test.ts new file mode 100644 index 000000000..2efaa943c --- /dev/null +++ b/packages/server/src/server/speech/providers/local/worker-bytes.test.ts @@ -0,0 +1,15 @@ +import { describe, expect, it } from "vitest"; + +import { bufferToWorkerBytes, workerBytesToBuffer } from "./worker-bytes.js"; + +describe("local speech worker bytes", () => { + it("turns Buffer views into process-local byte payloads", () => { + const source = Buffer.from([0, 1, 2, 3, 4]).subarray(1, 5); + expect(source.byteOffset % 2).toBe(1); + + const received = workerBytesToBuffer(bufferToWorkerBytes(source)); + + expect(received).toEqual(Buffer.from([1, 2, 3, 4])); + expect(received.byteOffset).toBe(0); + }); +}); diff --git a/packages/server/src/server/speech/providers/local/worker-bytes.ts b/packages/server/src/server/speech/providers/local/worker-bytes.ts new file mode 100644 index 000000000..7b190d9c4 --- /dev/null +++ b/packages/server/src/server/speech/providers/local/worker-bytes.ts @@ -0,0 +1,11 @@ +export type LocalSpeechWorkerBytes = ArrayBuffer; + +export function bufferToWorkerBytes(buffer: Buffer): LocalSpeechWorkerBytes { + const bytes = new Uint8Array(buffer.byteLength); + bytes.set(buffer); + return bytes.buffer; +} + +export function workerBytesToBuffer(bytes: LocalSpeechWorkerBytes): Buffer { + return Buffer.from(bytes); +} diff --git a/packages/server/src/server/speech/providers/local/worker-client.test.ts b/packages/server/src/server/speech/providers/local/worker-client.test.ts index edc2cdedd..27b37bd15 100644 --- a/packages/server/src/server/speech/providers/local/worker-client.test.ts +++ b/packages/server/src/server/speech/providers/local/worker-client.test.ts @@ -13,6 +13,7 @@ import type { LocalSpeechWorkerRequest, LocalSpeechWorkerToParentMessage, } from "./worker-protocol.js"; +import { bufferToWorkerBytes, workerBytesToBuffer } from "./worker-bytes.js"; class FakeLocalSpeechWorker extends EventEmitter { public connected = true; @@ -101,7 +102,7 @@ describe("LocalSpeechWorkerClient", () => { }); workers[0].respond(request, { - audio: Buffer.from([1, 2, 3, 4]), + audio: bufferToWorkerBytes(Buffer.from([1, 2, 3, 4])), format: "pcm;rate=24000", }); @@ -130,11 +131,16 @@ describe("LocalSpeechWorkerClient", () => { session.appendPcm16(Buffer.from([9, 8, 7, 6])); await waitForMicrotasks(); - expect(workers[0].sent[1]).toMatchObject({ + const appendRequest = workers[0].sent[1]; + expect(appendRequest).toMatchObject({ type: "session.append", sessionId: createRequest.sessionId, - audio: Buffer.from([9, 8, 7, 6]), }); + if (appendRequest.type !== "session.append") { + throw new Error("Expected session.append request"); + } + expect(Buffer.from(appendRequest.audio)).toEqual(Buffer.from([9, 8, 7, 6])); + expect(workerBytesToBuffer(appendRequest.audio).byteOffset).toBe(0); session.commit(); await waitForMicrotasks(); @@ -193,7 +199,7 @@ describe("LocalSpeechWorkerClient", () => { const first = client.synthesizeSpeech("first"); workers[0].respond(workers[0].sent[0], { - audio: Buffer.from([1]), + audio: bufferToWorkerBytes(Buffer.from([1])), format: "pcm;rate=24000", }); await first; @@ -204,7 +210,7 @@ describe("LocalSpeechWorkerClient", () => { const second = client.synthesizeSpeech("second"); expect(workers).toHaveLength(2); workers[1].respond(workers[1].sent[0], { - audio: Buffer.from([2]), + audio: bufferToWorkerBytes(Buffer.from([2])), format: "pcm;rate=24000", }); await second; diff --git a/packages/server/src/server/speech/providers/local/worker-client.ts b/packages/server/src/server/speech/providers/local/worker-client.ts index 786c82727..c6a6f95bb 100644 --- a/packages/server/src/server/speech/providers/local/worker-client.ts +++ b/packages/server/src/server/speech/providers/local/worker-client.ts @@ -23,6 +23,7 @@ import type { LocalSpeechWorkerResponse, LocalSpeechWorkerToParentMessage, } from "./worker-protocol.js"; +import { bufferToWorkerBytes, workerBytesToBuffer } from "./worker-bytes.js"; const DEFAULT_REQUEST_TIMEOUT_MS = 30000; const DEFAULT_IDLE_TTL_MS = 5 * 60 * 1000; @@ -125,7 +126,7 @@ export class LocalSpeechWorkerClient { text, }); return { - stream: Readable.from([result.audio]), + stream: Readable.from([workerBytesToBuffer(result.audio)]), format: result.format, }; } @@ -135,7 +136,7 @@ export class LocalSpeechWorkerClient { type: "stt.transcribe", config: this.config, model: "voice", - audio, + audio: bufferToWorkerBytes(audio), format, }); } @@ -164,7 +165,11 @@ export class LocalSpeechWorkerClient { } appendSessionAudio(sessionId: string, audio: Buffer): void { - void this.sendRequest({ type: "session.append", sessionId, audio }).catch((err) => { + void this.sendRequest({ + type: "session.append", + sessionId, + audio: bufferToWorkerBytes(audio), + }).catch((err) => { this.emitSessionError(sessionId, err); }); } 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 new file mode 100644 index 000000000..9634a7533 --- /dev/null +++ b/packages/server/src/server/speech/providers/local/worker-process.local.e2e.test.ts @@ -0,0 +1,291 @@ +import { fork, type ChildProcess } from "node:child_process"; +import { existsSync } from "node:fs"; +import { readFile } from "node:fs/promises"; +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 { parsePcm16MonoWav, wordSimilarity } from "../../../test-utils/dictation-e2e.js"; +import { ensureSherpaOnnxModels } from "./sherpa/model-downloader.js"; +import { applySherpaLoaderEnv } from "./sherpa/sherpa-runtime-env.js"; +import type { + LocalSpeechWorkerConfig, + LocalSpeechWorkerRequest, + LocalSpeechWorkerToParentMessage, +} from "./worker-protocol.js"; +import { bufferToWorkerBytes } from "./worker-bytes.js"; + +const modelsDir = + process.env.PASEO_LOCAL_MODELS_DIR ?? path.join(homedir(), ".paseo", "models", "local-speech"); +const shouldDownload = process.env.PASEO_SPEECH_E2E_DOWNLOAD === "1"; +const workerSpeechTest = hasParakeetModel(modelsDir) || shouldDownload ? test : test.skip; + +function hasParakeetModel(dir: string): boolean { + return ( + existsSync(path.join(dir, "sherpa-onnx-nemo-parakeet-tdt-0.6b-v2-int8", "encoder.int8.onnx")) && + existsSync(path.join(dir, "sherpa-onnx-nemo-parakeet-tdt-0.6b-v2-int8", "tokens.txt")) + ); +} + +function fixturePath(fileName: string): string { + return path.resolve(process.cwd(), "..", "app", "e2e", "fixtures", fileName); +} + +function resolveWorkerUrl(): URL { + const currentUrl = import.meta.url; + if (currentUrl.endsWith(".ts")) { + return new URL("./worker-process.ts", currentUrl); + } + return new URL("./worker-process.js", currentUrl); +} + +function resolveWorkerExecArgv(): string[] { + if (!import.meta.url.endsWith(".ts")) { + return []; + } + const loaderUrl = new URL("../../../../terminal/terminal-ts-loader.mjs", import.meta.url).href; + const importSource = [ + 'import { register } from "node:module";', + 'import { pathToFileURL } from "node:url";', + `register(${JSON.stringify(loaderUrl)}, pathToFileURL("./"));`, + ].join(" "); + return [ + "--experimental-strip-types", + "--import", + `data:text/javascript,${encodeURIComponent(importSource)}`, + ]; +} + +function forkWorker(): ChildProcess { + const env = { ...process.env, PASEO_LOG_LEVEL: "silent" }; + applySherpaLoaderEnv(env); + const worker = fork(fileURLToPath(resolveWorkerUrl()), [], { + env, + execArgv: resolveWorkerExecArgv(), + serialization: "advanced", + stdio: ["ignore", "ignore", "pipe", "ipc"], + }); + worker.setMaxListeners(100); + return worker; +} + +workerSpeechTest( + "transcribes PCM through the real local speech worker process", + async () => { + if (!hasParakeetModel(modelsDir)) { + await ensureSherpaOnnxModels({ + modelsDir, + modelIds: ["parakeet-tdt-0.6b-v2-int8"], + }); + } + + const worker = forkWorker(); + let stderr = ""; + worker.stderr?.on("data", (chunk) => { + stderr += Buffer.isBuffer(chunk) ? chunk.toString("utf8") : String(chunk); + }); + + const messages: LocalSpeechWorkerToParentMessage[] = []; + worker.on("message", (message: LocalSpeechWorkerToParentMessage) => { + 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(); + + 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, + ), + ); + } + 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(); + } + } + }, + 120_000, +); + +type RequestInput = LocalSpeechWorkerRequest extends infer Request + ? Request extends LocalSpeechWorkerRequest + ? Omit + : never + : never; + +async function sendRequest( + worker: ChildProcess, + messages: LocalSpeechWorkerToParentMessage[], + input: RequestInput, + getStderr: () => string, +): Promise { + const requestId = randomUUID(); + const message = { ...input, requestId } as LocalSpeechWorkerRequest; + worker.send(message); + return waitForResponse(worker, messages, requestId, getStderr); +} + +function waitForResponse( + worker: ChildProcess, + messages: LocalSpeechWorkerToParentMessage[], + requestId: string, + getStderr: () => string, +): Promise { + const existing = messages.find( + (message) => message.type === "response" && message.requestId === requestId, + ); + if (existing?.type === "response") { + if (existing.ok) { + return Promise.resolve(existing.result); + } + return Promise.reject(new Error(existing.error)); + } + + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + cleanup(); + reject(new Error(`Timed out waiting for local speech worker response: ${requestId}`)); + }, 30_000); + + const onMessage = (message: LocalSpeechWorkerToParentMessage) => { + if (message.type !== "response" || message.requestId !== requestId) { + return; + } + cleanup(); + if (message.ok) { + resolve(message.result); + } else { + reject(new Error(message.error)); + } + }; + const onExit = (code: number | null, signal: NodeJS.Signals | null) => { + cleanup(); + reject( + new Error( + `Local speech worker exited before response: code=${code} signal=${signal}\n${getStderr()}`, + ), + ); + }; + + function cleanup() { + clearTimeout(timeout); + worker.off("message", onMessage); + worker.off("exit", onExit); + } + + worker.on("message", onMessage); + worker.once("exit", onExit); + }); +} + +function waitForFinalTranscript( + worker: ChildProcess, + messages: LocalSpeechWorkerToParentMessage[], + sessionId: string, + stderr: string, +): Promise { + const existing = messages.find( + (message) => + message.type === "session.transcript" && + message.sessionId === sessionId && + message.payload.isFinal, + ); + if (existing?.type === "session.transcript") { + return Promise.resolve(existing.payload.transcript); + } + + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + cleanup(); + reject(new Error(`Timed out waiting for final transcript. Worker stderr:\n${stderr}`)); + }, 60_000); + + const onMessage = (message: LocalSpeechWorkerToParentMessage) => { + if (message.type === "session.error" && message.sessionId === sessionId) { + cleanup(); + reject(new Error(message.error)); + return; + } + if ( + message.type !== "session.transcript" || + message.sessionId !== sessionId || + !message.payload.isFinal + ) { + return; + } + cleanup(); + resolve(message.payload.transcript); + }; + const onExit = (code: number | null, signal: NodeJS.Signals | null) => { + cleanup(); + reject( + new Error( + `Local speech worker exited before final transcript: code=${code} signal=${signal}\n${stderr}`, + ), + ); + }; + + function cleanup() { + clearTimeout(timeout); + worker.off("message", onMessage); + worker.off("exit", onExit); + } + + worker.on("message", onMessage); + worker.once("exit", onExit); + }); +} diff --git a/packages/server/src/server/speech/providers/local/worker-process.ts b/packages/server/src/server/speech/providers/local/worker-process.ts index 6cb17ad4d..957ee6aaa 100644 --- a/packages/server/src/server/speech/providers/local/worker-process.ts +++ b/packages/server/src/server/speech/providers/local/worker-process.ts @@ -16,6 +16,7 @@ import type { LocalSpeechWorkerRequest, LocalSpeechWorkerToParentMessage, } from "./worker-protocol.js"; +import { bufferToWorkerBytes, workerBytesToBuffer } from "./worker-bytes.js"; process.title = "Paseo Voice"; @@ -251,7 +252,7 @@ function handleSessionRequest(message: LocalSpeechWorkerSessionRequest): void { const session = sessions.get(message.sessionId); switch (message.type) { case "session.append": - session?.appendPcm16(message.audio); + session?.appendPcm16(workerBytesToBuffer(message.audio)); break; case "session.commit": if (session && "commit" in session) { @@ -284,13 +285,16 @@ async function handleRequest(message: LocalSpeechWorkerRequest): Promise { for await (const chunk of result.stream) { chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); } - sendOk(message.requestId, { audio: Buffer.concat(chunks), format: result.format }); + sendOk(message.requestId, { + audio: bufferToWorkerBytes(Buffer.concat(chunks)), + format: result.format, + }); return; } if (message.type === "stt.transcribe") { const result = await getSttProvider(message.config, message.model).transcribeAudio( - message.audio, + workerBytesToBuffer(message.audio), message.format, ); sendOk(message.requestId, result); diff --git a/packages/server/src/server/speech/providers/local/worker-protocol.ts b/packages/server/src/server/speech/providers/local/worker-protocol.ts index 7c9a85075..9e58c1033 100644 --- a/packages/server/src/server/speech/providers/local/worker-protocol.ts +++ b/packages/server/src/server/speech/providers/local/worker-protocol.ts @@ -3,6 +3,7 @@ import type { StreamingTranscriptionEvent, TranscriptionResult, } from "../../speech-provider.js"; +import type { LocalSpeechWorkerBytes } from "./worker-bytes.js"; export interface LocalSpeechWorkerConfig { modelsDir: string; @@ -27,7 +28,7 @@ export type LocalSpeechWorkerRequest = requestId: string; config: LocalSpeechWorkerConfig; model: "voice" | "dictation"; - audio: Buffer; + audio: LocalSpeechWorkerBytes; format: string; } | { @@ -41,7 +42,7 @@ export type LocalSpeechWorkerRequest = type: "session.append"; requestId: string; sessionId: string; - audio: Buffer; + audio: LocalSpeechWorkerBytes; } | { type: "session.commit"; @@ -115,7 +116,7 @@ export interface LocalSpeechCreateSessionResult { } export interface LocalSpeechTtsResult { - audio: Buffer; + audio: LocalSpeechWorkerBytes; format: string; }