From 7504d20b67431dfc772698525ce05f779b7c64a6 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 30 Mar 2026 14:07:55 +0700 Subject: [PATCH] refactor(server): extract SpeechService and defer init until after listen MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Speech runtime was leaking 8+ individual resolvers into bootstrap and blocking server startup for ~3s with synchronous Sherpa native model loading. Refactor into a self-contained SpeechService that owns its full lifecycle (init, download, monitor, cleanup) and defer start() until after httpServer.listen() so native loading doesn't block accepting connections. Server startup drops from ~3.2s to ~0.5s. Also removes unused client-triggered speech model download/list path (CLI commands, session handlers, message types) — the service manages its own models. --- packages/cli/src/commands/speech/download.ts | 63 --------- packages/cli/src/commands/speech/index.ts | 19 +-- packages/cli/src/commands/speech/models.ts | 67 --------- packages/server/src/client/daemon-client.ts | 30 ---- packages/server/src/server/bootstrap.ts | 32 ++--- packages/server/src/server/session.ts | 128 +----------------- .../src/server/speech/speech-runtime.test.ts | 20 +-- .../src/server/speech/speech-runtime.ts | 83 ++++++++---- .../server/src/server/websocket-server.ts | 55 +++----- packages/server/src/shared/messages.ts | 47 ------- 10 files changed, 100 insertions(+), 444 deletions(-) delete mode 100644 packages/cli/src/commands/speech/download.ts delete mode 100644 packages/cli/src/commands/speech/models.ts diff --git a/packages/cli/src/commands/speech/download.ts b/packages/cli/src/commands/speech/download.ts deleted file mode 100644 index ec5f71a4c..000000000 --- a/packages/cli/src/commands/speech/download.ts +++ /dev/null @@ -1,63 +0,0 @@ -import type { Command } from "commander"; -import type { CommandError, CommandOptions, ListResult, OutputSchema } from "../../output/index.js"; -import { connectToDaemon } from "../../utils/client.js"; - -interface SpeechDownloadRow { - modelId: string; - status: "downloaded"; -} - -const speechDownloadSchema: OutputSchema = { - idField: "modelId", - columns: [ - { header: "MODEL", field: "modelId", width: 36 }, - { header: "STATUS", field: "status", width: 12, color: () => "green" }, - ], -}; - -export type SpeechDownloadResult = ListResult; - -export interface SpeechDownloadOptions extends CommandOptions { - host?: string; - model?: string[]; -} - -export async function runSpeechDownloadCommand( - options: SpeechDownloadOptions, - _command: Command, -): Promise { - const client = await connectToDaemon({ host: options.host }); - try { - const response = await client.downloadSpeechModels({ - modelIds: options.model && options.model.length > 0 ? options.model : undefined, - }); - if (response.error) { - const commandError: CommandError = { - code: "SPEECH_MODELS_DOWNLOAD_FAILED", - message: response.error, - }; - throw commandError; - } - - return { - type: "list", - data: response.downloadedModelIds.map((modelId) => ({ - modelId, - status: "downloaded" as const, - })), - schema: speechDownloadSchema, - }; - } catch (error) { - if (typeof error === "object" && error && "code" in error && "message" in error) { - throw error; - } - const message = error instanceof Error ? error.message : String(error); - const commandError: CommandError = { - code: "SPEECH_MODELS_DOWNLOAD_FAILED", - message: `Failed to download speech models: ${message}`, - }; - throw commandError; - } finally { - await client.close().catch(() => {}); - } -} diff --git a/packages/cli/src/commands/speech/index.ts b/packages/cli/src/commands/speech/index.ts index 4fd451d29..e84d55b60 100644 --- a/packages/cli/src/commands/speech/index.ts +++ b/packages/cli/src/commands/speech/index.ts @@ -1,22 +1,5 @@ import { Command } from "commander"; -import { withOutput } from "../../output/index.js"; -import { runSpeechModelsCommand } from "./models.js"; -import { runSpeechDownloadCommand } from "./download.js"; -import { addJsonAndDaemonHostOptions, collectMultiple } from "../../utils/command-options.js"; export function createSpeechCommand(): Command { - const speech = new Command("speech").description("Manage local speech models"); - - addJsonAndDaemonHostOptions( - speech.command("models").description("List local speech model download status"), - ).action(withOutput(runSpeechModelsCommand)); - - addJsonAndDaemonHostOptions( - speech - .command("download") - .description("Download local speech models") - .option("--model ", "Model ID to download (repeatable)", collectMultiple, []), - ).action(withOutput(runSpeechDownloadCommand)); - - return speech; + return new Command("speech").description("Speech commands"); } diff --git a/packages/cli/src/commands/speech/models.ts b/packages/cli/src/commands/speech/models.ts deleted file mode 100644 index 9b5ca5d50..000000000 --- a/packages/cli/src/commands/speech/models.ts +++ /dev/null @@ -1,67 +0,0 @@ -import type { Command } from "commander"; -import type { CommandError, CommandOptions, ListResult, OutputSchema } from "../../output/index.js"; -import { connectToDaemon } from "../../utils/client.js"; - -interface SpeechModelListItem { - id: string; - kind: string; - status: "downloaded" | "missing"; - modelDir: string; - missingFiles: string; -} - -const speechModelsSchema: OutputSchema = { - idField: "id", - columns: [ - { header: "MODEL", field: "id", width: 36 }, - { header: "KIND", field: "kind", width: 12 }, - { - header: "STATUS", - field: "status", - width: 12, - color: (value) => (value === "downloaded" ? "green" : "yellow"), - }, - { header: "MODEL DIR", field: "modelDir", width: 44 }, - { header: "MISSING FILES", field: "missingFiles", width: 40 }, - ], -}; - -export type SpeechModelsResult = ListResult; - -export interface SpeechModelsOptions extends CommandOptions { - host?: string; -} - -export async function runSpeechModelsCommand( - options: SpeechModelsOptions, - _command: Command, -): Promise { - const client = await connectToDaemon({ host: options.host }); - try { - const response = await client.listSpeechModels(); - const rows: SpeechModelListItem[] = response.models - .slice() - .sort((a, b) => a.kind.localeCompare(b.kind) || a.id.localeCompare(b.id)) - .map((model) => ({ - id: model.id, - kind: model.kind, - status: model.isDownloaded ? "downloaded" : "missing", - modelDir: model.modelDir, - missingFiles: model.missingFiles?.join(", ") ?? "", - })); - return { - type: "list", - data: rows, - schema: speechModelsSchema, - }; - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - const commandError: CommandError = { - code: "SPEECH_MODELS_LIST_FAILED", - message: `Failed to list speech models: ${message}`, - }; - throw commandError; - } finally { - await client.close().catch(() => {}); - } -} diff --git a/packages/server/src/client/daemon-client.ts b/packages/server/src/client/daemon-client.ts index a9c12c341..450fa05da 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -39,8 +39,6 @@ import type { ListCommandsResponse, ListProviderModelsResponseMessage, ListAvailableProvidersResponse, - SpeechModelsListResponse, - SpeechModelsDownloadResponse, ListTerminalsResponse, CreateTerminalResponse, SubscribeTerminalResponse, @@ -217,8 +215,6 @@ type FileExplorerPayload = FileExplorerResponse["payload"]; type FileDownloadTokenPayload = FileDownloadTokenResponse["payload"]; type ListProviderModelsPayload = ListProviderModelsResponseMessage["payload"]; type ListAvailableProvidersPayload = ListAvailableProvidersResponse["payload"]; -type SpeechModelsListPayload = SpeechModelsListResponse["payload"]; -type SpeechModelsDownloadPayload = SpeechModelsDownloadResponse["payload"]; type ListCommandsPayload = ListCommandsResponse["payload"]; type ListCommandsDraftConfig = Pick< AgentSessionConfig, @@ -2487,32 +2483,6 @@ export class DaemonClient { }); } - async listSpeechModels(requestId?: string): Promise { - return this.sendCorrelatedSessionRequest({ - requestId, - message: { - type: "speech_models_list_request", - }, - responseType: "speech_models_list_response", - timeout: 30000, - }); - } - - async downloadSpeechModels(options?: { - modelIds?: string[]; - requestId?: string; - }): Promise { - return this.sendCorrelatedSessionRequest({ - requestId: options?.requestId, - message: { - type: "speech_models_download_request", - modelIds: options?.modelIds, - }, - responseType: "speech_models_download_response", - timeout: 30 * 60 * 1000, - }); - } - async listCommands(agentId: string, requestId?: string): Promise; async listCommands(agentId: string, options?: ListCommandsOptions): Promise; async listCommands( diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 0451a6b36..e73fe74a6 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -91,7 +91,7 @@ import { DownloadTokenStore } from "./file-download/token-store.js"; import type { OpenAiSpeechProviderConfig } from "./speech/providers/openai/config.js"; import type { LocalSpeechProviderConfig } from "./speech/providers/local/config.js"; import type { RequestedSpeechProviders } from "./speech/speech-types.js"; -import { initializeSpeechRuntime } from "./speech/speech-runtime.js"; +import { createSpeechService } from "./speech/speech-runtime.js"; import { AgentManager } from "./agent/agent-manager.js"; import { AgentStorage } from "./agent/agent-storage.js"; import { attachAgentStoragePersistence } from "./persistence-hooks.js"; @@ -426,7 +426,6 @@ export async function createPaseoDaemon( logger.info({ elapsed: elapsed() }, "Preparing voice and MCP runtime"); let wsServer: VoiceAssistantWebSocketServer | null = null; let voiceMcpBridgeManager: VoiceMcpSocketBridgeManager | null = null; - let unsubscribeSpeechReadiness: (() => void) | null = null; // Create in-memory transport for Session's Agent MCP client (voice assistant tools) const createInMemoryAgentMcpTransport = async (): Promise => { @@ -589,21 +588,12 @@ export async function createPaseoDaemon( }); }, }); - const { - resolveVoiceTurnDetection, - resolveVoiceStt, - resolveVoiceTts, - resolveDictationStt, - getSpeechReadiness, - subscribeSpeechReadiness, - cleanup: cleanupSpeechRuntime, - localModelConfig, - } = await initializeSpeechRuntime({ + const speechService = createSpeechService({ logger, openaiConfig: config.openai, speechConfig: config.speech, }); - logger.info({ elapsed: elapsed() }, "Speech runtime initialized"); + logger.info({ elapsed: elapsed() }, "Speech service created"); wsServer = new VoiceAssistantWebSocketServer( httpServer, @@ -615,7 +605,7 @@ export async function createPaseoDaemon( config.paseoHome, createInMemoryAgentMcpTransport, { allowedOrigins, allowedHosts: config.allowedHosts }, - { turnDetection: resolveVoiceTurnDetection, stt: resolveVoiceStt, tts: resolveVoiceTts }, + speechService, terminalManager, { voiceAgentMcpStdio: { @@ -633,9 +623,6 @@ export async function createPaseoDaemon( }, { finalTimeoutMs: config.dictationFinalTimeoutMs, - stt: resolveDictationStt, - localModels: localModelConfig ?? undefined, - getSpeechReadiness, }, config.agentProviderSettings, daemonVersion, @@ -653,9 +640,6 @@ export async function createPaseoDaemon( scheduleService, checkoutDiffManager, ); - unsubscribeSpeechReadiness = subscribeSpeechReadiness((snapshot) => { - wsServer?.publishSpeechReadiness(snapshot); - }); logger.info({ elapsed: elapsed() }, "Bootstrap complete, ready to start listening"); @@ -741,6 +725,10 @@ export async function createPaseoDaemon( httpServer.listen(listenTarget.path); } }); + + // Start speech service after listening so synchronous Sherpa native + // model loading doesn't block the server from accepting connections. + speechService.start(); }; const stop = async () => { @@ -752,9 +740,7 @@ export async function createPaseoDaemon( runtimeSettings: config.agentProviderSettings, }); terminalManager.killAll(); - unsubscribeSpeechReadiness?.(); - unsubscribeSpeechReadiness = null; - cleanupSpeechRuntime(); + speechService.stop(); await scheduleService.stop().catch(() => undefined); await relayTransport?.stop().catch(() => undefined); if (wsServer) { diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 24ac670cc..61d593c92 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -170,12 +170,7 @@ import { toCheckoutError, } from "./checkout-git-utils.js"; import { CheckoutDiffManager } from "./checkout-diff-manager.js"; -import { - ensureLocalSpeechModels, - getLocalSpeechModelDir, - listLocalSpeechModels, - type LocalSpeechModelId, -} from "./speech/providers/local/models.js"; +import type { LocalSpeechModelId } from "./speech/providers/local/models.js"; import { toResolver, type Resolvable } from "./speech/provider-resolver.js"; import type { SpeechReadinessSnapshot, SpeechReadinessState } from "./speech/speech-runtime.js"; import type pino from "pino"; @@ -401,10 +396,6 @@ export type SessionOptions = { dictation?: { finalTimeoutMs?: number; stt?: Resolvable; - localModels?: { - modelsDir: string; - defaultModelIds: LocalSpeechModelId[]; - }; getSpeechReadiness?: () => SpeechReadinessSnapshot; }; agentProviderRuntimeSettings?: AgentProviderRuntimeSettingsMap; @@ -596,8 +587,6 @@ export class Session { private readonly checkoutDiffSubscriptions = new Map void>(); private readonly workspaceGitWatchTargets = new Map(); private readonly voiceAgentMcpStdio: VoiceMcpStdioConfig | null; - private readonly localSpeechModelsDir: string; - private readonly defaultLocalSpeechModelIds: LocalSpeechModelId[]; private readonly registerVoiceSpeakHandler?: ( agentId: string, handler: VoiceSpeakHandler, @@ -669,15 +658,6 @@ export class Session { } this.voiceAgentMcpStdio = voice?.voiceAgentMcpStdio ?? null; this.resolveVoiceTurnDetection = toResolver(voice?.turnDetection ?? null); - const configuredModelsDir = dictation?.localModels?.modelsDir?.trim(); - this.localSpeechModelsDir = - configuredModelsDir && configuredModelsDir.length > 0 - ? configuredModelsDir - : join(this.paseoHome, "models", "local-speech"); - this.defaultLocalSpeechModelIds = - dictation?.localModels?.defaultModelIds && dictation.localModels.defaultModelIds.length > 0 - ? [...new Set(dictation.localModels.defaultModelIds)] - : ["parakeet-tdt-0.6b-v2-int8", "kokoro-en-v0_19"]; this.registerVoiceSpeakHandler = voiceBridge?.registerVoiceSpeakHandler; this.unregisterVoiceSpeakHandler = voiceBridge?.unregisterVoiceSpeakHandler; this.registerVoiceCallerContext = voiceBridge?.registerVoiceCallerContext; @@ -1665,14 +1645,6 @@ export class Session { await this.handleListAvailableProvidersRequest(msg); break; - case "speech_models_list_request": - await this.handleSpeechModelsListRequest(msg); - break; - - case "speech_models_download_request": - await this.handleSpeechModelsDownloadRequest(msg); - break; - case "clear_agent_attention": await this.handleClearAgentAttention(msg.agentId); break; @@ -3062,104 +3034,6 @@ export class Session { } } - private async handleSpeechModelsListRequest( - msg: Extract, - ): Promise { - const modelsDir = this.localSpeechModelsDir; - - const models = await Promise.all( - listLocalSpeechModels().map(async (model) => { - const modelDir = getLocalSpeechModelDir(modelsDir, model.id); - const missingFiles: string[] = []; - for (const rel of model.requiredFiles) { - const filePath = join(modelDir, rel); - try { - const fileStat = await stat(filePath); - if (fileStat.isDirectory()) { - continue; - } - if (!fileStat.isFile() || fileStat.size <= 0) { - missingFiles.push(rel); - } - } catch { - missingFiles.push(rel); - } - } - - return { - id: model.id, - kind: model.kind, - description: model.description, - modelDir, - isDownloaded: missingFiles.length === 0, - ...(missingFiles.length > 0 ? { missingFiles } : {}), - }; - }), - ); - - this.emit({ - type: "speech_models_list_response", - payload: { - modelsDir, - models, - requestId: msg.requestId, - }, - }); - } - - private async handleSpeechModelsDownloadRequest( - msg: Extract, - ): Promise { - const modelsDir = this.localSpeechModelsDir; - - const modelIdsRaw = - msg.modelIds && msg.modelIds.length > 0 ? msg.modelIds : this.defaultLocalSpeechModelIds; - - const allModelIds = new Set(listLocalSpeechModels().map((m) => m.id)); - const invalid = modelIdsRaw.filter((id) => !allModelIds.has(id as LocalSpeechModelId)); - if (invalid.length > 0) { - this.emit({ - type: "speech_models_download_response", - payload: { - modelsDir, - downloadedModelIds: [], - error: `Unknown speech model id(s): ${invalid.join(", ")}`, - requestId: msg.requestId, - }, - }); - return; - } - - const modelIds = modelIdsRaw as LocalSpeechModelId[]; - try { - await ensureLocalSpeechModels({ - modelsDir, - modelIds, - logger: this.sessionLogger, - }); - this.emit({ - type: "speech_models_download_response", - payload: { - modelsDir, - downloadedModelIds: modelIds, - error: null, - requestId: msg.requestId, - }, - }); - } catch (error) { - this.sessionLogger.error({ err: error, modelIds }, "Failed to download speech models"); - this.emit({ - type: "speech_models_download_response", - payload: { - modelsDir, - downloadedModelIds: [], - error: error instanceof Error ? error.message : String(error), - requestId: msg.requestId, - }, - }); - } - } - private normalizeGitOptions( gitOptions?: GitSetupOptions, legacyWorktreeName?: string, diff --git a/packages/server/src/server/speech/speech-runtime.test.ts b/packages/server/src/server/speech/speech-runtime.test.ts index e1ddf2ae2..a891e7656 100644 --- a/packages/server/src/server/speech/speech-runtime.test.ts +++ b/packages/server/src/server/speech/speech-runtime.test.ts @@ -5,7 +5,7 @@ import type { PaseoSpeechConfig } from "../bootstrap.js"; import type { InitializedLocalSpeech } from "./providers/local/runtime.js"; import type { SpeechToTextProvider, TextToSpeechProvider } from "./speech-provider.js"; import type { TurnDetectionProvider } from "./turn-detection-provider.js"; -import { initializeSpeechRuntime } from "./speech-runtime.js"; +import { createSpeechService } from "./speech-runtime.js"; const { initializeLocalSpeechServicesMock } = vi.hoisted(() => ({ initializeLocalSpeechServicesMock: vi.fn<(args: unknown) => Promise>(), @@ -70,7 +70,7 @@ function createSpeechConfig(providers: PaseoSpeechConfig["providers"]): PaseoSpe return { providers }; } -describe("initializeSpeechRuntime readiness", () => { +describe("createSpeechService readiness", () => { beforeEach(() => { initializeLocalSpeechServicesMock.mockReset(); }); @@ -92,7 +92,7 @@ describe("initializeSpeechRuntime readiness", () => { cleanup: () => {}, }); - const runtime = await initializeSpeechRuntime({ + const runtime = createSpeechService({ logger: pino({ level: "silent" }), speechConfig: createSpeechConfig({ dictationStt: { provider: "local", enabled: true, explicit: true }, @@ -101,14 +101,16 @@ describe("initializeSpeechRuntime readiness", () => { voiceTts: { provider: "local", enabled: false, explicit: true }, }), }); + runtime.start(); + await runtime.ready; - const readiness = runtime.getSpeechReadiness(); + const readiness = runtime.getReadiness(); expect(readiness.dictation.available).toBe(true); expect(readiness.realtimeVoice.reasonCode).toBe("disabled"); expect(readiness.voiceFeature.available).toBe(true); expect(readiness.voiceFeature.reasonCode).toBe("ready"); - runtime.cleanup(); + runtime.stop(); }); it("keeps voice feature available when only realtime voice is enabled and ready", async () => { @@ -130,7 +132,7 @@ describe("initializeSpeechRuntime readiness", () => { cleanup: () => {}, }); - const runtime = await initializeSpeechRuntime({ + const runtime = createSpeechService({ logger: pino({ level: "silent" }), speechConfig: createSpeechConfig({ dictationStt: { provider: "local", enabled: false, explicit: true }, @@ -139,13 +141,15 @@ describe("initializeSpeechRuntime readiness", () => { voiceTts: { provider: "local", enabled: true, explicit: true }, }), }); + runtime.start(); + await runtime.ready; - const readiness = runtime.getSpeechReadiness(); + const readiness = runtime.getReadiness(); expect(readiness.realtimeVoice.available).toBe(true); expect(readiness.dictation.reasonCode).toBe("disabled"); expect(readiness.voiceFeature.available).toBe(true); expect(readiness.voiceFeature.reasonCode).toBe("ready"); - runtime.cleanup(); + runtime.stop(); }); }); diff --git a/packages/server/src/server/speech/speech-runtime.ts b/packages/server/src/server/speech/speech-runtime.ts index 2781361b5..c56c19296 100644 --- a/packages/server/src/server/speech/speech-runtime.ts +++ b/packages/server/src/server/speech/speech-runtime.ts @@ -336,25 +336,23 @@ function resolveEffectiveProviderIds(params: { }; } -export type InitializedSpeechRuntime = { - resolveVoiceTurnDetection: () => TurnDetectionProvider | null; - resolveVoiceStt: () => SpeechToTextProvider | null; - resolveVoiceTts: () => TextToSpeechProvider | null; +export type SpeechService = { + resolveStt: () => SpeechToTextProvider | null; + resolveTts: () => TextToSpeechProvider | null; + resolveTurnDetection: () => TurnDetectionProvider | null; resolveDictationStt: () => SpeechToTextProvider | null; - getSpeechReadiness: () => SpeechReadinessSnapshot; - subscribeSpeechReadiness: (listener: (snapshot: SpeechReadinessSnapshot) => void) => () => void; - cleanup: () => void; - localModelConfig: { - modelsDir: string; - defaultModelIds: LocalSpeechModelId[]; - } | null; + getReadiness: () => SpeechReadinessSnapshot; + onReadinessChange: (listener: (snapshot: SpeechReadinessSnapshot) => void) => () => void; + start: () => void; + stop: () => void; + ready: Promise; }; -export async function initializeSpeechRuntime(params: { +export function createSpeechService(params: { logger: Logger; openaiConfig?: PaseoOpenAIConfig; speechConfig?: PaseoSpeechConfig; -}): Promise { +}): SpeechService { const logger = params.logger.child({ module: "speech-runtime" }); const speechConfig = params.speechConfig ?? null; const openaiConfig = params.openaiConfig; @@ -397,6 +395,14 @@ export async function initializeSpeechRuntime(params: { const readinessListeners = new Set<(snapshot: SpeechReadinessSnapshot) => void>(); let lastReadinessFingerprint: string | null = null; let lastPublishedReadinessSnapshot: SpeechReadinessSnapshot | null = null; + let started = false; + let readySettled = false; + let resolveReady!: () => void; + let rejectReady!: (error: unknown) => void; + const ready = new Promise((resolve, reject) => { + resolveReady = resolve; + rejectReady = reject; + }); const computeReadinessSnapshot = (): SpeechReadinessSnapshot => { const realtimeVoice = buildRealtimeVoiceReadiness({ @@ -658,16 +664,36 @@ export async function initializeSpeechRuntime(params: { } }; - await runReconcile(); - const snapshot = computeReadinessSnapshot(); - if (snapshot.voiceFeature.enabled && !snapshot.voiceFeature.available) { - if (missingLocalModelIds.length > 0) { - startBackgroundDownload(); + const start = (): void => { + if (started || stopped) { + return; } - scheduleMonitor(); - } + started = true; + void (async () => { + try { + await runReconcile(); + const snapshot = computeReadinessSnapshot(); + if (snapshot.voiceFeature.enabled && !snapshot.voiceFeature.available) { + if (missingLocalModelIds.length > 0) { + startBackgroundDownload(); + } + scheduleMonitor(); + } + if (!readySettled) { + readySettled = true; + resolveReady(); + } + } catch (error) { + if (!readySettled) { + readySettled = true; + rejectReady(error); + } + logger.error({ err: error }, "Speech runtime failed during initial reconcile"); + } + })(); + }; - const cleanup = (): void => { + const stop = (): void => { stopped = true; if (monitorTimeout) { clearTimeout(monitorTimeout); @@ -677,13 +703,14 @@ export async function initializeSpeechRuntime(params: { }; return { - resolveVoiceTurnDetection: () => turnDetectionService, - resolveVoiceStt: () => sttService, - resolveVoiceTts: () => ttsService, + resolveTurnDetection: () => turnDetectionService, + resolveStt: () => sttService, + resolveTts: () => ttsService, resolveDictationStt: () => dictationSttService, - getSpeechReadiness: () => lastPublishedReadinessSnapshot ?? computeReadinessSnapshot(), - subscribeSpeechReadiness, - cleanup, - localModelConfig, + getReadiness: () => lastPublishedReadinessSnapshot ?? computeReadinessSnapshot(), + onReadinessChange: subscribeSpeechReadiness, + start, + stop, + ready, }; } diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index ddd6a29ba..ab9d0f787 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -33,11 +33,7 @@ import type { AgentProvider } from "./agent/agent-sdk-types.js"; import type { AgentProviderRuntimeSettingsMap } from "./agent/provider-launch-config.js"; import { PushTokenStore } from "./push/token-store.js"; import { PushService } from "./push/push-service.js"; -import type { SpeechToTextProvider, TextToSpeechProvider } from "./speech/speech-provider.js"; -import type { TurnDetectionProvider } from "./speech/turn-detection-provider.js"; -import type { Resolvable } from "./speech/provider-resolver.js"; -import type { SpeechReadinessSnapshot } from "./speech/speech-runtime.js"; -import type { LocalSpeechModelId } from "./speech/providers/local/models.js"; +import type { SpeechReadinessSnapshot, SpeechService } from "./speech/speech-runtime.js"; import type { VoiceCallerContext, VoiceMcpStdioConfig, VoiceSpeakHandler } from "./voice-types.js"; import { computeShouldNotifyClient, @@ -243,18 +239,10 @@ export class VoiceAssistantWebSocketServer { private readonly pushTokenStore: PushTokenStore; private readonly pushService: PushService; private readonly createAgentMcpTransport: AgentMcpTransportFactory; - private readonly stt: Resolvable; - private readonly tts: Resolvable; - private readonly turnDetection: Resolvable; + private readonly speech: SpeechService | null; private readonly terminalManager: TerminalManager | null; private readonly dictation: { finalTimeoutMs?: number; - stt?: Resolvable; - localModels?: { - modelsDir: string; - defaultModelIds: LocalSpeechModelId[]; - }; - getSpeechReadiness?: () => SpeechReadinessSnapshot; } | null; private readonly voice: { voiceAgentMcpStdio?: VoiceMcpStdioConfig | null; @@ -288,6 +276,7 @@ export class VoiceAssistantWebSocketServer { private readonly inboundSessionRequestCounts = new Map(); private readonly requestLatencies = new Map(); private runtimeMetricsInterval: ReturnType | null = null; + private unsubscribeSpeechReadiness: (() => void) | null = null; constructor( server: HTTPServer, @@ -299,11 +288,7 @@ export class VoiceAssistantWebSocketServer { paseoHome: string, createAgentMcpTransport: AgentMcpTransportFactory, wsConfig: WebSocketServerConfig, - speech?: { - turnDetection: Resolvable; - stt: Resolvable; - tts: Resolvable; - }, + speech?: SpeechService | null, terminalManager?: TerminalManager | null, voice?: { voiceAgentMcpStdio?: VoiceMcpStdioConfig | null; @@ -312,12 +297,6 @@ export class VoiceAssistantWebSocketServer { }, dictation?: { finalTimeoutMs?: number; - stt?: Resolvable; - localModels?: { - modelsDir: string; - defaultModelIds: LocalSpeechModelId[]; - }; - getSpeechReadiness?: () => SpeechReadinessSnapshot; }, agentProviderRuntimeSettings?: AgentProviderRuntimeSettingsMap, daemonVersion?: string, @@ -358,17 +337,18 @@ export class VoiceAssistantWebSocketServer { this.downloadTokenStore = downloadTokenStore; this.paseoHome = paseoHome; this.createAgentMcpTransport = createAgentMcpTransport; - this.turnDetection = speech?.turnDetection ?? null; - this.stt = speech?.stt ?? null; - this.tts = speech?.tts ?? null; + this.speech = speech ?? null; this.terminalManager = terminalManager ?? null; this.voice = voice ?? null; this.dictation = dictation ?? null; this.agentProviderRuntimeSettings = agentProviderRuntimeSettings; this.onLifecycleIntent = onLifecycleIntent ?? null; this.serverCapabilities = buildServerCapabilities({ - readiness: this.dictation?.getSpeechReadiness?.() ?? null, + readiness: this.speech?.getReadiness() ?? null, }); + this.unsubscribeSpeechReadiness = this.speech?.onReadinessChange((snapshot) => { + this.publishSpeechReadiness(snapshot); + }) ?? null; const pushLogger = this.logger.child({ module: "push" }); this.pushTokenStore = new PushTokenStore(pushLogger, join(paseoHome, "push-tokens.json")); @@ -457,6 +437,8 @@ export class VoiceAssistantWebSocketServer { } public async close(): Promise { + this.unsubscribeSpeechReadiness?.(); + this.unsubscribeSpeechReadiness = null; if (this.runtimeMetricsInterval) { clearInterval(this.runtimeMetricsInterval); this.runtimeMetricsInterval = null; @@ -658,12 +640,12 @@ export class VoiceAssistantWebSocketServer { scheduleService: this.scheduleService, checkoutDiffManager: this.checkoutDiffManager, createAgentMcpTransport: this.createAgentMcpTransport, - stt: this.stt, - tts: this.tts, + stt: () => this.speech?.resolveStt() ?? null, + tts: () => this.speech?.resolveTts() ?? null, terminalManager: this.terminalManager, voice: { ...(this.voice ?? {}), - turnDetection: this.turnDetection, + turnDetection: () => this.speech?.resolveTurnDetection() ?? null, }, voiceBridge: { registerVoiceSpeakHandler: (agentId, handler) => { @@ -681,7 +663,14 @@ export class VoiceAssistantWebSocketServer { ensureVoiceMcpSocketForAgent: this.voice?.ensureVoiceMcpSocketForAgent, removeVoiceMcpSocketForAgent: this.voice?.removeVoiceMcpSocketForAgent, }, - dictation: this.dictation ?? undefined, + dictation: + this.dictation || this.speech + ? { + finalTimeoutMs: this.dictation?.finalTimeoutMs, + stt: () => this.speech?.resolveDictationStt() ?? null, + getSpeechReadiness: () => this.speech!.getReadiness(), + } + : undefined, agentProviderRuntimeSettings: this.agentProviderRuntimeSettings, }); diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index 6ac99e5e8..9885aaf27 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -726,17 +726,6 @@ export const ListAvailableProvidersRequestMessageSchema = z.object({ requestId: z.string(), }); -export const SpeechModelsListRequestSchema = z.object({ - type: z.literal("speech_models_list_request"), - requestId: z.string(), -}); - -export const SpeechModelsDownloadRequestSchema = z.object({ - type: z.literal("speech_models_download_request"), - modelIds: z.array(z.string()).optional(), - requestId: z.string(), -}); - export const ResumeAgentRequestMessageSchema = z.object({ type: z.literal("resume_agent_request"), handle: AgentPersistenceHandleSchema, @@ -1203,8 +1192,6 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ CreateAgentRequestMessageSchema, ListProviderModelsRequestMessageSchema, ListAvailableProvidersRequestMessageSchema, - SpeechModelsListRequestSchema, - SpeechModelsDownloadRequestSchema, ResumeAgentRequestMessageSchema, RefreshAgentRequestMessageSchema, CancelAgentRequestMessageSchema, @@ -2104,34 +2091,6 @@ export const ListAvailableProvidersResponseSchema = z.object({ }), }); -export const SpeechModelsListResponseSchema = z.object({ - type: z.literal("speech_models_list_response"), - payload: z.object({ - modelsDir: z.string(), - models: z.array( - z.object({ - id: z.string(), - kind: z.string(), - description: z.string(), - modelDir: z.string(), - isDownloaded: z.boolean(), - missingFiles: z.array(z.string()).optional(), - }), - ), - requestId: z.string(), - }), -}); - -export const SpeechModelsDownloadResponseSchema = z.object({ - type: z.literal("speech_models_download_response"), - payload: z.object({ - modelsDir: z.string(), - downloadedModelIds: z.array(z.string()), - error: z.string().nullable(), - requestId: z.string(), - }), -}); - const AgentSlashCommandSchema = z.object({ name: z.string(), description: z.string(), @@ -2311,8 +2270,6 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ FileDownloadTokenResponseSchema, ListProviderModelsResponseMessageSchema, ListAvailableProvidersResponseSchema, - SpeechModelsListResponseSchema, - SpeechModelsDownloadResponseSchema, ListCommandsResponseSchema, ListTerminalsResponseSchema, TerminalsChangedSchema, @@ -2381,8 +2338,6 @@ export type ListProviderModelsResponseMessage = z.infer< typeof ListProviderModelsResponseMessageSchema >; export type ListAvailableProvidersResponse = z.infer; -export type SpeechModelsListResponse = z.infer; -export type SpeechModelsDownloadResponse = z.infer; export type ChatCreateResponse = z.infer; export type ChatListResponse = z.infer; export type ChatInspectResponse = z.infer; @@ -2443,8 +2398,6 @@ export type LoopListRequest = z.infer; export type LoopInspectRequest = z.infer; export type LoopLogsRequest = z.infer; export type LoopStopRequest = z.infer; -export type SpeechModelsListRequestMessage = z.infer; -export type SpeechModelsDownloadRequestMessage = z.infer; export type ResumeAgentRequestMessage = z.infer; export type DeleteAgentRequestMessage = z.infer; export type UpdateAgentRequestMessage = z.infer;