From f9633dd359ce2fdcd108fb5b169aa7417363c60b Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 18 Aug 2026 01:28:21 +0200 Subject: [PATCH] fix(server): stop background speech downloads --- packages/server/src/server/bootstrap.ts | 2 +- .../server/speech/providers/local/models.ts | 2 + .../local/sherpa/model-downloader.ts | 18 ++++-- .../src/server/speech/speech-runtime.test.ts | 58 +++++++++++++++++-- .../src/server/speech/speech-runtime.ts | 15 ++++- 5 files changed, 83 insertions(+), 12 deletions(-) diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index c1197754e..20d860cc1 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1704,7 +1704,7 @@ export async function createPaseoDaemon( await agentStorage.flush().catch(() => undefined); await providerSnapshotManager.shutdown(); terminalManager.killAll(); - speechService.stop(); + await speechService.stop(); await scheduleService.stop().catch(() => undefined); await relayRuntime?.stop().catch(() => undefined); if (wsServer) { diff --git a/packages/server/src/server/speech/providers/local/models.ts b/packages/server/src/server/speech/providers/local/models.ts index fbbaaa8d6..f6c7cc4e4 100644 --- a/packages/server/src/server/speech/providers/local/models.ts +++ b/packages/server/src/server/speech/providers/local/models.ts @@ -34,10 +34,12 @@ export async function ensureLocalSpeechModels(options: { modelsDir: string; modelIds: LocalSpeechModelId[]; logger: import("pino").Logger; + signal?: AbortSignal; }): Promise> { return ensureSherpaOnnxModels({ modelsDir: options.modelsDir, modelIds: options.modelIds, logger: options.logger, + signal: options.signal, }); } diff --git a/packages/server/src/server/speech/providers/local/sherpa/model-downloader.ts b/packages/server/src/server/speech/providers/local/sherpa/model-downloader.ts index 393237d37..8ce21f5e0 100644 --- a/packages/server/src/server/speech/providers/local/sherpa/model-downloader.ts +++ b/packages/server/src/server/speech/providers/local/sherpa/model-downloader.ts @@ -12,6 +12,7 @@ export interface EnsureSherpaOnnxModelOptions { modelsDir: string; modelId: SherpaOnnxModelId; logger: pino.Logger; + signal?: AbortSignal; } export function getSherpaOnnxModelDir(modelsDir: string, modelId: SherpaOnnxModelId): string { @@ -40,11 +41,12 @@ async function hasRequiredFiles(modelDir: string, requiredFiles: string[]): Prom interface DownloadToFileOptions { url: string; outputPath: string; + signal?: AbortSignal; } async function downloadToFile(options: DownloadToFileOptions): Promise { const { url, outputPath } = options; - const res = await fetch(url); + const res = await fetch(url, { signal: options.signal }); if (!res.ok) { throw new Error(`Failed to download ${url}: ${res.status} ${res.statusText}`); } @@ -60,7 +62,7 @@ async function downloadToFile(options: DownloadToFileOptions): Promise { const nodeStream = Readable.fromWeb(res.body as any); try { - await pipeline(nodeStream, createWriteStream(tmpPath)); + await pipeline(nodeStream, createWriteStream(tmpPath), { signal: options.signal }); await rename(tmpPath, outputPath); } catch (error) { await rm(tmpPath, { force: true }).catch(() => undefined); @@ -68,12 +70,17 @@ async function downloadToFile(options: DownloadToFileOptions): Promise { } } -async function extractTarArchive(archivePath: string, destDir: string): Promise { +async function extractTarArchive( + archivePath: string, + destDir: string, + signal?: AbortSignal, +): Promise { await mkdir(destDir, { recursive: true }); await new Promise((resolve, reject) => { const child = spawnProcess("tar", ["xf", archivePath, "-C", destDir], { stdio: "inherit", + signal, }); child.on("error", reject); child.on("exit", (code) => { @@ -119,6 +126,7 @@ export async function ensureSherpaOnnxModel( await downloadToFile({ url: spec.archiveUrl, outputPath: archivePath, + signal: options.signal, }); } @@ -130,7 +138,7 @@ export async function ensureSherpaOnnxModel( }, "Extracting model archive", ); - await extractTarArchive(archivePath, options.modelsDir); + await extractTarArchive(archivePath, options.modelsDir, options.signal); logger.info( { @@ -170,6 +178,7 @@ export async function ensureSherpaOnnxModels(options: { modelsDir: string; modelIds: SherpaOnnxModelId[]; logger: pino.Logger; + signal?: AbortSignal; }): Promise> { const uniq = Array.from(new Set(options.modelIds)); const entries: Array<[SherpaOnnxModelId, string]> = await Promise.all( @@ -178,6 +187,7 @@ export async function ensureSherpaOnnxModels(options: { modelsDir: options.modelsDir, modelId: id, logger: options.logger, + signal: options.signal, }); return [id, modelPath] as [SherpaOnnxModelId, string]; }), diff --git a/packages/server/src/server/speech/speech-runtime.test.ts b/packages/server/src/server/speech/speech-runtime.test.ts index f368d5977..428ecd27a 100644 --- a/packages/server/src/server/speech/speech-runtime.test.ts +++ b/packages/server/src/server/speech/speech-runtime.test.ts @@ -7,7 +7,8 @@ import type { SpeechToTextProvider, TextToSpeechProvider } from "./speech-provid import type { TurnDetectionProvider } from "./turn-detection-provider.js"; import { createSpeechService } from "./speech-runtime.js"; -const { initializeLocalSpeechServicesMock } = vi.hoisted(() => ({ +const { ensureLocalSpeechModelsMock, initializeLocalSpeechServicesMock } = vi.hoisted(() => ({ + ensureLocalSpeechModelsMock: vi.fn(async () => ({})), initializeLocalSpeechServicesMock: vi.fn<(args: unknown) => Promise>(), })); @@ -34,7 +35,7 @@ vi.mock("./providers/openai/runtime.js", () => ({ })); vi.mock("./providers/local/models.js", () => ({ - ensureLocalSpeechModels: vi.fn(async () => {}), + ensureLocalSpeechModels: ensureLocalSpeechModelsMock, getLocalSpeechModelDir: vi.fn(() => ""), listLocalSpeechModels: vi.fn(() => []), })); @@ -66,6 +67,12 @@ function createStubTurnDetection(id: string): TurnDetectionProvider { }; } +function rejectWhenAborted(signal?: AbortSignal): Promise { + return new Promise((_resolve, reject) => { + signal?.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); +} + function createSpeechConfig(providers: PaseoSpeechConfig["providers"]): PaseoSpeechConfig { return { providers, @@ -78,6 +85,8 @@ function createSpeechConfig(providers: PaseoSpeechConfig["providers"]): PaseoSpe describe("createSpeechService readiness", () => { beforeEach(() => { + ensureLocalSpeechModelsMock.mockReset(); + ensureLocalSpeechModelsMock.mockResolvedValue({}); initializeLocalSpeechServicesMock.mockReset(); }); @@ -116,7 +125,7 @@ describe("createSpeechService readiness", () => { expect(readiness.voiceFeature.available).toBe(true); expect(readiness.voiceFeature.reasonCode).toBe("ready"); - runtime.stop(); + await runtime.stop(); }); it("keeps voice feature available when only realtime voice is enabled and ready", async () => { @@ -156,6 +165,47 @@ describe("createSpeechService readiness", () => { expect(readiness.voiceFeature.available).toBe(true); expect(readiness.voiceFeature.reasonCode).toBe("ready"); - runtime.stop(); + await runtime.stop(); + }); + + it("aborts and joins an in-flight model download when stopped", async () => { + let downloadSignal: AbortSignal | undefined; + ensureLocalSpeechModelsMock.mockImplementation(({ signal }: { signal?: AbortSignal }) => { + downloadSignal = signal; + return rejectWhenAborted(signal); + }); + initializeLocalSpeechServicesMock.mockResolvedValue({ + turnDetectionService: null, + sttService: null, + ttsService: null, + dictationSttService: null, + localVoiceTtsProvider: null, + localModelConfig: { + modelsDir: "/tmp/missing-local-speech-models", + defaultModelIds: ["parakeet-tdt-0.6b-v2-int8"], + }, + availability: { + configured: true, + modelsDir: "/tmp/missing-local-speech-models", + }, + cleanup: () => {}, + }); + + const runtime = createSpeechService({ + logger: pino({ level: "silent" }), + speechConfig: createSpeechConfig({ + dictationStt: { provider: "local", enabled: true, explicit: true }, + voiceTurnDetection: { provider: "local", enabled: false, explicit: true }, + voiceStt: { provider: "local", enabled: false, explicit: true }, + voiceTts: { provider: "local", enabled: false, explicit: true }, + }), + }); + runtime.start(); + await runtime.ready; + expect(ensureLocalSpeechModelsMock).toHaveBeenCalledOnce(); + + await runtime.stop(); + + expect(downloadSignal?.aborted).toBe(true); }); }); diff --git a/packages/server/src/server/speech/speech-runtime.ts b/packages/server/src/server/speech/speech-runtime.ts index 8da70afd8..6e56a9e95 100644 --- a/packages/server/src/server/speech/speech-runtime.ts +++ b/packages/server/src/server/speech/speech-runtime.ts @@ -350,7 +350,7 @@ export interface SpeechService { getReadiness: () => SpeechReadinessSnapshot; onReadinessChange: (listener: (snapshot: SpeechReadinessSnapshot) => void) => () => void; start: () => void; - stop: () => void; + stop: () => Promise; ready: Promise; } @@ -394,6 +394,8 @@ export function createSpeechService(params: { let missingLocalModelIds: LocalSpeechModelId[] = []; let backgroundDownloadInProgress = false; + let backgroundDownloadAbortController: AbortController | null = null; + let backgroundDownloadPromise: Promise | null = null; let backgroundDownloadError: string | null = null; let stopped = false; let monitorTimeout: ReturnType | null = null; @@ -609,12 +611,15 @@ export function createSpeechService(params: { "Starting background download for missing local speech models", ); - void (async () => { + const abortController = new AbortController(); + backgroundDownloadAbortController = abortController; + backgroundDownloadPromise = (async () => { try { await ensureLocalSpeechModels({ modelsDir, modelIds, logger, + signal: abortController.signal, }); await runReconcile(); backgroundDownloadError = null; @@ -629,6 +634,7 @@ export function createSpeechService(params: { "Background local speech model download failed", ); } finally { + backgroundDownloadAbortController = null; backgroundDownloadInProgress = false; await refreshMissingLocalModels().catch((error) => { logger.warn({ err: error }, "Failed to refresh local speech model status after download"); @@ -699,13 +705,16 @@ export function createSpeechService(params: { })(); }; - const stop = (): void => { + const stop = async (): Promise => { stopped = true; if (monitorTimeout) { clearTimeout(monitorTimeout); monitorTimeout = null; } localCleanup(); + backgroundDownloadAbortController?.abort(); + await backgroundDownloadPromise?.catch(() => undefined); + backgroundDownloadPromise = null; }; return {