feat(voice): first-class voice input and local TTS across web, desktop, and mobile (#2018)
Complete rebuild of voice input on a server-authoritative streaming architecture, replacing the legacy Web Speech / whole-blob / WASM engines and the dead voice-agent layer (~4k lines removed). Speech-to-text (dictation): - Client streams 16 kHz mono PCM16 chunks over /api/dictation/ws with seq/ack ordering; buffered audio is retained and replayed on reconnect - Server transcribes and streams live partial transcripts back; segments auto-commit every ~15s with silence suppression and adaptive finalization timeouts - Local provider (default, zero config): sherpa-onnx models in a forked worker process — auto-download with progress, staged extraction with verification, corrupt-model auto-recovery, idle shutdown after 5 min - Model catalog with settings picker (accuracy/speed ratings, sizes, download/delete): Parakeet TDT v2 (English) and v3 (25 European languages, auto-detected), Whisper base and tiny (multilingual, light) - OpenAI-compatible provider for any Whisper endpoint - Composer overlay with live transcript, volume meter, timer, and cancel / insert / insert-and-send actions; failed transcriptions keep their audio for retry or accepting the partial text as-is - Configurable keyboard shortcut (default mod+alt+v) toggles dictation; Enter confirms and Escape cancels while recording - Overlay is pixel-aligned with the composer (measured footer height, matching paddings/typography/gaps) — no layout shift when toggling Text-to-speech: - Local Kokoro provider (English, 11 voices) synthesized in the same worker via /api/dictation/tts/speak, managed by the shared model pipeline; sentence-pipelined playback keeps time-to-first-audio at ~1 sentence regardless of message length, and stop cancels in-flight synthesis - Sanitizer keeps inline-code content (strips backticks only), reads interword slashes aloud, and removes only absolute file paths Settings: - Voice page unified: a single read-aloud toggle owns all playback options (the confusing "Enable Voice Mode" is gone); a new "Enable voice input" toggle (default on, persisted to settings.json) hides the composer mic entirely when disabled Mobile and transport: - iOS/Android microphone permissions added (dictation was previously impossible on mobile) - Fixed Android WebSocket upgrades: the Capacitor WebView origin (https://localhost) was missing from the packaged-client allowlist, 403-ing every WS connection — root cause of the old mobile SSE lock, which is now removed for all transports Security and conventions: - All HTTP routes sit behind the global /api auth gate; the WS upgrade explicitly validates the UI session and origin, with oc_url_token narrowly allowlisted and covered by tests; the dictation socket mints a fresh URL token before connecting - Routes register before the generic OpenCode proxy; the client goes through runtimeFetch/getRuntimeUrlResolver, and runtime switches reset the dictation socket - VS Code deliberately reports dictation as unavailable (no server process in that runtime) CI: workflow Node bumped 20 -> 22 to match the repo engines and fix better-sqlite3 installs broken by node-gyp@latest on Node 20. New dependency: sherpa-onnx-node (prebuilt N-API; macOS/Linux x64+arm64, Windows x64 — Windows-on-ARM falls back to the OpenAI-compatible provider)
This commit is contained in:
Bohdan Triapitsyn
committed by
GitHub
GitHub
parent
3f5151d424
commit
de1b85ac56
@@ -38,6 +38,7 @@ import { prepareNotificationLastMessage } from './lib/notifications/index.js';
|
||||
import { registerTtsRoutes } from './lib/tts/routes.js';
|
||||
import { detectSayTtsCapability } from './lib/tts/capability-runtime.js';
|
||||
import { createTerminalRuntime } from './lib/terminal/runtime.js';
|
||||
import { createDictationRuntime } from './lib/dictation/runtime.js';
|
||||
import {
|
||||
createGlobalUiEventBroadcaster,
|
||||
createGlobalMessageStreamHub,
|
||||
@@ -495,6 +496,7 @@ const tunnelAuthController = createTunnelAuth();
|
||||
let runtimeManagedRemoteTunnelToken = '';
|
||||
let runtimeManagedRemoteTunnelHostname = '';
|
||||
let terminalRuntime = null;
|
||||
let dictationRuntime = null;
|
||||
let messageStreamRuntime = null;
|
||||
const userProvidedOpenCodePassword = hmrStateRuntime.getUserProvidedOpenCodePassword(hmrState);
|
||||
const initialOpenCodeAuthState = hmrStateRuntime.resolveOpenCodeAuthFromState({
|
||||
@@ -898,6 +900,7 @@ const tunnelWiringRuntime = createTunnelWiringRuntime({
|
||||
});
|
||||
const startupPipelineRuntime = createStartupPipelineRuntime({
|
||||
createTerminalRuntime,
|
||||
createDictationRuntime,
|
||||
createMessageStreamWsRuntime,
|
||||
createServerStartupRuntime,
|
||||
});
|
||||
@@ -1357,8 +1360,10 @@ async function main(options = {}) {
|
||||
tunnelRuntimeContext,
|
||||
attachSignals,
|
||||
apiOnly,
|
||||
dictationModelsDir: path.join(OPENCHAMBER_USER_CONFIG_ROOT, 'speech-models'),
|
||||
});
|
||||
terminalRuntime = startupPipelineResult.terminalRuntime;
|
||||
dictationRuntime = startupPipelineResult.dictationRuntime;
|
||||
messageStreamRuntime = startupPipelineResult.messageStreamRuntime;
|
||||
|
||||
try {
|
||||
@@ -1397,6 +1402,11 @@ async function main(options = {}) {
|
||||
},
|
||||
stop: (shutdownOptions = {}) => {
|
||||
realtimeProxyRuntime.stop();
|
||||
try {
|
||||
dictationRuntime?.stop?.();
|
||||
} catch {
|
||||
// best-effort shutdown of the dictation worker
|
||||
}
|
||||
return gracefulShutdown({ exitProcess: shutdownOptions.exitProcess ?? false });
|
||||
}
|
||||
};
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
# Dictation module
|
||||
|
||||
Server-authoritative streaming speech-to-text for the chat composer, plus
|
||||
local text-to-speech. The client streams 16 kHz mono PCM16 chunks (base64)
|
||||
over a WebSocket; the server runs the transcription and streams live partial
|
||||
transcripts back.
|
||||
|
||||
Local TTS (Kokoro via sherpa-onnx OfflineTts) runs in the same worker process
|
||||
and is exposed as `POST /api/dictation/tts/speak` (JSON `{text, speakerId?,
|
||||
speed?, model?}` → WAV bytes; 503 with `reasonCode` while the model is
|
||||
downloading). TTS models live in the same catalog/downloader as STT models
|
||||
(`local/model-catalog.js` `LOCAL_TTS_MODEL_CATALOG`) and are managed by the
|
||||
same status/download/delete routes.
|
||||
|
||||
## Ownership
|
||||
|
||||
- `runtime.js` — registers `GET /api/dictation/status`,
|
||||
`POST /api/dictation/models/:modelId/download`, and the
|
||||
`/api/dictation/ws` WebSocket endpoint (auth-gated the same way as the
|
||||
terminal WS: UI session token or `oc_url_token`, plus origin check).
|
||||
Created from the startup pipeline (`startup-pipeline-runtime.js`) before
|
||||
the generic OpenCode proxy so routes are not shadowed.
|
||||
- `stream-manager.js` — `DictationStreamManager`, one per WS connection.
|
||||
Chunk reordering by `seq` + ack, resampling to the provider rate,
|
||||
auto-commit every ~15 s of audio, silence suppression by PCM peak,
|
||||
partial-transcript concatenation, adaptive finalization timeout.
|
||||
- `service.js` — provider resolution and readiness. Providers:
|
||||
- `local` (default): sherpa-onnx Parakeet TDT in a forked worker process.
|
||||
Models auto-download in the background on first use; while missing, the
|
||||
stream fails with `reasonCode: 'model_download_in_progress'` and the
|
||||
status route reports per-model install/download state.
|
||||
- `openai-compatible`: buffered per-segment transcription against any
|
||||
OpenAI-compatible `/v1/audio/transcriptions` endpoint
|
||||
(`openai-compatible-session.js`, reuses `../tts/stt.js`).
|
||||
- `local/` — worker process + client (IPC, idle shutdown TTL), sherpa
|
||||
recognizer engine and realtime session (throttled re-decode for partials),
|
||||
model catalog and downloader. The native `sherpa-onnx-node` addon is only
|
||||
ever loaded inside the worker process.
|
||||
- `audio.js` — PCM16 helpers: format parsing, peak, WAV wrapping, streaming
|
||||
linear resampler.
|
||||
|
||||
## WebSocket protocol (JSON text frames)
|
||||
|
||||
Client → server: `start {dictationId, format, options}`,
|
||||
`chunk {dictationId, seq, audio}`, `finish {dictationId, finalSeq}`,
|
||||
`cancel {dictationId}`, `ping`.
|
||||
|
||||
Server → client: `ready`, `ack {ackSeq}`, `partial {text}`,
|
||||
`finish_accepted {timeoutMs}`, `final {text}`,
|
||||
`error {error, retryable, reasonCode?}`, `pong`.
|
||||
|
||||
`options` in `start` carries the client-selected provider config:
|
||||
`{ provider: 'local' | 'openai-compatible', language?, localModel?,
|
||||
openaiCompatible?: { baseUrl, model, apiKey } }`.
|
||||
|
||||
## Invariants
|
||||
|
||||
- Never load `sherpa-onnx-node` in the main server process.
|
||||
- The stream manager acks only the highest contiguous seq; the client is
|
||||
expected to retain unacked segments for retry/replay.
|
||||
- Silence-only segments (peak < 300) are cleared, never committed, so
|
||||
Whisper-style providers do not hallucinate on silence.
|
||||
- Model files live under `~/.config/openchamber/speech-models`.
|
||||
@@ -0,0 +1,195 @@
|
||||
/**
|
||||
* PCM16 audio helpers for the dictation streaming pipeline.
|
||||
*
|
||||
* All dictation audio travels as 16-bit little-endian mono PCM. The client
|
||||
* captures at 16 kHz; providers may require a different rate, so chunks are
|
||||
* resampled with Pcm16MonoResampler before being appended to an STT session.
|
||||
*/
|
||||
|
||||
/**
|
||||
* Parse the sample rate out of a format string like "audio/pcm;rate=16000;bits=16".
|
||||
* @param {string} format
|
||||
* @param {number|null} [fallback]
|
||||
* @returns {number|null}
|
||||
*/
|
||||
export function parsePcmRateFromFormat(format, fallback = null) {
|
||||
const match = /(?:^|[;,\s])rate\s*=\s*(\d+)(?:$|[;,\s])/i.exec(String(format || ''));
|
||||
if (!match) {
|
||||
return fallback;
|
||||
}
|
||||
const rate = Number.parseInt(match[1], 10);
|
||||
return Number.isFinite(rate) && rate > 0 ? rate : fallback;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return an Int16Array view over a PCM16LE buffer, copying when the buffer's
|
||||
* byteOffset is not 2-byte aligned (IPC-transferred buffers can be views at
|
||||
* odd offsets, and Int16Array requires an even start offset).
|
||||
* @param {Buffer} pcm16le
|
||||
* @returns {Int16Array}
|
||||
*/
|
||||
function toInt16Samples(pcm16le) {
|
||||
if (pcm16le.byteOffset % 2 !== 0) {
|
||||
const copy = Buffer.from(pcm16le);
|
||||
return new Int16Array(copy.buffer, copy.byteOffset, copy.byteLength / 2);
|
||||
}
|
||||
return new Int16Array(pcm16le.buffer, pcm16le.byteOffset, pcm16le.byteLength / 2);
|
||||
}
|
||||
|
||||
/**
|
||||
* Peak absolute sample value of a PCM16LE buffer. Used for silence detection.
|
||||
* @param {Buffer} pcm16le
|
||||
* @returns {number}
|
||||
*/
|
||||
export function pcm16lePeakAbs(pcm16le) {
|
||||
if (!pcm16le || pcm16le.length === 0) {
|
||||
return 0;
|
||||
}
|
||||
if (pcm16le.length % 2 !== 0) {
|
||||
throw new Error(`PCM16 chunk byteLength must be even, got ${pcm16le.length}`);
|
||||
}
|
||||
const samples = toInt16Samples(pcm16le);
|
||||
let peak = 0;
|
||||
for (let i = 0; i < samples.length; i += 1) {
|
||||
const v = samples[i];
|
||||
const abs = v < 0 ? -v : v;
|
||||
if (abs > peak) {
|
||||
peak = abs;
|
||||
if (peak >= 32767) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
return peak;
|
||||
}
|
||||
|
||||
/**
|
||||
* Convert PCM16LE to Float32 samples in [-1, 1], with optional gain.
|
||||
* @param {Buffer} pcm16le
|
||||
* @param {number} [gain]
|
||||
* @returns {Float32Array}
|
||||
*/
|
||||
export function pcm16leToFloat32(pcm16le, gain = 1) {
|
||||
if (pcm16le.length % 2 !== 0) {
|
||||
throw new Error(`PCM16 chunk byteLength must be even, got ${pcm16le.length}`);
|
||||
}
|
||||
const int16 = toInt16Samples(pcm16le);
|
||||
const out = new Float32Array(int16.length);
|
||||
for (let i = 0; i < int16.length; i += 1) {
|
||||
const v = (int16[i] / 32768.0) * gain;
|
||||
out[i] = Math.max(-1, Math.min(1, v));
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Wrap raw PCM16LE mono audio in a WAV container.
|
||||
* @param {Buffer} pcmBuffer
|
||||
* @param {number} sampleRate
|
||||
* @returns {Buffer}
|
||||
*/
|
||||
export function pcm16ToWav(pcmBuffer, sampleRate) {
|
||||
const channels = 1;
|
||||
const bitsPerSample = 16;
|
||||
const headerSize = 44;
|
||||
const wavBuffer = Buffer.alloc(headerSize + pcmBuffer.length);
|
||||
const byteRate = (sampleRate * channels * bitsPerSample) / 8;
|
||||
const blockAlign = (channels * bitsPerSample) / 8;
|
||||
|
||||
wavBuffer.write('RIFF', 0);
|
||||
wavBuffer.writeUInt32LE(36 + pcmBuffer.length, 4);
|
||||
wavBuffer.write('WAVE', 8);
|
||||
wavBuffer.write('fmt ', 12);
|
||||
wavBuffer.writeUInt32LE(16, 16);
|
||||
wavBuffer.writeUInt16LE(1, 20);
|
||||
wavBuffer.writeUInt16LE(channels, 22);
|
||||
wavBuffer.writeUInt32LE(sampleRate, 24);
|
||||
wavBuffer.writeUInt32LE(byteRate, 28);
|
||||
wavBuffer.writeUInt16LE(blockAlign, 32);
|
||||
wavBuffer.writeUInt16LE(bitsPerSample, 34);
|
||||
wavBuffer.write('data', 36);
|
||||
wavBuffer.writeUInt32LE(pcmBuffer.length, 40);
|
||||
pcmBuffer.copy(wavBuffer, 44);
|
||||
|
||||
return wavBuffer;
|
||||
}
|
||||
|
||||
/**
|
||||
* Streaming linear-interpolation resampler for PCM16LE mono audio.
|
||||
* Carries one sample across chunk boundaries so consecutive chunks resample
|
||||
* without seams.
|
||||
*/
|
||||
export class Pcm16MonoResampler {
|
||||
/**
|
||||
* @param {{ inputRate: number, outputRate: number }} params
|
||||
*/
|
||||
constructor({ inputRate, outputRate }) {
|
||||
this.inputRate = inputRate;
|
||||
this.outputRate = outputRate;
|
||||
this.step = inputRate / outputRate;
|
||||
this.pos = 0;
|
||||
this.carrySample = null;
|
||||
}
|
||||
|
||||
reset() {
|
||||
this.pos = 0;
|
||||
this.carrySample = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {Buffer} pcm16le
|
||||
* @returns {Buffer}
|
||||
*/
|
||||
processChunk(pcm16le) {
|
||||
if (pcm16le.length === 0) {
|
||||
return Buffer.alloc(0);
|
||||
}
|
||||
if (pcm16le.length % 2 !== 0) {
|
||||
throw new Error(`PCM16 chunk byteLength must be even, got ${pcm16le.length}`);
|
||||
}
|
||||
|
||||
const srcChunk = toInt16Samples(pcm16le);
|
||||
|
||||
const hasCarry = this.carrySample !== null;
|
||||
const srcLen = srcChunk.length + (hasCarry ? 1 : 0);
|
||||
if (srcLen < 2) {
|
||||
this.carrySample = srcChunk.length ? srcChunk[srcChunk.length - 1] : this.carrySample;
|
||||
return Buffer.alloc(0);
|
||||
}
|
||||
|
||||
const src = new Float32Array(srcLen);
|
||||
let offset = 0;
|
||||
if (hasCarry) {
|
||||
src[0] = this.carrySample / 32768;
|
||||
offset = 1;
|
||||
}
|
||||
for (let i = 0; i < srcChunk.length; i += 1) {
|
||||
src[offset + i] = srcChunk[i] / 32768;
|
||||
}
|
||||
|
||||
const out = [];
|
||||
const maxPos = src.length - 1;
|
||||
|
||||
while (this.pos < maxPos) {
|
||||
const i = Math.floor(this.pos);
|
||||
const frac = this.pos - i;
|
||||
const s0 = src[i];
|
||||
const s1 = src[i + 1];
|
||||
const sample = s0 + (s1 - s0) * frac;
|
||||
const clamped = Math.max(-1, Math.min(1, sample));
|
||||
out.push(Math.round(clamped * 32767));
|
||||
this.pos += this.step;
|
||||
}
|
||||
|
||||
this.carrySample = srcChunk[srcChunk.length - 1];
|
||||
|
||||
const shift = src.length - 1;
|
||||
this.pos = this.pos - shift;
|
||||
if (this.pos < 0) {
|
||||
this.pos = 0;
|
||||
}
|
||||
|
||||
const outArr = Int16Array.from(out);
|
||||
return Buffer.from(outArr.buffer, outArr.byteOffset, outArr.byteLength);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,141 @@
|
||||
/**
|
||||
* Catalog of local sherpa-onnx STT models available for dictation.
|
||||
* Models are downloaded on demand from the k2-fsa GitHub releases and
|
||||
* extracted under the OpenChamber speech-models directory.
|
||||
*
|
||||
* `type` selects the recognizer construction path in the worker:
|
||||
* - 'nemo_transducer': encoder/decoder/joiner transducer (Parakeet)
|
||||
* - 'whisper': encoder/decoder Whisper export
|
||||
* `files` maps logical roles to file names inside the extracted directory.
|
||||
*/
|
||||
|
||||
import path from 'path';
|
||||
|
||||
export const LOCAL_STT_MODEL_CATALOG = {
|
||||
'parakeet-tdt-0.6b-v2-int8': {
|
||||
type: 'nemo_transducer',
|
||||
archiveUrl:
|
||||
'https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-nemo-parakeet-tdt-0.6b-v2-int8.tar.bz2',
|
||||
extractedDir: 'sherpa-onnx-nemo-parakeet-tdt-0.6b-v2-int8',
|
||||
files: {
|
||||
encoder: 'encoder.int8.onnx',
|
||||
decoder: 'decoder.int8.onnx',
|
||||
joiner: 'joiner.int8.onnx',
|
||||
tokens: 'tokens.txt',
|
||||
},
|
||||
description: 'NVIDIA Parakeet TDT v2 (English)',
|
||||
},
|
||||
'parakeet-tdt-0.6b-v3-int8': {
|
||||
type: 'nemo_transducer',
|
||||
archiveUrl:
|
||||
'https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-nemo-parakeet-tdt-0.6b-v3-int8.tar.bz2',
|
||||
extractedDir: 'sherpa-onnx-nemo-parakeet-tdt-0.6b-v3-int8',
|
||||
files: {
|
||||
encoder: 'encoder.int8.onnx',
|
||||
decoder: 'decoder.int8.onnx',
|
||||
joiner: 'joiner.int8.onnx',
|
||||
tokens: 'tokens.txt',
|
||||
},
|
||||
description: 'NVIDIA Parakeet TDT v3 (25 European languages, auto-detected)',
|
||||
},
|
||||
'whisper-base-int8': {
|
||||
type: 'whisper',
|
||||
archiveUrl:
|
||||
'https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-whisper-base.tar.bz2',
|
||||
extractedDir: 'sherpa-onnx-whisper-base',
|
||||
files: {
|
||||
encoder: 'base-encoder.int8.onnx',
|
||||
decoder: 'base-decoder.int8.onnx',
|
||||
tokens: 'base-tokens.txt',
|
||||
},
|
||||
description: 'OpenAI Whisper base (multilingual, smaller and lighter)',
|
||||
},
|
||||
'whisper-tiny-int8': {
|
||||
type: 'whisper',
|
||||
archiveUrl:
|
||||
'https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-whisper-tiny.tar.bz2',
|
||||
extractedDir: 'sherpa-onnx-whisper-tiny',
|
||||
files: {
|
||||
encoder: 'tiny-encoder.int8.onnx',
|
||||
decoder: 'tiny-decoder.int8.onnx',
|
||||
tokens: 'tiny-tokens.txt',
|
||||
},
|
||||
description: 'OpenAI Whisper tiny (multilingual, fastest and lightest)',
|
||||
},
|
||||
};
|
||||
|
||||
/**
|
||||
* Local text-to-speech models (sherpa-onnx OfflineTts). Downloaded and
|
||||
* managed through the same pipeline as the STT models.
|
||||
*/
|
||||
export const LOCAL_TTS_MODEL_CATALOG = {
|
||||
'kokoro-en-v0_19': {
|
||||
type: 'kokoro',
|
||||
archiveUrl:
|
||||
'https://github.com/k2-fsa/sherpa-onnx/releases/download/tts-models/kokoro-en-v0_19.tar.bz2',
|
||||
extractedDir: 'kokoro-en-v0_19',
|
||||
files: {
|
||||
model: 'model.onnx',
|
||||
voices: 'voices.bin',
|
||||
tokens: 'tokens.txt',
|
||||
espeakData: 'espeak-ng-data',
|
||||
},
|
||||
description: 'Kokoro TTS (English, natural voices)',
|
||||
},
|
||||
};
|
||||
|
||||
export const DEFAULT_LOCAL_STT_MODEL = 'parakeet-tdt-0.6b-v2-int8';
|
||||
export const DEFAULT_LOCAL_TTS_MODEL = 'kokoro-en-v0_19';
|
||||
|
||||
export const LOCAL_STT_MODEL_IDS = Object.keys(LOCAL_STT_MODEL_CATALOG);
|
||||
export const LOCAL_TTS_MODEL_IDS = Object.keys(LOCAL_TTS_MODEL_CATALOG);
|
||||
|
||||
/**
|
||||
* @param {string} modelId
|
||||
* @returns {boolean}
|
||||
*/
|
||||
export function isLocalSttModelId(modelId) {
|
||||
return typeof modelId === 'string' && Object.hasOwn(LOCAL_STT_MODEL_CATALOG, modelId);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} modelId
|
||||
* @returns {boolean}
|
||||
*/
|
||||
export function isLocalTtsModelId(modelId) {
|
||||
return typeof modelId === 'string' && Object.hasOwn(LOCAL_TTS_MODEL_CATALOG, modelId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Any managed local model (STT or TTS).
|
||||
* @param {string} modelId
|
||||
* @returns {boolean}
|
||||
*/
|
||||
export function isLocalModelId(modelId) {
|
||||
return isLocalSttModelId(modelId) || isLocalTtsModelId(modelId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Spec lookup across both catalogs (STT and TTS).
|
||||
* @param {string} modelId
|
||||
*/
|
||||
export function getLocalSttModelSpec(modelId) {
|
||||
const spec = LOCAL_STT_MODEL_CATALOG[modelId] ?? LOCAL_TTS_MODEL_CATALOG[modelId];
|
||||
if (!spec) {
|
||||
throw new Error(`Unknown local speech model id: ${modelId}`);
|
||||
}
|
||||
return {
|
||||
id: modelId,
|
||||
...spec,
|
||||
requiredFiles: Object.values(spec.files),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} modelsDir
|
||||
* @param {string} modelId
|
||||
* @returns {string}
|
||||
*/
|
||||
export function getLocalSttModelDir(modelsDir, modelId) {
|
||||
return path.join(modelsDir, getLocalSttModelSpec(modelId).extractedDir);
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
/**
|
||||
* Downloads and extracts local sherpa-onnx STT model archives.
|
||||
* Archives (.tar.bz2) come from the k2-fsa GitHub releases and are extracted
|
||||
* with the system `tar` into the speech-models directory.
|
||||
*/
|
||||
|
||||
import { createWriteStream } from 'fs';
|
||||
import { mkdir, rename, rm, stat } from 'fs/promises';
|
||||
import path from 'path';
|
||||
import { Readable } from 'stream';
|
||||
import { pipeline } from 'stream/promises';
|
||||
import { spawn } from 'child_process';
|
||||
|
||||
import { getLocalSttModelSpec } from './model-catalog.js';
|
||||
|
||||
async function hasRequiredFiles(modelDir, requiredFiles) {
|
||||
const results = await Promise.all(
|
||||
requiredFiles.map(async (rel) => {
|
||||
try {
|
||||
const s = await stat(path.join(modelDir, rel));
|
||||
if (s.isDirectory()) {
|
||||
return true;
|
||||
}
|
||||
return s.isFile() && s.size > 0;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}),
|
||||
);
|
||||
return results.every(Boolean);
|
||||
}
|
||||
|
||||
async function downloadToFile(url, outputPath, onProgress) {
|
||||
const res = await fetch(url);
|
||||
if (!res.ok) {
|
||||
throw new Error(`Failed to download ${url}: ${res.status} ${res.statusText}`);
|
||||
}
|
||||
if (!res.body) {
|
||||
throw new Error(`Failed to download ${url}: missing response body`);
|
||||
}
|
||||
|
||||
const totalBytes = Number.parseInt(res.headers.get('content-length') || '', 10) || null;
|
||||
let downloadedBytes = 0;
|
||||
|
||||
const tmpPath = `${outputPath}.tmp-${Date.now()}`;
|
||||
await mkdir(path.dirname(outputPath), { recursive: true });
|
||||
|
||||
const nodeStream = Readable.fromWeb(res.body);
|
||||
if (typeof onProgress === 'function') {
|
||||
nodeStream.on('data', (chunk) => {
|
||||
downloadedBytes += chunk.length;
|
||||
onProgress(downloadedBytes, totalBytes);
|
||||
});
|
||||
}
|
||||
|
||||
try {
|
||||
await pipeline(nodeStream, createWriteStream(tmpPath));
|
||||
await rename(tmpPath, outputPath);
|
||||
} catch (error) {
|
||||
await rm(tmpPath, { force: true }).catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function extractTarArchive(archivePath, destDir) {
|
||||
await mkdir(destDir, { recursive: true });
|
||||
|
||||
await new Promise((resolve, reject) => {
|
||||
const child = spawn('tar', ['xf', archivePath, '-C', destDir], {
|
||||
stdio: 'ignore',
|
||||
windowsHide: true,
|
||||
});
|
||||
child.on('error', reject);
|
||||
child.on('exit', (code) => {
|
||||
if (code === 0) {
|
||||
resolve();
|
||||
} else {
|
||||
reject(new Error(`tar exited with code ${code}`));
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async function isNonEmptyFile(filePath) {
|
||||
try {
|
||||
const s = await stat(filePath);
|
||||
return s.isFile() && s.size > 0;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check whether a model is fully installed (all required files present).
|
||||
* @param {string} modelsDir
|
||||
* @param {string} modelId
|
||||
* @returns {Promise<boolean>}
|
||||
*/
|
||||
export async function isLocalSttModelInstalled(modelsDir, modelId) {
|
||||
const spec = getLocalSttModelSpec(modelId);
|
||||
return hasRequiredFiles(path.join(modelsDir, spec.extractedDir), spec.requiredFiles);
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensure a model is downloaded and extracted. Resolves with the model dir.
|
||||
*
|
||||
* Extraction is staged: the archive unpacks into a temporary directory and is
|
||||
* verified before being renamed into place. An interrupted or failed tar must
|
||||
* never leave partial files at the final path — the installed check only
|
||||
* verifies file presence, so a truncated .onnx there would be treated as an
|
||||
* installed model forever ("Protobuf parsing failed" at load time).
|
||||
*
|
||||
* @param {{ modelsDir: string, modelId: string,
|
||||
* onProgress?: (downloadedBytes: number, totalBytes: number | null) => void }} options
|
||||
* @returns {Promise<string>}
|
||||
*/
|
||||
export async function ensureLocalSttModel({ modelsDir, modelId, onProgress }) {
|
||||
const spec = getLocalSttModelSpec(modelId);
|
||||
const modelDir = path.join(modelsDir, spec.extractedDir);
|
||||
if (await hasRequiredFiles(modelDir, spec.requiredFiles)) {
|
||||
return modelDir;
|
||||
}
|
||||
|
||||
// A directory that exists but fails the required-files check is a partial
|
||||
// extraction from an earlier interrupted attempt — remove it before retrying.
|
||||
await rm(modelDir, { recursive: true, force: true }).catch(() => undefined);
|
||||
|
||||
const downloadsDir = path.join(modelsDir, '.downloads');
|
||||
const archiveFilename = path.basename(new URL(spec.archiveUrl).pathname);
|
||||
const archivePath = path.join(downloadsDir, archiveFilename);
|
||||
|
||||
if (!(await isNonEmptyFile(archivePath))) {
|
||||
await downloadToFile(spec.archiveUrl, archivePath, onProgress);
|
||||
}
|
||||
|
||||
const stagingDir = path.join(modelsDir, `.staging-${spec.extractedDir}-${Date.now()}`);
|
||||
try {
|
||||
await extractTarArchive(archivePath, stagingDir);
|
||||
|
||||
const stagedModelDir = path.join(stagingDir, spec.extractedDir);
|
||||
if (!(await hasRequiredFiles(stagedModelDir, spec.requiredFiles))) {
|
||||
// Bad archive (truncated download / corrupt cache): drop it so the next
|
||||
// attempt re-downloads instead of re-extracting the same broken bytes.
|
||||
await rm(archivePath, { force: true }).catch(() => undefined);
|
||||
throw new Error(
|
||||
`Extracted ${archiveFilename}, but required model files are missing or empty. The archive was discarded; retry to re-download.`,
|
||||
);
|
||||
}
|
||||
|
||||
await rename(stagedModelDir, modelDir);
|
||||
} catch (error) {
|
||||
await rm(stagingDir, { recursive: true, force: true }).catch(() => undefined);
|
||||
// Any extraction failure means the cached archive can't be trusted
|
||||
// (corrupt bz2, truncated download). Discard it so retry re-downloads.
|
||||
await rm(archivePath, { force: true }).catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
await rm(stagingDir, { recursive: true, force: true }).catch(() => undefined);
|
||||
|
||||
await rm(archivePath, { force: true }).catch(() => undefined);
|
||||
|
||||
return modelDir;
|
||||
}
|
||||
@@ -0,0 +1,136 @@
|
||||
/**
|
||||
* Loader for the sherpa-onnx-node native addon.
|
||||
*
|
||||
* sherpa-onnx-node ships its native addon and shared libraries in a
|
||||
* platform-specific package (e.g. sherpa-onnx-darwin-arm64). The shared
|
||||
* libraries must be findable via the platform's dynamic-loader search path,
|
||||
* so the loader prepends the platform package directory to LD_LIBRARY_PATH /
|
||||
* DYLD_LIBRARY_PATH / PATH before requiring the addon.
|
||||
*/
|
||||
|
||||
import { createRequire } from 'module';
|
||||
import path from 'path';
|
||||
import { existsSync } from 'fs';
|
||||
|
||||
const require = createRequire(import.meta.url);
|
||||
|
||||
let cached = null;
|
||||
|
||||
function sherpaPlatformPackageName(platform = process.platform, arch = process.arch) {
|
||||
const normalizedPlatform = platform === 'win32' ? 'win' : platform;
|
||||
return `sherpa-onnx-${normalizedPlatform}-${arch}`;
|
||||
}
|
||||
|
||||
function sherpaLoaderEnvKey(platform = process.platform) {
|
||||
if (platform === 'linux') {
|
||||
return 'LD_LIBRARY_PATH';
|
||||
}
|
||||
if (platform === 'darwin') {
|
||||
return 'DYLD_LIBRARY_PATH';
|
||||
}
|
||||
if (platform === 'win32') {
|
||||
return 'PATH';
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function prependEnvPath(existing, value) {
|
||||
const parts = String(existing ?? '').split(path.delimiter).filter(Boolean);
|
||||
if (parts.includes(value)) {
|
||||
return parts.join(path.delimiter);
|
||||
}
|
||||
return [value, ...parts].join(path.delimiter);
|
||||
}
|
||||
|
||||
/**
|
||||
* Case-insensitive env key lookup: on Windows `{...process.env}` yields a
|
||||
* plain object where PATH may be stored as `Path`. Using a hardcoded 'PATH'
|
||||
* would create a duplicate key and break the child process PATH.
|
||||
*/
|
||||
function findEnvKey(env, key) {
|
||||
const lower = key.toLowerCase();
|
||||
for (const k of Object.keys(env)) {
|
||||
if (k.toLowerCase() === lower) {
|
||||
return k;
|
||||
}
|
||||
}
|
||||
return key;
|
||||
}
|
||||
|
||||
function resolveSherpaLibDir(platform = process.platform, arch = process.arch) {
|
||||
const packageName = sherpaPlatformPackageName(platform, arch);
|
||||
try {
|
||||
const pkgJson = require.resolve(`${packageName}/package.json`);
|
||||
// Electron packages node_modules inside app.asar, but native addons and
|
||||
// their shared libraries are extracted to app.asar.unpacked. The dynamic
|
||||
// loader (dlopen/DYLD/LD) cannot read from the asar archive, so point the
|
||||
// search path at the unpacked copy.
|
||||
const dir = path.dirname(pkgJson);
|
||||
const unpacked = dir.replace(`app.asar${path.sep}`, `app.asar.unpacked${path.sep}`);
|
||||
return existsSync(unpacked) ? unpacked : dir;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Prepend the sherpa platform package dir to the loader search path env var.
|
||||
* Mutates the provided env object.
|
||||
* @param {NodeJS.ProcessEnv} env
|
||||
*/
|
||||
export function applySherpaLoaderEnv(env) {
|
||||
const key = sherpaLoaderEnvKey();
|
||||
const libDir = resolveSherpaLibDir();
|
||||
if (!key || !libDir) {
|
||||
return { key: null, libDir: null };
|
||||
}
|
||||
const actualKey = findEnvKey(env, key);
|
||||
env[actualKey] = prependEnvPath(env[actualKey], libDir);
|
||||
return { key, libDir };
|
||||
}
|
||||
|
||||
/**
|
||||
* Load the sherpa-onnx-node module, trying the upstream entry first and then
|
||||
* the platform addon directly.
|
||||
*/
|
||||
export function loadSherpaOnnxNode() {
|
||||
if (cached) {
|
||||
return cached;
|
||||
}
|
||||
|
||||
const attempts = [];
|
||||
|
||||
try {
|
||||
cached = require('sherpa-onnx-node');
|
||||
return cached;
|
||||
} catch (error) {
|
||||
attempts.push(`sherpa-onnx-node: ${error?.message || String(error)}`);
|
||||
}
|
||||
|
||||
const libDir = resolveSherpaLibDir();
|
||||
if (libDir) {
|
||||
applySherpaLoaderEnv(process.env);
|
||||
const addonPath = path.join(libDir, 'sherpa-onnx.node');
|
||||
if (existsSync(addonPath)) {
|
||||
try {
|
||||
cached = require(addonPath);
|
||||
return cached;
|
||||
} catch (error) {
|
||||
attempts.push(`${addonPath}: ${error?.message || String(error)}`);
|
||||
}
|
||||
} else {
|
||||
attempts.push(`${addonPath}: file not found`);
|
||||
}
|
||||
} else {
|
||||
attempts.push(`${sherpaPlatformPackageName()}: platform package not installed`);
|
||||
}
|
||||
|
||||
throw new Error(
|
||||
[
|
||||
`Failed to load sherpa-onnx-node for ${process.platform}-${process.arch}.`,
|
||||
`Node ${process.version} (ABI ${process.versions.modules}).`,
|
||||
'Load attempts:',
|
||||
...attempts.map((line) => `- ${line}`),
|
||||
].join('\n'),
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,277 @@
|
||||
/**
|
||||
* Sherpa-onnx offline recognizer engine (NeMo transducer / Parakeet) plus a
|
||||
* realtime streaming transcription session that re-decodes the accumulated
|
||||
* segment audio on a throttle to produce live partial transcripts.
|
||||
*
|
||||
* Runs inside the dictation worker process only — never load the native
|
||||
* addon in the main server process.
|
||||
*/
|
||||
|
||||
import { EventEmitter } from 'events';
|
||||
import { existsSync } from 'fs';
|
||||
import { randomUUID } from 'crypto';
|
||||
|
||||
import { loadSherpaOnnxNode } from './sherpa-loader.js';
|
||||
import { pcm16lePeakAbs, pcm16leToFloat32 } from '../audio.js';
|
||||
|
||||
function assertFileExists(filePath, label) {
|
||||
if (!existsSync(filePath)) {
|
||||
throw new Error(`Missing ${label}: ${filePath}`);
|
||||
}
|
||||
}
|
||||
|
||||
export class SherpaOfflineRecognizerEngine {
|
||||
/**
|
||||
* @param {{ type: 'nemo_transducer' | 'whisper',
|
||||
* encoder: string, decoder: string, joiner?: string, tokens: string,
|
||||
* numThreads?: number }} config
|
||||
*/
|
||||
constructor(config) {
|
||||
assertFileExists(config.encoder, 'offline encoder');
|
||||
assertFileExists(config.decoder, 'offline decoder');
|
||||
if (config.type === 'nemo_transducer') {
|
||||
assertFileExists(config.joiner, 'offline joiner');
|
||||
}
|
||||
assertFileExists(config.tokens, 'tokens');
|
||||
|
||||
const sherpa = loadSherpaOnnxNode();
|
||||
|
||||
const modelConfig =
|
||||
config.type === 'whisper'
|
||||
? {
|
||||
whisper: {
|
||||
encoder: config.encoder,
|
||||
decoder: config.decoder,
|
||||
// Empty language auto-detects for multilingual Whisper exports.
|
||||
language: '',
|
||||
task: 'transcribe',
|
||||
tailPaddings: -1,
|
||||
},
|
||||
tokens: config.tokens,
|
||||
modelType: 'whisper',
|
||||
numThreads: config.numThreads ?? 2,
|
||||
provider: 'cpu',
|
||||
debug: 0,
|
||||
}
|
||||
: {
|
||||
transducer: {
|
||||
encoder: config.encoder,
|
||||
decoder: config.decoder,
|
||||
joiner: config.joiner,
|
||||
},
|
||||
tokens: config.tokens,
|
||||
modelType: 'nemo_transducer',
|
||||
numThreads: config.numThreads ?? 2,
|
||||
provider: 'cpu',
|
||||
debug: 0,
|
||||
};
|
||||
|
||||
const recognizerConfig = {
|
||||
featConfig: {
|
||||
sampleRate: 16000,
|
||||
featureDim: 80,
|
||||
},
|
||||
modelConfig,
|
||||
decodingMethod: 'greedy_search',
|
||||
maxActivePaths: 4,
|
||||
};
|
||||
|
||||
this.recognizer = new sherpa.OfflineRecognizer(recognizerConfig);
|
||||
const sr = this.recognizer?.config?.featConfig?.sampleRate;
|
||||
this.sampleRate =
|
||||
typeof sr === 'number' && Number.isFinite(sr) && sr > 0
|
||||
? sr
|
||||
: recognizerConfig.featConfig.sampleRate;
|
||||
}
|
||||
|
||||
createStream() {
|
||||
return this.recognizer.createStream();
|
||||
}
|
||||
|
||||
acceptWaveform(stream, sampleRate, samples) {
|
||||
if (!stream || typeof stream.acceptWaveform !== 'function') {
|
||||
throw new Error('Unexpected sherpa offline stream: missing acceptWaveform()');
|
||||
}
|
||||
// sherpa-onnx-node expects acceptWaveform({ samples, sampleRate });
|
||||
// the WASM build expects acceptWaveform(sampleRate, samples).
|
||||
if (stream.acceptWaveform.length <= 1) {
|
||||
stream.acceptWaveform({ samples, sampleRate });
|
||||
} else {
|
||||
stream.acceptWaveform(sampleRate, samples);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Decode a full PCM16 segment and return its text.
|
||||
* Applies auto-gain when the peak is low so quiet microphones still decode.
|
||||
* @param {Buffer} pcm16
|
||||
* @returns {string}
|
||||
*/
|
||||
decodePcm16(pcm16) {
|
||||
if (pcm16.length === 0) {
|
||||
return '';
|
||||
}
|
||||
|
||||
const peak = pcm16lePeakAbs(pcm16);
|
||||
const peakFloat = peak / 32768.0;
|
||||
const targetPeak = 0.6;
|
||||
const maxGain = 50;
|
||||
const gain =
|
||||
peakFloat > 0 && peakFloat < targetPeak ? Math.min(maxGain, targetPeak / peakFloat) : 1;
|
||||
|
||||
const stream = this.createStream();
|
||||
try {
|
||||
const floatSamples = pcm16leToFloat32(pcm16, gain);
|
||||
this.acceptWaveform(stream, this.sampleRate, floatSamples);
|
||||
this.recognizer.decode(stream);
|
||||
const result = this.recognizer.getResult(stream);
|
||||
const text =
|
||||
typeof result === 'object' && result && 'text' in result ? result.text : result;
|
||||
return String(text ?? '').trim();
|
||||
} finally {
|
||||
try {
|
||||
stream.free?.();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
free() {
|
||||
try {
|
||||
this.recognizer?.free?.();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Streaming transcription session backed by the offline recognizer.
|
||||
* Accumulates the current segment's PCM and re-decodes it at most every
|
||||
* `minDecodeIntervalMs` to emit non-final partial transcripts; `commit()`
|
||||
* finalizes the segment and starts a new one.
|
||||
*
|
||||
* Implements the StreamingTranscriptionSession contract used by
|
||||
* DictationStreamManager.
|
||||
*/
|
||||
export class SherpaRealtimeTranscriptionSession extends EventEmitter {
|
||||
/**
|
||||
* @param {{ engine: SherpaOfflineRecognizerEngine, minDecodeIntervalMs?: number }} params
|
||||
*/
|
||||
constructor({ engine, minDecodeIntervalMs }) {
|
||||
super();
|
||||
this.engine = engine;
|
||||
this.requiredSampleRate = engine.sampleRate;
|
||||
this.minDecodeIntervalMs = minDecodeIntervalMs ?? 350;
|
||||
this.connected = false;
|
||||
this.currentSegmentId = null;
|
||||
this.previousSegmentId = null;
|
||||
this.lastPartialText = '';
|
||||
this.pcm16 = Buffer.alloc(0);
|
||||
this.lastDecodeAt = 0;
|
||||
this.decoding = false;
|
||||
this.pendingDecode = false;
|
||||
}
|
||||
|
||||
async connect() {
|
||||
if (this.connected) {
|
||||
return;
|
||||
}
|
||||
this.currentSegmentId = randomUUID();
|
||||
this.connected = true;
|
||||
}
|
||||
|
||||
appendPcm16(chunk) {
|
||||
if (!this.connected || !this.currentSegmentId) {
|
||||
this.emit('error', new Error('Sherpa realtime session not connected'));
|
||||
return;
|
||||
}
|
||||
this.pcm16 = this.pcm16.length === 0 ? chunk : Buffer.concat([this.pcm16, chunk]);
|
||||
this.maybeDecode(false).catch((err) => {
|
||||
this.emit('error', err instanceof Error ? err : new Error(String(err)));
|
||||
});
|
||||
}
|
||||
|
||||
commit() {
|
||||
if (!this.connected || !this.currentSegmentId) {
|
||||
this.emit('error', new Error('Sherpa realtime session not connected'));
|
||||
return;
|
||||
}
|
||||
|
||||
void (async () => {
|
||||
try {
|
||||
await this.maybeDecode(true);
|
||||
const finalText = this.lastPartialText;
|
||||
const segmentId = this.currentSegmentId;
|
||||
const previousSegmentId = this.previousSegmentId;
|
||||
|
||||
this.emit('committed', { segmentId, previousSegmentId });
|
||||
this.emit('transcript', { segmentId, transcript: finalText, isFinal: true });
|
||||
|
||||
this.previousSegmentId = segmentId;
|
||||
this.currentSegmentId = randomUUID();
|
||||
this.lastPartialText = '';
|
||||
this.pcm16 = Buffer.alloc(0);
|
||||
} catch (err) {
|
||||
this.emit('error', err instanceof Error ? err : new Error(String(err)));
|
||||
}
|
||||
})();
|
||||
}
|
||||
|
||||
clear() {
|
||||
if (!this.connected) {
|
||||
return;
|
||||
}
|
||||
this.pcm16 = Buffer.alloc(0);
|
||||
this.currentSegmentId = randomUUID();
|
||||
this.lastPartialText = '';
|
||||
}
|
||||
|
||||
close() {
|
||||
this.connected = false;
|
||||
this.currentSegmentId = null;
|
||||
this.pcm16 = Buffer.alloc(0);
|
||||
}
|
||||
|
||||
async maybeDecode(force) {
|
||||
if (!this.connected || !this.currentSegmentId) {
|
||||
return;
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
if (!force && now - this.lastDecodeAt < this.minDecodeIntervalMs) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.decoding) {
|
||||
this.pendingDecode = true;
|
||||
return;
|
||||
}
|
||||
|
||||
this.decoding = true;
|
||||
try {
|
||||
const decodeStartedAt = Date.now();
|
||||
const text = this.engine.decodePcm16(this.pcm16);
|
||||
this.lastDecodeAt = Date.now();
|
||||
// Adaptive throttle: on slow hardware (or heavy models) re-decoding the
|
||||
// growing segment every 350ms would monopolize the worker. Space partial
|
||||
// decodes to ~1.5x the observed decode time.
|
||||
this.minDecodeIntervalMs = Math.max(350, (this.lastDecodeAt - decodeStartedAt) * 1.5);
|
||||
if (text !== this.lastPartialText) {
|
||||
this.lastPartialText = text;
|
||||
this.emit('transcript', {
|
||||
segmentId: this.currentSegmentId,
|
||||
transcript: text,
|
||||
isFinal: false,
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
this.decoding = false;
|
||||
if (this.pendingDecode) {
|
||||
this.pendingDecode = false;
|
||||
await this.maybeDecode(true);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
/**
|
||||
* Sherpa-onnx offline TTS (Kokoro). Runs inside the dictation worker process
|
||||
* only — never load the native addon in the main server process.
|
||||
*/
|
||||
|
||||
import { existsSync } from 'fs';
|
||||
import path from 'path';
|
||||
|
||||
import { loadSherpaOnnxNode } from './sherpa-loader.js';
|
||||
|
||||
function assertFileExists(filePath, label) {
|
||||
if (!existsSync(filePath)) {
|
||||
throw new Error(`Missing ${label}: ${filePath}`);
|
||||
}
|
||||
}
|
||||
|
||||
function float32ToPcm16le(samples) {
|
||||
const out = new Int16Array(samples.length);
|
||||
for (let i = 0; i < samples.length; i += 1) {
|
||||
const clamped = Math.max(-1, Math.min(1, samples[i]));
|
||||
out[i] = Math.round(clamped * 32767);
|
||||
}
|
||||
return Buffer.from(out.buffer, out.byteOffset, out.byteLength);
|
||||
}
|
||||
|
||||
export class SherpaTtsEngine {
|
||||
/**
|
||||
* @param {{ modelDir: string, files: { model: string, voices: string, tokens: string, espeakData: string }, numThreads?: number }} config
|
||||
*/
|
||||
constructor(config) {
|
||||
const modelPath = path.join(config.modelDir, config.files.model);
|
||||
const voicesPath = path.join(config.modelDir, config.files.voices);
|
||||
const tokensPath = path.join(config.modelDir, config.files.tokens);
|
||||
const dataDir = path.join(config.modelDir, config.files.espeakData);
|
||||
|
||||
assertFileExists(modelPath, 'TTS model');
|
||||
assertFileExists(voicesPath, 'TTS voices');
|
||||
assertFileExists(tokensPath, 'TTS tokens');
|
||||
assertFileExists(dataDir, 'TTS espeak-ng dataDir');
|
||||
|
||||
const sherpa = loadSherpaOnnxNode();
|
||||
if (typeof sherpa.OfflineTts !== 'function') {
|
||||
throw new Error('sherpa-onnx-node OfflineTts is unavailable');
|
||||
}
|
||||
|
||||
this.tts = new sherpa.OfflineTts({
|
||||
model: {
|
||||
kokoro: {
|
||||
model: modelPath,
|
||||
voices: voicesPath,
|
||||
tokens: tokensPath,
|
||||
dataDir,
|
||||
lengthScale: 1.0,
|
||||
},
|
||||
},
|
||||
numThreads: config.numThreads ?? 2,
|
||||
provider: 'cpu',
|
||||
maxNumSentences: 1,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Synthesize text to PCM16LE.
|
||||
* @param {string} text
|
||||
* @param {{ speakerId?: number, speed?: number }} [options]
|
||||
* @returns {{ pcm16: Buffer, sampleRate: number }}
|
||||
*/
|
||||
synthesize(text, options = {}) {
|
||||
const trimmed = String(text || '').trim();
|
||||
if (!trimmed) {
|
||||
throw new Error('Cannot synthesize empty text');
|
||||
}
|
||||
|
||||
const audio = this.tts.generate({
|
||||
text: trimmed,
|
||||
sid: Number.isInteger(options.speakerId) ? options.speakerId : 0,
|
||||
speed: typeof options.speed === 'number' && options.speed > 0 ? options.speed : 1.0,
|
||||
// Request a copied buffer from sherpa itself: native external-backed
|
||||
// typed arrays are rejected by Electron.
|
||||
enableExternalBuffer: false,
|
||||
});
|
||||
|
||||
let samples = null;
|
||||
if (audio && audio.samples instanceof Float32Array) {
|
||||
samples = Float32Array.from(audio.samples);
|
||||
} else if (audio && Array.isArray(audio.samples)) {
|
||||
samples = Float32Array.from(audio.samples);
|
||||
}
|
||||
if (!samples) {
|
||||
throw new Error('Unexpected sherpa TTS output: missing Float32 samples');
|
||||
}
|
||||
|
||||
const sampleRate =
|
||||
audio && typeof audio.sampleRate === 'number' && audio.sampleRate > 0
|
||||
? audio.sampleRate
|
||||
: typeof this.tts.sampleRate === 'number' && this.tts.sampleRate > 0
|
||||
? this.tts.sampleRate
|
||||
: 24000;
|
||||
|
||||
return { pcm16: float32ToPcm16le(samples), sampleRate };
|
||||
}
|
||||
|
||||
free() {
|
||||
try {
|
||||
this.tts?.free?.();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,352 @@
|
||||
/**
|
||||
* Client for the dictation local-speech worker process.
|
||||
*
|
||||
* Lazily forks the worker on first use, correlates request/response messages
|
||||
* by requestId, routes session events to per-session EventEmitters, and
|
||||
* shuts the worker down after an idle TTL so the ONNX runtime does not sit
|
||||
* in memory while dictation is unused.
|
||||
*/
|
||||
|
||||
import { fork } from 'child_process';
|
||||
import { randomUUID } from 'crypto';
|
||||
import { EventEmitter } from 'events';
|
||||
import { fileURLToPath } from 'url';
|
||||
|
||||
import { applySherpaLoaderEnv } from './sherpa-loader.js';
|
||||
|
||||
const DEFAULT_REQUEST_TIMEOUT_MS = 30000;
|
||||
const DEFAULT_IDLE_TTL_MS = 5 * 60 * 1000;
|
||||
const DEFAULT_LOCAL_SAMPLE_RATE = 16000;
|
||||
const STDERR_TAIL_MAX_CHARS = 2000;
|
||||
|
||||
function forkDictationWorker() {
|
||||
const env = { ...process.env };
|
||||
applySherpaLoaderEnv(env);
|
||||
return fork(fileURLToPath(new URL('./worker-process.js', import.meta.url)), [], {
|
||||
env,
|
||||
serialization: 'advanced',
|
||||
stdio: ['ignore', 'ignore', 'pipe', 'ipc'],
|
||||
windowsHide: true,
|
||||
});
|
||||
}
|
||||
|
||||
export class DictationWorkerClient {
|
||||
/**
|
||||
* @param {{ requestTimeoutMs?: number, idleTtlMs?: number }} [options]
|
||||
*/
|
||||
constructor(options = {}) {
|
||||
this.requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS;
|
||||
this.idleTtlMs = options.idleTtlMs ?? DEFAULT_IDLE_TTL_MS;
|
||||
this.pendingRequests = new Map();
|
||||
this.sessionEmitters = new Map();
|
||||
this.worker = null;
|
||||
this.stderrTail = '';
|
||||
this.inFlightRequests = 0;
|
||||
this.idleTimer = null;
|
||||
this.intentionalCloses = new WeakSet();
|
||||
}
|
||||
|
||||
/**
|
||||
* Synthesize speech in the worker. Returns WAV bytes.
|
||||
* @param {{ modelsDir: string, modelId: string, text: string, speakerId?: number, speed?: number }} params
|
||||
* @returns {Promise<{ audio: Buffer, format: string }>}
|
||||
*/
|
||||
async synthesizeSpeech(params) {
|
||||
// Long texts on slow hardware can exceed the default request timeout.
|
||||
const result = await this.sendRequest(
|
||||
{ type: 'tts.synthesize', ...params },
|
||||
{ timeoutMs: 120000 },
|
||||
);
|
||||
return {
|
||||
audio: Buffer.isBuffer(result.audio) ? result.audio : Buffer.from(result.audio),
|
||||
format: result.format || 'audio/wav',
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a streaming STT session in the worker.
|
||||
* @param {{ modelsDir: string, modelId: string }} params
|
||||
* @param {EventEmitter} emitter receives 'committed' | 'transcript' | 'error'
|
||||
* @returns {Promise<{ sessionId: string, requiredSampleRate: number }>}
|
||||
*/
|
||||
async createSession({ modelsDir, modelId }, emitter) {
|
||||
const sessionId = randomUUID();
|
||||
this.sessionEmitters.set(sessionId, emitter);
|
||||
try {
|
||||
const result = await this.sendRequest({
|
||||
type: 'session.create',
|
||||
sessionId,
|
||||
modelsDir,
|
||||
modelId,
|
||||
});
|
||||
return { sessionId, requiredSampleRate: result?.requiredSampleRate ?? DEFAULT_LOCAL_SAMPLE_RATE };
|
||||
} catch (err) {
|
||||
this.sessionEmitters.delete(sessionId);
|
||||
this.scheduleIdleShutdownIfReady();
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
appendSessionAudio(sessionId, audio) {
|
||||
void this.sendRequest({ type: 'session.append', sessionId, audio }).catch((err) => {
|
||||
this.emitSessionError(sessionId, err);
|
||||
});
|
||||
}
|
||||
|
||||
commitSession(sessionId) {
|
||||
void this.sendRequest({ type: 'session.commit', sessionId }).catch((err) => {
|
||||
this.emitSessionError(sessionId, err);
|
||||
});
|
||||
}
|
||||
|
||||
clearSession(sessionId) {
|
||||
void this.sendRequest({ type: 'session.clear', sessionId }).catch((err) => {
|
||||
this.emitSessionError(sessionId, err);
|
||||
});
|
||||
}
|
||||
|
||||
closeSession(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() {
|
||||
this.clearIdleTimer();
|
||||
this.rejectAllPending(new Error('Dictation worker shut down'));
|
||||
this.sessionEmitters.clear();
|
||||
const worker = this.worker;
|
||||
this.worker = null;
|
||||
if (worker && !worker.killed) {
|
||||
this.intentionalCloses.add(worker);
|
||||
try {
|
||||
worker.disconnect();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
try {
|
||||
worker.kill();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
sendRequest(input, options = {}) {
|
||||
const worker = this.ensureWorker();
|
||||
const requestId = randomUUID();
|
||||
const message = { ...input, requestId };
|
||||
this.inFlightRequests += 1;
|
||||
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(`Dictation worker request timed out: ${input.type}`));
|
||||
}, options.timeoutMs ?? this.requestTimeoutMs);
|
||||
|
||||
this.pendingRequests.set(requestId, { resolve, reject, timeout });
|
||||
|
||||
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);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
ensureWorker() {
|
||||
if (this.worker && !this.worker.killed && this.worker.connected) {
|
||||
return this.worker;
|
||||
}
|
||||
const worker = forkDictationWorker();
|
||||
this.worker = worker;
|
||||
this.stderrTail = '';
|
||||
worker.stderr?.on('data', (chunk) => {
|
||||
const text = Buffer.isBuffer(chunk) ? chunk.toString('utf8') : String(chunk);
|
||||
this.stderrTail = (this.stderrTail + text).slice(-STDERR_TAIL_MAX_CHARS);
|
||||
});
|
||||
worker.on('message', (message) => this.handleWorkerMessage(message));
|
||||
worker.on('close', (code, signal) => this.handleWorkerExit(worker, code, signal));
|
||||
return worker;
|
||||
}
|
||||
|
||||
handleWorkerMessage(message) {
|
||||
if (message?.type === 'response') {
|
||||
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 || 'Dictation worker request failed'));
|
||||
}
|
||||
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.error':
|
||||
emitter.emit('error', new Error(message.error));
|
||||
return;
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
handleWorkerExit(worker, code, signal) {
|
||||
const wasCurrentWorker = this.worker === worker;
|
||||
const wasIntentional = this.intentionalCloses.has(worker);
|
||||
this.intentionalCloses.delete(worker);
|
||||
if (!wasCurrentWorker || wasIntentional) {
|
||||
if (wasCurrentWorker) {
|
||||
this.worker = null;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const stderr = this.stderrTail.trim();
|
||||
const error = new Error(
|
||||
`Dictation worker exited (code ${code ?? 'null'}${signal ? `, signal ${signal}` : ''}).` +
|
||||
(stderr ? ` Last stderr: ${stderr.slice(-500)}` : ''),
|
||||
);
|
||||
|
||||
this.worker = null;
|
||||
this.clearIdleTimer();
|
||||
this.rejectAllPending(error);
|
||||
for (const emitter of this.sessionEmitters.values()) {
|
||||
if (emitter.listenerCount('error') > 0) {
|
||||
emitter.emit('error', error);
|
||||
}
|
||||
}
|
||||
this.sessionEmitters.clear();
|
||||
this.inFlightRequests = 0;
|
||||
}
|
||||
|
||||
rejectAllPending(error) {
|
||||
for (const [requestId, pending] of this.pendingRequests) {
|
||||
clearTimeout(pending.timeout);
|
||||
pending.reject(error);
|
||||
this.pendingRequests.delete(requestId);
|
||||
}
|
||||
}
|
||||
|
||||
emitSessionError(sessionId, error) {
|
||||
const emitter = this.sessionEmitters.get(sessionId);
|
||||
if (emitter && emitter.listenerCount('error') > 0) {
|
||||
emitter.emit('error', error instanceof Error ? error : new Error(String(error)));
|
||||
}
|
||||
}
|
||||
|
||||
scheduleIdleShutdownIfReady() {
|
||||
if (!this.worker || this.inFlightRequests > 0 || this.sessionEmitters.size > 0) {
|
||||
return;
|
||||
}
|
||||
this.clearIdleTimer();
|
||||
this.idleTimer = setTimeout(() => {
|
||||
if (this.inFlightRequests === 0 && this.sessionEmitters.size === 0) {
|
||||
this.shutdown();
|
||||
}
|
||||
}, this.idleTtlMs);
|
||||
}
|
||||
|
||||
clearIdleTimer() {
|
||||
if (this.idleTimer) {
|
||||
clearTimeout(this.idleTimer);
|
||||
this.idleTimer = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* StreamingTranscriptionSession backed by the worker process.
|
||||
* Matches the session contract consumed by DictationStreamManager.
|
||||
*/
|
||||
export class WorkerBackedTranscriptionSession extends EventEmitter {
|
||||
/**
|
||||
* @param {DictationWorkerClient} client
|
||||
* @param {{ modelsDir: string, modelId: string }} modelConfig
|
||||
*/
|
||||
constructor(client, modelConfig) {
|
||||
super();
|
||||
this.client = client;
|
||||
this.modelConfig = modelConfig;
|
||||
this.requiredSampleRate = DEFAULT_LOCAL_SAMPLE_RATE;
|
||||
this.connectedSessionId = null;
|
||||
this.connecting = null;
|
||||
}
|
||||
|
||||
async connect() {
|
||||
if (this.connectedSessionId) {
|
||||
return;
|
||||
}
|
||||
if (!this.connecting) {
|
||||
this.connecting = (async () => {
|
||||
try {
|
||||
const result = await this.client.createSession(this.modelConfig, this);
|
||||
this.connectedSessionId = result.sessionId;
|
||||
this.requiredSampleRate = result.requiredSampleRate;
|
||||
} finally {
|
||||
this.connecting = null;
|
||||
}
|
||||
})();
|
||||
}
|
||||
await this.connecting;
|
||||
}
|
||||
|
||||
appendPcm16(pcm16le) {
|
||||
if (!this.connectedSessionId) {
|
||||
this.emit('error', new Error('Local STT session not connected'));
|
||||
return;
|
||||
}
|
||||
this.client.appendSessionAudio(this.connectedSessionId, pcm16le);
|
||||
}
|
||||
|
||||
commit() {
|
||||
if (!this.connectedSessionId) {
|
||||
this.emit('error', new Error('Local STT session not connected'));
|
||||
return;
|
||||
}
|
||||
this.client.commitSession(this.connectedSessionId);
|
||||
}
|
||||
|
||||
clear() {
|
||||
if (this.connectedSessionId) {
|
||||
this.client.clearSession(this.connectedSessionId);
|
||||
}
|
||||
}
|
||||
|
||||
close() {
|
||||
const sessionId = this.connectedSessionId;
|
||||
this.connectedSessionId = null;
|
||||
if (sessionId) {
|
||||
this.client.closeSession(sessionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,197 @@
|
||||
/**
|
||||
* Dictation local-speech worker process.
|
||||
*
|
||||
* Hosts the sherpa-onnx native inference (Parakeet STT) in a separate process
|
||||
* so ONNX decoding never blocks the main OpenChamber server. Communicates
|
||||
* with the parent over child_process IPC (advanced serialization, so Buffers
|
||||
* survive the trip as Uint8Array).
|
||||
*
|
||||
* Request/response protocol (parent -> worker):
|
||||
* { type: 'session.create', requestId, sessionId, modelsDir, modelId }
|
||||
* { type: 'session.append', requestId, sessionId, audio }
|
||||
* { type: 'session.commit' | 'session.clear' | 'session.close', requestId, sessionId }
|
||||
* Worker -> parent:
|
||||
* { type: 'response', requestId, ok, result?, error? }
|
||||
* { type: 'session.committed' | 'session.transcript' | 'session.error', sessionId, ... }
|
||||
*/
|
||||
|
||||
import {
|
||||
SherpaOfflineRecognizerEngine,
|
||||
SherpaRealtimeTranscriptionSession,
|
||||
} from './sherpa-recognizer.js';
|
||||
import { SherpaTtsEngine } from './sherpa-tts.js';
|
||||
import { getLocalSttModelDir, getLocalSttModelSpec } from './model-catalog.js';
|
||||
import { pcm16ToWav } from '../audio.js';
|
||||
import path from 'path';
|
||||
|
||||
process.title = 'OpenChamber Dictation';
|
||||
|
||||
const engines = new Map();
|
||||
const ttsEngines = new Map();
|
||||
const sessions = new Map();
|
||||
let ipcClosing = false;
|
||||
|
||||
function sendToParent(message) {
|
||||
if (ipcClosing || !process.connected || !process.send) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
process.send(message, (error) => {
|
||||
if (error) {
|
||||
ipcClosing = true;
|
||||
}
|
||||
});
|
||||
} catch {
|
||||
ipcClosing = true;
|
||||
}
|
||||
}
|
||||
|
||||
function sendOk(requestId, result) {
|
||||
sendToParent({ type: 'response', requestId, ok: true, ...(result !== undefined ? { result } : {}) });
|
||||
}
|
||||
|
||||
function getEngine(modelsDir, modelId) {
|
||||
const key = `${modelsDir}:${modelId}`;
|
||||
const existing = engines.get(key);
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
const modelDir = getLocalSttModelDir(modelsDir, modelId);
|
||||
const spec = getLocalSttModelSpec(modelId);
|
||||
const created = new SherpaOfflineRecognizerEngine({
|
||||
type: spec.type,
|
||||
encoder: path.join(modelDir, spec.files.encoder),
|
||||
decoder: path.join(modelDir, spec.files.decoder),
|
||||
...(spec.files.joiner ? { joiner: path.join(modelDir, spec.files.joiner) } : {}),
|
||||
tokens: path.join(modelDir, spec.files.tokens),
|
||||
numThreads: 2,
|
||||
});
|
||||
engines.set(key, created);
|
||||
return created;
|
||||
}
|
||||
|
||||
function cleanupSession(sessionId) {
|
||||
const session = sessions.get(sessionId);
|
||||
sessions.delete(sessionId);
|
||||
try {
|
||||
session?.close();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
|
||||
function toBuffer(audio) {
|
||||
if (Buffer.isBuffer(audio)) {
|
||||
return audio;
|
||||
}
|
||||
if (audio instanceof Uint8Array) {
|
||||
return Buffer.from(audio.buffer, audio.byteOffset, audio.byteLength);
|
||||
}
|
||||
if (audio && typeof audio === 'object' && audio.type === 'Buffer' && Array.isArray(audio.data)) {
|
||||
return Buffer.from(audio.data);
|
||||
}
|
||||
throw new Error('Unsupported audio payload in dictation worker');
|
||||
}
|
||||
|
||||
function getTtsEngine(modelsDir, modelId) {
|
||||
const key = `${modelsDir}:${modelId}`;
|
||||
const existing = ttsEngines.get(key);
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
const spec = getLocalSttModelSpec(modelId);
|
||||
const created = new SherpaTtsEngine({
|
||||
modelDir: getLocalSttModelDir(modelsDir, modelId),
|
||||
files: spec.files,
|
||||
numThreads: 2,
|
||||
});
|
||||
ttsEngines.set(key, created);
|
||||
return created;
|
||||
}
|
||||
|
||||
async function handleRequest(message) {
|
||||
switch (message.type) {
|
||||
case 'tts.synthesize': {
|
||||
const engine = getTtsEngine(message.modelsDir, message.modelId);
|
||||
const { pcm16, sampleRate } = engine.synthesize(message.text, {
|
||||
speakerId: message.speakerId,
|
||||
speed: message.speed,
|
||||
});
|
||||
sendOk(message.requestId, {
|
||||
audio: pcm16ToWav(pcm16, sampleRate),
|
||||
format: 'audio/wav',
|
||||
});
|
||||
return;
|
||||
}
|
||||
case 'session.create': {
|
||||
cleanupSession(message.sessionId);
|
||||
const engine = getEngine(message.modelsDir, message.modelId);
|
||||
const session = new SherpaRealtimeTranscriptionSession({ engine });
|
||||
session.on('committed', (payload) => {
|
||||
sendToParent({ type: 'session.committed', sessionId: message.sessionId, payload });
|
||||
});
|
||||
session.on('transcript', (payload) => {
|
||||
sendToParent({ type: 'session.transcript', sessionId: message.sessionId, payload });
|
||||
});
|
||||
session.on('error', (err) => {
|
||||
sendToParent({
|
||||
type: 'session.error',
|
||||
sessionId: message.sessionId,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
});
|
||||
await session.connect();
|
||||
sessions.set(message.sessionId, session);
|
||||
sendOk(message.requestId, { requiredSampleRate: session.requiredSampleRate });
|
||||
return;
|
||||
}
|
||||
case 'session.append': {
|
||||
sessions.get(message.sessionId)?.appendPcm16(toBuffer(message.audio));
|
||||
sendOk(message.requestId);
|
||||
return;
|
||||
}
|
||||
case 'session.commit': {
|
||||
sessions.get(message.sessionId)?.commit();
|
||||
sendOk(message.requestId);
|
||||
return;
|
||||
}
|
||||
case 'session.clear': {
|
||||
sessions.get(message.sessionId)?.clear();
|
||||
sendOk(message.requestId);
|
||||
return;
|
||||
}
|
||||
case 'session.close': {
|
||||
cleanupSession(message.sessionId);
|
||||
sendOk(message.requestId);
|
||||
return;
|
||||
}
|
||||
default: {
|
||||
throw new Error(`Unknown dictation worker request: ${message?.type}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
process.on('message', (message) => {
|
||||
void handleRequest(message).catch((error) => {
|
||||
sendToParent({
|
||||
type: 'response',
|
||||
requestId: message?.requestId,
|
||||
ok: false,
|
||||
error: error instanceof Error ? error.message : 'Dictation worker request failed',
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
process.once('disconnect', () => {
|
||||
ipcClosing = true;
|
||||
for (const sessionId of Array.from(sessions.keys())) {
|
||||
cleanupSession(sessionId);
|
||||
}
|
||||
for (const engine of engines.values()) {
|
||||
engine.free();
|
||||
}
|
||||
for (const tts of ttsEngines.values()) {
|
||||
tts.free();
|
||||
}
|
||||
process.exit(0);
|
||||
});
|
||||
@@ -0,0 +1,98 @@
|
||||
/**
|
||||
* Pseudo-streaming transcription session for OpenAI-compatible Whisper
|
||||
* endpoints (faster-whisper, whisper.cpp, OpenAI, ...).
|
||||
*
|
||||
* The Whisper HTTP API cannot stream, so audio is buffered per segment and
|
||||
* transcribed on commit(). Live partials therefore only advance at segment
|
||||
* boundaries (the DictationStreamManager auto-commits every ~15s of speech).
|
||||
*
|
||||
* Implements the StreamingTranscriptionSession contract used by
|
||||
* DictationStreamManager.
|
||||
*/
|
||||
|
||||
import { EventEmitter } from 'events';
|
||||
import { randomUUID } from 'crypto';
|
||||
|
||||
import { transcribeAudio } from '../tts/stt.js';
|
||||
import { pcm16ToWav } from './audio.js';
|
||||
|
||||
const OPENAI_COMPATIBLE_SAMPLE_RATE = 16000;
|
||||
|
||||
export class OpenAICompatibleTranscriptionSession extends EventEmitter {
|
||||
/**
|
||||
* @param {{ baseURL: string, model: string, apiKey?: string, language?: string, prompt?: string }} config
|
||||
*/
|
||||
constructor(config) {
|
||||
super();
|
||||
this.config = config;
|
||||
this.requiredSampleRate = OPENAI_COMPATIBLE_SAMPLE_RATE;
|
||||
this.connected = false;
|
||||
this.segmentId = randomUUID();
|
||||
this.previousSegmentId = null;
|
||||
this.pcm16 = Buffer.alloc(0);
|
||||
}
|
||||
|
||||
async connect() {
|
||||
if (!this.config.baseURL) {
|
||||
throw new Error('Custom STT server URL is not configured');
|
||||
}
|
||||
if (!this.config.model) {
|
||||
throw new Error('STT model is not configured');
|
||||
}
|
||||
this.connected = true;
|
||||
}
|
||||
|
||||
appendPcm16(chunk) {
|
||||
if (!this.connected) {
|
||||
this.emit('error', new Error('STT session not connected'));
|
||||
return;
|
||||
}
|
||||
this.pcm16 = this.pcm16.length === 0 ? chunk : Buffer.concat([this.pcm16, chunk]);
|
||||
}
|
||||
|
||||
commit() {
|
||||
if (!this.connected) {
|
||||
this.emit('error', new Error('STT session not connected'));
|
||||
return;
|
||||
}
|
||||
|
||||
const committedId = this.segmentId;
|
||||
const previousSegmentId = this.previousSegmentId;
|
||||
const committedPcm16 = this.pcm16;
|
||||
this.previousSegmentId = committedId;
|
||||
this.segmentId = randomUUID();
|
||||
this.pcm16 = Buffer.alloc(0);
|
||||
this.emit('committed', { segmentId: committedId, previousSegmentId });
|
||||
|
||||
void (async () => {
|
||||
try {
|
||||
const wav = pcm16ToWav(committedPcm16, OPENAI_COMPATIBLE_SAMPLE_RATE);
|
||||
const text = await transcribeAudio({
|
||||
audioBuffer: wav,
|
||||
mimeType: 'audio/wav',
|
||||
model: this.config.model,
|
||||
baseURL: this.config.baseURL,
|
||||
apiKey: this.config.apiKey,
|
||||
language: this.config.language,
|
||||
});
|
||||
this.emit('transcript', {
|
||||
segmentId: committedId,
|
||||
transcript: (text ?? '').trim(),
|
||||
isFinal: true,
|
||||
});
|
||||
} catch (err) {
|
||||
this.emit('error', err instanceof Error ? err : new Error(String(err)));
|
||||
}
|
||||
})();
|
||||
}
|
||||
|
||||
clear() {
|
||||
this.pcm16 = Buffer.alloc(0);
|
||||
this.segmentId = randomUUID();
|
||||
}
|
||||
|
||||
close() {
|
||||
this.connected = false;
|
||||
this.pcm16 = Buffer.alloc(0);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,278 @@
|
||||
/**
|
||||
* Dictation runtime: registers the streaming dictation WebSocket endpoint and
|
||||
* the HTTP status/model routes.
|
||||
*
|
||||
* WebSocket protocol (JSON text frames) on /api/dictation/ws:
|
||||
* client -> server:
|
||||
* { type: 'start', dictationId, format, options? }
|
||||
* options: { provider?, language?, localModel?, openaiCompatible? }
|
||||
* { type: 'chunk', dictationId, seq, audio } // audio: base64 PCM16LE
|
||||
* { type: 'finish', dictationId, finalSeq }
|
||||
* { type: 'cancel', dictationId }
|
||||
* { type: 'ping' }
|
||||
* server -> client:
|
||||
* { type: 'ready' }
|
||||
* { type: 'ack', dictationId, ackSeq }
|
||||
* { type: 'partial', dictationId, text }
|
||||
* { type: 'finish_accepted', dictationId, timeoutMs }
|
||||
* { type: 'final', dictationId, text }
|
||||
* { type: 'error', dictationId, error, retryable, reasonCode? }
|
||||
* { type: 'pong' }
|
||||
*/
|
||||
|
||||
import { WebSocketServer } from 'ws';
|
||||
|
||||
import { DictationStreamManager } from './stream-manager.js';
|
||||
import { createDictationService } from './service.js';
|
||||
|
||||
const DICTATION_WS_PATH = '/api/dictation/ws';
|
||||
|
||||
const DICTATION_WS_MAX_PAYLOAD_BYTES = 512 * 1024;
|
||||
const DICTATION_WS_HEARTBEAT_INTERVAL_MS = 30000;
|
||||
|
||||
const parseRequestPathname = (url) => {
|
||||
try {
|
||||
return new URL(url, 'http://localhost').pathname;
|
||||
} catch {
|
||||
return typeof url === 'string' ? url.split('?')[0] : '';
|
||||
}
|
||||
};
|
||||
|
||||
export function createDictationRuntime({
|
||||
app,
|
||||
server,
|
||||
express,
|
||||
uiAuthController,
|
||||
isRequestOriginAllowed,
|
||||
rejectWebSocketUpgrade,
|
||||
modelsDir,
|
||||
}) {
|
||||
const service = createDictationService({ modelsDir });
|
||||
|
||||
// Local text-to-speech (Kokoro in the dictation worker). Returns WAV bytes;
|
||||
// 503 with a reason code while the model is still downloading.
|
||||
app.post('/api/dictation/tts/speak', express.json({ limit: '1mb' }), async (req, res) => {
|
||||
try {
|
||||
const text = typeof req.body?.text === 'string' ? req.body.text.trim() : '';
|
||||
if (!text) {
|
||||
res.status(400).json({ error: 'Text is required' });
|
||||
return;
|
||||
}
|
||||
const result = await service.synthesizeSpeech({
|
||||
text,
|
||||
model: typeof req.body?.model === 'string' ? req.body.model : undefined,
|
||||
speakerId: Number.isInteger(req.body?.speakerId) ? req.body.speakerId : undefined,
|
||||
speed: typeof req.body?.speed === 'number' ? req.body.speed : undefined,
|
||||
});
|
||||
if (result.error) {
|
||||
res.status(503).json({
|
||||
error: result.error,
|
||||
retryable: result.retryable !== false,
|
||||
...(result.reasonCode ? { reasonCode: result.reasonCode } : {}),
|
||||
});
|
||||
return;
|
||||
}
|
||||
res.setHeader('Content-Type', result.format || 'audio/wav');
|
||||
res.send(result.audio);
|
||||
} catch (error) {
|
||||
res.status(500).json({ error: error?.message || 'Failed to synthesize speech' });
|
||||
}
|
||||
});
|
||||
|
||||
app.get('/api/dictation/status', async (req, res) => {
|
||||
try {
|
||||
const provider = typeof req.query.provider === 'string' ? req.query.provider : undefined;
|
||||
const localModel = typeof req.query.localModel === 'string' ? req.query.localModel : undefined;
|
||||
const status = await service.getStatus({ provider, localModel });
|
||||
res.json(status);
|
||||
} catch (error) {
|
||||
res.status(500).json({ error: error?.message || 'Failed to read dictation status' });
|
||||
}
|
||||
});
|
||||
|
||||
app.post('/api/dictation/models/:modelId/download', async (req, res) => {
|
||||
try {
|
||||
const result = await service.requestModelDownload(req.params.modelId);
|
||||
if (!result.ok) {
|
||||
res.status(400).json({ error: result.error });
|
||||
return;
|
||||
}
|
||||
res.json(result);
|
||||
} catch (error) {
|
||||
res.status(500).json({ error: error?.message || 'Failed to start model download' });
|
||||
}
|
||||
});
|
||||
|
||||
app.delete('/api/dictation/models/:modelId', async (req, res) => {
|
||||
try {
|
||||
const result = await service.deleteModel(req.params.modelId);
|
||||
if (!result.ok) {
|
||||
res.status(400).json({ error: result.error });
|
||||
return;
|
||||
}
|
||||
res.json(result);
|
||||
} catch (error) {
|
||||
res.status(500).json({ error: error?.message || 'Failed to delete model' });
|
||||
}
|
||||
});
|
||||
|
||||
const wsServer = new WebSocketServer({
|
||||
noServer: true,
|
||||
maxPayload: DICTATION_WS_MAX_PAYLOAD_BYTES,
|
||||
});
|
||||
|
||||
wsServer.on('connection', (socket) => {
|
||||
const send = (msg) => {
|
||||
if (socket.readyState !== 1) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
socket.send(JSON.stringify(msg));
|
||||
} catch {
|
||||
// socket is going away; the manager cleanup on close handles state
|
||||
}
|
||||
};
|
||||
|
||||
const manager = new DictationStreamManager({
|
||||
emit: ({ type, payload }) => send({ type, ...payload }),
|
||||
createSttSession: (options) => service.createSttSession(options),
|
||||
});
|
||||
|
||||
send({ type: 'ready' });
|
||||
|
||||
const heartbeatInterval = setInterval(() => {
|
||||
if (socket.readyState !== 1) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
socket.ping();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}, DICTATION_WS_HEARTBEAT_INTERVAL_MS);
|
||||
|
||||
socket.on('message', (raw, isBinary) => {
|
||||
if (isBinary) {
|
||||
return;
|
||||
}
|
||||
let message;
|
||||
try {
|
||||
message = JSON.parse(raw.toString('utf8'));
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
if (!message || typeof message !== 'object') {
|
||||
return;
|
||||
}
|
||||
|
||||
switch (message.type) {
|
||||
case 'start': {
|
||||
if (typeof message.dictationId !== 'string' || typeof message.format !== 'string') {
|
||||
return;
|
||||
}
|
||||
const options =
|
||||
message.options && typeof message.options === 'object' ? message.options : {};
|
||||
void manager.handleStart(message.dictationId, message.format, options);
|
||||
return;
|
||||
}
|
||||
case 'chunk': {
|
||||
if (
|
||||
typeof message.dictationId !== 'string' ||
|
||||
typeof message.seq !== 'number' ||
|
||||
typeof message.audio !== 'string'
|
||||
) {
|
||||
return;
|
||||
}
|
||||
manager.handleChunk({
|
||||
dictationId: message.dictationId,
|
||||
seq: message.seq,
|
||||
audioBase64: message.audio,
|
||||
});
|
||||
return;
|
||||
}
|
||||
case 'finish': {
|
||||
if (typeof message.dictationId !== 'string' || typeof message.finalSeq !== 'number') {
|
||||
return;
|
||||
}
|
||||
manager.handleFinish(message.dictationId, message.finalSeq);
|
||||
return;
|
||||
}
|
||||
case 'cancel': {
|
||||
if (typeof message.dictationId !== 'string') {
|
||||
return;
|
||||
}
|
||||
manager.handleCancel(message.dictationId);
|
||||
return;
|
||||
}
|
||||
case 'ping': {
|
||||
send({ type: 'pong' });
|
||||
return;
|
||||
}
|
||||
default:
|
||||
}
|
||||
});
|
||||
|
||||
socket.on('close', () => {
|
||||
clearInterval(heartbeatInterval);
|
||||
manager.cleanupAll();
|
||||
});
|
||||
|
||||
socket.on('error', () => {
|
||||
// 'close' follows and performs cleanup.
|
||||
});
|
||||
});
|
||||
|
||||
const upgradeHandler = (req, socket, head) => {
|
||||
const pathname = parseRequestPathname(req.url);
|
||||
if (pathname !== DICTATION_WS_PATH) {
|
||||
return;
|
||||
}
|
||||
|
||||
const handleUpgrade = async () => {
|
||||
try {
|
||||
if (uiAuthController?.enabled) {
|
||||
const sessionToken = await uiAuthController?.ensureSessionToken?.(req, null);
|
||||
if (!sessionToken) {
|
||||
rejectWebSocketUpgrade(socket, 401, 'UI authentication required');
|
||||
return;
|
||||
}
|
||||
|
||||
const originAllowed = await isRequestOriginAllowed(req);
|
||||
if (!originAllowed) {
|
||||
rejectWebSocketUpgrade(socket, 403, 'Invalid origin');
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
wsServer.handleUpgrade(req, socket, head, (ws) => {
|
||||
wsServer.emit('connection', ws, req);
|
||||
});
|
||||
} catch {
|
||||
rejectWebSocketUpgrade(socket, 500, 'Upgrade failed');
|
||||
}
|
||||
};
|
||||
|
||||
void handleUpgrade();
|
||||
};
|
||||
|
||||
server.on('upgrade', upgradeHandler);
|
||||
|
||||
const stop = () => {
|
||||
server.off('upgrade', upgradeHandler);
|
||||
for (const client of wsServer.clients) {
|
||||
try {
|
||||
client.close(1001, 'server shutting down');
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
try {
|
||||
wsServer.close();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
service.shutdown();
|
||||
};
|
||||
|
||||
return { stop };
|
||||
}
|
||||
@@ -0,0 +1,302 @@
|
||||
/**
|
||||
* Dictation service: resolves STT providers, tracks local model download
|
||||
* state, and exposes a readiness snapshot for the status route.
|
||||
*
|
||||
* Providers:
|
||||
* - 'local' (default): sherpa-onnx Parakeet running in a worker process.
|
||||
* Models auto-download in the background on first use.
|
||||
* - 'openai-compatible': any OpenAI-compatible /v1/audio/transcriptions
|
||||
* endpoint (faster-whisper, whisper.cpp, OpenAI).
|
||||
*/
|
||||
|
||||
import { rm } from 'fs/promises';
|
||||
|
||||
import { DictationWorkerClient, WorkerBackedTranscriptionSession } from './local/worker-client.js';
|
||||
import { OpenAICompatibleTranscriptionSession } from './openai-compatible-session.js';
|
||||
import {
|
||||
DEFAULT_LOCAL_STT_MODEL,
|
||||
DEFAULT_LOCAL_TTS_MODEL,
|
||||
LOCAL_STT_MODEL_CATALOG,
|
||||
LOCAL_STT_MODEL_IDS,
|
||||
LOCAL_TTS_MODEL_CATALOG,
|
||||
LOCAL_TTS_MODEL_IDS,
|
||||
getLocalSttModelDir,
|
||||
isLocalModelId,
|
||||
isLocalSttModelId,
|
||||
isLocalTtsModelId,
|
||||
} from './local/model-catalog.js';
|
||||
import { ensureLocalSttModel, isLocalSttModelInstalled } from './local/model-downloader.js';
|
||||
|
||||
export function createDictationService({ modelsDir }) {
|
||||
const workerClient = new DictationWorkerClient();
|
||||
/** modelId -> 'downloading' | 'error' */
|
||||
const downloadStates = new Map();
|
||||
/** modelId -> last download error message */
|
||||
const downloadErrors = new Map();
|
||||
/** modelId -> in-flight ensure promise */
|
||||
const downloadPromises = new Map();
|
||||
/** modelId -> 0..100 download percent (null while size unknown) */
|
||||
const downloadProgress = new Map();
|
||||
|
||||
const startModelDownload = (modelId) => {
|
||||
const existing = downloadPromises.get(modelId);
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
downloadStates.set(modelId, 'downloading');
|
||||
downloadErrors.delete(modelId);
|
||||
downloadProgress.set(modelId, 0);
|
||||
const promise = ensureLocalSttModel({
|
||||
modelsDir,
|
||||
modelId,
|
||||
onProgress: (downloadedBytes, totalBytes) => {
|
||||
downloadProgress.set(
|
||||
modelId,
|
||||
totalBytes ? Math.min(100, Math.round((downloadedBytes / totalBytes) * 100)) : null,
|
||||
);
|
||||
},
|
||||
})
|
||||
.then(() => {
|
||||
downloadStates.delete(modelId);
|
||||
downloadPromises.delete(modelId);
|
||||
downloadProgress.delete(modelId);
|
||||
})
|
||||
.catch((error) => {
|
||||
downloadStates.set(modelId, 'error');
|
||||
downloadErrors.set(modelId, error?.message || String(error));
|
||||
downloadPromises.delete(modelId);
|
||||
downloadProgress.delete(modelId);
|
||||
});
|
||||
downloadPromises.set(modelId, promise);
|
||||
return promise;
|
||||
};
|
||||
|
||||
const resolveLocalModelId = (requested) => {
|
||||
return isLocalSttModelId(requested) ? requested : DEFAULT_LOCAL_STT_MODEL;
|
||||
};
|
||||
|
||||
/**
|
||||
* Create a connected StreamingTranscriptionSession for one dictation.
|
||||
* Returns { session } on success or { error, retryable, reasonCode } when
|
||||
* the provider is not ready.
|
||||
*
|
||||
* @param {{ provider?: string, language?: string, localModel?: string,
|
||||
* openaiCompatible?: { baseUrl?: string, model?: string, apiKey?: string } }} options
|
||||
*/
|
||||
const createSttSession = async (options = {}) => {
|
||||
const provider = options.provider === 'openai-compatible' ? 'openai-compatible' : 'local';
|
||||
|
||||
if (provider === 'openai-compatible') {
|
||||
const config = options.openaiCompatible || {};
|
||||
const session = new OpenAICompatibleTranscriptionSession({
|
||||
baseURL: config.baseUrl,
|
||||
model: config.model,
|
||||
apiKey: config.apiKey || undefined,
|
||||
language: options.language || undefined,
|
||||
});
|
||||
try {
|
||||
await session.connect();
|
||||
} catch (error) {
|
||||
return {
|
||||
error: error?.message || String(error),
|
||||
retryable: false,
|
||||
reasonCode: 'stt_not_configured',
|
||||
};
|
||||
}
|
||||
return { session };
|
||||
}
|
||||
|
||||
const modelId = resolveLocalModelId(options.localModel);
|
||||
const installed = await isLocalSttModelInstalled(modelsDir, modelId);
|
||||
if (!installed) {
|
||||
const state = downloadStates.get(modelId);
|
||||
if (state === 'error') {
|
||||
const message = downloadErrors.get(modelId) || 'Model download failed';
|
||||
// Allow a retry on the next attempt.
|
||||
downloadStates.delete(modelId);
|
||||
return {
|
||||
error: `Failed to download dictation model: ${message}`,
|
||||
retryable: true,
|
||||
reasonCode: 'model_download_failed',
|
||||
};
|
||||
}
|
||||
void startModelDownload(modelId);
|
||||
return {
|
||||
error: 'Dictation model is downloading',
|
||||
retryable: true,
|
||||
reasonCode: 'model_download_in_progress',
|
||||
};
|
||||
}
|
||||
|
||||
const session = new WorkerBackedTranscriptionSession(workerClient, { modelsDir, modelId });
|
||||
try {
|
||||
await session.connect();
|
||||
} catch (error) {
|
||||
const message = error?.message || String(error);
|
||||
// A model that passes the file-presence check but fails to load is
|
||||
// corrupt on disk (e.g. truncated by an interrupted extraction). Remove
|
||||
// it so the next attempt re-downloads instead of crashing forever.
|
||||
if (/Load model|Protobuf parsing failed/i.test(message)) {
|
||||
await rm(getLocalSttModelDir(modelsDir, modelId), { recursive: true, force: true })
|
||||
.catch(() => undefined);
|
||||
return {
|
||||
error: 'Dictation model files were corrupt and have been removed; retry to re-download',
|
||||
retryable: true,
|
||||
reasonCode: 'model_corrupt',
|
||||
};
|
||||
}
|
||||
return {
|
||||
error: message,
|
||||
retryable: true,
|
||||
reasonCode: 'stt_unavailable',
|
||||
};
|
||||
}
|
||||
return { session };
|
||||
};
|
||||
|
||||
/**
|
||||
* Readiness snapshot for the status route and UI gating.
|
||||
* @param {{ provider?: string, localModel?: string }} [options]
|
||||
*/
|
||||
const getStatus = async (options = {}) => {
|
||||
const provider = options.provider === 'openai-compatible' ? 'openai-compatible' : 'local';
|
||||
const modelId = resolveLocalModelId(options.localModel);
|
||||
|
||||
const describeModel = async (id, catalog) => ({
|
||||
id,
|
||||
description: catalog[id].description,
|
||||
installed: await isLocalSttModelInstalled(modelsDir, id),
|
||||
downloading: downloadStates.get(id) === 'downloading',
|
||||
downloadProgress: downloadProgress.get(id) ?? null,
|
||||
downloadError: downloadErrors.get(id) || null,
|
||||
});
|
||||
|
||||
const models = await Promise.all(
|
||||
LOCAL_STT_MODEL_IDS.map((id) => describeModel(id, LOCAL_STT_MODEL_CATALOG)),
|
||||
);
|
||||
const ttsModels = await Promise.all(
|
||||
LOCAL_TTS_MODEL_IDS.map((id) => describeModel(id, LOCAL_TTS_MODEL_CATALOG)),
|
||||
);
|
||||
|
||||
if (provider === 'openai-compatible') {
|
||||
return { provider, available: true, models, ttsModels };
|
||||
}
|
||||
|
||||
const model = models.find((entry) => entry.id === modelId) || null;
|
||||
if (model?.installed) {
|
||||
return { provider, available: true, activeModel: modelId, models, ttsModels };
|
||||
}
|
||||
if (model?.downloading) {
|
||||
return {
|
||||
provider,
|
||||
available: false,
|
||||
reasonCode: 'model_download_in_progress',
|
||||
activeModel: modelId,
|
||||
models,
|
||||
ttsModels,
|
||||
};
|
||||
}
|
||||
if (model?.downloadError) {
|
||||
return {
|
||||
provider,
|
||||
available: false,
|
||||
reasonCode: 'model_download_failed',
|
||||
error: model.downloadError,
|
||||
activeModel: modelId,
|
||||
models,
|
||||
ttsModels,
|
||||
};
|
||||
}
|
||||
return {
|
||||
provider,
|
||||
available: false,
|
||||
reasonCode: 'models_missing',
|
||||
activeModel: modelId,
|
||||
models,
|
||||
ttsModels,
|
||||
};
|
||||
};
|
||||
|
||||
/**
|
||||
* Synthesize speech with the local TTS model. Returns WAV bytes, or a
|
||||
* readiness error while the model is missing/downloading.
|
||||
* @param {{ text: string, model?: string, speakerId?: number, speed?: number }} options
|
||||
*/
|
||||
const synthesizeSpeech = async ({ text, model, speakerId, speed }) => {
|
||||
const modelId = isLocalTtsModelId(model) ? model : DEFAULT_LOCAL_TTS_MODEL;
|
||||
const installed = await isLocalSttModelInstalled(modelsDir, modelId);
|
||||
if (!installed) {
|
||||
const state = downloadStates.get(modelId);
|
||||
if (state === 'error') {
|
||||
const message = downloadErrors.get(modelId) || 'Model download failed';
|
||||
downloadStates.delete(modelId);
|
||||
return {
|
||||
error: `Failed to download TTS model: ${message}`,
|
||||
retryable: true,
|
||||
reasonCode: 'model_download_failed',
|
||||
};
|
||||
}
|
||||
void startModelDownload(modelId);
|
||||
return {
|
||||
error: 'TTS model is downloading',
|
||||
retryable: true,
|
||||
reasonCode: 'model_download_in_progress',
|
||||
};
|
||||
}
|
||||
|
||||
const result = await workerClient.synthesizeSpeech({
|
||||
modelsDir,
|
||||
modelId,
|
||||
text,
|
||||
speakerId,
|
||||
speed,
|
||||
});
|
||||
return { audio: result.audio, format: result.format };
|
||||
};
|
||||
|
||||
/**
|
||||
* Kick off a background download for a model (used by the status route's
|
||||
* download action so Settings can pre-download models).
|
||||
*/
|
||||
const requestModelDownload = async (modelId) => {
|
||||
if (!isLocalModelId(modelId)) {
|
||||
return { ok: false, error: 'Unknown model id' };
|
||||
}
|
||||
if (await isLocalSttModelInstalled(modelsDir, modelId)) {
|
||||
return { ok: true, installed: true };
|
||||
}
|
||||
void startModelDownload(modelId);
|
||||
return { ok: true, installed: false };
|
||||
};
|
||||
|
||||
/**
|
||||
* Delete an installed model from disk. A model that is mid-download cannot
|
||||
* be deleted. An engine already loaded in the worker keeps its in-memory
|
||||
* copy until the worker's idle shutdown; the files are simply re-downloaded
|
||||
* on the next use if the model is selected again.
|
||||
*/
|
||||
const deleteModel = async (modelId) => {
|
||||
if (!isLocalModelId(modelId)) {
|
||||
return { ok: false, error: 'Unknown model id' };
|
||||
}
|
||||
if (downloadStates.get(modelId) === 'downloading') {
|
||||
return { ok: false, error: 'Model is downloading' };
|
||||
}
|
||||
await rm(getLocalSttModelDir(modelsDir, modelId), { recursive: true, force: true });
|
||||
downloadErrors.delete(modelId);
|
||||
return { ok: true };
|
||||
};
|
||||
|
||||
const shutdown = () => {
|
||||
workerClient.shutdown();
|
||||
};
|
||||
|
||||
return {
|
||||
createSttSession,
|
||||
synthesizeSpeech,
|
||||
getStatus,
|
||||
requestModelDownload,
|
||||
deleteModel,
|
||||
shutdown,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,461 @@
|
||||
/**
|
||||
* DictationStreamManager
|
||||
*
|
||||
* Server-authoritative streaming dictation state machine. One manager owns
|
||||
* all dictation streams for a single WebSocket connection.
|
||||
*
|
||||
* Responsibilities:
|
||||
* - Reorders inbound chunks by `seq` and acks the highest contiguous seq.
|
||||
* - Resamples client PCM (16 kHz by default) to the provider's required rate.
|
||||
* - Auto-commits a segment every `autoCommitSeconds` of audio, but clears
|
||||
* silence-only segments instead of committing them.
|
||||
* - Concatenates per-segment transcripts into live partials and emits the
|
||||
* final text once every committed segment has a final transcript.
|
||||
* - Applies an adaptive finalization timeout budget based on pending work.
|
||||
*/
|
||||
|
||||
import { Pcm16MonoResampler, parsePcmRateFromFormat, pcm16lePeakAbs } from './audio.js';
|
||||
|
||||
const DEFAULT_FINAL_TIMEOUT_MS = 10000;
|
||||
const DEFAULT_AUTO_COMMIT_SECONDS = 15;
|
||||
const FINAL_TIMEOUT_MAX_MS = 5 * 60 * 1000;
|
||||
const FINAL_TIMEOUT_PER_PENDING_SEGMENT_MS = 15 * 1000;
|
||||
const FINAL_TIMEOUT_PER_PENDING_AUDIO_SECOND_MS = 1500;
|
||||
const FINAL_TIMEOUT_PER_MISSING_SEQ_MS = 250;
|
||||
const SILENCE_PEAK_THRESHOLD = 300;
|
||||
|
||||
export class DictationStreamManager {
|
||||
/**
|
||||
* @param {object} params
|
||||
* @param {(msg: { type: string, payload: object }) => void} params.emit
|
||||
* @param {(startOptions: object) => Promise<{ session: object } | { error: string, retryable: boolean, reasonCode?: string }>} params.createSttSession
|
||||
* Resolves a connected streaming transcription session for one dictation.
|
||||
* The streaming transcription session contract:
|
||||
* { requiredSampleRate, appendPcm16(buf), commit(), clear(), close(), on(event, handler) }
|
||||
* @param {number} [params.finalTimeoutMs]
|
||||
* @param {number} [params.autoCommitSeconds]
|
||||
*/
|
||||
constructor({ emit, createSttSession, finalTimeoutMs, autoCommitSeconds }) {
|
||||
this.emit = emit;
|
||||
this.createSttSession = createSttSession;
|
||||
this.finalTimeoutMs = finalTimeoutMs ?? DEFAULT_FINAL_TIMEOUT_MS;
|
||||
this.autoCommitSeconds = autoCommitSeconds ?? DEFAULT_AUTO_COMMIT_SECONDS;
|
||||
this.streams = new Map();
|
||||
}
|
||||
|
||||
cleanupAll() {
|
||||
for (const dictationId of Array.from(this.streams.keys())) {
|
||||
this.cleanupStream(dictationId);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} dictationId
|
||||
* @param {string} format e.g. "audio/pcm;rate=16000;bits=16"
|
||||
* @param {object} startOptions provider/config options forwarded to createSttSession
|
||||
*/
|
||||
async handleStart(dictationId, format, startOptions = {}) {
|
||||
this.cleanupStream(dictationId);
|
||||
|
||||
const inputRate = parsePcmRateFromFormat(format, 16000) ?? 16000;
|
||||
if (!Number.isFinite(inputRate) || inputRate <= 0) {
|
||||
this.failStream(dictationId, `Invalid dictation input rate in format: ${format}`, false);
|
||||
return;
|
||||
}
|
||||
|
||||
let resolved;
|
||||
try {
|
||||
resolved = await this.createSttSession(startOptions);
|
||||
} catch (error) {
|
||||
this.failStream(dictationId, error?.message || String(error), true);
|
||||
return;
|
||||
}
|
||||
if (!resolved || resolved.error) {
|
||||
this.failStream(
|
||||
dictationId,
|
||||
resolved?.error || 'Dictation STT not configured',
|
||||
Boolean(resolved?.retryable),
|
||||
resolved?.reasonCode,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const stt = resolved.session;
|
||||
|
||||
stt.on('committed', ({ segmentId }) => {
|
||||
const state = this.streams.get(dictationId);
|
||||
if (!state) {
|
||||
return;
|
||||
}
|
||||
state.committedSegmentIds.push(segmentId);
|
||||
state.bytesSinceCommit = 0;
|
||||
state.peakSinceCommit = 0;
|
||||
|
||||
if (state.finishRequested && state.awaitingFinalCommit) {
|
||||
state.awaitingFinalCommit = false;
|
||||
}
|
||||
|
||||
this.maybeFinalizeStream(dictationId);
|
||||
});
|
||||
|
||||
stt.on('transcript', ({ segmentId, transcript, isFinal }) => {
|
||||
const state = this.streams.get(dictationId);
|
||||
if (!state) {
|
||||
return;
|
||||
}
|
||||
state.transcriptsBySegmentId.set(segmentId, transcript);
|
||||
if (isFinal) {
|
||||
state.finalTranscriptSegmentIds.add(segmentId);
|
||||
}
|
||||
|
||||
if (state.finishRequested && state.awaitingFinalCommit && isFinal) {
|
||||
state.awaitingFinalCommit = false;
|
||||
}
|
||||
|
||||
const orderedIds = state.committedSegmentIds.includes(segmentId)
|
||||
? state.committedSegmentIds
|
||||
: [...state.committedSegmentIds, segmentId];
|
||||
const partialText = orderedIds
|
||||
.map((id) => state.transcriptsBySegmentId.get(id) ?? '')
|
||||
.join(' ')
|
||||
.trim();
|
||||
this.emit({ type: 'partial', payload: { dictationId, text: partialText } });
|
||||
|
||||
this.maybeSealStreamFinish(dictationId);
|
||||
this.maybeFinalizeStream(dictationId);
|
||||
});
|
||||
|
||||
stt.on('error', (err) => {
|
||||
const message = err?.message || String(err);
|
||||
this.failAndCleanupStream(dictationId, message, true);
|
||||
});
|
||||
|
||||
this.streams.set(dictationId, {
|
||||
dictationId,
|
||||
inputFormat: format,
|
||||
stt,
|
||||
inputRate,
|
||||
outputRate: stt.requiredSampleRate,
|
||||
resampler:
|
||||
inputRate === stt.requiredSampleRate
|
||||
? null
|
||||
: new Pcm16MonoResampler({ inputRate, outputRate: stt.requiredSampleRate }),
|
||||
receivedChunks: new Map(),
|
||||
nextSeqToForward: 0,
|
||||
ackSeq: -1,
|
||||
autoCommitBytes:
|
||||
this.autoCommitSeconds > 0
|
||||
? Math.max(1, Math.round(this.autoCommitSeconds * stt.requiredSampleRate * 2))
|
||||
: 0,
|
||||
bytesSinceCommit: 0,
|
||||
peakSinceCommit: 0,
|
||||
committedSegmentIds: [],
|
||||
transcriptsBySegmentId: new Map(),
|
||||
finalTranscriptSegmentIds: new Set(),
|
||||
awaitingFinalCommit: false,
|
||||
finishRequested: false,
|
||||
finishSealed: false,
|
||||
finalSeq: null,
|
||||
finalTimeout: null,
|
||||
});
|
||||
|
||||
this.emitAck(dictationId, -1);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {{ dictationId: string, seq: number, audioBase64: string }} params
|
||||
*/
|
||||
handleChunk({ dictationId, seq, audioBase64 }) {
|
||||
const state = this.streams.get(dictationId);
|
||||
if (!state) {
|
||||
this.failStream(dictationId, 'Dictation stream not started', true);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!Number.isInteger(seq) || seq < 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (seq < state.nextSeqToForward) {
|
||||
this.emitAck(dictationId, state.ackSeq);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!state.receivedChunks.has(seq)) {
|
||||
let chunk;
|
||||
try {
|
||||
chunk = Buffer.from(audioBase64, 'base64');
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
if (chunk.length % 2 !== 0) {
|
||||
chunk = chunk.subarray(0, chunk.length - 1);
|
||||
}
|
||||
state.receivedChunks.set(seq, chunk);
|
||||
}
|
||||
|
||||
while (state.receivedChunks.has(state.nextSeqToForward)) {
|
||||
const nextSeq = state.nextSeqToForward;
|
||||
const pcm16 = state.receivedChunks.get(nextSeq);
|
||||
state.receivedChunks.delete(nextSeq);
|
||||
|
||||
const resampled = state.resampler ? state.resampler.processChunk(pcm16) : pcm16;
|
||||
if (resampled.length > 0) {
|
||||
state.stt.appendPcm16(resampled);
|
||||
state.bytesSinceCommit += resampled.length;
|
||||
state.peakSinceCommit = Math.max(state.peakSinceCommit, pcm16lePeakAbs(resampled));
|
||||
try {
|
||||
this.maybeAutoCommitSegment(state);
|
||||
} catch (error) {
|
||||
this.failAndCleanupStream(dictationId, error?.message || String(error), true);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
state.nextSeqToForward += 1;
|
||||
state.ackSeq = state.nextSeqToForward - 1;
|
||||
}
|
||||
|
||||
this.emitAck(dictationId, state.ackSeq);
|
||||
this.maybeSealStreamFinish(dictationId);
|
||||
this.maybeFinalizeStream(dictationId);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {string} dictationId
|
||||
* @param {number} finalSeq highest seq the client sent (or -1 if none)
|
||||
*/
|
||||
handleFinish(dictationId, finalSeq) {
|
||||
const state = this.streams.get(dictationId);
|
||||
if (!state) {
|
||||
this.failStream(dictationId, 'Dictation stream not started', true);
|
||||
return;
|
||||
}
|
||||
|
||||
state.finishRequested = true;
|
||||
state.finalSeq = finalSeq;
|
||||
|
||||
if (
|
||||
finalSeq >= 0 &&
|
||||
state.ackSeq < 0 &&
|
||||
state.nextSeqToForward === 0 &&
|
||||
state.receivedChunks.size === 0
|
||||
) {
|
||||
this.failStream(
|
||||
dictationId,
|
||||
'Dictation finished but no audio chunks were received',
|
||||
true,
|
||||
);
|
||||
this.cleanupStream(dictationId);
|
||||
return;
|
||||
}
|
||||
|
||||
this.maybeSealStreamFinish(dictationId);
|
||||
this.maybeFinalizeStream(dictationId);
|
||||
|
||||
const updatedState = this.streams.get(dictationId);
|
||||
if (!updatedState) {
|
||||
return;
|
||||
}
|
||||
|
||||
const timeoutMs = this.estimateFinalizationTimeout(updatedState);
|
||||
if (updatedState.finalTimeout) {
|
||||
clearTimeout(updatedState.finalTimeout);
|
||||
}
|
||||
updatedState.finalTimeout = setTimeout(() => {
|
||||
this.failAndCleanupStream(dictationId, 'Timed out waiting for final transcription', true);
|
||||
}, timeoutMs);
|
||||
|
||||
this.emit({ type: 'finish_accepted', payload: { dictationId, timeoutMs } });
|
||||
}
|
||||
|
||||
handleCancel(dictationId) {
|
||||
this.cleanupStream(dictationId);
|
||||
}
|
||||
|
||||
emitAck(dictationId, ackSeq) {
|
||||
this.emit({ type: 'ack', payload: { dictationId, ackSeq } });
|
||||
}
|
||||
|
||||
failStream(dictationId, error, retryable, reasonCode) {
|
||||
this.emit({
|
||||
type: 'error',
|
||||
payload: {
|
||||
dictationId,
|
||||
error,
|
||||
retryable,
|
||||
...(reasonCode ? { reasonCode } : {}),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
failAndCleanupStream(dictationId, error, retryable) {
|
||||
this.failStream(dictationId, error, retryable);
|
||||
this.cleanupStream(dictationId);
|
||||
}
|
||||
|
||||
cleanupStream(dictationId) {
|
||||
const state = this.streams.get(dictationId);
|
||||
if (!state) {
|
||||
return;
|
||||
}
|
||||
if (state.finalTimeout) {
|
||||
clearTimeout(state.finalTimeout);
|
||||
}
|
||||
try {
|
||||
state.stt.close();
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
this.streams.delete(dictationId);
|
||||
}
|
||||
|
||||
estimateFinalizationTimeout(state) {
|
||||
const bytesPerSecond = Math.max(1, state.outputRate * 2);
|
||||
const pendingCommittedSegments = state.committedSegmentIds.reduce((count, segmentId) => {
|
||||
return state.finalTranscriptSegmentIds.has(segmentId) ? count : count + 1;
|
||||
}, 0);
|
||||
const committedSet = new Set(state.committedSegmentIds);
|
||||
const pendingUncommittedTranscriptSegments = Array.from(
|
||||
state.transcriptsBySegmentId.keys(),
|
||||
).reduce((count, segmentId) => {
|
||||
if (committedSet.has(segmentId)) {
|
||||
return count;
|
||||
}
|
||||
return state.finalTranscriptSegmentIds.has(segmentId) ? count : count + 1;
|
||||
}, 0);
|
||||
const pendingSegments =
|
||||
pendingCommittedSegments +
|
||||
pendingUncommittedTranscriptSegments +
|
||||
(state.awaitingFinalCommit ? 1 : 0);
|
||||
const pendingAudioSeconds = Math.ceil(Math.max(0, state.bytesSinceCommit) / bytesPerSecond);
|
||||
const missingSeqCount =
|
||||
state.finalSeq === null ? 0 : Math.max(0, state.finalSeq - state.ackSeq);
|
||||
|
||||
const extraMs =
|
||||
pendingSegments * FINAL_TIMEOUT_PER_PENDING_SEGMENT_MS +
|
||||
pendingAudioSeconds * FINAL_TIMEOUT_PER_PENDING_AUDIO_SECOND_MS +
|
||||
missingSeqCount * FINAL_TIMEOUT_PER_MISSING_SEQ_MS;
|
||||
|
||||
return Math.max(
|
||||
this.finalTimeoutMs,
|
||||
Math.min(FINAL_TIMEOUT_MAX_MS, this.finalTimeoutMs + extraMs),
|
||||
);
|
||||
}
|
||||
|
||||
maybeAutoCommitSegment(state) {
|
||||
if (state.finishRequested) {
|
||||
return;
|
||||
}
|
||||
if (state.autoCommitBytes <= 0 || state.bytesSinceCommit < state.autoCommitBytes) {
|
||||
return;
|
||||
}
|
||||
if (state.peakSinceCommit < SILENCE_PEAK_THRESHOLD) {
|
||||
state.stt.clear();
|
||||
state.bytesSinceCommit = 0;
|
||||
state.peakSinceCommit = 0;
|
||||
return;
|
||||
}
|
||||
|
||||
state.bytesSinceCommit = 0;
|
||||
state.peakSinceCommit = 0;
|
||||
state.stt.commit();
|
||||
}
|
||||
|
||||
maybeSealStreamFinish(dictationId) {
|
||||
const state = this.streams.get(dictationId);
|
||||
if (!state) {
|
||||
return;
|
||||
}
|
||||
if (!state.finishRequested || state.finalSeq === null) {
|
||||
return;
|
||||
}
|
||||
if (state.ackSeq < state.finalSeq) {
|
||||
return;
|
||||
}
|
||||
if (state.finishSealed) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (state.bytesSinceCommit > 0) {
|
||||
if (state.peakSinceCommit < SILENCE_PEAK_THRESHOLD) {
|
||||
state.stt.clear();
|
||||
state.bytesSinceCommit = 0;
|
||||
state.peakSinceCommit = 0;
|
||||
state.awaitingFinalCommit = false;
|
||||
this.dropUncommittedNonFinalTranscripts(state);
|
||||
} else {
|
||||
state.awaitingFinalCommit = true;
|
||||
try {
|
||||
state.stt.commit();
|
||||
} catch (error) {
|
||||
this.failAndCleanupStream(dictationId, error?.message || String(error), true);
|
||||
return;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
state.awaitingFinalCommit = false;
|
||||
}
|
||||
|
||||
state.finishSealed = true;
|
||||
}
|
||||
|
||||
dropUncommittedNonFinalTranscripts(state) {
|
||||
const committedSet = new Set(state.committedSegmentIds);
|
||||
for (const segmentId of Array.from(state.transcriptsBySegmentId.keys())) {
|
||||
if (committedSet.has(segmentId)) {
|
||||
continue;
|
||||
}
|
||||
if (state.finalTranscriptSegmentIds.has(segmentId)) {
|
||||
continue;
|
||||
}
|
||||
state.transcriptsBySegmentId.delete(segmentId);
|
||||
}
|
||||
}
|
||||
|
||||
maybeFinalizeStream(dictationId) {
|
||||
const state = this.streams.get(dictationId);
|
||||
if (!state) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (!state.finishRequested || state.finalSeq === null) {
|
||||
return;
|
||||
}
|
||||
if (state.ackSeq < state.finalSeq) {
|
||||
return;
|
||||
}
|
||||
if (state.awaitingFinalCommit) {
|
||||
return;
|
||||
}
|
||||
|
||||
const committedSet = new Set(state.committedSegmentIds);
|
||||
const orderedSegmentIds = [...state.committedSegmentIds];
|
||||
for (const segmentId of state.transcriptsBySegmentId.keys()) {
|
||||
if (!committedSet.has(segmentId)) {
|
||||
orderedSegmentIds.push(segmentId);
|
||||
}
|
||||
}
|
||||
|
||||
if (orderedSegmentIds.length === 0) {
|
||||
this.emit({ type: 'final', payload: { dictationId, text: '' } });
|
||||
this.cleanupStream(dictationId);
|
||||
return;
|
||||
}
|
||||
|
||||
const allTranscriptsReady = orderedSegmentIds.every((segmentId) =>
|
||||
state.finalTranscriptSegmentIds.has(segmentId),
|
||||
);
|
||||
if (!allTranscriptsReady) {
|
||||
return;
|
||||
}
|
||||
|
||||
const orderedText = orderedSegmentIds
|
||||
.map((segmentId) => state.transcriptsBySegmentId.get(segmentId) ?? '')
|
||||
.join(' ')
|
||||
.trim();
|
||||
|
||||
this.emit({ type: 'final', payload: { dictationId, text: orderedText } });
|
||||
this.cleanupStream(dictationId);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,194 @@
|
||||
import { describe, it, expect } from 'bun:test';
|
||||
import { EventEmitter } from 'events';
|
||||
|
||||
import { DictationStreamManager } from './stream-manager.js';
|
||||
|
||||
const FORMAT = 'audio/pcm;rate=16000;bits=16';
|
||||
|
||||
class FakeSttSession extends EventEmitter {
|
||||
constructor({ transcriptBySegment = () => 'hello world' } = {}) {
|
||||
super();
|
||||
this.requiredSampleRate = 16000;
|
||||
this.appended = [];
|
||||
this.commits = 0;
|
||||
this.clears = 0;
|
||||
this.closed = false;
|
||||
this.segmentCounter = 0;
|
||||
this.transcriptBySegment = transcriptBySegment;
|
||||
}
|
||||
|
||||
async connect() {}
|
||||
|
||||
appendPcm16(buf) {
|
||||
this.appended.push(buf);
|
||||
}
|
||||
|
||||
commit() {
|
||||
this.commits += 1;
|
||||
const segmentId = `seg-${this.segmentCounter}`;
|
||||
this.segmentCounter += 1;
|
||||
this.emit('committed', { segmentId, previousSegmentId: null });
|
||||
setTimeout(() => {
|
||||
this.emit('transcript', {
|
||||
segmentId,
|
||||
transcript: this.transcriptBySegment(segmentId),
|
||||
isFinal: true,
|
||||
});
|
||||
}, 0);
|
||||
}
|
||||
|
||||
clear() {
|
||||
this.clears += 1;
|
||||
}
|
||||
|
||||
close() {
|
||||
this.closed = true;
|
||||
}
|
||||
}
|
||||
|
||||
function loudChunkBase64(samples = 1600, amplitude = 8000) {
|
||||
const arr = new Int16Array(samples);
|
||||
for (let i = 0; i < samples; i += 1) {
|
||||
arr[i] = i % 2 === 0 ? amplitude : -amplitude;
|
||||
}
|
||||
return Buffer.from(arr.buffer).toString('base64');
|
||||
}
|
||||
|
||||
function silentChunkBase64(samples = 1600) {
|
||||
return Buffer.from(new Int16Array(samples).buffer).toString('base64');
|
||||
}
|
||||
|
||||
function createManager(session) {
|
||||
const messages = [];
|
||||
const manager = new DictationStreamManager({
|
||||
emit: (msg) => messages.push(msg),
|
||||
createSttSession: async () => ({ session }),
|
||||
});
|
||||
return { manager, messages };
|
||||
}
|
||||
|
||||
function waitFor(predicate, timeoutMs = 1000) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const startedAt = Date.now();
|
||||
const tick = () => {
|
||||
if (predicate()) {
|
||||
resolve(undefined);
|
||||
return;
|
||||
}
|
||||
if (Date.now() - startedAt > timeoutMs) {
|
||||
reject(new Error('waitFor timed out'));
|
||||
return;
|
||||
}
|
||||
setTimeout(tick, 5);
|
||||
};
|
||||
tick();
|
||||
});
|
||||
}
|
||||
|
||||
describe('DictationStreamManager', () => {
|
||||
it('transcribes ordered chunks and emits final text', async () => {
|
||||
const session = new FakeSttSession();
|
||||
const { manager, messages } = createManager(session);
|
||||
|
||||
await manager.handleStart('d1', FORMAT, {});
|
||||
manager.handleChunk({ dictationId: 'd1', seq: 0, audioBase64: loudChunkBase64() });
|
||||
manager.handleChunk({ dictationId: 'd1', seq: 1, audioBase64: loudChunkBase64() });
|
||||
manager.handleFinish('d1', 1);
|
||||
|
||||
await waitFor(() => messages.some((m) => m.type === 'final'));
|
||||
|
||||
const final = messages.find((m) => m.type === 'final');
|
||||
expect(final.payload.text).toBe('hello world');
|
||||
expect(session.commits).toBe(1);
|
||||
expect(session.closed).toBe(true);
|
||||
|
||||
const acks = messages.filter((m) => m.type === 'ack');
|
||||
expect(acks[acks.length - 1].payload.ackSeq).toBe(1);
|
||||
});
|
||||
|
||||
it('reorders out-of-order chunks before appending', async () => {
|
||||
const session = new FakeSttSession();
|
||||
const { manager, messages } = createManager(session);
|
||||
|
||||
await manager.handleStart('d1', FORMAT, {});
|
||||
manager.handleChunk({ dictationId: 'd1', seq: 1, audioBase64: loudChunkBase64() });
|
||||
expect(session.appended.length).toBe(0);
|
||||
manager.handleChunk({ dictationId: 'd1', seq: 0, audioBase64: loudChunkBase64() });
|
||||
expect(session.appended.length).toBe(2);
|
||||
manager.handleFinish('d1', 1);
|
||||
|
||||
await waitFor(() => messages.some((m) => m.type === 'final'));
|
||||
});
|
||||
|
||||
it('clears silence-only tails instead of committing', async () => {
|
||||
const session = new FakeSttSession();
|
||||
const { manager, messages } = createManager(session);
|
||||
|
||||
await manager.handleStart('d1', FORMAT, {});
|
||||
manager.handleChunk({ dictationId: 'd1', seq: 0, audioBase64: silentChunkBase64() });
|
||||
manager.handleFinish('d1', 0);
|
||||
|
||||
await waitFor(() => messages.some((m) => m.type === 'final'));
|
||||
|
||||
const final = messages.find((m) => m.type === 'final');
|
||||
expect(final.payload.text).toBe('');
|
||||
expect(session.commits).toBe(0);
|
||||
expect(session.clears).toBe(1);
|
||||
});
|
||||
|
||||
it('fails fast when finish arrives with no chunks', async () => {
|
||||
const session = new FakeSttSession();
|
||||
const { manager, messages } = createManager(session);
|
||||
|
||||
await manager.handleStart('d1', FORMAT, {});
|
||||
manager.handleFinish('d1', 3);
|
||||
|
||||
const error = messages.find((m) => m.type === 'error');
|
||||
expect(error).toBeDefined();
|
||||
expect(error.payload.retryable).toBe(true);
|
||||
expect(session.closed).toBe(true);
|
||||
});
|
||||
|
||||
it('reports provider readiness errors from createSttSession', async () => {
|
||||
const messages = [];
|
||||
const manager = new DictationStreamManager({
|
||||
emit: (msg) => messages.push(msg),
|
||||
createSttSession: async () => ({
|
||||
error: 'Dictation model is downloading',
|
||||
retryable: true,
|
||||
reasonCode: 'model_download_in_progress',
|
||||
}),
|
||||
});
|
||||
|
||||
await manager.handleStart('d1', FORMAT, {});
|
||||
const error = messages.find((m) => m.type === 'error');
|
||||
expect(error.payload.reasonCode).toBe('model_download_in_progress');
|
||||
expect(error.payload.retryable).toBe(true);
|
||||
});
|
||||
|
||||
it('emits partials as segment transcripts arrive', async () => {
|
||||
let segment = 0;
|
||||
const session = new FakeSttSession({
|
||||
transcriptBySegment: () => {
|
||||
segment += 1;
|
||||
return segment === 1 ? 'first part' : 'second part';
|
||||
},
|
||||
});
|
||||
const { manager, messages } = createManager(session);
|
||||
// Force auto-commit after ~0.05s of audio so two segments form.
|
||||
manager.autoCommitSeconds = 0.05;
|
||||
|
||||
await manager.handleStart('d1', FORMAT, {});
|
||||
manager.handleChunk({ dictationId: 'd1', seq: 0, audioBase64: loudChunkBase64(1600) });
|
||||
await waitFor(() => session.commits >= 1);
|
||||
manager.handleChunk({ dictationId: 'd1', seq: 1, audioBase64: loudChunkBase64(1600) });
|
||||
manager.handleFinish('d1', 1);
|
||||
|
||||
await waitFor(() => messages.some((m) => m.type === 'final'));
|
||||
|
||||
const final = messages.find((m) => m.type === 'final');
|
||||
expect(final.payload.text).toBe('first part second part');
|
||||
const partials = messages.filter((m) => m.type === 'partial');
|
||||
expect(partials.length).toBeGreaterThan(0);
|
||||
});
|
||||
});
|
||||
@@ -721,10 +721,18 @@ export const createSettingsHelpers = (dependencies) => {
|
||||
}
|
||||
}
|
||||
|
||||
if (typeof candidate.dictationEnabled === 'boolean') {
|
||||
result.dictationEnabled = candidate.dictationEnabled;
|
||||
}
|
||||
if (typeof candidate.sttProvider === 'string') {
|
||||
const provider = candidate.sttProvider.trim();
|
||||
if (provider === 'browser' || provider === 'server' || provider === 'wasm') {
|
||||
if (provider === 'local' || provider === 'openai-compatible') {
|
||||
result.sttProvider = provider;
|
||||
} else if (provider === 'server') {
|
||||
// Legacy provider migration: 'server' was the OpenAI-compatible endpoint.
|
||||
result.sttProvider = 'openai-compatible';
|
||||
} else if (provider === 'browser' || provider === 'wasm') {
|
||||
result.sttProvider = 'local';
|
||||
}
|
||||
}
|
||||
if (typeof candidate.sttServerUrl === 'string') {
|
||||
@@ -739,10 +747,10 @@ export const createSettingsHelpers = (dependencies) => {
|
||||
result.sttModel = trimmed;
|
||||
}
|
||||
}
|
||||
if (typeof candidate.wasmSttModel === 'string') {
|
||||
const trimmed = candidate.wasmSttModel.trim();
|
||||
if (trimmed.length <= 256) {
|
||||
result.wasmSttModel = trimmed;
|
||||
if (typeof candidate.sttLocalModel === 'string') {
|
||||
const trimmed = candidate.sttLocalModel.trim();
|
||||
if (trimmed.length <= STT_MODEL_MAX_LENGTH) {
|
||||
result.sttLocalModel = trimmed;
|
||||
}
|
||||
}
|
||||
if (typeof candidate.sttLanguage === 'string') {
|
||||
@@ -751,15 +759,6 @@ export const createSettingsHelpers = (dependencies) => {
|
||||
result.sttLanguage = trimmed;
|
||||
}
|
||||
}
|
||||
if (typeof candidate.sttSilenceThresholdDb === 'number' && Number.isFinite(candidate.sttSilenceThresholdDb)) {
|
||||
result.sttSilenceThresholdDb = Math.max(-100, Math.min(0, candidate.sttSilenceThresholdDb));
|
||||
}
|
||||
if (typeof candidate.sttSilenceHoldMs === 'number' && Number.isFinite(candidate.sttSilenceHoldMs)) {
|
||||
result.sttSilenceHoldMs = Math.max(250, Math.min(10000, Math.round(candidate.sttSilenceHoldMs)));
|
||||
}
|
||||
if (typeof candidate.sttTranscribeOnStop === 'boolean') {
|
||||
result.sttTranscribeOnStop = candidate.sttTranscribeOnStop;
|
||||
}
|
||||
|
||||
return result;
|
||||
};
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
export const createStartupPipelineRuntime = (dependencies) => {
|
||||
const {
|
||||
createTerminalRuntime,
|
||||
createDictationRuntime,
|
||||
createMessageStreamWsRuntime,
|
||||
createServerStartupRuntime,
|
||||
} = dependencies;
|
||||
@@ -52,6 +53,7 @@ export const createStartupPipelineRuntime = (dependencies) => {
|
||||
tunnelRuntimeContext,
|
||||
attachSignals,
|
||||
apiOnly,
|
||||
dictationModelsDir,
|
||||
} = options;
|
||||
|
||||
const terminalRuntime = createTerminalRuntime({
|
||||
@@ -71,6 +73,16 @@ export const createStartupPipelineRuntime = (dependencies) => {
|
||||
TERMINAL_INPUT_WS_MAX_REBINDS_PER_WINDOW: terminalMaxRebindsPerWindow,
|
||||
});
|
||||
|
||||
const dictationRuntime = createDictationRuntime({
|
||||
app,
|
||||
server,
|
||||
express,
|
||||
uiAuthController,
|
||||
isRequestOriginAllowed,
|
||||
rejectWebSocketUpgrade,
|
||||
modelsDir: dictationModelsDir,
|
||||
});
|
||||
|
||||
const messageStreamRuntime = createMessageStreamWsRuntime({
|
||||
server,
|
||||
uiAuthController,
|
||||
@@ -125,6 +137,7 @@ export const createStartupPipelineRuntime = (dependencies) => {
|
||||
|
||||
return {
|
||||
terminalRuntime,
|
||||
dictationRuntime,
|
||||
messageStreamRuntime,
|
||||
};
|
||||
};
|
||||
|
||||
@@ -1,6 +1,16 @@
|
||||
export const createRequestSecurityRuntime = (deps) => {
|
||||
const { readSettingsFromDiskMigrated } = deps;
|
||||
const packagedClientOrigins = new Set(['openchamber-ui://app', 'capacitor://localhost']);
|
||||
// Origins of packaged (non-browser) clients whose WebView origin never
|
||||
// matches the server host: the desktop shell, the iOS Capacitor WebView
|
||||
// (capacitor://localhost), and the Android Capacitor WebView, which uses
|
||||
// androidScheme 'https' and therefore reports 'https://localhost'. Missing
|
||||
// the Android origin 403'd every WebSocket upgrade from the Android app
|
||||
// (message stream, terminal, dictation) while SSE kept working.
|
||||
const packagedClientOrigins = new Set([
|
||||
'openchamber-ui://app',
|
||||
'capacitor://localhost',
|
||||
'https://localhost',
|
||||
]);
|
||||
|
||||
const getUiSessionTokenFromRequest = (req) => {
|
||||
const cookieHeader = req?.headers?.cookie;
|
||||
|
||||
@@ -24,5 +24,26 @@ describe('request security runtime', () => {
|
||||
},
|
||||
socket: {},
|
||||
})).resolves.toBe(true);
|
||||
|
||||
// Android Capacitor WebView (androidScheme 'https') reports this origin.
|
||||
await expect(runtime.isRequestOriginAllowed({
|
||||
headers: {
|
||||
origin: 'https://localhost',
|
||||
host: '192.168.1.130:1202',
|
||||
},
|
||||
socket: {},
|
||||
})).resolves.toBe(true);
|
||||
});
|
||||
|
||||
test('rejects unknown origins', async () => {
|
||||
const runtime = createRuntime();
|
||||
|
||||
await expect(runtime.isRequestOriginAllowed({
|
||||
headers: {
|
||||
origin: 'https://evil.example.com',
|
||||
host: '192.168.1.130:1202',
|
||||
},
|
||||
socket: {},
|
||||
})).resolves.toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -310,6 +310,7 @@ const isUrlAuthWebSocketPath = (pathname) => {
|
||||
|| pathname === '/api/global/event/ws'
|
||||
|| pathname === '/api/openchamber/realtime-proxy/ws'
|
||||
|| pathname === '/api/terminal/ws'
|
||||
|| pathname === '/api/dictation/ws'
|
||||
|| pathname.startsWith('/api/preview/proxy/');
|
||||
};
|
||||
|
||||
|
||||
@@ -213,6 +213,28 @@ describe('ui auth client credential seam', () => {
|
||||
});
|
||||
expect(mountedServeCalled).toBe(true);
|
||||
|
||||
const dictationWsReq = {
|
||||
method: 'GET',
|
||||
path: '/api/dictation/ws',
|
||||
url: `/api/dictation/ws?oc_url_token=${encodeURIComponent(urlToken)}`,
|
||||
headers: { upgrade: 'websocket' },
|
||||
};
|
||||
expect(await auth.ensureSessionToken(dictationWsReq, null)).toBe('client:device-1');
|
||||
|
||||
const dictationHttpReq = {
|
||||
method: 'GET',
|
||||
path: '/api/dictation/ws',
|
||||
url: `/api/dictation/ws?oc_url_token=${encodeURIComponent(urlToken)}`,
|
||||
headers: { accept: 'application/json' },
|
||||
};
|
||||
const dictationHttpRes = createResponse();
|
||||
let dictationHttpCalled = false;
|
||||
await auth.requireAuth(dictationHttpReq, dictationHttpRes, () => {
|
||||
dictationHttpCalled = true;
|
||||
});
|
||||
expect(dictationHttpCalled).toBe(false);
|
||||
expect(dictationHttpRes.statusCode).toBe(401);
|
||||
|
||||
const arbitraryGetReq = { method: 'GET', path: '/api/config/settings', url: `/api/config/settings?oc_url_token=${encodeURIComponent(urlToken)}`, headers: { accept: 'application/json' } };
|
||||
const arbitraryGetRes = createResponse();
|
||||
let arbitraryGetCalled = false;
|
||||
|
||||
Reference in New Issue
Block a user