jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(server): stop background speech downloads

Browse Source
This commit is contained in:
Mohamed Boudra
2026-08-18 01:28:21 +02:00
parent 78629f92ef
commit f9633dd359
5 changed files with 83 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/bootstrap.ts
+1 -1
View File
@@ -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) {
packages/server/src/server/speech/providers/local/models.ts
+2
View File
@@ -34,10 +34,12 @@ export async function ensureLocalSpeechModels(options: {
modelsDir: string;
modelIds: LocalSpeechModelId[];
logger: import("pino").Logger;
signal?: AbortSignal;
}): Promise<Record<LocalSpeechModelId, string>> {
return ensureSherpaOnnxModels({
modelsDir: options.modelsDir,
modelIds: options.modelIds,
logger: options.logger,
signal: options.signal,
});
}
packages/server/src/server/speech/providers/local/sherpa/model-downloader.ts
+14 -4
View File
@@ -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<void> {
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<void> {
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<void> {
}
}
async function extractTarArchive(archivePath: string, destDir: string): Promise<void> {
async function extractTarArchive(
archivePath: string,
destDir: string,
signal?: AbortSignal,
): Promise<void> {
await mkdir(destDir, { recursive: true });
await new Promise<void>((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<Record<SherpaOnnxModelId, string>> {
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];
}),
packages/server/src/server/speech/speech-runtime.test.ts
+54 -4
View File
@@ -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<InitializedLocalSpeech>>(),
}));
@@ -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<never> {
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);
});
});
packages/server/src/server/speech/speech-runtime.ts
+12 -3
View File
@@ -350,7 +350,7 @@ export interface SpeechService {
getReadiness: () => SpeechReadinessSnapshot;
onReadinessChange: (listener: (snapshot: SpeechReadinessSnapshot) => void) => () => void;
start: () => void;
stop: () => void;
stop: () => Promise<void>;
ready: Promise<void>;
}
@@ -394,6 +394,8 @@ export function createSpeechService(params: {
let missingLocalModelIds: LocalSpeechModelId[] = [];
let backgroundDownloadInProgress = false;
let backgroundDownloadAbortController: AbortController | null = null;
let backgroundDownloadPromise: Promise<void> | null = null;
let backgroundDownloadError: string | null = null;
let stopped = false;
let monitorTimeout: ReturnType<typeof setTimeout> | 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<void> => {
stopped = true;
if (monitorTimeout) {
clearTimeout(monitorTimeout);
monitorTimeout = null;
}
localCleanup();
backgroundDownloadAbortController?.abort();
await backgroundDownloadPromise?.catch(() => undefined);
backgroundDownloadPromise = null;
};
return {