From 04fb0f9b82e1f0015515b1fecf397f94a11b2617 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 29 May 2026 13:27:28 +0800 Subject: [PATCH] Keep local voice memory out of the daemon (#1217) * Move local speech work out of daemon Run local speech models in a lazily spawned worker so native model memory and blocking synthesis/transcription work stay out of the daemon process. The worker exits after idle time so local speech RSS can be reclaimed. * Update supervisor stop test for process title --- .../tests/22-daemon-stop-supervisor.test.ts | 20 +- .../server/scripts/supervisor-entrypoint.ts | 2 + packages/server/src/server/daemon-worker.ts | 2 + .../server/speech/providers/local/runtime.ts | 247 ++------ .../providers/local/worker-client.test.ts | 212 +++++++ .../speech/providers/local/worker-client.ts | 538 ++++++++++++++++++ .../speech/providers/local/worker-process.ts | 335 +++++++++++ .../speech/providers/local/worker-protocol.ts | 122 ++++ 8 files changed, 1279 insertions(+), 199 deletions(-) create mode 100644 packages/server/src/server/speech/providers/local/worker-client.test.ts create mode 100644 packages/server/src/server/speech/providers/local/worker-client.ts create mode 100644 packages/server/src/server/speech/providers/local/worker-process.ts create mode 100644 packages/server/src/server/speech/providers/local/worker-protocol.ts diff --git a/packages/cli/tests/22-daemon-stop-supervisor.test.ts b/packages/cli/tests/22-daemon-stop-supervisor.test.ts index 71c129c77..1dc2abb91 100644 --- a/packages/cli/tests/22-daemon-stop-supervisor.test.ts +++ b/packages/cli/tests/22-daemon-stop-supervisor.test.ts @@ -6,7 +6,7 @@ */ import assert from "node:assert"; -import { spawn, spawnSync, type ChildProcess } from "node:child_process"; +import { spawn, type ChildProcess } from "node:child_process"; import { mkdtemp, readFile, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -59,15 +59,6 @@ async function readPidLockState(paseoHome: string): Promise { } } -function readProcessCommand(pid: number): string | null { - const result = spawnSync("ps", ["-p", String(pid), "-o", "command="], { encoding: "utf8" }); - if (result.status !== 0 || result.error) { - return null; - } - const command = result.stdout.trim(); - return command.length > 0 ? command : null; -} - interface DaemonStatus { localDaemon: string | null; pid: number | null; @@ -168,11 +159,10 @@ try { assert(isProcessRunning(daemonPid), "daemon process should be running"); const pidLockBeforeStop = await readPidLockState(paseoHome); assert.strictEqual(pidLockBeforeStop.pid, daemonPid, "pid lock should match status pid"); - const command = readProcessCommand(daemonPid); - assert(command !== null, "pid lock pid should resolve to a running process command"); - assert( - command.includes("supervisor-entrypoint.ts") || command.includes("supervisor-entrypoint.js"), - `pid lock pid should be supervisor-entrypoint process, got: ${command}`, + assert.strictEqual( + daemonPid, + supervisorProcess.pid, + "pid lock pid should be the supervisor-entrypoint process", ); console.log(`✓ dev daemon started with daemon pid ${daemonPid}\n`); diff --git a/packages/server/scripts/supervisor-entrypoint.ts b/packages/server/scripts/supervisor-entrypoint.ts index bf8b3c67d..d7d23d41b 100644 --- a/packages/server/scripts/supervisor-entrypoint.ts +++ b/packages/server/scripts/supervisor-entrypoint.ts @@ -13,6 +13,8 @@ import { runSupervisor } from "./supervisor.js"; import { resolveSupervisorLogFile } from "./supervisor-log-config.js"; import { applySherpaLoaderEnv } from "../src/server/speech/providers/local/sherpa/sherpa-runtime-env.js"; +process.title = "Paseo Supervisor"; + interface DaemonRunnerConfig { devMode: boolean; workerArgs: string[]; diff --git a/packages/server/src/server/daemon-worker.ts b/packages/server/src/server/daemon-worker.ts index eca0b368d..2a74cd46e 100644 --- a/packages/server/src/server/daemon-worker.ts +++ b/packages/server/src/server/daemon-worker.ts @@ -4,6 +4,8 @@ import { resolvePaseoHome } from "./paseo-home.js"; import { createRootLogger } from "./logger.js"; import type { DaemonLifecycleIntent } from "./bootstrap.js"; +process.title = "Paseo Daemon"; + type SupervisorLifecycleMessage = | { type: "paseo:shutdown"; diff --git a/packages/server/src/server/speech/providers/local/runtime.ts b/packages/server/src/server/speech/providers/local/runtime.ts index 23c6d3bc1..0653a75e5 100644 --- a/packages/server/src/server/speech/providers/local/runtime.ts +++ b/packages/server/src/server/speech/providers/local/runtime.ts @@ -5,7 +5,6 @@ import type { SpeechToTextProvider, TextToSpeechProvider } from "../../speech-pr import type { RequestedSpeechProviders } from "../../speech-types.js"; import type { TurnDetectionProvider } from "../../turn-detection-provider.js"; import { - getLocalSpeechModelDir, DEFAULT_LOCAL_STT_MODEL, DEFAULT_LOCAL_TTS_MODEL, LocalSttModelIdSchema, @@ -14,16 +13,12 @@ import { type LocalSttModelId, type LocalTtsModelId, } from "./models.js"; -import { SherpaOfflineRecognizerEngine } from "./sherpa/sherpa-offline-recognizer.js"; -import { SherpaOnnxParakeetSTT } from "./sherpa/sherpa-parakeet-stt.js"; -import { SherpaParakeetRealtimeTranscriptionSession } from "./sherpa/sherpa-parakeet-realtime-session.js"; -import { SherpaOnnxTTS } from "./sherpa/sherpa-tts.js"; import { - ensureSileroVadModel, - SherpaSileroTurnDetectionProvider, -} from "./sherpa/silero-vad-provider.js"; - -type LocalSttEngine = SherpaOfflineRecognizerEngine; + LocalSpeechWorkerClient, + WorkerBackedSpeechToTextProvider, + WorkerBackedTextToSpeechProvider, + WorkerBackedTurnDetectionProvider, +} from "./worker-client.js"; interface ResolvedLocalModels { dictationLocalSttModel: LocalSttModelId; @@ -50,10 +45,6 @@ export interface InitializedLocalSpeech { cleanup: () => void; } -function buildModelDownloadHint(modelId: LocalSpeechModelId): string { - return `Use 'paseo speech download --model ${modelId}' to download this model.`; -} - function resolveConfiguredLocalModels(speechConfig: PaseoSpeechConfig | null): ResolvedLocalModels { return { dictationLocalSttModel: LocalSttModelIdSchema.parse( @@ -104,131 +95,43 @@ function computeRequiredLocalModelIds(params: { return Array.from(ids); } -async function createLocalSttEngine(params: { - modelId: LocalSttModelId; - modelsDir: string; - logger: Logger; -}): Promise { - const { modelId, modelsDir, logger } = params; - - const modelDir = getLocalSpeechModelDir(modelsDir, modelId); - return new SherpaOfflineRecognizerEngine( - { - model: { - kind: "nemo_transducer", - encoder: `${modelDir}/encoder.int8.onnx`, - decoder: `${modelDir}/decoder.int8.onnx`, - joiner: `${modelDir}/joiner.int8.onnx`, - tokens: `${modelDir}/tokens.txt`, - }, - numThreads: 2, - debug: 0, - }, - logger, - ); -} - -type LocalConfig = NonNullable; - function isLocalProviderEnabled(provider: { enabled?: boolean; provider: string }): boolean { return provider.enabled !== false && provider.provider === "local"; } -async function initializeLocalTurnDetection( - localConfig: LocalConfig | null, - logger: Logger, -): Promise { - let vadModelPath: string | undefined; - if (localConfig) { - try { - vadModelPath = await ensureSileroVadModel(localConfig.modelsDir, logger); - } catch (err) { - logger.warn({ err }, "Failed to provision Silero VAD model, falling back to bundled"); - } - } - return new SherpaSileroTurnDetectionProvider({ modelPath: vadModelPath }, logger); +function warnLocalConfigMissing(logger: Logger, feature: string): void { + logger.warn( + { configured: false }, + `Local ${feature} selected but local provider config is missing; ${feature} will be unavailable`, + ); } -async function initializeLocalVoiceStt(params: { - localConfig: LocalConfig | null; - modelId: LocalSttModelId; - logger: Logger; - getLocalSttEngine: (modelId: LocalSttModelId) => Promise; -}): Promise { - const { localConfig, modelId, logger, getLocalSttEngine } = params; - if (!localConfig) { - logger.warn( - { configured: false }, - "Local STT selected for voice but local provider config is missing; STT will be unavailable", - ); - return null; - } - const voiceEngine = await getLocalSttEngine(modelId); - return voiceEngine ? new SherpaOnnxParakeetSTT({ engine: voiceEngine }, logger) : null; +function initializeLocalTurnDetection(params: { + client: LocalSpeechWorkerClient; +}): TurnDetectionProvider { + const { client } = params; + return new WorkerBackedTurnDetectionProvider(client); } -async function initializeLocalDictationStt(params: { - localConfig: LocalConfig | null; - modelId: LocalSttModelId; - logger: Logger; - getLocalSttEngine: (modelId: LocalSttModelId) => Promise; -}): Promise { - const { localConfig, modelId, logger, getLocalSttEngine } = params; - if (!localConfig) { - logger.warn( - { configured: false }, - "Local STT selected for dictation but local provider config is missing; dictation STT will be unavailable", - ); - return null; - } - const dictationEngine = await getLocalSttEngine(modelId); - if (dictationEngine) { - return { - id: "local", - createSession: () => - new SherpaParakeetRealtimeTranscriptionSession({ engine: dictationEngine }), - }; - } - return null; +function initializeLocalVoiceStt(params: { + client: LocalSpeechWorkerClient; +}): SpeechToTextProvider { + const { client } = params; + return new WorkerBackedSpeechToTextProvider(client, "voiceStt"); } -async function initializeLocalVoiceTts(params: { - localConfig: LocalConfig | null; - speechConfig: PaseoSpeechConfig | null; - localModels: ResolvedLocalModels; - logger: Logger; -}): Promise { - const { localConfig, speechConfig, localModels, logger } = params; - if (!localConfig) { - logger.warn( - { configured: false }, - "Local TTS selected for voice but local provider config is missing; TTS will be unavailable", - ); - return null; - } - try { - const modelDir = getLocalSpeechModelDir(localConfig.modelsDir, localModels.voiceLocalTtsModel); - return new SherpaOnnxTTS( - { - preset: localModels.voiceLocalTtsModel, - modelDir, - speakerId: speechConfig?.local?.models.voiceTtsSpeakerId, - speed: speechConfig?.local?.models.voiceTtsSpeed, - }, - logger, - ); - } catch (err) { - logger.warn( - { - err, - modelsDir: localConfig.modelsDir, - modelId: localModels.voiceLocalTtsModel, - hint: buildModelDownloadHint(localModels.voiceLocalTtsModel), - }, - "Local TTS engine unavailable", - ); - return null; - } +function initializeLocalDictationStt(params: { + client: LocalSpeechWorkerClient; +}): SpeechToTextProvider { + const { client } = params; + return new WorkerBackedSpeechToTextProvider(client, "dictationStt"); +} + +function initializeLocalVoiceTts(params: { + client: LocalSpeechWorkerClient; +}): TextToSpeechProvider { + const { client } = params; + return new WorkerBackedTextToSpeechProvider(client); } export async function initializeLocalSpeechServices(params: { @@ -251,80 +154,56 @@ export async function initializeLocalSpeechServices(params: { models: localModels, }); - const localSttEngines = new Map(); - - const getLocalSttEngine = async (modelId: LocalSttModelId): Promise => { - const existing = localSttEngines.get(modelId); - if (existing) { - return existing; - } - if (!localConfig) { - return null; - } - try { - const created = await createLocalSttEngine({ - modelId, - modelsDir: localConfig.modelsDir, - logger, - }); - localSttEngines.set(modelId, created); - return created; - } catch (err) { - logger.warn( - { - err, + const workerClient = localConfig + ? new LocalSpeechWorkerClient({ + config: { modelsDir: localConfig.modelsDir, - modelId, - hint: buildModelDownloadHint(modelId), + voiceSttModel: localModels.voiceLocalSttModel, + dictationSttModel: localModels.dictationLocalSttModel, + voiceTtsModel: localModels.voiceLocalTtsModel, + voiceTtsSpeakerId: speechConfig?.local?.models.voiceTtsSpeakerId, + voiceTtsSpeed: speechConfig?.local?.models.voiceTtsSpeed, }, - "Local STT engine unavailable", - ); - return null; - } - }; + }) + : null; if (isLocalProviderEnabled(providers.voiceTurnDetection)) { - turnDetectionService = await initializeLocalTurnDetection(localConfig, logger); + if (workerClient) { + turnDetectionService = initializeLocalTurnDetection({ client: workerClient }); + } else { + warnLocalConfigMissing(logger, "turn detection"); + } } if (isLocalProviderEnabled(providers.voiceStt)) { - sttService = await initializeLocalVoiceStt({ - localConfig, - modelId: localModels.voiceLocalSttModel, - logger, - getLocalSttEngine, - }); + if (workerClient) { + sttService = initializeLocalVoiceStt({ client: workerClient }); + } else { + warnLocalConfigMissing(logger, "voice STT"); + } } if (isLocalProviderEnabled(providers.dictationStt)) { - dictationSttService = await initializeLocalDictationStt({ - localConfig, - modelId: localModels.dictationLocalSttModel, - logger, - getLocalSttEngine, - }); + if (workerClient) { + dictationSttService = initializeLocalDictationStt({ client: workerClient }); + } else { + warnLocalConfigMissing(logger, "dictation STT"); + } } if (isLocalProviderEnabled(providers.voiceTts)) { - localVoiceTtsProvider = await initializeLocalVoiceTts({ - localConfig, - speechConfig, - localModels, - logger, - }); + if (workerClient) { + localVoiceTtsProvider = initializeLocalVoiceTts({ client: workerClient }); + } else { + warnLocalConfigMissing(logger, "voice TTS"); + } if (localVoiceTtsProvider) { ttsService = localVoiceTtsProvider; } } const cleanup = () => { - const maybeFreeable = localVoiceTtsProvider as unknown as { free?: () => void } | null; - if (typeof maybeFreeable?.free === "function") { - maybeFreeable.free(); - } - for (const engine of localSttEngines.values()) { - engine.free(); - } + workerClient?.shutdown(); }; return { 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 new file mode 100644 index 000000000..edc2cdedd --- /dev/null +++ b/packages/server/src/server/speech/providers/local/worker-client.test.ts @@ -0,0 +1,212 @@ +import { EventEmitter } from "node:events"; +import { once } from "node:events"; +import pino from "pino"; +import { describe, expect, it } from "vitest"; + +import { + LocalSpeechWorkerClient, + WorkerBackedSpeechToTextProvider, + WorkerBackedTextToSpeechProvider, + WorkerBackedTurnDetectionProvider, +} from "./worker-client.js"; +import type { + LocalSpeechWorkerRequest, + LocalSpeechWorkerToParentMessage, +} from "./worker-protocol.js"; + +class FakeLocalSpeechWorker extends EventEmitter { + public connected = true; + public killed = false; + public readonly sent: LocalSpeechWorkerRequest[] = []; + public disconnects = 0; + public kills = 0; + + send(message: LocalSpeechWorkerRequest, callback: (error: Error | null) => void): boolean { + this.sent.push(message); + queueMicrotask(() => callback(null)); + return true; + } + + disconnect(): void { + this.disconnects++; + this.connected = false; + } + + kill(): boolean { + this.kills++; + this.killed = true; + this.connected = false; + return true; + } + + respond(request: LocalSpeechWorkerRequest, result?: unknown): void { + this.emit("message", { + type: "response", + requestId: request.requestId, + ok: true, + result, + } satisfies LocalSpeechWorkerToParentMessage); + } + + emitWorkerMessage(message: LocalSpeechWorkerToParentMessage): void { + this.emit("message", message); + } +} + +function createClient(options?: { idleTtlMs?: number }) { + const workers: FakeLocalSpeechWorker[] = []; + const client = new LocalSpeechWorkerClient({ + config: { + modelsDir: "/tmp/models", + voiceSttModel: "parakeet-tdt-0.6b-v2-int8", + dictationSttModel: "parakeet-tdt-0.6b-v2-int8", + voiceTtsModel: "kokoro-en-v0_19", + }, + requestTimeoutMs: 1000, + idleTtlMs: options?.idleTtlMs ?? 1000, + forkWorker: () => { + const worker = new FakeLocalSpeechWorker(); + workers.push(worker); + return worker; + }, + }); + return { client, workers }; +} + +async function waitForMicrotasks(): Promise { + await new Promise((resolve) => setTimeout(resolve, 0)); +} + +describe("LocalSpeechWorkerClient", () => { + it("does not spawn the worker until first local speech use", () => { + const { workers } = createClient(); + + expect(workers).toHaveLength(0); + }); + + it("sends TTS requests through the worker and returns the audio stream", async () => { + const { client, workers } = createClient(); + const provider = new WorkerBackedTextToSpeechProvider(client); + + const pending = provider.synthesizeSpeech("hello"); + expect(workers).toHaveLength(1); + const request = workers[0].sent[0]; + expect(request).toMatchObject({ + type: "tts.synthesize", + text: "hello", + config: { + modelsDir: "/tmp/models", + voiceTtsModel: "kokoro-en-v0_19", + }, + }); + + workers[0].respond(request, { + audio: Buffer.from([1, 2, 3, 4]), + format: "pcm;rate=24000", + }); + + const result = await pending; + const chunks: Buffer[] = []; + for await (const chunk of result.stream) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + } + expect(result.format).toBe("pcm;rate=24000"); + expect(Buffer.concat(chunks)).toEqual(Buffer.from([1, 2, 3, 4])); + }); + + it("forwards STT session audio and transcript events through IPC", async () => { + const { client, workers } = createClient(); + const provider = new WorkerBackedSpeechToTextProvider(client, "voiceStt"); + const session = provider.createSession({ logger: pino({ level: "silent" }) }); + + const transcriptPromise = once(session as EventEmitter, "transcript"); + const committedPromise = once(session as EventEmitter, "committed"); + + const connect = session.connect(); + const createRequest = workers[0].sent[0]; + expect(createRequest).toMatchObject({ type: "session.create", kind: "voiceStt" }); + workers[0].respond(createRequest, { requiredSampleRate: 16000 }); + await connect; + + session.appendPcm16(Buffer.from([9, 8, 7, 6])); + await waitForMicrotasks(); + expect(workers[0].sent[1]).toMatchObject({ + type: "session.append", + sessionId: createRequest.sessionId, + audio: Buffer.from([9, 8, 7, 6]), + }); + + session.commit(); + await waitForMicrotasks(); + expect(workers[0].sent[2]).toMatchObject({ + type: "session.commit", + sessionId: createRequest.sessionId, + }); + + workers[0].emitWorkerMessage({ + type: "session.committed", + sessionId: createRequest.sessionId, + payload: { segmentId: "seg-1", previousSegmentId: null }, + }); + workers[0].emitWorkerMessage({ + type: "session.transcript", + sessionId: createRequest.sessionId, + payload: { segmentId: "seg-1", transcript: "hello", isFinal: true }, + }); + + await expect(committedPromise).resolves.toEqual([ + { segmentId: "seg-1", previousSegmentId: null }, + ]); + await expect(transcriptPromise).resolves.toEqual([ + { segmentId: "seg-1", transcript: "hello", isFinal: true }, + ]); + }); + + it("forwards VAD session events through the shared worker", async () => { + const { client, workers } = createClient(); + const provider = new WorkerBackedTurnDetectionProvider(client); + const session = provider.createSession({ logger: pino({ level: "silent" }) }); + const startedPromise = once(session as EventEmitter, "speech_started"); + const stoppedPromise = once(session as EventEmitter, "speech_stopped"); + + const connect = session.connect(); + const createRequest = workers[0].sent[0]; + expect(createRequest).toMatchObject({ type: "session.create", kind: "vad" }); + workers[0].respond(createRequest, { requiredSampleRate: 16000 }); + await connect; + + workers[0].emitWorkerMessage({ + type: "session.speech_started", + sessionId: createRequest.sessionId, + }); + workers[0].emitWorkerMessage({ + type: "session.speech_stopped", + sessionId: createRequest.sessionId, + }); + + await expect(startedPromise).resolves.toEqual([]); + await expect(stoppedPromise).resolves.toEqual([]); + }); + + it("kills an idle worker and respawns on later use", async () => { + const { client, workers } = createClient({ idleTtlMs: 5 }); + + const first = client.synthesizeSpeech("first"); + workers[0].respond(workers[0].sent[0], { + audio: Buffer.from([1]), + format: "pcm;rate=24000", + }); + await first; + + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(workers[0].kills).toBe(1); + + const second = client.synthesizeSpeech("second"); + expect(workers).toHaveLength(2); + workers[1].respond(workers[1].sent[0], { + audio: 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 new file mode 100644 index 000000000..786c82727 --- /dev/null +++ b/packages/server/src/server/speech/providers/local/worker-client.ts @@ -0,0 +1,538 @@ +import { fork } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { EventEmitter } from "node:events"; +import { Readable } from "node:stream"; +import { fileURLToPath } from "node:url"; +import type pino from "pino"; + +import type { + SpeechStreamResult, + SpeechToTextProvider, + StreamingTranscriptionSession, + TextToSpeechProvider, +} from "../../speech-provider.js"; +import type { TurnDetectionProvider, TurnDetectionSession } from "../../turn-detection-provider.js"; +import { applySherpaLoaderEnv } from "./sherpa/sherpa-runtime-env.js"; +import type { + LocalSpeechCreateSessionResult, + LocalSpeechSessionKind, + LocalSpeechTranscriptionResult, + LocalSpeechTtsResult, + LocalSpeechWorkerConfig, + LocalSpeechWorkerRequest, + LocalSpeechWorkerResponse, + LocalSpeechWorkerToParentMessage, +} from "./worker-protocol.js"; + +const DEFAULT_REQUEST_TIMEOUT_MS = 30000; +const DEFAULT_IDLE_TTL_MS = 5 * 60 * 1000; +const DEFAULT_LOCAL_SAMPLE_RATE = 16000; + +type LocalSpeechWorkerRequestInput = LocalSpeechWorkerRequest extends infer Request + ? Request extends LocalSpeechWorkerRequest + ? Omit + : never + : never; + +interface LocalSpeechWorkerProcess { + connected: boolean; + killed: boolean; + send(message: LocalSpeechWorkerRequest, callback: (error: Error | null) => void): boolean; + disconnect(): void; + kill(): boolean; + on(event: "message", listener: (message: LocalSpeechWorkerToParentMessage) => void): this; + on(event: "exit", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this; +} + +interface PendingRequest { + resolve: (value: unknown) => void; + reject: (error: Error) => void; + timeout: ReturnType; +} + +interface LocalSpeechWorkerClientOptions { + config: LocalSpeechWorkerConfig; + requestTimeoutMs?: number; + idleTtlMs?: number; + forkWorker?: () => LocalSpeechWorkerProcess; +} + +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 forkLocalSpeechWorker(): LocalSpeechWorkerProcess { + const env = { ...process.env }; + applySherpaLoaderEnv(env); + return fork(fileURLToPath(resolveWorkerUrl()), [], { + env, + execArgv: resolveWorkerExecArgv(), + serialization: "advanced", + stdio: ["ignore", "ignore", "inherit", "ipc"], + }) as LocalSpeechWorkerProcess; +} + +function isResponse( + message: LocalSpeechWorkerToParentMessage, +): message is LocalSpeechWorkerResponse { + return message.type === "response"; +} + +export class LocalSpeechWorkerClient { + private readonly config: LocalSpeechWorkerConfig; + private readonly requestTimeoutMs: number; + private readonly idleTtlMs: number; + private readonly forkWorker: () => LocalSpeechWorkerProcess; + private readonly pendingRequests = new Map(); + private readonly activeSessionIds = new Set(); + private readonly sessionEmitters = new Map(); + private worker: LocalSpeechWorkerProcess | null = null; + private inFlightRequests = 0; + private idleTimer: ReturnType | null = null; + + constructor(options: LocalSpeechWorkerClientOptions) { + this.config = options.config; + this.requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS; + this.idleTtlMs = options.idleTtlMs ?? DEFAULT_IDLE_TTL_MS; + this.forkWorker = options.forkWorker ?? forkLocalSpeechWorker; + } + + async synthesizeSpeech(text: string): Promise { + const result = await this.sendRequest({ + type: "tts.synthesize", + config: this.config, + text, + }); + return { + stream: Readable.from([result.audio]), + format: result.format, + }; + } + + transcribeVoice(audio: Buffer, format: string): Promise { + return this.sendRequest({ + type: "stt.transcribe", + config: this.config, + model: "voice", + audio, + format, + }); + } + + async createSession( + kind: LocalSpeechSessionKind, + emitter: EventEmitter, + ): Promise<{ sessionId: string; requiredSampleRate: number }> { + const sessionId = randomUUID(); + this.activeSessionIds.add(sessionId); + this.sessionEmitters.set(sessionId, emitter); + try { + const result = await this.sendRequest({ + type: "session.create", + config: this.config, + sessionId, + kind, + }); + return { sessionId, requiredSampleRate: result.requiredSampleRate }; + } catch (err) { + this.activeSessionIds.delete(sessionId); + this.sessionEmitters.delete(sessionId); + this.scheduleIdleShutdownIfReady(); + throw err; + } + } + + appendSessionAudio(sessionId: string, audio: Buffer): void { + void this.sendRequest({ type: "session.append", sessionId, audio }).catch((err) => { + this.emitSessionError(sessionId, err); + }); + } + + commitSession(sessionId: string): void { + void this.sendRequest({ type: "session.commit", sessionId }).catch((err) => { + this.emitSessionError(sessionId, err); + }); + } + + clearSession(sessionId: string): void { + void this.sendRequest({ type: "session.clear", sessionId }).catch((err) => { + this.emitSessionError(sessionId, err); + }); + } + + flushSession(sessionId: string): void { + void this.sendRequest({ type: "session.flush", sessionId }).catch((err) => { + this.emitSessionError(sessionId, err); + }); + } + + resetSession(sessionId: string): void { + void this.sendRequest({ type: "session.reset", sessionId }).catch((err) => { + this.emitSessionError(sessionId, err); + }); + } + + closeSession(sessionId: string): void { + this.activeSessionIds.delete(sessionId); + this.sessionEmitters.delete(sessionId); + void this.sendRequest({ type: "session.close", sessionId }).catch(() => { + // Closing is best-effort; the parent already dropped the session. + }); + this.scheduleIdleShutdownIfReady(); + } + + shutdown(): void { + this.clearIdleTimer(); + this.rejectAllPending(new Error("Local speech worker shut down")); + this.activeSessionIds.clear(); + this.sessionEmitters.clear(); + const worker = this.worker; + this.worker = null; + if (worker && !worker.killed) { + try { + worker.disconnect(); + } catch { + // ignore + } + try { + worker.kill(); + } catch { + // ignore + } + } + } + + private sendRequest(input: LocalSpeechWorkerRequestInput): Promise { + const worker = this.ensureWorker(); + const requestId = randomUUID(); + const message = { ...input, requestId } as LocalSpeechWorkerRequest; + this.inFlightRequests++; + this.clearIdleTimer(); + + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + this.pendingRequests.delete(requestId); + this.inFlightRequests = Math.max(0, this.inFlightRequests - 1); + this.scheduleIdleShutdownIfReady(); + reject(new Error(`Local speech worker request timed out: ${input.type}`)); + }, this.requestTimeoutMs); + this.pendingRequests.set(requestId, { + resolve: (value) => resolve(value as T), + reject, + timeout, + }); + + const sent = worker.send(message, (error) => { + if (!error) { + return; + } + const pending = this.pendingRequests.get(requestId); + if (!pending) { + return; + } + clearTimeout(pending.timeout); + this.pendingRequests.delete(requestId); + this.inFlightRequests = Math.max(0, this.inFlightRequests - 1); + this.scheduleIdleShutdownIfReady(); + pending.reject(error); + }); + if (!sent) { + const pending = this.pendingRequests.get(requestId); + if (pending) { + clearTimeout(pending.timeout); + this.pendingRequests.delete(requestId); + this.inFlightRequests = Math.max(0, this.inFlightRequests - 1); + this.scheduleIdleShutdownIfReady(); + pending.reject(new Error("Local speech worker IPC channel is not writable")); + } + } + }); + } + + private ensureWorker(): LocalSpeechWorkerProcess { + if (this.worker && !this.worker.killed && this.worker.connected) { + return this.worker; + } + const worker = this.forkWorker(); + this.worker = worker; + worker.on("message", (message) => this.handleWorkerMessage(message)); + worker.on("exit", () => this.handleWorkerExit()); + return worker; + } + + private handleWorkerMessage(message: LocalSpeechWorkerToParentMessage): void { + if (isResponse(message)) { + const pending = this.pendingRequests.get(message.requestId); + if (!pending) { + return; + } + clearTimeout(pending.timeout); + this.pendingRequests.delete(message.requestId); + this.inFlightRequests = Math.max(0, this.inFlightRequests - 1); + this.scheduleIdleShutdownIfReady(); + if (message.ok) { + pending.resolve(message.result); + } else { + pending.reject(new Error(message.error)); + } + return; + } + + const emitter = this.sessionEmitters.get(message.sessionId); + if (!emitter) { + return; + } + switch (message.type) { + case "session.committed": + emitter.emit("committed", message.payload); + return; + case "session.transcript": + emitter.emit("transcript", message.payload); + return; + case "session.speech_started": + emitter.emit("speech_started"); + return; + case "session.speech_stopped": + emitter.emit("speech_stopped"); + return; + case "session.error": + emitter.emit("error", new Error(message.error)); + return; + } + } + + private handleWorkerExit(): void { + this.worker = null; + this.clearIdleTimer(); + this.rejectAllPending(new Error("Local speech worker exited")); + for (const [sessionId, emitter] of this.sessionEmitters) { + if (this.activeSessionIds.has(sessionId)) { + emitter.emit("error", new Error("Local speech worker exited")); + } + } + this.activeSessionIds.clear(); + this.sessionEmitters.clear(); + this.inFlightRequests = 0; + } + + private rejectAllPending(error: Error): void { + for (const [requestId, pending] of this.pendingRequests) { + clearTimeout(pending.timeout); + pending.reject(error); + this.pendingRequests.delete(requestId); + } + } + + private emitSessionError(sessionId: string, error: unknown): void { + const emitter = this.sessionEmitters.get(sessionId); + if (!emitter) { + return; + } + emitter.emit("error", error instanceof Error ? error : new Error(String(error))); + } + + private scheduleIdleShutdownIfReady(): void { + if (!this.worker || this.inFlightRequests > 0 || this.activeSessionIds.size > 0) { + return; + } + this.clearIdleTimer(); + this.idleTimer = setTimeout(() => { + if (this.inFlightRequests === 0 && this.activeSessionIds.size === 0) { + this.shutdown(); + } + }, this.idleTtlMs); + } + + private clearIdleTimer(): void { + if (this.idleTimer) { + clearTimeout(this.idleTimer); + this.idleTimer = null; + } + } +} + +export class WorkerBackedTextToSpeechProvider implements TextToSpeechProvider { + constructor(private readonly client: LocalSpeechWorkerClient) {} + + synthesizeSpeech(text: string): Promise { + return this.client.synthesizeSpeech(text); + } +} + +export class WorkerBackedSpeechToTextProvider implements SpeechToTextProvider { + public readonly id = "local" as const; + + constructor( + private readonly client: LocalSpeechWorkerClient, + private readonly kind: Extract, + ) {} + + createSession(_params: { + logger: pino.Logger; + language?: string; + prompt?: string; + }): StreamingTranscriptionSession { + return new WorkerBackedTranscriptionSession(this.client, this.kind); + } +} + +export class WorkerBackedTurnDetectionProvider implements TurnDetectionProvider { + public readonly id = "local" as const; + + constructor(private readonly client: LocalSpeechWorkerClient) {} + + createSession(_params: { logger: pino.Logger }): TurnDetectionSession { + return new WorkerBackedTurnDetectionSession(this.client); + } +} + +class WorkerBackedTranscriptionSession + extends EventEmitter + implements StreamingTranscriptionSession +{ + public requiredSampleRate = DEFAULT_LOCAL_SAMPLE_RATE; + private connectedSessionId: string | null = null; + private connecting: Promise | null = null; + + constructor( + private readonly client: LocalSpeechWorkerClient, + private readonly kind: Extract, + ) { + super(); + } + + async connect(): Promise { + if (this.connectedSessionId) { + return; + } + if (!this.connecting) { + this.connecting = this.connectRemoteSession(); + } + await this.connecting; + } + + private async connectRemoteSession(): Promise { + try { + const result = await this.client.createSession(this.kind, this); + this.connectedSessionId = result.sessionId; + this.requiredSampleRate = result.requiredSampleRate; + } finally { + this.connecting = null; + } + } + + appendPcm16(pcm16le: Buffer): void { + const sessionId = this.connectedSessionId; + if (!sessionId) { + this.emit("error", new Error("Local STT session not connected")); + return; + } + this.client.appendSessionAudio(sessionId, pcm16le); + } + + commit(): void { + const sessionId = this.connectedSessionId; + if (!sessionId) { + this.emit("error", new Error("Local STT session not connected")); + return; + } + this.client.commitSession(sessionId); + } + + clear(): void { + const sessionId = this.connectedSessionId; + if (sessionId) { + this.client.clearSession(sessionId); + } + } + + close(): void { + const sessionId = this.connectedSessionId; + this.connectedSessionId = null; + if (sessionId) { + this.client.closeSession(sessionId); + } + } +} + +class WorkerBackedTurnDetectionSession extends EventEmitter implements TurnDetectionSession { + public requiredSampleRate = DEFAULT_LOCAL_SAMPLE_RATE; + private connectedSessionId: string | null = null; + private connecting: Promise | null = null; + + constructor(private readonly client: LocalSpeechWorkerClient) { + super(); + } + + async connect(): Promise { + if (this.connectedSessionId) { + return; + } + if (!this.connecting) { + this.connecting = this.connectRemoteSession(); + } + await this.connecting; + } + + private async connectRemoteSession(): Promise { + try { + const result = await this.client.createSession("vad", this); + this.connectedSessionId = result.sessionId; + this.requiredSampleRate = result.requiredSampleRate; + } finally { + this.connecting = null; + } + } + + appendPcm16(pcm16le: Buffer): void { + const sessionId = this.connectedSessionId; + if (!sessionId) { + this.emit("error", new Error("Local turn-detection session not connected")); + return; + } + this.client.appendSessionAudio(sessionId, pcm16le); + } + + flush(): void { + const sessionId = this.connectedSessionId; + if (sessionId) { + this.client.flushSession(sessionId); + } + } + + reset(): void { + const sessionId = this.connectedSessionId; + if (sessionId) { + this.client.resetSession(sessionId); + } + } + + close(): void { + const sessionId = this.connectedSessionId; + this.connectedSessionId = null; + if (sessionId) { + this.client.closeSession(sessionId); + } + } +} diff --git a/packages/server/src/server/speech/providers/local/worker-process.ts b/packages/server/src/server/speech/providers/local/worker-process.ts new file mode 100644 index 000000000..6cb17ad4d --- /dev/null +++ b/packages/server/src/server/speech/providers/local/worker-process.ts @@ -0,0 +1,335 @@ +import pino from "pino"; + +import type { StreamingTranscriptionSession } from "../../speech-provider.js"; +import type { TurnDetectionSession } from "../../turn-detection-provider.js"; +import { getLocalSpeechModelDir, type LocalSttModelId, type LocalTtsModelId } from "./models.js"; +import { SherpaOfflineRecognizerEngine } from "./sherpa/sherpa-offline-recognizer.js"; +import { SherpaOnnxParakeetSTT } from "./sherpa/sherpa-parakeet-stt.js"; +import { SherpaParakeetRealtimeTranscriptionSession } from "./sherpa/sherpa-parakeet-realtime-session.js"; +import { SherpaOnnxTTS } from "./sherpa/sherpa-tts.js"; +import { + ensureSileroVadModel, + SherpaSileroTurnDetectionProvider, +} from "./sherpa/silero-vad-provider.js"; +import type { + LocalSpeechWorkerConfig, + LocalSpeechWorkerRequest, + LocalSpeechWorkerToParentMessage, +} from "./worker-protocol.js"; + +process.title = "Paseo Voice"; + +type LocalSttEngine = SherpaOfflineRecognizerEngine; + +const logger = pino({ + level: process.env.PASEO_LOG_LEVEL ?? "info", +}).child({ module: "speech", component: "local-worker" }); + +const sttEngines = new Map(); +const sttProviders = new Map(); +const ttsProviders = new Map(); +const sessions = new Map(); +const unsubscribeBySessionId = new Map void>>(); +let ipcClosing = false; + +function sendToParent(message: LocalSpeechWorkerToParentMessage): void { + if (ipcClosing || !process.connected || !process.send) { + return; + } + try { + process.send(message, (error) => { + if (error) { + ipcClosing = true; + } + }); + } catch { + ipcClosing = true; + } +} + +function sttModelId( + config: LocalSpeechWorkerConfig, + model: "voice" | "dictation", +): LocalSttModelId { + return (model === "voice" ? config.voiceSttModel : config.dictationSttModel) as LocalSttModelId; +} + +function ttsModelId(config: LocalSpeechWorkerConfig): LocalTtsModelId { + return config.voiceTtsModel as LocalTtsModelId; +} + +function sttEngineKey(config: LocalSpeechWorkerConfig, modelId: LocalSttModelId): string { + return `${config.modelsDir}:${modelId}`; +} + +function ttsKey(config: LocalSpeechWorkerConfig): string { + return [ + config.modelsDir, + config.voiceTtsModel, + config.voiceTtsSpeakerId ?? 0, + config.voiceTtsSpeed ?? 1, + ].join(":"); +} + +function getSttEngine( + config: LocalSpeechWorkerConfig, + model: "voice" | "dictation", +): LocalSttEngine { + const modelId = sttModelId(config, model); + const key = sttEngineKey(config, modelId); + const existing = sttEngines.get(key); + if (existing) { + return existing; + } + const modelDir = getLocalSpeechModelDir(config.modelsDir, modelId); + const created = new SherpaOfflineRecognizerEngine( + { + model: { + kind: "nemo_transducer", + encoder: `${modelDir}/encoder.int8.onnx`, + decoder: `${modelDir}/decoder.int8.onnx`, + joiner: `${modelDir}/joiner.int8.onnx`, + tokens: `${modelDir}/tokens.txt`, + }, + numThreads: 2, + debug: 0, + }, + logger, + ); + sttEngines.set(key, created); + return created; +} + +function getSttProvider( + config: LocalSpeechWorkerConfig, + model: "voice" | "dictation", +): SherpaOnnxParakeetSTT { + const modelId = sttModelId(config, model); + const key = sttEngineKey(config, modelId); + const existing = sttProviders.get(key); + if (existing) { + return existing; + } + const created = new SherpaOnnxParakeetSTT({ engine: getSttEngine(config, model) }, logger); + sttProviders.set(key, created); + return created; +} + +function getTtsProvider(config: LocalSpeechWorkerConfig): SherpaOnnxTTS { + const key = ttsKey(config); + const existing = ttsProviders.get(key); + if (existing) { + return existing; + } + const modelDir = getLocalSpeechModelDir(config.modelsDir, ttsModelId(config)); + const created = new SherpaOnnxTTS( + { + preset: ttsModelId(config), + modelDir, + speakerId: config.voiceTtsSpeakerId, + speed: config.voiceTtsSpeed, + }, + logger, + ); + ttsProviders.set(key, created); + return created; +} + +function cleanupSession(sessionId: string): void { + const unsubscribe = unsubscribeBySessionId.get(sessionId); + if (unsubscribe) { + for (const fn of unsubscribe) { + try { + fn(); + } catch { + // ignore + } + } + } + unsubscribeBySessionId.delete(sessionId); + const session = sessions.get(sessionId); + sessions.delete(sessionId); + try { + session?.close(); + } catch { + // ignore + } +} + +function trackTranscriptionSession( + sessionId: string, + session: StreamingTranscriptionSession, +): void { + session.on("committed", (payload) => { + sendToParent({ type: "session.committed", sessionId, payload }); + }); + session.on("transcript", (payload) => { + sendToParent({ type: "session.transcript", sessionId, payload }); + }); + session.on("error", (err) => { + sendToParent({ + type: "session.error", + sessionId, + error: err instanceof Error ? err.message : String(err), + }); + }); + unsubscribeBySessionId.set(sessionId, []); +} + +function trackTurnDetectionSession(sessionId: string, session: TurnDetectionSession): void { + session.on("speech_started", () => { + sendToParent({ type: "session.speech_started", sessionId }); + }); + session.on("speech_stopped", () => { + sendToParent({ type: "session.speech_stopped", sessionId }); + }); + session.on("error", (err) => { + sendToParent({ + type: "session.error", + sessionId, + error: err instanceof Error ? err.message : String(err), + }); + }); + unsubscribeBySessionId.set(sessionId, []); +} + +async function createSession( + message: Extract, +) { + cleanupSession(message.sessionId); + if (message.kind === "vad") { + let vadModelPath: string | undefined; + try { + vadModelPath = await ensureSileroVadModel(message.config.modelsDir, logger); + } catch (err) { + logger.warn({ err }, "Failed to provision Silero VAD model, falling back to bundled"); + } + const provider = new SherpaSileroTurnDetectionProvider({ modelPath: vadModelPath }, logger); + const session = provider.createSession({ logger }); + trackTurnDetectionSession(message.sessionId, session); + await session.connect(); + sessions.set(message.sessionId, session); + return { requiredSampleRate: session.requiredSampleRate }; + } + + const model = message.kind === "voiceStt" ? "voice" : "dictation"; + const engine = getSttEngine(message.config, model); + const session = + message.kind === "voiceStt" + ? getSttProvider(message.config, "voice").createSession({ logger }) + : new SherpaParakeetRealtimeTranscriptionSession({ engine }); + trackTranscriptionSession(message.sessionId, session); + await session.connect(); + sessions.set(message.sessionId, session); + return { requiredSampleRate: session.requiredSampleRate }; +} + +function sendOk(requestId: string, result?: unknown): void { + sendToParent({ type: "response", requestId, ok: true, result }); +} + +type LocalSpeechWorkerSessionRequest = Extract< + LocalSpeechWorkerRequest, + { + type: + | "session.append" + | "session.commit" + | "session.clear" + | "session.flush" + | "session.reset" + | "session.close"; + } +>; + +function handleSessionRequest(message: LocalSpeechWorkerSessionRequest): void { + if (message.type === "session.close") { + cleanupSession(message.sessionId); + sendOk(message.requestId); + return; + } + + const session = sessions.get(message.sessionId); + switch (message.type) { + case "session.append": + session?.appendPcm16(message.audio); + break; + case "session.commit": + if (session && "commit" in session) { + session.commit(); + } + break; + case "session.clear": + if (session && "clear" in session) { + session.clear(); + } + break; + case "session.flush": + if (session && "flush" in session) { + session.flush(); + } + break; + case "session.reset": + if (session && "reset" in session) { + session.reset(); + } + break; + } + sendOk(message.requestId); +} + +async function handleRequest(message: LocalSpeechWorkerRequest): Promise { + if (message.type === "tts.synthesize") { + const result = await getTtsProvider(message.config).synthesizeSpeech(message.text); + const chunks: Buffer[] = []; + 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 }); + return; + } + + if (message.type === "stt.transcribe") { + const result = await getSttProvider(message.config, message.model).transcribeAudio( + message.audio, + message.format, + ); + sendOk(message.requestId, result); + return; + } + + if (message.type === "session.create") { + const result = await createSession(message); + sendOk(message.requestId, result); + return; + } + + handleSessionRequest(message); +} + +process.on("message", (message: LocalSpeechWorkerRequest) => { + void handleRequest(message).catch((error: unknown) => { + sendToParent({ + type: "response", + requestId: message.requestId, + ok: false, + error: error instanceof Error ? error.message : "Local speech worker request failed", + }); + }); +}); + +process.once("disconnect", () => { + ipcClosing = true; + for (const sessionId of Array.from(sessions.keys())) { + cleanupSession(sessionId); + } + for (const tts of ttsProviders.values()) { + try { + tts.free(); + } catch { + // ignore + } + } + for (const engine of sttEngines.values()) { + engine.free(); + } +}); diff --git a/packages/server/src/server/speech/providers/local/worker-protocol.ts b/packages/server/src/server/speech/providers/local/worker-protocol.ts new file mode 100644 index 000000000..7c9a85075 --- /dev/null +++ b/packages/server/src/server/speech/providers/local/worker-protocol.ts @@ -0,0 +1,122 @@ +import type { + StreamingTranscriptionCommittedEvent, + StreamingTranscriptionEvent, + TranscriptionResult, +} from "../../speech-provider.js"; + +export interface LocalSpeechWorkerConfig { + modelsDir: string; + voiceSttModel: string; + dictationSttModel: string; + voiceTtsModel: string; + voiceTtsSpeakerId?: number; + voiceTtsSpeed?: number; +} + +export type LocalSpeechSessionKind = "voiceStt" | "dictationStt" | "vad"; + +export type LocalSpeechWorkerRequest = + | { + type: "tts.synthesize"; + requestId: string; + config: LocalSpeechWorkerConfig; + text: string; + } + | { + type: "stt.transcribe"; + requestId: string; + config: LocalSpeechWorkerConfig; + model: "voice" | "dictation"; + audio: Buffer; + format: string; + } + | { + type: "session.create"; + requestId: string; + config: LocalSpeechWorkerConfig; + sessionId: string; + kind: LocalSpeechSessionKind; + } + | { + type: "session.append"; + requestId: string; + sessionId: string; + audio: Buffer; + } + | { + type: "session.commit"; + requestId: string; + sessionId: string; + } + | { + type: "session.clear"; + requestId: string; + sessionId: string; + } + | { + type: "session.flush"; + requestId: string; + sessionId: string; + } + | { + type: "session.reset"; + requestId: string; + sessionId: string; + } + | { + type: "session.close"; + requestId: string; + sessionId: string; + }; + +export type LocalSpeechWorkerResponse = + | { + type: "response"; + requestId: string; + ok: true; + result?: unknown; + } + | { + type: "response"; + requestId: string; + ok: false; + error: string; + }; + +export type LocalSpeechWorkerEvent = + | { + type: "session.committed"; + sessionId: string; + payload: StreamingTranscriptionCommittedEvent; + } + | { + type: "session.transcript"; + sessionId: string; + payload: StreamingTranscriptionEvent; + } + | { + type: "session.speech_started"; + sessionId: string; + } + | { + type: "session.speech_stopped"; + sessionId: string; + } + | { + type: "session.error"; + sessionId: string; + error: string; + }; + +export type LocalSpeechWorkerToParentMessage = LocalSpeechWorkerResponse | LocalSpeechWorkerEvent; + +export interface LocalSpeechCreateSessionResult { + requiredSampleRate: number; +} + +export interface LocalSpeechTtsResult { + audio: Buffer; + format: string; +} + +export type LocalSpeechTranscriptionResult = TranscriptionResult;