Fix local speech worker audio transport

Worker IPC now carries plain byte payloads instead of Node Buffer objects, so provider code receives process-local Buffers instead of transport-shaped views.

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

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

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;

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

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<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();
+      }
+    }
+  },
+  120_000,
+);
+
+type RequestInput = LocalSpeechWorkerRequest extends infer Request
+  ? Request extends LocalSpeechWorkerRequest
+    ? Omit<Request, "requestId">
+    : never
+  : never;
+
+async function sendRequest(
+  worker: ChildProcess,
+  messages: LocalSpeechWorkerToParentMessage[],
+  input: RequestInput,
+  getStderr: () => string,
+): Promise<unknown> {
+  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<unknown> {
+  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<string> {
+  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);
+  });
+}

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<void> {
     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);

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