1// @effect-diagnostics nodeBuiltinImport:off globalTimers:off - native inference and IPC cleanup need killable Promise boundaries outside the Effect runtime.
2import * as NodeChildProcess from "node:child_process";
3import * as NodeSea from "node:sea";
4import type { SpeechStreamText } from "@t3tools/contracts";
5import { createSpeechAudio } from "./audio.ts";
6
7// Inline the small child entry so the packaged CLI does not need a separate worker artifact.
8const entry = `
9const createSpeechAudio = ${createSpeechAudio.toString()};
10let speech;
11let model;
12let vadModel;
13let vad;
14let session;
15let stream;
16let audio;
17let options;
18let nativeStreaming;
19let useVad;
20let scored;
21let revision;
22let texts;
23let heardAudio;
24const warnVad = () => console.warn("Silero silence filtering unavailable; preserving raw recording and using bounded ASR chunks.");
25const resetAsr = () => {
26 if (stream) stream.reset();
27 if (session) session.dispose();
28 stream = session = undefined;
29};
30const newAudio = (filterSilence) => createSpeechAudio({
31 frameSamples: 512, filterSilence, nativeStreaming: model.capabilities.supportsStreaming,
32 longForm: model.supports("long_form"), maxAudioMs: model.capabilities.maxAudioMs,
33});
34const openAsr = async () => {
35 if (nativeStreaming) {
36 session = model.createSession();
37 stream = await session.stream(options);
38 }
39};
40const begin = async (message, live) => {
41 resetAsr();
42 options = message.options;
43 nativeStreaming = live && model.capabilities.supportsStreaming;
44 texts = [];
45 scored = revision = 0;
46 heardAudio = false;
47 useVad = Boolean(message.vadPath) && (message.filterSilence ||
48 (!model.capabilities.supportsStreaming && !model.supports("long_form")));
49 if (useVad) {
50 try {
51 if (!vadModel) {
52 vadModel = await speech.TranscribeModel.load(message.vadPath, { backend: "cpu" });
53 if (vadModel.vadInfo.sampleRate !== 16000 || vadModel.vadInfo.frameSamples !== 512)
54 throw new Error("Unsupported Silero model frames.");
55 vad = vadModel.createVadSession({ nThreads: 1 });
56 }
57 await vad.streamReset();
58 } catch {
59 vad?.dispose();
60 vadModel?.dispose();
61 vad = vadModel = undefined;
62 useVad = false;
63 warnVad();
64 }
65 }
66 audio = newAudio(useVad && message.filterSilence);
67 await openAsr();
68};
69const transcribe = async (pcm) => {
70 try {
71 const result = await model.transcribe(pcm, options);
72 texts.push(result.text);
73 } catch (error) {
74 if ((error?.constructor?.name !== "OutputTruncated" &&
75 error?.constructor?.name !== "InputTooLong") || pcm.length <= 4000) throw error;
76 const middle = Math.floor(pcm.length / 2);
77 await transcribe(pcm.subarray(0, middle));
78 await transcribe(pcm.subarray(middle));
79 }
80};
81const decode = async (final) => {
82 let preview = null;
83 for (const pcm of audio.take(final)) {
84 if (!pcm.length) continue;
85 heardAudio = true;
86 if (nativeStreaming) {
87 const update = await stream.feed(pcm);
88 if (update.committedChanged || update.tentativeChanged) {
89 const text = stream.text;
90 preview = { committed: text.committed, tentative: text.tentative };
91 }
92 } else {
93 await transcribe(pcm);
94 preview = { committed: texts.join(" ").trim(), tentative: "" };
95 }
96 }
97 return preview;
98};
99const feed = async (pcm, final = false) => {
100 if (pcm) audio.append(pcm);
101 try {
102 if (useVad) {
103 if (pcm) {
104 const result = await vad.streamFeed(pcm);
105 audio.score(result.firstFrame, result.probs);
106 if (result.firstFrame + result.probs.length !== Math.floor(audio.samples() / 512))
107 throw new Error("Incomplete Silero frames.");
108 }
109 if (final) {
110 const result = await vad.streamFlush();
111 audio.score(result.firstFrame, result.probs);
112 if (result.firstFrame + result.probs.length !== Math.ceil(audio.samples() / 512))
113 throw new Error("Incomplete Silero tail.");
114 }
115 } else {
116 const count = final ? Math.ceil(audio.samples() / 512) : Math.floor(audio.samples() / 512);
117 audio.score(scored, new Array(count - scored).fill(1));
118 scored = count;
119 }
120 } catch {
121 // Replay on optional VAD failure, including audio it had withheld; ASR errors stay separate.
122 const raw = audio.audio();
123 useVad = false;
124 warnVad();
125 resetAsr();
126 texts = [];
127 heardAudio = false;
128 audio = newAudio(false);
129 audio.append(raw);
130 scored = final ? Math.ceil(raw.length / 512) : Math.floor(raw.length / 512);
131 audio.score(0, new Array(scored).fill(1));
132 await openAsr();
133 }
134 return await decode(final);
135};
136process.on("message", async (message) => {
137 try {
138 if (message.kind === "load") {
139 speech = await loadSpeechModule(message.moduleUrl);
140 const { TranscribeModel, getAvailableBackends } = speech;
141 const device = message.acceleration.startsWith("gpu:")
142 ? getAvailableBackends().find((item) =>
143 (item.deviceType === "gpu" || item.deviceType === "igpu") &&
144 JSON.stringify([item.kind, item.deviceId ?? item.name]) === message.acceleration.slice(4))
145 : undefined;
146 if (message.acceleration.startsWith("gpu:") && !device)
147 throw new Error("The selected speech GPU is unavailable.");
148 model = await TranscribeModel.load(message.path, device ? { device } : { backend: message.acceleration });
149 process.send({ type: "t3-speech-reply", ok: true, backend: model.backend,
150 languages: model.capabilities.languages,
151 supportsStreaming: model.capabilities.supportsStreaming,
152 supportsLongForm: model.supports("long_form"),
153 supportsTranslation: model.capabilities.supportsTranslate,
154 supportsVocabulary: model.supports("vocabulary") });
155 } else if (message.kind === "begin") {
156 await begin(message, true);
157 process.send({ type: "t3-speech-reply", ok: true });
158 } else if (message.kind === "feed") {
159 const preview = await feed(message.pcm);
160 process.send({ type: "t3-speech-reply", ok: true, revision: ++revision, preview });
161 } else if (message.kind === "finish") {
162 await feed(undefined, true);
163 if (nativeStreaming && heardAudio) await stream.finalize();
164 const text = nativeStreaming ? (heardAudio ? stream.text.full : "") : texts.join(" ").trim();
165 resetAsr();
166 audio = undefined;
167 process.send({ type: "t3-speech-reply", ok: true, text });
168 } else if (message.kind === "reset") {
169 resetAsr();
170 if (vad) await vad.streamReset();
171 audio = undefined;
172 process.send({ type: "t3-speech-reply", ok: true });
173 } else if (message.kind === "transcribe") {
174 await begin(message, false);
175 await feed(message.pcm, true);
176 const text = texts.join(" ").trim();
177 audio = undefined;
178 process.send({ type: "t3-speech-reply", ok: true, text });
179 }
180 } catch (error) {
181 process.send({ type: "t3-speech-reply", ok: false,
182 error: error instanceof Error ? error.stack ?? error.message : String(error) });
183 }
184});
185process.on("disconnect", () => process.exit(0));
186`;
187
188type Reply = {
189 readonly type: "t3-speech-reply";
190 readonly ok: boolean;
191 readonly text?: string;
192 readonly error?: string;
193 readonly supportsStreaming?: boolean;
194 readonly supportsLongForm?: boolean;
195 readonly supportsTranslation?: boolean;
196 readonly supportsVocabulary?: boolean;
197 readonly backend?: string;
198 readonly languages?: unknown;
199 readonly revision?: number;
200 readonly preview?: SpeechStreamText | null;
201};
202
203// transcribe-cpp has an import-only export. Load its disk entry with require so
204// Node SEA can reach it, including its native dependencies.
205const loaderEntry = `
206 const { createRequire, findPackageJSON } = await import("node:module");
207 const { dirname, join } = await import("node:path");
208 const loadSpeechModule = async (moduleUrl) => {
209 let speech;
210 if (moduleUrl) speech = await import(moduleUrl);
211 else {
212 const packagePath = findPackageJSON("transcribe-cpp", baseUrl);
213 if (!packagePath) throw new Error("The native speech package is unavailable.");
214 speech = createRequire(packagePath)(join(dirname(packagePath), "dist/index.js"));
215 }
216 if (process.versions.electron && !process.env.TRANSCRIBE_LIBRARY) {
217 const directory = speech.artifactDir();
218 const unpacked = directory.replace(/\\.asar([\\\\/])/g, ".asar.unpacked$1");
219 // Koffi opens native libraries outside Electron's ASAR-aware filesystem.
220 if (unpacked !== directory) process.env.TRANSCRIBE_LIBRARY = join(unpacked,
221 process.platform === "win32" ? "transcribe.dll"
222 : process.platform === "darwin" ? "libtranscribe.dylib" : "libtranscribe.so");
223 }
224 return speech;
225 };
226`;
227
228const discoveryEntry = `
229 const { getAvailableBackends } = await loadSpeechModule(moduleUrl);
230 process.send(getAvailableBackends().filter((device) =>
231 device.deviceType === "gpu" || device.deviceType === "igpu").map((device) => ({
232 id: JSON.stringify([device.kind, device.deviceId ?? device.name]),
233 name: device.description || device.name,
234 })));
235`;
236
237export async function runNativeSpeechWorker(mode: "discover" | "model", moduleUrl?: string) {
238 if (!process.send) throw new Error("Speech workers require an IPC channel.");
239 const run = new Function(
240 "baseUrl",
241 "moduleUrl",
242 `return (async () => {${loaderEntry}${mode === "discover" ? discoveryEntry : entry}})();`,
243 ) as (baseUrl: string, moduleUrl?: string) => Promise<void>;
244 await run(import.meta.url, moduleUrl);
245}
246
247function speechWorkerArgs(mode: "discover" | "model", moduleUrl?: string) {
248 if (NodeSea.isSea()) return ["__speech-worker", mode, ...(moduleUrl ? [moduleUrl] : [])];
249 // Normal Node and Electron hosts can evaluate the same entry without a sibling artifact.
250 const loader = `const baseUrl = ${JSON.stringify(import.meta.url)};
251 const moduleUrl = process.argv[1];${loaderEntry}`;
252 return [
253 "--input-type=module",
254 "-e",
255 loader + (mode === "discover" ? discoveryEntry : entry),
256 ...(moduleUrl ? [moduleUrl] : []),
257 ];
258}
259
260export async function listNativeSpeechGpuDevices(moduleUrl?: string) {
261 const child = NodeChildProcess.spawn(process.execPath, speechWorkerArgs("discover", moduleUrl), {
262 stdio: ["ignore", "ignore", "inherit", "ipc"],
263 env: { ...process.env, ELECTRON_RUN_AS_NODE: "1" },
264 });
265 return await new Promise<{ id: string; name: string }[]>((resolve, reject) => {
266 const timer = setTimeout(() => {
267 child.kill();
268 reject(new Error("Speech device discovery timed out."));
269 }, 10_000);
270 child.on("message", (devices: unknown) => {
271 if (
272 Array.isArray(devices) &&
273 devices.every(
274 (device: unknown) =>
275 typeof device === "object" &&
276 device !== null &&
277 "id" in device &&
278 typeof device.id === "string" &&
279 "name" in device &&
280 typeof device.name === "string",
281 )
282 ) {
283 clearTimeout(timer);
284 resolve(devices);
285 child.kill();
286 }
287 });
288 child.once("error", (error) => {
289 clearTimeout(timer);
290 child.kill();
291 reject(error);
292 });
293 child.once("exit", (code) => {
294 clearTimeout(timer);
295 reject(new Error(`Speech device discovery exited (${code}).`));
296 });
297 });
298}
299
300/** Loads a model in an isolated process. "auto" retries on the CPU when GPU setup fails. */
301export async function loadNativeSpeechModel(
302 path: string,
303 signal: AbortSignal,
304 moduleUrl?: string,
305 acceleration = "auto",
306) {
307 try {
308 return await loadNativeSpeechModelOnce(path, signal, moduleUrl, acceleration);
309 } catch (error) {
310 if (acceleration !== "auto" || signal.aborted) throw error;
311 return await loadNativeSpeechModelOnce(path, signal, moduleUrl, "cpu");
312 }
313}
314
315async function loadNativeSpeechModelOnce(
316 path: string,
317 signal: AbortSignal,
318 moduleUrl?: string,
319 acceleration = "auto",
320) {
321 signal.throwIfAborted();
322 const child = NodeChildProcess.spawn(process.execPath, speechWorkerArgs("model"), {
323 stdio: ["ignore", "ignore", "inherit", "ipc"],
324 serialization: "advanced",
325 env: {
326 ...process.env,
327 ELECTRON_RUN_AS_NODE: "1",
328 // Backend registration can initialize drivers even when model placement is CPU.
329 ...(acceleration === "cpu" ? { TRANSCRIBE_BACKENDS: "cpu" } : {}),
330 },
331 });
332 let stopped = false;
333 let pending: ReturnType<typeof Promise.withResolvers<Reply>> | undefined;
334 const exited = Promise.withResolvers<void>();
335 const fail = (error: Error) => {
336 stopped = true;
337 pending?.reject(error);
338 pending = undefined;
339 };
340 child.on("error", (error) => {
341 fail(error);
342 exited.resolve();
343 });
344 child.on("exit", (code, exitSignal) => {
345 fail(new Error(`Speech process exited (${exitSignal ?? code}).`));
346 signal.removeEventListener("abort", stop);
347 exited.resolve();
348 });
349 child.on("message", (message: unknown) => {
350 if (
351 typeof message !== "object" ||
352 message === null ||
353 !("type" in message) ||
354 message.type !== "t3-speech-reply"
355 )
356 return;
357 const reply = message as Reply;
358 const request = pending;
359 pending = undefined;
360 if (reply.ok) request?.resolve(reply);
361 else request?.reject(new Error(reply.error ?? "Native speech failed."));
362 });
363 function stop() {
364 fail(new Error("Speech process stopped."));
365 // Never call native dispose while inference is active. The OS owns cleanup of the isolated process.
366 child.kill("SIGKILL");
367 }
368 signal.addEventListener("abort", stop, { once: true });
369 if (signal.aborted) stop();
370 const send = (message: object) => {
371 if (stopped) return Promise.reject(new Error("Speech process stopped."));
372 if (pending) return Promise.reject(new Error("Speech process is busy."));
373 const request = Promise.withResolvers<Reply>();
374 pending = request;
375 child.send(message, (error) => {
376 if (error) {
377 request.reject(error);
378 if (pending === request) pending = undefined;
379 }
380 });
381 return request.promise;
382 };
383 const dispose = async () => {
384 stop();
385 await exited.promise;
386 };
387 let loaded: Reply;
388 try {
389 loaded = await send({ kind: "load", path, moduleUrl, acceleration });
390 } catch (error) {
391 await dispose();
392 throw error;
393 }
394 return {
395 backend: loaded.backend ?? "unknown",
396 languages: Array.isArray(loaded.languages)
397 ? loaded.languages.filter((entry): entry is string => typeof entry === "string")
398 : [],
399 supportsStreaming: loaded.supportsStreaming === true,
400 supportsLongForm: loaded.supportsLongForm === true,
401 supportsTranslation: loaded.supportsTranslation === true,
402 supportsVocabulary: loaded.supportsVocabulary === true,
403 begin: async (options: NativeSpeechOptions, filterSilence: boolean, vadPath?: string) => {
404 await send({ kind: "begin", options, filterSilence, vadPath });
405 },
406 feed: async (pcm: Float32Array) => {
407 const reply = await send({ kind: "feed", pcm });
408 if (typeof reply.revision !== "number" || reply.preview === undefined)
409 throw new Error("Invalid speech stream response.");
410 return { revision: reply.revision, text: reply.preview };
411 },
412 finish: async () => {
413 const reply = await send({ kind: "finish" });
414 if (typeof reply.text !== "string") throw new Error("Invalid speech stream response.");
415 return reply.text;
416 },
417 reset: async () => {
418 const inFlight = pending?.promise;
419 if (inFlight) {
420 let timeout: ReturnType<typeof setTimeout> | undefined;
421 try {
422 await Promise.race([
423 inFlight.catch(() => undefined),
424 new Promise<never>((_, reject) => {
425 timeout = setTimeout(
426 () => reject(new Error("Speech stream reset timed out.")),
427 1_000,
428 );
429 }),
430 ]);
431 } finally {
432 clearTimeout(timeout);
433 }
434 }
435 await send({ kind: "reset" });
436 },
437 transcribe: async (
438 pcm: Float32Array,
439 options: NativeSpeechOptions,
440 filterSilence: boolean,
441 vadPath?: string,
442 ) => {
443 const result = await send({ kind: "transcribe", pcm, options, filterSilence, vadPath });
444 if (typeof result.text !== "string") throw new Error("Invalid speech process response.");
445 return { text: result.text };
446 },
447 dispose,
448 };
449}
450
451type NativeSpeechOptions = {
452 readonly timestamps: "none";
453 readonly task?: "translate";
454 readonly targetLanguage?: "en";
455 readonly language?: string;
456 readonly vocabulary?: readonly string[];
457};