diff --git a/config/max-lines-baseline.txt b/config/max-lines-baseline.txt index 75e536a2198..8d2496f1d69 100644 --- a/config/max-lines-baseline.txt +++ b/config/max-lines-baseline.txt @@ -28,7 +28,6 @@ inline src/main/runtime/orca-runtime.ts inline src/main/runtime/rpc/methods/orchestration.ts inline src/main/runtime/runtime-rpc.ts inline src/main/source-control/hosted-review-creation.ts -inline src/main/speech/stt-service.ts inline src/main/ssh/ssh-channel-multiplexer.ts inline src/main/ssh/ssh-connection.ts inline src/main/ssh/ssh-relay-deploy.ts diff --git a/src/main/speech/stt-service.ts b/src/main/speech/stt-service.ts index a840d7aa256..21d3fba2714 100644 --- a/src/main/speech/stt-service.ts +++ b/src/main/speech/stt-service.ts @@ -1,16 +1,9 @@ -/* eslint-disable max-lines -- Why: speech worker ownership, warm reuse, and -timeout teardown stay co-located so dictation lifecycle state cannot drift. */ -import { Worker } from 'node:worker_threads' -import { getCatalogModel } from './model-catalog' import type { ModelManager } from './model-manager' -import { OpenAiTranscriptionSession } from './openai-transcription-client' -import { readOpenAiSpeechApiKey } from './openai-api-key-store' -import { getSherpaModulePath, getSttWorkerPath } from './stt-worker-paths' -import { waitForSttWorkerStop, type SttWorkerStopOutcome } from './stt-worker-stop' +import { startSttDictation } from './stt-session-start' +import { createSttSessionState, type SttSessionState } from './stt-session-state' +import { prepareSttModelForDeletion, stopSttDictation } from './stt-session-stop' -export const START_DICTATION_TIMEOUT_MS = 60_000 -const STOP_DICTATION_TIMEOUT_MS = 60_000 -export const IDLE_WORKER_TEARDOWN_MS = 60 * 60 * 1000 +export { IDLE_WORKER_TEARDOWN_MS, START_DICTATION_TIMEOUT_MS } from './stt-session-timeouts' export type SttEvent = | { type: 'ready' } @@ -21,481 +14,58 @@ export type SttEvent = export type SttEventSink = (event: SttEvent) => void -type StopInFlight = { - worker: Worker - owner: string - promise: Promise -} - export class SttService { - private worker: Worker | null = null - private cloudSession: OpenAiTranscriptionSession | null = null - private modelManager: ModelManager - private activeModelId: string | null = null - private activeHotwordsFilePath: string | undefined - private activeOwner: string | null = null - private startingOwner: string | null = null - private startingModelId: string | null = null - private starting = false - private canceledOwners = new Set() - private eventSink: SttEventSink | null = null - private idleTeardownTimer: NodeJS.Timeout | null = null - private stopInFlight: StopInFlight | null = null - // Why: stop resolves only after the worker flushes; in-flight feedAudio IPC - // must not enqueue samples after that flush or they stick on the warm worker - // and contaminate the next dictation session. - private stopping = false - // Why: warm workers intentionally keep lifecycle listeners while reusable; - // stale workers must not retain this service after error, exit, or teardown. - private cleanupWorkerLifecycleListeners: (() => void) | null = null + private readonly state: SttSessionState constructor(modelManager: ModelManager) { - this.modelManager = modelManager + this.state = createSttSessionState(modelManager) } - async startDictation( + startDictation( modelId: string, sink: SttEventSink, hotwordsFilePath?: string, owner = 'desktop' ): Promise { - if (this.starting) { - if (this.startingOwner !== owner) { - throw new Error('dictation_already_active') - } - return - } - if ((this.worker || this.cloudSession) && this.activeOwner && this.activeOwner !== owner) { - throw new Error('dictation_already_active') - } - this.starting = true - this.startingOwner = owner - this.startingModelId = modelId - this.clearIdleTeardownTimer() - - try { - await this._startDictation(modelId, sink, hotwordsFilePath, owner) - if (this.canceledOwners.delete(owner)) { - await this.stopDictation(owner, { cancelStarting: false }) - throw new Error('dictation_canceled') - } - this.activeOwner = owner - } finally { - this.starting = false - this.startingOwner = null - this.startingModelId = null - this.canceledOwners.delete(owner) - } - } - - private async _startDictation( - modelId: string, - sink: SttEventSink, - hotwordsFilePath?: string, - owner = 'desktop' - ): Promise { - const manifest = getCatalogModel(modelId) - if (!manifest) { - throw new Error(`Unknown model: ${modelId}`) - } - - if (manifest.provider === 'openai') { - if (this.worker) { - const existingWorker = this.worker - await this.stopDictation(owner, { cancelStarting: false }) - await this.teardownWorker(existingWorker) - } - - const modelState = await this.modelManager.getModelState(modelId) - if (modelState.status !== 'ready') { - throw new Error(`Model not ready: ${modelState.status}`) - } - - this.cloudSession = new OpenAiTranscriptionSession(modelId, readOpenAiSpeechApiKey) - this.activeModelId = modelId - this.activeHotwordsFilePath = undefined - this.eventSink = sink - sink({ type: 'ready' }) - return - } - - if (this.cloudSession) { - await this.stopDictation(owner, { cancelStarting: false }) - } - - const reusableWorker = this.worker - if ( - reusableWorker && - this.activeModelId === modelId && - this.activeHotwordsFilePath === hotwordsFilePath && - this.stopInFlight?.worker !== reusableWorker - ) { - const worker = reusableWorker - if (!this.activeOwner) { - const modelState = await this.modelManager.getModelState(modelId) - if (modelState.status !== 'ready') { - await this.teardownWorker(worker) - throw new Error(`Model not ready: ${modelState.status}`) - } - } - if ( - this.worker === worker && - this.activeModelId === modelId && - this.activeHotwordsFilePath === hotwordsFilePath && - this.stopInFlight?.worker !== worker - ) { - this.eventSink = sink - sink({ type: 'ready' }) - return - } - } - - if (this.worker) { - const existingWorker = this.worker - await this.stopDictation(owner, { cancelStarting: false }) - await this.teardownWorker(existingWorker) - } - - const modelState = await this.modelManager.getModelState(modelId) - if (modelState.status !== 'ready') { - throw new Error(`Model not ready: ${modelState.status}`) - } - - const workerPath = getSttWorkerPath() - const sherpaModulePath = getSherpaModulePath() - - this.worker = new Worker(workerPath, { - workerData: { sherpaModulePath } - }) - const worker = this.worker - - this.activeModelId = modelId - this.activeHotwordsFilePath = hotwordsFilePath - this.eventSink = sink - - const readyPromise = new Promise((resolve, reject) => { - let settled = false - let startupTimeout: ReturnType | null = null - const cleanup = () => { - if (startupTimeout) { - clearTimeout(startupTimeout) - startupTimeout = null - } - worker.off('message', onReadyOrError) - worker.off('error', onStartupError) - worker.off('exit', onStartupExit) - } - const failStartup = (error: Error): void => { - if (settled) { - return - } - settled = true - cleanup() - reject(error) - } - const onReadyOrError = (msg: { type: string; text?: string; error?: string }) => { - if (settled) { - return - } - if (msg.type === 'ready') { - settled = true - cleanup() - resolve() - } else if (msg.type === 'error') { - failStartup(new Error(msg.error ?? 'Speech worker failed to initialize')) - } - } - const onStartupError = (err: Error) => { - failStartup(err) - } - const onStartupExit = (code: number) => { - failStartup(new Error(`Speech worker exited before ready: ${code}`)) - } - worker.on('message', onReadyOrError) - worker.on('error', onStartupError) - worker.on('exit', onStartupExit) - // Why: a native STT worker can wedge while loading model bindings without - // emitting ready/error/exit; startup must leave the UI's Starting state. - startupTimeout = setTimeout(() => { - failStartup(new Error('Speech worker timed out while starting.')) - }, START_DICTATION_TIMEOUT_MS) - startupTimeout.unref?.() - }) - - const onWorkerMessage = (msg: SttEvent) => { - if (this.worker === worker) { - this.eventSink?.(msg) - } - } - - const onWorkerError = (err: Error) => { - if (this.worker === worker) { - this.eventSink?.({ type: 'error', error: String(err) }) - this.cleanupActiveWorkerLifecycleListeners() - this.worker = null - this.activeModelId = null - this.activeHotwordsFilePath = undefined - this.activeOwner = null - this.eventSink = null - } - } - - const onWorkerExit = () => { - if (this.worker === worker) { - this.cleanupActiveWorkerLifecycleListeners() - this.worker = null - this.activeModelId = null - this.activeHotwordsFilePath = undefined - this.activeOwner = null - this.eventSink = null - } - } - - worker.on('message', onWorkerMessage) - worker.on('error', onWorkerError) - worker.on('exit', onWorkerExit) - this.cleanupWorkerLifecycleListeners = () => { - worker.off('message', onWorkerMessage) - worker.off('error', onWorkerError) - worker.off('exit', onWorkerExit) - } - - const modelDir = this.modelManager.getModelDir(modelId) - worker.postMessage({ - type: 'init', - modelDir, - modelType: manifest.type, - streaming: manifest.streaming, - sampleRate: manifest.sampleRate, - files: manifest.files ?? [], - hotwordsFilePath, - modelingUnit: manifest.modelingUnit - }) - - try { - await readyPromise - } catch (error) { - this.cleanupActiveWorkerLifecycleListeners() - worker.removeAllListeners() - void worker.terminate() - if (this.worker === worker) { - this.worker = null - this.activeModelId = null - this.activeHotwordsFilePath = undefined - this.activeOwner = null - this.eventSink = null - } - throw error - } + return startSttDictation(this.state, modelId, sink, hotwordsFilePath, owner) } feedAudio(samples: Float32Array, sampleRate: number, owner = 'desktop'): void { - if (this.stopping) { + if (this.state.stopping) { return } - const currentOwner = this.activeOwner ?? this.startingOwner + const currentOwner = this.state.activeOwner ?? this.state.startingOwner if (!currentOwner) { return } if (currentOwner !== owner) { throw new Error('dictation_owner_mismatch') } - if (this.cloudSession) { - this.cloudSession.feedAudio(samples, sampleRate) + if (this.state.cloudSession) { + this.state.cloudSession.feedAudio(samples, sampleRate) return } - this.worker?.postMessage({ type: 'feed', samples, sampleRate }, [samples.buffer as ArrayBuffer]) + this.state.worker?.postMessage({ type: 'feed', samples, sampleRate }, [ + samples.buffer as ArrayBuffer + ]) } - async stopDictation( + stopDictation( owner = 'desktop', options: { cancelStarting?: boolean } = { cancelStarting: true } ): Promise { - if (options.cancelStarting !== false && this.startingOwner === owner) { - this.canceledOwners.add(owner) - } - if (!this.worker && !this.cloudSession) { - return - } - const currentOwner = this.activeOwner ?? this.startingOwner - if (currentOwner && currentOwner !== owner) { - throw new Error('dictation_owner_mismatch') - } - - if (this.cloudSession) { - this.stopping = true - try { - const session = this.cloudSession - this.cloudSession = null - try { - const text = await session.finish() - if (text) { - this.eventSink?.({ type: 'final', text }) - } - } catch (error) { - this.eventSink?.({ - type: 'error', - error: error instanceof Error ? error.message : String(error) - }) - } finally { - this.eventSink?.({ type: 'stopped' }) - this.activeModelId = null - this.activeHotwordsFilePath = undefined - this.activeOwner = null - this.eventSink = null - } - } finally { - this.stopping = false - } - return - } - - const worker = this.worker - if (!worker) { - return - } - if (this.stopInFlight?.worker === worker) { - if (this.stopInFlight.owner !== owner) { - throw new Error('dictation_owner_mismatch') - } - return this.stopInFlight.promise - } - - const capturedSink = this.eventSink - let stopPromise!: Promise - stopPromise = this.createStopPromise(worker, capturedSink).finally(() => { - if (this.stopInFlight?.worker === worker && this.stopInFlight.promise === stopPromise) { - this.stopInFlight = null - } - }) - this.stopInFlight = { worker, owner, promise: stopPromise } - this.stopping = true - try { - // Why: keep the stop message inside the try so a postMessage throw still - // clears `stopping` — otherwise feedAudio silently drops all future audio. - worker.postMessage({ type: 'stop' }) - await stopPromise - } finally { - this.stopping = false - } - } - - private createStopPromise(worker: Worker, capturedSink: SttEventSink | null): Promise { - return waitForSttWorkerStop({ - worker, - capturedSink, - timeoutMs: STOP_DICTATION_TIMEOUT_MS, - finish: (outcome) => this.finishWorkerStop(worker, outcome) - }) - } - - private finishWorkerStop(worker: Worker, outcome: SttWorkerStopOutcome): void { - if (outcome === 'stopped') { - if (this.worker === worker) { - this.activeOwner = null - this.eventSink = null - this.scheduleIdleTeardown() - } - return - } - // Why: a worker that cannot finish dictation is no longer reusable; drop - // its lifecycle listeners so a stale worker can't retain this service. - this.cleanupActiveWorkerLifecycleListeners() - worker.removeAllListeners() - if (outcome !== 'exit') { - void worker.terminate().catch(() => undefined) - } - if (this.worker === worker) { - this.worker = null - this.activeModelId = null - this.activeHotwordsFilePath = undefined - this.activeOwner = null - this.eventSink = null - } + return stopSttDictation(this.state, owner, options) } isActive(): boolean { - return this.worker !== null || this.cloudSession !== null + return this.state.worker !== null || this.state.cloudSession !== null } getActiveModelId(): string | null { - return this.activeModelId + return this.state.activeModelId } - async prepareModelForDeletion(modelId: string): Promise { - if (this.startingModelId === modelId || (this.activeOwner && this.activeModelId === modelId)) { - throw new Error('voice_model_in_use') - } - if (this.worker && this.activeModelId === modelId) { - await this.teardownIdleWorker({ ignoreTerminateErrors: false }) - if (this.worker && this.activeModelId === modelId) { - throw new Error('voice_model_in_use') - } - } - } - - private clearIdleTeardownTimer(): void { - if (this.idleTeardownTimer) { - clearTimeout(this.idleTeardownTimer) - this.idleTeardownTimer = null - } - } - - private scheduleIdleTeardown(): void { - this.clearIdleTeardownTimer() - // Why: keep the native recognizer warm for repeated dictations, but release - // the ONNX model after a quiet period so long-running Orca sessions don't - // pin speech memory forever. - this.idleTeardownTimer = setTimeout(() => { - void this.teardownIdleWorker() - }, IDLE_WORKER_TEARDOWN_MS) - this.idleTeardownTimer.unref?.() - } - - private async teardownIdleWorker( - options: { ignoreTerminateErrors?: boolean } = { ignoreTerminateErrors: true } - ): Promise { - this.clearIdleTeardownTimer() - if (!this.worker || this.activeOwner || this.startingOwner) { - return - } - await this.teardownWorker(this.worker, options) - } - - private async teardownWorker( - worker: Worker, - options: { ignoreTerminateErrors?: boolean } = { ignoreTerminateErrors: true } - ): Promise { - this.clearIdleTeardownTimer() - if (this.stopInFlight?.worker === worker) { - await this.stopInFlight.promise - } - try { - worker.postMessage({ type: 'teardown' }) - } catch { - // The worker may already have exited on a forced stop path. - } - this.cleanupActiveWorkerLifecycleListeners() - worker.removeAllListeners() - try { - await worker.terminate() - } catch (error) { - if (!options.ignoreTerminateErrors) { - throw error - } - } - if (this.worker === worker) { - this.worker = null - this.activeModelId = null - this.activeHotwordsFilePath = undefined - this.activeOwner = null - this.eventSink = null - } - } - - private cleanupActiveWorkerLifecycleListeners(): void { - const cleanup = this.cleanupWorkerLifecycleListeners - this.cleanupWorkerLifecycleListeners = null - cleanup?.() + prepareModelForDeletion(modelId: string): Promise { + return prepareSttModelForDeletion(this.state, modelId) } } diff --git a/src/main/speech/stt-session-start.ts b/src/main/speech/stt-session-start.ts new file mode 100644 index 00000000000..5377699b382 --- /dev/null +++ b/src/main/speech/stt-session-start.ts @@ -0,0 +1,164 @@ +import { Worker } from 'node:worker_threads' +import { getCatalogModel } from './model-catalog' +import { OpenAiTranscriptionSession } from './openai-transcription-client' +import { readOpenAiSpeechApiKey } from './openai-api-key-store' +import type { SttEventSink } from './stt-service' +import type { SttSessionState } from './stt-session-state' +import { + clearSttIdleTeardownTimer, + cleanupActiveSttWorkerLifecycleListeners, + handleSttWorkerFailure, + stopSttDictation, + teardownSttWorker +} from './stt-session-stop' +import { getSherpaModulePath, getSttWorkerPath } from './stt-worker-paths' +import { + attachSttWorkerLifecycle, + initializeSttWorker, + waitForSttWorkerReady +} from './stt-worker-startup' +import { START_DICTATION_TIMEOUT_MS } from './stt-session-timeouts' + +export async function startSttDictation( + state: SttSessionState, + modelId: string, + sink: SttEventSink, + hotwordsFilePath?: string, + owner = 'desktop' +): Promise { + if (state.starting) { + if (state.startingOwner !== owner) { + throw new Error('dictation_already_active') + } + return + } + if ((state.worker || state.cloudSession) && state.activeOwner && state.activeOwner !== owner) { + throw new Error('dictation_already_active') + } + state.starting = true + state.startingOwner = owner + state.startingModelId = modelId + clearSttIdleTeardownTimer(state) + + try { + await startSttSession(state, modelId, sink, hotwordsFilePath, owner) + if (state.canceledOwners.delete(owner)) { + await stopSttDictation(state, owner, { cancelStarting: false }) + throw new Error('dictation_canceled') + } + state.activeOwner = owner + } finally { + state.starting = false + state.startingOwner = null + state.startingModelId = null + state.canceledOwners.delete(owner) + } +} + +async function startSttSession( + state: SttSessionState, + modelId: string, + sink: SttEventSink, + hotwordsFilePath: string | undefined, + owner: string +): Promise { + const manifest = getCatalogModel(modelId) + if (!manifest) { + throw new Error(`Unknown model: ${modelId}`) + } + + if (manifest.provider === 'openai') { + if (state.worker) { + const existingWorker = state.worker + await stopSttDictation(state, owner, { cancelStarting: false }) + await teardownSttWorker(state, existingWorker) + } + const modelState = await state.modelManager.getModelState(modelId) + if (modelState.status !== 'ready') { + throw new Error(`Model not ready: ${modelState.status}`) + } + state.cloudSession = new OpenAiTranscriptionSession(modelId, readOpenAiSpeechApiKey) + state.activeModelId = modelId + state.activeHotwordsFilePath = undefined + state.eventSink = sink + sink({ type: 'ready' }) + return + } + + if (state.cloudSession) { + await stopSttDictation(state, owner, { cancelStarting: false }) + } + const reusableWorker = state.worker + if ( + reusableWorker && + state.activeModelId === modelId && + state.activeHotwordsFilePath === hotwordsFilePath && + state.stopInFlight?.worker !== reusableWorker + ) { + if (!state.activeOwner) { + const modelState = await state.modelManager.getModelState(modelId) + if (modelState.status !== 'ready') { + await teardownSttWorker(state, reusableWorker) + throw new Error(`Model not ready: ${modelState.status}`) + } + } + if ( + state.worker === reusableWorker && + state.activeModelId === modelId && + state.activeHotwordsFilePath === hotwordsFilePath && + state.stopInFlight?.worker !== reusableWorker + ) { + state.eventSink = sink + sink({ type: 'ready' }) + return + } + } + + if (state.worker) { + const existingWorker = state.worker + await stopSttDictation(state, owner, { cancelStarting: false }) + await teardownSttWorker(state, existingWorker) + } + const modelState = await state.modelManager.getModelState(modelId) + if (modelState.status !== 'ready') { + throw new Error(`Model not ready: ${modelState.status}`) + } + + const worker = new Worker(getSttWorkerPath(), { + workerData: { sherpaModulePath: getSherpaModulePath() } + }) + state.worker = worker + state.activeModelId = modelId + state.activeHotwordsFilePath = hotwordsFilePath + state.eventSink = sink + + const readyPromise = waitForSttWorkerReady(worker, START_DICTATION_TIMEOUT_MS) + state.cleanupWorkerLifecycleListeners = attachSttWorkerLifecycle({ + worker, + isCurrent: () => state.worker === worker, + onMessage: (event) => state.eventSink?.(event), + onError: (error) => handleSttWorkerFailure(state, error), + onExit: () => handleSttWorkerFailure(state) + }) + initializeSttWorker(worker, { + modelDir: state.modelManager.getModelDir(modelId), + modelType: manifest.type, + streaming: manifest.streaming, + sampleRate: manifest.sampleRate, + files: manifest.files ?? [], + hotwordsFilePath, + modelingUnit: manifest.modelingUnit + }) + + try { + await readyPromise + } catch (error) { + cleanupActiveSttWorkerLifecycleListeners(state) + worker.removeAllListeners() + void worker.terminate() + if (state.worker === worker) { + handleSttWorkerFailure(state) + } + throw error + } +} diff --git a/src/main/speech/stt-session-state.ts b/src/main/speech/stt-session-state.ts new file mode 100644 index 00000000000..f9054b41b8d --- /dev/null +++ b/src/main/speech/stt-session-state.ts @@ -0,0 +1,48 @@ +import type { Worker } from 'node:worker_threads' +import type { ModelManager } from './model-manager' +import type { OpenAiTranscriptionSession } from './openai-transcription-client' +import type { SttEventSink } from './stt-service' + +export type StopInFlight = { + worker: Worker + owner: string + promise: Promise +} + +export type SttSessionState = { + worker: Worker | null + cloudSession: OpenAiTranscriptionSession | null + modelManager: ModelManager + activeModelId: string | null + activeHotwordsFilePath: string | undefined + activeOwner: string | null + startingOwner: string | null + startingModelId: string | null + starting: boolean + canceledOwners: Set + eventSink: SttEventSink | null + idleTeardownTimer: NodeJS.Timeout | null + stopInFlight: StopInFlight | null + stopping: boolean + cleanupWorkerLifecycleListeners: (() => void) | null +} + +export function createSttSessionState(modelManager: ModelManager): SttSessionState { + return { + worker: null, + cloudSession: null, + modelManager, + activeModelId: null, + activeHotwordsFilePath: undefined, + activeOwner: null, + startingOwner: null, + startingModelId: null, + starting: false, + canceledOwners: new Set(), + eventSink: null, + idleTeardownTimer: null, + stopInFlight: null, + stopping: false, + cleanupWorkerLifecycleListeners: null + } +} diff --git a/src/main/speech/stt-session-stop.ts b/src/main/speech/stt-session-stop.ts new file mode 100644 index 00000000000..f8f1893a493 --- /dev/null +++ b/src/main/speech/stt-session-stop.ts @@ -0,0 +1,197 @@ +import type { Worker } from 'node:worker_threads' +import type { SttSessionState } from './stt-session-state' +import { waitForSttWorkerStop, type SttWorkerStopOutcome } from './stt-worker-stop' +import { IDLE_WORKER_TEARDOWN_MS } from './stt-session-timeouts' + +const STOP_DICTATION_TIMEOUT_MS = 60_000 + +export async function stopSttDictation( + state: SttSessionState, + owner = 'desktop', + options: { cancelStarting?: boolean } = { cancelStarting: true } +): Promise { + if (options.cancelStarting !== false && state.startingOwner === owner) { + state.canceledOwners.add(owner) + } + if (!state.worker && !state.cloudSession) { + return + } + const currentOwner = state.activeOwner ?? state.startingOwner + if (currentOwner && currentOwner !== owner) { + throw new Error('dictation_owner_mismatch') + } + + if (state.cloudSession) { + state.stopping = true + try { + const session = state.cloudSession + state.cloudSession = null + try { + const text = await session.finish() + if (text) { + state.eventSink?.({ type: 'final', text }) + } + } catch (error) { + state.eventSink?.({ + type: 'error', + error: error instanceof Error ? error.message : String(error) + }) + } finally { + state.eventSink?.({ type: 'stopped' }) + state.activeModelId = null + state.activeHotwordsFilePath = undefined + state.activeOwner = null + state.eventSink = null + } + } finally { + state.stopping = false + } + return + } + + const worker = state.worker + if (!worker) { + return + } + if (state.stopInFlight?.worker === worker) { + if (state.stopInFlight.owner !== owner) { + throw new Error('dictation_owner_mismatch') + } + return state.stopInFlight.promise + } + + const capturedSink = state.eventSink + let stopPromise!: Promise + stopPromise = waitForSttWorkerStop({ + worker, + capturedSink, + timeoutMs: STOP_DICTATION_TIMEOUT_MS, + finish: (outcome) => finishSttWorkerStop(state, worker, outcome) + }).finally(() => { + if (state.stopInFlight?.worker === worker && state.stopInFlight.promise === stopPromise) { + state.stopInFlight = null + } + }) + state.stopInFlight = { worker, owner, promise: stopPromise } + state.stopping = true + try { + worker.postMessage({ type: 'stop' }) + await stopPromise + } finally { + state.stopping = false + } +} + +function finishSttWorkerStop( + state: SttSessionState, + worker: Worker, + outcome: SttWorkerStopOutcome +): void { + if (outcome === 'stopped') { + if (state.worker === worker) { + state.activeOwner = null + state.eventSink = null + scheduleSttIdleTeardown(state) + } + return + } + cleanupActiveSttWorkerLifecycleListeners(state) + worker.removeAllListeners() + if (outcome !== 'exit') { + void worker.terminate().catch(() => undefined) + } + if (state.worker === worker) { + clearSttWorkerState(state) + } +} + +export async function prepareSttModelForDeletion( + state: SttSessionState, + modelId: string +): Promise { + if (state.startingModelId === modelId || (state.activeOwner && state.activeModelId === modelId)) { + throw new Error('voice_model_in_use') + } + if (state.worker && state.activeModelId === modelId) { + await teardownIdleSttWorker(state, { ignoreTerminateErrors: false }) + if (state.worker && state.activeModelId === modelId) { + throw new Error('voice_model_in_use') + } + } +} + +export function clearSttIdleTeardownTimer(state: SttSessionState): void { + if (state.idleTeardownTimer) { + clearTimeout(state.idleTeardownTimer) + state.idleTeardownTimer = null + } +} + +function scheduleSttIdleTeardown(state: SttSessionState): void { + clearSttIdleTeardownTimer(state) + state.idleTeardownTimer = setTimeout(() => { + void teardownIdleSttWorker(state) + }, IDLE_WORKER_TEARDOWN_MS) + state.idleTeardownTimer.unref?.() +} + +async function teardownIdleSttWorker( + state: SttSessionState, + options: { ignoreTerminateErrors?: boolean } = { ignoreTerminateErrors: true } +): Promise { + clearSttIdleTeardownTimer(state) + if (!state.worker || state.activeOwner || state.startingOwner) { + return + } + await teardownSttWorker(state, state.worker, options) +} + +export async function teardownSttWorker( + state: SttSessionState, + worker: Worker, + options: { ignoreTerminateErrors?: boolean } = { ignoreTerminateErrors: true } +): Promise { + clearSttIdleTeardownTimer(state) + if (state.stopInFlight?.worker === worker) { + await state.stopInFlight.promise + } + try { + worker.postMessage({ type: 'teardown' }) + } catch { + // The worker may already have exited on a forced stop path. + } + cleanupActiveSttWorkerLifecycleListeners(state) + worker.removeAllListeners() + try { + await worker.terminate() + } catch (error) { + if (!options.ignoreTerminateErrors) { + throw error + } + } + if (state.worker === worker) { + clearSttWorkerState(state) + } +} + +export function handleSttWorkerFailure(state: SttSessionState, error?: Error): void { + if (error) { + state.eventSink?.({ type: 'error', error: String(error) }) + } + cleanupActiveSttWorkerLifecycleListeners(state) + clearSttWorkerState(state) +} + +export function cleanupActiveSttWorkerLifecycleListeners(state: SttSessionState): void { + const cleanup = state.cleanupWorkerLifecycleListeners + state.cleanupWorkerLifecycleListeners = null + cleanup?.() +} + +function clearSttWorkerState(state: SttSessionState): void { + state.worker = null + state.activeModelId = null + state.activeHotwordsFilePath = undefined + state.activeOwner = null + state.eventSink = null +} diff --git a/src/main/speech/stt-session-timeouts.ts b/src/main/speech/stt-session-timeouts.ts new file mode 100644 index 00000000000..75dfddb845f --- /dev/null +++ b/src/main/speech/stt-session-timeouts.ts @@ -0,0 +1,3 @@ +/** Single source for the dictation timeouts the start and stop paths share. */ +export const START_DICTATION_TIMEOUT_MS = 60_000 +export const IDLE_WORKER_TEARDOWN_MS = 60 * 60 * 1000 diff --git a/src/main/speech/stt-worker-startup.ts b/src/main/speech/stt-worker-startup.ts new file mode 100644 index 00000000000..18653da1bd5 --- /dev/null +++ b/src/main/speech/stt-worker-startup.ts @@ -0,0 +1,97 @@ +import type { Worker } from 'node:worker_threads' +import type { SttEvent } from './stt-service' + +export function waitForSttWorkerReady(worker: Worker, timeoutMs: number): Promise { + const { promise, resolve, reject } = Promise.withResolvers() + let settled = false + let startupTimeout: ReturnType | null = null + const cleanup = (): void => { + if (startupTimeout) { + clearTimeout(startupTimeout) + startupTimeout = null + } + worker.off('message', onReadyOrError) + worker.off('error', onStartupError) + worker.off('exit', onStartupExit) + } + const failStartup = (error: Error): void => { + if (settled) { + return + } + settled = true + cleanup() + reject(error) + } + const onReadyOrError = (message: { type: string; error?: string }): void => { + if (settled) { + return + } + if (message.type === 'ready') { + settled = true + cleanup() + resolve() + } else if (message.type === 'error') { + failStartup(new Error(message.error ?? 'Speech worker failed to initialize')) + } + } + const onStartupError = (error: Error): void => failStartup(error) + const onStartupExit = (code: number): void => { + failStartup(new Error(`Speech worker exited before ready: ${code}`)) + } + worker.on('message', onReadyOrError) + worker.on('error', onStartupError) + worker.on('exit', onStartupExit) + startupTimeout = setTimeout( + () => failStartup(new Error('Speech worker timed out while starting.')), + timeoutMs + ) + startupTimeout.unref?.() + return promise +} + +export function attachSttWorkerLifecycle(args: { + worker: Worker + isCurrent: () => boolean + onMessage: (event: SttEvent) => void + onError: (error: Error) => void + onExit: () => void +}): () => void { + const onWorkerMessage = (event: SttEvent): void => { + if (args.isCurrent()) { + args.onMessage(event) + } + } + const onWorkerError = (error: Error): void => { + if (args.isCurrent()) { + args.onError(error) + } + } + const onWorkerExit = (): void => { + if (args.isCurrent()) { + args.onExit() + } + } + args.worker.on('message', onWorkerMessage) + args.worker.on('error', onWorkerError) + args.worker.on('exit', onWorkerExit) + return () => { + args.worker.off('message', onWorkerMessage) + args.worker.off('error', onWorkerError) + args.worker.off('exit', onWorkerExit) + } +} + +export function initializeSttWorker( + worker: Worker, + input: { + modelDir: string + modelType: string + streaming: boolean + sampleRate: number + files: readonly string[] + hotwordsFilePath?: string + modelingUnit?: string + } +): void { + worker.postMessage({ type: 'init', ...input }) +}