1import {
2 SPEECH_STREAM_MAX_CHUNK_BYTES,
3 type EnvironmentSpeechModel,
4 type EnvironmentSpeechStatus,
5 type SpeechAcceleration,
6 type SpeechModelUnloadTimeout,
7 type SpeechTranscriptionOptions,
8} from "@t3tools/contracts";
9import * as HostProcess from "@t3tools/shared/HostProcess";
10import * as Cause from "effect/Cause";
11import * as Clock from "effect/Clock";
12import * as Context from "effect/Context";
13import * as Deferred from "effect/Deferred";
14import * as Duration from "effect/Duration";
15import * as Effect from "effect/Effect";
16import * as Exit from "effect/Exit";
17import * as Fiber from "effect/Fiber";
18import * as Layer from "effect/Layer";
19import * as Queue from "effect/Queue";
20import * as Schema from "effect/Schema";
21import * as Scope from "effect/Scope";
22import * as Stream from "effect/Stream";
23
24import * as ServerSettings from "../serverSettings.ts";
25import { makeSpeechTextCorrector, normalizeSpeechCustomWords } from "./customWords.ts";
26import { removeSpeechFillerWords } from "./fillerWords.ts";
27import {
28 DEFAULT_SPEECH_MODEL_ID,
29 effectiveSpeechLanguage,
30 getSpeechModel,
31 SPEECH_MODELS,
32 type SpeechModel,
33} from "./model.ts";
34import { listNativeSpeechGpuDevices, loadNativeSpeechModel } from "./native.ts";
35import { nativeSpeechLanguage } from "./nativeLanguage.ts";
36import { SILERO_VAD } from "./silero.ts";
37import { isSpeechPlatformSupported } from "./platform.ts";
38import * as SpeechModelStore from "./SpeechModelStore.ts";
39
40const SAMPLE_RATE = 16_000;
41const MAX_SPEECH_DURATION_SECONDS = 5 * 60;
42export const MAX_SPEECH_BYTES =
43 SAMPLE_RATE * Float32Array.BYTES_PER_ELEMENT * MAX_SPEECH_DURATION_SECONDS;
44
45export class SpeechInvalidAudioError extends Schema.TaggedError<SpeechInvalidAudioError>()(
46 "SpeechInvalidAudioError",
47 { byteLength: Schema.Number, message: Schema.String },
48) {}
49
50export function decodeSpeechPcm(pcmBytes: Uint8Array): Float32Array {
51 if (
52 pcmBytes.byteLength === 0 ||
53 pcmBytes.byteLength > MAX_SPEECH_BYTES ||
54 pcmBytes.byteLength % Float32Array.BYTES_PER_ELEMENT !== 0
55 ) {
56 throw new SpeechInvalidAudioError({
57 byteLength: pcmBytes.byteLength,
58 message: "audio must contain at most five minutes of 16 kHz mono Float32 PCM",
59 });
60 }
61 // WebSocket frames can be pooled Node Buffers whose backing storage is larger
62 // than the frame. Copy the view so only the received bytes are decoded.
63 const pcm = new Float32Array(Uint8Array.from(pcmBytes).buffer);
64 for (const sample of pcm) {
65 if (!Number.isFinite(sample) || sample < -1 || sample > 1) {
66 throw new SpeechInvalidAudioError({
67 byteLength: pcmBytes.byteLength,
68 message: "audio contains an invalid PCM sample",
69 });
70 }
71 }
72 return pcm;
73}
74
75export class SpeechOperationError extends Schema.TaggedError<SpeechOperationError>()(
76 "SpeechOperationError",
77 { operation: Schema.String, cause: Schema.Defect() },
78) {
79 override get message(): string {
80 return `Environment speech ${this.operation} failed.`;
81 }
82}
83
84export class SpeechUnsupportedPlatformError extends Schema.TaggedError<SpeechUnsupportedPlatformError>()(
85 "SpeechUnsupportedPlatformError",
86 { platform: Schema.String, architecture: Schema.String },
87) {}
88
89export class SpeechBusyError extends Schema.TaggedError<SpeechBusyError>()("SpeechBusyError", {
90 operation: Schema.String,
91}) {}
92
93export class SpeechModelNotFoundError extends Schema.TaggedError<SpeechModelNotFoundError>()(
94 "SpeechModelNotFoundError",
95 { modelId: Schema.String },
96) {}
97
98export class SpeechDownloadCancelledError extends Schema.TaggedError<SpeechDownloadCancelledError>()(
99 "SpeechDownloadCancelledError",
100 { modelId: Schema.String },
101) {}
102
103export type SpeechError =
104 | SpeechOperationError
105 | SpeechInvalidAudioError
106 | SpeechUnsupportedPlatformError
107 | SpeechModelNotFoundError
108 | SpeechDownloadCancelledError
109 | SpeechBusyError;
110
111const isSpeechError = Schema.is(
112 Schema.Union([
113 SpeechInvalidAudioError,
114 SpeechUnsupportedPlatformError,
115 SpeechBusyError,
116 SpeechModelNotFoundError,
117 SpeechDownloadCancelledError,
118 SpeechOperationError,
119 ]),
120);
121
122type LoadedModel = Awaited<ReturnType<typeof loadNativeSpeechModel>>;
123type SpeechModels = { readonly models: ReadonlyArray<EnvironmentSpeechModel> };
124
125export type SpeechStream = {
126 readonly feed: (
127 pcm: Uint8Array,
128 ) => Effect.Effect<Awaited<ReturnType<LoadedModel["feed"]>>, SpeechError>;
129 readonly finish: Effect.Effect<string, SpeechError>;
130};
131
132export class SpeechService extends Context.Service<
133 SpeechService,
134 {
135 readonly status: Effect.Effect<EnvironmentSpeechStatus, SpeechError>;
136 readonly models: (includeBenchmarks: boolean) => Effect.Effect<SpeechModels, SpeechError>;
137 readonly downloadModel: (modelId: string) => Effect.Effect<SpeechModels, SpeechError>;
138 readonly selectModel: (modelId: string) => Effect.Effect<SpeechModels, SpeechError>;
139 readonly cancelDownload: (modelId: string) => Effect.Effect<SpeechModels, SpeechError>;
140 readonly removeModel: (modelId: string) => Effect.Effect<SpeechModels, SpeechError>;
141 /** Uses catalog model files already on this machine, from a file or folder path. */
142 readonly linkModels: (source: string) => Effect.Effect<SpeechModels, SpeechError>;
143 readonly prepareModel: Effect.Effect<void, SpeechError>;
144 readonly transcribe: (
145 pcmBytes: Uint8Array,
146 options: SpeechTranscriptionOptions,
147 ) => Effect.Effect<string, SpeechError>;
148 readonly startStream: (
149 options: SpeechTranscriptionOptions,
150 ) => Effect.Effect<SpeechStream, SpeechError, Scope.Scope>;
151 readonly updateAcceleration: (
152 acceleration: SpeechAcceleration,
153 ) => Effect.Effect<EnvironmentSpeechStatus, SpeechError>;
154 readonly updateModelUnloadTimeout: (
155 timeout: SpeechModelUnloadTimeout,
156 ) => Effect.Effect<EnvironmentSpeechStatus, SpeechError>;
157 }
158>()("t3/speech/SpeechService") {}
159
160const UNLOAD_MILLISECONDS: Record<Exclude<SpeechModelUnloadTimeout, "never">, number> = {
161 immediately: 0,
162 min_2: 2 * 60_000,
163 min_5: 5 * 60_000,
164 min_10: 10 * 60_000,
165 min_15: 15 * 60_000,
166 hour_1: 60 * 60_000,
167};
168
169const make = Effect.gen(function* () {
170 const platform = yield* HostProcess.Platform;
171 const architecture = yield* HostProcess.Architecture;
172 const serverSettings = yield* ServerSettings.ServerSettingsService;
173 const store = yield* SpeechModelStore.SpeechModelStore;
174 const scope = yield* Effect.scope;
175 const clock = yield* Clock.Clock;
176 const now = () => clock.currentTimeMillisUnsafe();
177 const libc = platform === "linux" ? yield* HostProcess.LinuxLibc : undefined;
178 const unsupportedReason = isSpeechPlatformSupported(platform, architecture, libc)
179 ? null
180 : `voice transcription is not available on ${platform}-${architecture}`;
181
182 let model: LoadedModel | undefined;
183 let loadedModelId: string | undefined;
184 let loadedAcceleration: string | undefined;
185 let autoCpuModel: string | undefined;
186 let loading: Promise<LoadedModel> | undefined;
187 const discoverGpuDevices = yield* Effect.tryPromise({
188 try: () => listNativeSpeechGpuDevices(),
189 catch: (cause) => new SpeechOperationError({ operation: "discover GPU devices", cause }),
190 }).pipe(Effect.cachedWithTTL((exit) => (Exit.isSuccess(exit) ? Duration.infinity : "1 minute")));
191 let downloading:
192 | {
193 modelId: string;
194 downloaded: number;
195 verifying: boolean;
196 fiber: Deferred.Deferred<Fiber.Fiber<string, SpeechError>>;
197 }
198 | undefined;
199 let activeTranscriptions = 0;
200 // One native request at a time; the slot is a promise so unload waits for it.
201 let activeOperation: Promise<unknown> | undefined;
202 const operationBusy = () => Boolean(activeOperation || downloading);
203 let closing = false;
204 // @effect-diagnostics-next-line abortControllerInEffect:off - Cached native workers must also stop on service shutdown.
205 const lifetime = new AbortController();
206 let unloadTimeout: SpeechModelUnloadTimeout = "min_15";
207
208 // Deadlines are absolute: the consumer fiber registers its sleep a few scheduler
209 // hops after the request, so a relative delay would drift.
210 const unloadRequests = yield* Queue.sliding<{ deadline: number } | null>(1);
211 const clearUnloadTimer = () => Queue.offerUnsafe(unloadRequests, null);
212 const scheduleUnload = (minimumDelayMs = 0) => {
213 clearUnloadTimer();
214 if (!model || closing || unloadTimeout === "never") return;
215 Queue.offerUnsafe(unloadRequests, {
216 deadline: now() + Math.max(minimumDelayMs, UNLOAD_MILLISECONDS[unloadTimeout]),
217 });
218 };
219
220 const forgetModel = (target: LoadedModel) => {
221 if (model !== target) return;
222 model = undefined;
223 loadedModelId = undefined;
224 loadedAcceleration = undefined;
225 };
226 const disposeModel = async () => {
227 const target = model;
228 if (!target) return;
229 forgetModel(target);
230 await target.dispose();
231 };
232 // Stops the isolated process without waiting for native work to settle.
233 const dropModel = (target: LoadedModel) => {
234 forgetModel(target);
235 void Promise.resolve()
236 .then(() => target.dispose())
237 .catch(() => undefined);
238 };
239
240 // Auto placement may land on a GPU the native engine cannot actually run; one retry on
241 // this host's CPU recovers it. A GPU the person chose, and CPU itself, report the error.
242 const canRetryOnCpu = (acceleration: SpeechAcceleration, loaded: LoadedModel) =>
243 acceleration === "auto" && loaded.backend.toLowerCase() !== "cpu";
244
245 yield* Effect.addFinalizer(() =>
246 Effect.promise(async () => {
247 closing = true;
248 clearUnloadTimer();
249 lifetime.abort();
250 await disposeModel();
251 }),
252 );
253
254 yield* Stream.fromQueue(unloadRequests).pipe(
255 Stream.switchMap((request) =>
256 request === null
257 ? Stream.empty
258 : Stream.fromEffect(
259 Effect.gen(function* () {
260 yield* Effect.sleep(Math.max(0, request.deadline - now()));
261 if (closing) return;
262 if (operationBusy() || activeTranscriptions || loading) {
263 scheduleUnload(1000);
264 return;
265 }
266 yield* Effect.promise(() => disposeModel().catch(() => undefined)).pipe(
267 Effect.uninterruptible,
268 );
269 }),
270 ),
271 ),
272 Stream.runDrain,
273 Effect.forkScoped,
274 );
275
276 type SettingsSnapshot = Effect.Success<
277 ServerSettings.ServerSettingsService["Service"]["getSettings"]
278 >;
279
280 const selectedModel = (settings: SettingsSnapshot): SpeechModel =>
281 getSpeechModel(settings.speechModelId) ?? getSpeechModel(DEFAULT_SPEECH_MODEL_ID)!;
282
283 let silenceRetryAt = 0;
284 const acquireSilenceModel = Effect.fn("SpeechService.acquireSilenceModel")(function* (
285 purpose: "installation" | "recording",
286 ) {
287 if (purpose === "recording" && now() < silenceRetryAt) return undefined;
288 const path = yield* store.download(SILERO_VAD).pipe(
289 Effect.timeout(purpose === "recording" ? "15 seconds" : "2 minutes"),
290 Effect.orElseSucceed(() => undefined),
291 );
292 if (path !== undefined) {
293 silenceRetryAt = 0;
294 return path;
295 }
296 silenceRetryAt = now() + 10 * 60_000;
297 yield* Effect.logWarning("Silero unavailable; using raw audio with bounded ASR processing.");
298 });
299
300 // Publish the reservation before forking: a cached transfer can finish before forkIn returns.
301 const downloadFiber = Effect.fn("SpeechService.downloadFiber")(function* (modelId: string) {
302 const definition = getSpeechModel(modelId);
303 if (!definition) return yield* new SpeechModelNotFoundError({ modelId });
304 if (closing || activeOperation || (downloading && downloading.modelId !== modelId))
305 return yield* new SpeechBusyError({ operation: "model download" });
306 if (downloading) return yield* Deferred.await(downloading.fiber);
307 const transfer = {
308 modelId,
309 downloaded: 0,
310 verifying: false,
311 fiber: Deferred.makeUnsafe<Fiber.Fiber<string, SpeechError>>(),
312 };
313 downloading = transfer;
314 const fiber = yield* store
315 .download(definition, {
316 onProgress: (downloaded) => {
317 transfer.downloaded = downloaded;
318 transfer.verifying = false;
319 },
320 onVerificationStart: () => {
321 transfer.verifying = true;
322 },
323 })
324 .pipe(
325 (modelDownload) =>
326 Effect.all([modelDownload, acquireSilenceModel("installation")], { concurrency: 2 }),
327 Effect.map(([modelPath]) => modelPath),
328 Effect.mapError((cause) => speechError("model download", cause)),
329 Effect.forkIn(scope),
330 );
331 fiber.addObserver(() => {
332 if (downloading !== transfer) return;
333 downloading = undefined;
334 scheduleUnload();
335 });
336 yield* Deferred.succeed(transfer.fiber, fiber);
337 return fiber;
338 }, Effect.uninterruptible);
339
340 const download = Effect.fn("SpeechService.download")(function* (modelId: string) {
341 const exit = yield* Fiber.await(yield* downloadFiber(modelId));
342 if (Exit.isSuccess(exit)) return exit.value;
343 return yield* Cause.hasInterruptsOnly(exit.cause)
344 ? new SpeechDownloadCancelledError({ modelId })
345 : Effect.failCause(exit.cause);
346 });
347
348 const installedModelPath = (definition: SpeechModel) =>
349 store.installedPath(definition).pipe(
350 Effect.filterOrFail(
351 (modelPath) => modelPath !== undefined,
352 () => new SpeechModelNotFoundError({ modelId: definition.id }),
353 ),
354 );
355
356 const loadModel = async (
357 modelPath: string,
358 definition: SpeechModel,
359 signal = lifetime.signal,
360 requestedAcceleration = "auto",
361 ) => {
362 const acceleration =
363 requestedAcceleration === "auto" && autoCpuModel === definition.id
364 ? "cpu"
365 : requestedAcceleration;
366 if (model && loadedModelId === definition.id && loadedAcceleration === acceleration) {
367 scheduleUnload();
368 return model;
369 }
370 if (loading) return loading;
371 await disposeModel();
372 signal.throwIfAborted();
373 const pending = loadNativeSpeechModel(modelPath, signal, undefined, acceleration).then(
374 async (loaded) => {
375 if (closing || signal.aborted) {
376 await loaded.dispose();
377 signal.throwIfAborted();
378 throw new SpeechBusyError({ operation: "model preparation" });
379 }
380 model = loaded;
381 loadedModelId = definition.id;
382 loadedAcceleration = acceleration;
383 scheduleUnload();
384 return loaded;
385 },
386 );
387 const release = () => {
388 if (loading === pending) loading = undefined;
389 };
390 loading = pending;
391 signal.addEventListener("abort", release, { once: true });
392 try {
393 return await pending;
394 } finally {
395 signal.removeEventListener("abort", release);
396 release();
397 }
398 };
399
400 const speechError = (operation: string, cause: unknown): SpeechError =>
401 isSpeechError(cause) ? cause : new SpeechOperationError({ operation, cause });
402
403 const attempt = <A>(operation: string, run: (signal: AbortSignal) => Promise<A>) =>
404 Effect.tryPromise({ try: run, catch: (cause) => speechError(operation, cause) });
405
406 // The busy slot is released by a finalizer, so interrupting the fiber frees it at
407 // once instead of waiting for native work that may never settle.
408 const exclusive = <A, E, R>(operation: string, effect: Effect.Effect<A, E, R>) =>
409 Effect.uninterruptibleMask((restore) =>
410 Effect.suspend<A, E | SpeechBusyError, R>(() => {
411 if (closing || operationBusy()) return Effect.fail(new SpeechBusyError({ operation }));
412 const gate = Promise.withResolvers<void>();
413 activeOperation = gate.promise;
414 return restore(effect).pipe(
415 Effect.ensuring(
416 Effect.sync(() => {
417 if (activeOperation === gate.promise) activeOperation = undefined;
418 scheduleUnload(
419 operation === "model preparation" ? MAX_SPEECH_DURATION_SECONDS * 1000 : 0,
420 );
421 gate.resolve();
422 }),
423 ),
424 );
425 }),
426 );
427
428 const readSettings = (operation: string) =>
429 serverSettings.getSettings.pipe(
430 Effect.tap((settings) =>
431 Effect.sync(() => {
432 if (settings.speechModelId !== autoCpuModel || settings.speechAcceleration !== "auto")
433 autoCpuModel = undefined;
434 if (unloadTimeout !== settings.speechModelUnloadTimeout) {
435 unloadTimeout = settings.speechModelUnloadTimeout;
436 scheduleUnload();
437 }
438 }),
439 ),
440 Effect.mapError((cause) => new SpeechOperationError({ operation, cause })),
441 );
442
443 const writeSettings = (
444 operation: string,
445 patch: Parameters<ServerSettings.ServerSettingsService["Service"]["updateSettings"]>[0],
446 ) =>
447 serverSettings
448 .updateSettings(patch)
449 .pipe(Effect.mapError((cause) => new SpeechOperationError({ operation, cause })));
450
451 const currentStatus = Effect.fn("SpeechService.currentStatus")(function* (
452 settings: SettingsSnapshot,
453 ): Effect.fn.Return<EnvironmentSpeechStatus> {
454 if (unsupportedReason) return { supported: false, reason: unsupportedReason };
455 const definition = selectedModel(settings);
456 const ready = yield* store.isReady(definition);
457 const gpuDevices = yield* discoverGpuDevices.pipe(Effect.orElseSucceed(() => []));
458 const autoUsesCpu =
459 gpuDevices.length === 0 ||
460 autoCpuModel === definition.id ||
461 (loadedModelId === definition.id &&
462 loadedAcceleration === "auto" &&
463 model?.backend.toLowerCase() === "cpu");
464 return {
465 supported: true,
466 state: activeTranscriptions > 0 ? "transcribing" : ready ? "ready" : "missing-model",
467 modelId: definition.id,
468 model: definition.name,
469 size: definition.size,
470 supportsStreaming: definition.supportsStreaming,
471 supportsBufferedStreaming: true,
472 supportsModelLinking: true,
473 supportsTranslation: definition.supportsTranslation,
474 languages: definition.languages,
475 supportsLanguageDetection: definition.supportsLanguageDetection,
476 acceleration: settings.speechAcceleration,
477 benchmarkTier:
478 settings.speechAcceleration === "cpu" ||
479 (settings.speechAcceleration === "auto" && autoUsesCpu)
480 ? "cpu"
481 : "accelerated",
482 modelUnloadTimeout: settings.speechModelUnloadTimeout,
483 gpuDevices,
484 };
485 });
486
487 const listModels = Effect.fn("SpeechService.listModels")(function* (
488 settings: SettingsSnapshot,
489 includeBenchmarks: boolean,
490 ): Effect.fn.Return<SpeechModels> {
491 const selected = selectedModel(settings);
492 const models = yield* Effect.forEach(
493 SPEECH_MODELS,
494 (definition) =>
495 Effect.all([store.isReady(definition), store.linkedFrom(definition)]).pipe(
496 Effect.map(([ready, linkedFrom]): EnvironmentSpeechModel => {
497 const operation = downloading?.modelId === definition.id ? downloading : undefined;
498 return {
499 id: definition.id,
500 name: definition.name,
501 description: definition.description,
502 size: definition.size,
503 languages: definition.languages,
504 accuracy: definition.accuracy,
505 speed: definition.speed,
506 ...(includeBenchmarks ? { benchmark: definition.benchmark } : {}),
507 recommended: definition.recommended,
508 supportsStreaming: definition.supportsStreaming,
509 supportsLanguageDetection: definition.supportsLanguageDetection,
510 supportsTranslation: definition.supportsTranslation,
511 active: ready && selected.id === definition.id,
512 state: operation
513 ? operation.verifying
514 ? "verifying"
515 : "downloading"
516 : ready
517 ? "installed"
518 : "downloadable",
519 ...(operation ? { downloaded: operation.downloaded } : {}),
520 ...(ready && linkedFrom ? { linkedFrom } : {}),
521 };
522 }),
523 ),
524 { concurrency: "unbounded" },
525 );
526 return { models };
527 });
528
529 const freshStatus = (operation: string) =>
530 readSettings(operation).pipe(Effect.flatMap(currentStatus));
531 const readModels = (includeBenchmarks: boolean) =>
532 readSettings("model listing").pipe(
533 Effect.flatMap((settings) => listModels(settings, includeBenchmarks)),
534 );
535 const modelsEffect = readModels(false);
536
537 const requireSupported = Effect.suspend(() =>
538 unsupportedReason
539 ? Effect.fail(new SpeechUnsupportedPlatformError({ platform, architecture }))
540 : Effect.void,
541 );
542
543 const finalizeTranscript = (
544 text: string,
545 options: SpeechTranscriptionOptions,
546 language: string | undefined,
547 ) => {
548 return (
549 options.speechRemoveFillerWords
550 ? removeSpeechFillerWords(text, language, options.speechCorrectionWord)
551 : text
552 ).trim();
553 };
554
555 const prepareModel = readSettings("model preparation").pipe(
556 Effect.flatMap((settings) =>
557 exclusive(
558 "model preparation",
559 requireSupported.pipe(
560 Effect.andThen(installedModelPath(selectedModel(settings))),
561 Effect.flatMap((modelPath) =>
562 attempt("model preparation", (signal) =>
563 loadModel(
564 modelPath,
565 selectedModel(settings),
566 AbortSignal.any([signal, lifetime.signal]),
567 settings.speechAcceleration,
568 ),
569 ).pipe(
570 Effect.timeout("2 minutes"),
571 Effect.catchTags({
572 TimeoutError: (cause) =>
573 Effect.fail(new SpeechOperationError({ operation: "model preparation", cause })),
574 }),
575 ),
576 ),
577 ),
578 ),
579 ),
580 );
581
582 const silenceModelPath = Effect.fn("SpeechService.silenceModelPath")(function* (
583 loaded: LoadedModel,
584 options: SpeechTranscriptionOptions,
585 ) {
586 if (!options.speechSilenceFiltering && (loaded.supportsStreaming || loaded.supportsLongForm))
587 return undefined;
588 return yield* acquireSilenceModel("recording");
589 });
590
591 const transcribe: SpeechService["Service"]["transcribe"] = Effect.fn("SpeechService.transcribe")(
592 function* (pcmBytes, options) {
593 const settings = yield* readSettings("transcription");
594 return yield* exclusive(
595 "transcription",
596 Effect.gen(function* () {
597 yield* requireSupported;
598 const pcm = yield* Effect.try({
599 try: () => decodeSpeechPcm(pcmBytes),
600 catch: (cause) => speechError("transcription", cause),
601 });
602 if (pcm.length === 0) return "";
603 activeTranscriptions += 1;
604 return yield* Effect.gen(function* () {
605 const definition = selectedModel(settings);
606 const language = effectiveSpeechLanguage(definition, options.speechLanguage);
607 const translateToEnglish =
608 options.speechTranslateToEnglish &&
609 definition.supportsTranslation &&
610 language !== "en";
611 const modelPath = yield* installedModelPath(definition);
612 const loaded = yield* Effect.tryPromise({
613 try: (signal) =>
614 loadModel(
615 modelPath,
616 definition,
617 AbortSignal.any([signal, lifetime.signal]),
618 settings.speechAcceleration,
619 ),
620 catch: (cause) => speechError("model preparation", cause),
621 });
622 const vadPath = yield* silenceModelPath(loaded, options);
623 const dictionary = normalizeSpeechCustomWords(options.speechCustomWords);
624 const customWords = dictionary.map(({ term }) => term);
625 // The vocabulary capability guarantees plain transcription, not translation.
626 const vocabularyFor = (inference: LoadedModel) =>
627 inference.supportsVocabulary && !translateToEnglish ? customWords : [];
628 const transcribeWith = (inference: LoadedModel) =>
629 Effect.tryPromise({
630 try: (signal) => {
631 // Interruption kills the isolated process; the next request reloads the model.
632 signal.addEventListener("abort", () => dropModel(inference), { once: true });
633 const hint = nativeSpeechLanguage(inference.languages, language);
634 return inference.transcribe(
635 pcm,
636 {
637 timestamps: "none",
638 ...(hint ? { language: hint } : {}),
639 ...(translateToEnglish
640 ? { task: "translate" as const, targetLanguage: "en" as const }
641 : {}),
642 ...(vocabularyFor(inference).length > 0 ? { vocabulary: customWords } : {}),
643 },
644 options.speechSilenceFiltering,
645 vadPath,
646 );
647 },
648 catch: (cause) => {
649 dropModel(inference);
650 return speechError("inference", cause);
651 },
652 });
653 let inference = loaded;
654 const result = yield* transcribeWith(loaded).pipe(
655 Effect.catch((error) => {
656 if (!canRetryOnCpu(settings.speechAcceleration, loaded)) return Effect.fail(error);
657 return Effect.tryPromise({
658 try: (signal) =>
659 loadModel(
660 modelPath,
661 definition,
662 AbortSignal.any([signal, lifetime.signal]),
663 "cpu",
664 ),
665 catch: (cause) => speechError("model preparation", cause),
666 }).pipe(
667 Effect.flatMap((cpu) => {
668 autoCpuModel = definition.id;
669 inference = cpu;
670 return transcribeWith(cpu);
671 }),
672 );
673 }),
674 );
675 const correct = makeSpeechTextCorrector(
676 vocabularyFor(inference).length > 0 ? [] : customWords,
677 dictionary,
678 );
679 return finalizeTranscript(
680 correct(result.text),
681 options,
682 translateToEnglish ? "en" : language === "auto" ? undefined : language,
683 );
684 }).pipe(
685 Effect.ensuring(
686 Effect.sync(() => {
687 activeTranscriptions -= 1;
688 }),
689 ),
690 );
691 }),
692 );
693 },
694 );
695
696 // Stream setup shares the public startStream span.
697 const openStream = Effect.fnUntraced(function* (options: SpeechTranscriptionOptions) {
698 const operation = "streaming transcription";
699 yield* requireSupported;
700 const released = Promise.withResolvers<void>();
701 // @effect-diagnostics-next-line abortControllerInEffect:off - Stream cancellation must abort native work independently of the caller fiber.
702 const controller = new AbortController();
703 const signal = AbortSignal.any([controller.signal, lifetime.signal]);
704 let finished = false;
705 let closed = false;
706 let started = false;
707 let loaded: LoadedModel | undefined;
708 yield* Effect.acquireRelease(
709 Effect.suspend(() => {
710 if (closing || operationBusy()) return Effect.fail(new SpeechBusyError({ operation }));
711 activeOperation = released.promise;
712 activeTranscriptions += 1;
713 return Effect.void;
714 }),
715 () =>
716 Effect.sync(() => {
717 activeTranscriptions -= 1;
718 if (activeOperation === released.promise) activeOperation = undefined;
719 scheduleUnload();
720 released.resolve();
721 }),
722 );
723 yield* Effect.addFinalizer(() =>
724 Effect.gen(function* () {
725 closed = true;
726 if (!finished) {
727 const dispose = Effect.promise(async () => {
728 const pending = loading;
729 controller.abort();
730 await pending?.catch(() => undefined);
731 if (loaded) {
732 forgetModel(loaded);
733 await loaded.dispose();
734 }
735 });
736 if (loaded && started) {
737 yield* attempt("reset stream", () => loaded!.reset()).pipe(
738 Effect.timeout("2500 millis"),
739 Effect.catch(() => dispose),
740 );
741 } else {
742 yield* dispose;
743 }
744 }
745 }),
746 );
747 const settings = yield* readSettings(operation);
748 const definition = selectedModel(settings);
749 const language = effectiveSpeechLanguage(definition, options.speechLanguage);
750 const dictionary = normalizeSpeechCustomWords(options.speechCustomWords);
751 const customWords = dictionary.map(({ term }) => term);
752 let correct = makeSpeechTextCorrector(customWords, dictionary);
753 const modelPath = yield* installedModelPath(definition);
754 let streamModel = yield* attempt(operation, () =>
755 loadModel(modelPath, definition, signal, settings.speechAcceleration),
756 );
757 loaded = streamModel;
758 const vadPath = yield* silenceModelPath(streamModel, options);
759 let translateToEnglish = false;
760 const beginStream = async () => {
761 translateToEnglish =
762 options.speechTranslateToEnglish &&
763 !streamModel.supportsStreaming &&
764 streamModel.supportsTranslation &&
765 language !== "en";
766 const vocabulary =
767 streamModel.supportsVocabulary && !translateToEnglish && customWords.length
768 ? customWords
769 : undefined;
770 const hint = nativeSpeechLanguage(streamModel.languages, language);
771 await streamModel.begin(
772 {
773 timestamps: "none",
774 ...(hint ? { language: hint } : {}),
775 ...(translateToEnglish ? { task: "translate", targetLanguage: "en" } : {}),
776 ...(vocabulary ? { vocabulary } : {}),
777 },
778 options.speechSilenceFiltering,
779 vadPath,
780 );
781 correct = makeSpeechTextCorrector(vocabulary ? [] : customWords, dictionary);
782 started = true;
783 };
784 // Audio already fed is kept so a CPU restart can replay it; the limit caps it at one recording.
785 const received: Float32Array[] = [];
786 let retriedOnCpu = false;
787 const recoverOnCpu = async (cause: unknown) => {
788 started = false;
789 if (closed || signal.aborted || retriedOnCpu) throw cause;
790 if (!canRetryOnCpu(settings.speechAcceleration, streamModel)) throw cause;
791 retriedOnCpu = true;
792 forgetModel(streamModel);
793 await streamModel.dispose();
794 if (closed || signal.aborted) throw cause;
795 streamModel = await loadModel(modelPath, definition, signal, "cpu");
796 autoCpuModel = definition.id;
797 loaded = streamModel;
798 await beginStream();
799 let update: Awaited<ReturnType<LoadedModel["feed"]>> | undefined;
800 for (const chunk of received) {
801 if (closed || signal.aborted) throw new Error("Speech stream is closed.");
802 update = await streamModel.feed(chunk);
803 }
804 return update;
805 };
806 yield* attempt(operation, async () => {
807 try {
808 await beginStream();
809 } catch (cause) {
810 await recoverOnCpu(cause);
811 }
812 });
813 let byteLength = 0;
814 let busy = false;
815 // Client revisions stay monotonic across a native restart, whose own counter resets.
816 let revisionBase = 0;
817 let lastNativeRevision = 0;
818 const run = <A>(work: () => Promise<A>) =>
819 attempt(operation, async () => {
820 if (busy || finished || closed || !started || signal.aborted)
821 throw new Error("Speech stream is not ready.");
822 busy = true;
823 try {
824 return await work();
825 } finally {
826 busy = false;
827 }
828 });
829 return {
830 feed: (bytes: Uint8Array) =>
831 run(async () => {
832 if (
833 bytes.byteLength > SPEECH_STREAM_MAX_CHUNK_BYTES ||
834 byteLength + bytes.byteLength > MAX_SPEECH_BYTES
835 )
836 throw new SpeechInvalidAudioError({
837 byteLength: bytes.byteLength,
838 message: "Speech stream audio limit exceeded.",
839 });
840 const pcm = decodeSpeechPcm(bytes);
841 byteLength += bytes.byteLength;
842 received.push(pcm);
843 let update: Awaited<ReturnType<LoadedModel["feed"]>> | undefined;
844 try {
845 update = await streamModel.feed(pcm);
846 } catch (cause) {
847 update = await recoverOnCpu(cause);
848 revisionBase += lastNativeRevision;
849 lastNativeRevision = 0;
850 }
851 if (update) lastNativeRevision = update.revision;
852 return {
853 revision: revisionBase + (update?.revision ?? lastNativeRevision),
854 text: update?.text
855 ? {
856 committed: correct(update.text.committed),
857 tentative: correct(update.text.tentative),
858 }
859 : null,
860 };
861 }),
862 finish: run(async () => {
863 let text: string;
864 try {
865 text = await streamModel.finish().catch(async (cause: unknown) => {
866 await recoverOnCpu(cause);
867 if (closed || signal.aborted) throw cause;
868 return streamModel.finish();
869 });
870 } catch (cause) {
871 started = false;
872 throw cause;
873 }
874 finished = true;
875 return finalizeTranscript(
876 correct(text),
877 options,
878 translateToEnglish ? "en" : language === "auto" ? undefined : language,
879 );
880 }),
881 };
882 });
883
884 return SpeechService.of({
885 status: freshStatus("status"),
886 models: readModels,
887 prepareModel,
888 transcribe,
889 startStream: Effect.fn("SpeechService.startStream")(function* (options) {
890 const session = yield* Scope.fork(yield* Effect.scope);
891 return yield* openStream(options).pipe(
892 Scope.provide(session),
893 Effect.onError((cause) => Scope.close(session, Exit.failCause(cause))),
894 );
895 }),
896 downloadModel: Effect.fn("SpeechService.downloadModel")((modelId) =>
897 requireSupported.pipe(
898 Effect.andThen(download(modelId)),
899 Effect.catchTags({ SpeechDownloadCancelledError: () => Effect.void }),
900 Effect.andThen(modelsEffect),
901 ),
902 ),
903 selectModel: Effect.fn("SpeechService.selectModel")((modelId) =>
904 exclusive(
905 "model selection",
906 attempt("model selection", async () => {
907 if (!getSpeechModel(modelId)) throw new SpeechModelNotFoundError({ modelId });
908 await disposeModel();
909 }),
910 ).pipe(
911 Effect.andThen(writeSettings("model selection", { speechModelId: modelId })),
912 Effect.andThen(modelsEffect),
913 ),
914 ),
915 cancelDownload: Effect.fn("SpeechService.cancelDownload")((modelId) =>
916 Effect.suspend(() =>
917 downloading?.modelId === modelId
918 ? Deferred.await(downloading.fiber).pipe(Effect.flatMap(Fiber.interrupt))
919 : Effect.void,
920 ).pipe(Effect.andThen(modelsEffect)),
921 ),
922 removeModel: Effect.fn("SpeechService.removeModel")((modelId) =>
923 readSettings("model removal").pipe(
924 Effect.flatMap((settings) =>
925 exclusive(
926 "model removal",
927 Effect.gen(function* () {
928 const definition = getSpeechModel(modelId);
929 if (!definition) return yield* new SpeechModelNotFoundError({ modelId });
930 yield* attempt("model removal", async () => {
931 await loading?.catch(() => undefined);
932 if (loadedModelId === modelId) await disposeModel();
933 });
934 yield* store
935 .remove(definition)
936 .pipe(Effect.mapError((cause) => speechError("model removal", cause)));
937 if (selectedModel(settings).id !== modelId) return undefined;
938 const candidates = yield* Effect.forEach(
939 SPEECH_MODELS.filter((candidate) => candidate.id !== modelId),
940 (candidate) =>
941 store.isReady(candidate).pipe(Effect.map((ready) => ({ candidate, ready }))),
942 { concurrency: "unbounded" },
943 );
944 return candidates.find(({ ready }) => ready)?.candidate.id;
945 }),
946 ).pipe(
947 Effect.flatMap((replacementId) =>
948 replacementId
949 ? writeSettings("model removal", { speechModelId: replacementId }).pipe(
950 Effect.andThen(modelsEffect),
951 )
952 : modelsEffect,
953 ),
954 ),
955 ),
956 ),
957 ),
958 linkModels: Effect.fn("SpeechService.linkModels")(function* (source) {
959 yield* requireSupported;
960 const settings = yield* readSettings("model linking");
961 const linked = yield* exclusive(
962 "model linking",
963 store
964 .link(SPEECH_MODELS, source)
965 .pipe(Effect.mapError((cause) => speechError("model linking", cause))),
966 );
967 const first = linked[0];
968 if (first === undefined) return yield* new SpeechModelNotFoundError({ modelId: source });
969 if (!(yield* store.isReady(selectedModel(settings))))
970 yield* writeSettings("model linking", { speechModelId: first });
971 return yield* modelsEffect;
972 }),
973 updateAcceleration: Effect.fn("SpeechService.updateAcceleration")((acceleration) =>
974 writeSettings("acceleration update", { speechAcceleration: acceleration }).pipe(
975 Effect.andThen(freshStatus("acceleration update")),
976 ),
977 ),
978 updateModelUnloadTimeout: Effect.fn("SpeechService.updateModelUnloadTimeout")((timeout) =>
979 writeSettings("model unload timeout update", { speechModelUnloadTimeout: timeout }).pipe(
980 Effect.andThen(freshStatus("model unload timeout update")),
981 ),
982 ),
983 });
984});
985
986export const layer = Layer.effect(SpeechService, make);