mirror of
https://github.com/stablyai/orca.git
synced 2026-09-22 08:02:28 +00:00
Split speech session lifecycle (#17123)
* Split speech session lifecycle * Fix F3-speech for #17123
This commit is contained in:
Neil
@@ -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
|
||||
|
||||
+21
-451
@@ -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<void>
|
||||
}
|
||||
|
||||
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<string>()
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void>((resolve, reject) => {
|
||||
let settled = false
|
||||
let startupTimeout: ReturnType<typeof setTimeout> | 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<void> {
|
||||
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<void>
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
return prepareSttModelForDeletion(this.state, modelId)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -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<void>
|
||||
}
|
||||
|
||||
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<string>
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -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<void> {
|
||||
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<void>
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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
|
||||
}
|
||||
@@ -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
|
||||
@@ -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<void> {
|
||||
const { promise, resolve, reject } = Promise.withResolvers<void>()
|
||||
let settled = false
|
||||
let startupTimeout: ReturnType<typeof setTimeout> | 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 })
|
||||
}
|
||||
Reference in New Issue
Block a user