From 6aa73baaabddb4ad68b8a1175214322f0cdc8412 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 6 Jun 2026 21:45:33 +0800 Subject: [PATCH] Show useful local speech crash details (#1379) * Improve local speech worker crash diagnostics * Fix speech worker stderr tail diagnostics * Drain speech worker stderr before reporting exit * Avoid false speech worker crash logs on shutdown --- .../server/speech/providers/local/runtime.ts | 1 + .../providers/local/worker-client.test.ts | 95 ++++++- .../speech/providers/local/worker-client.ts | 244 +++++++++++++++++- 3 files changed, 330 insertions(+), 10 deletions(-) diff --git a/packages/server/src/server/speech/providers/local/runtime.ts b/packages/server/src/server/speech/providers/local/runtime.ts index 0653a75e5..ba9ec9aac 100644 --- a/packages/server/src/server/speech/providers/local/runtime.ts +++ b/packages/server/src/server/speech/providers/local/runtime.ts @@ -156,6 +156,7 @@ export async function initializeLocalSpeechServices(params: { const workerClient = localConfig ? new LocalSpeechWorkerClient({ + logger, config: { modelsDir: localConfig.modelsDir, voiceSttModel: localModels.voiceLocalSttModel, diff --git a/packages/server/src/server/speech/providers/local/worker-client.test.ts b/packages/server/src/server/speech/providers/local/worker-client.test.ts index b42ff4adc..3c028be13 100644 --- a/packages/server/src/server/speech/providers/local/worker-client.test.ts +++ b/packages/server/src/server/speech/providers/local/worker-client.test.ts @@ -20,6 +20,8 @@ import { bufferToWorkerBytes, workerBytesToBuffer } from "./worker-bytes.js"; class FakeLocalSpeechWorker extends EventEmitter { public connected = true; public killed = false; + public pid = 12345; + public readonly stderr = new EventEmitter() as NodeJS.ReadableStream; public readonly sent: LocalSpeechWorkerRequest[] = []; public disconnects = 0; public kills = 0; @@ -88,9 +90,9 @@ class PausedIpcWorker { } on(event: "message", listener: (message: LocalSpeechWorkerToParentMessage) => void): this; - on(event: "exit", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this; + on(event: "close", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this; on( - event: "message" | "exit", + event: "message" | "close", listener: | ((message: LocalSpeechWorkerToParentMessage) => void) | ((code: number | null, signal: NodeJS.Signals | null) => void), @@ -103,6 +105,7 @@ class PausedIpcWorker { function createClient(options?: { idleTtlMs?: number }) { const workers: FakeLocalSpeechWorker[] = []; const client = new LocalSpeechWorkerClient({ + logger: pino({ level: "silent" }), config: { modelsDir: "/tmp/models", voiceSttModel: "parakeet-tdt-0.6b-v2-int8", @@ -120,6 +123,19 @@ function createClient(options?: { idleTtlMs?: number }) { return { client, workers }; } +function createCapturingLogger(): { logger: pino.Logger; records: Array> } { + const records: Array> = []; + const logger = pino( + { level: "trace" }, + { + write(line: string) { + records.push(JSON.parse(line) as Record); + }, + }, + ); + return { logger, records }; +} + async function waitForMicrotasks(): Promise { await new Promise((resolve) => setTimeout(resolve, 0)); } @@ -217,6 +233,7 @@ describe("LocalSpeechWorkerClient", () => { it("does not surface real IPC backpressure when replaying native-sized dictation frames", async () => { const workers: PausedIpcWorker[] = []; const client = new LocalSpeechWorkerClient({ + logger: pino({ level: "silent" }), config: { modelsDir: "/tmp/models", voiceSttModel: "parakeet-tdt-0.6b-v2-int8", @@ -257,6 +274,80 @@ describe("LocalSpeechWorkerClient", () => { } }); + it("logs worker exit details and includes actionable context in the surfaced error", async () => { + const { logger, records } = createCapturingLogger(); + const workers: FakeLocalSpeechWorker[] = []; + const client = new LocalSpeechWorkerClient({ + logger, + config: { + modelsDir: "/tmp/models", + voiceSttModel: "parakeet-tdt-0.6b-v2-int8", + dictationSttModel: "parakeet-tdt-0.6b-v2-int8", + voiceTtsModel: "kokoro-en-v0_19", + }, + forkWorker: () => { + const worker = new FakeLocalSpeechWorker(); + workers.push(worker); + return worker; + }, + }); + const provider = new WorkerBackedSpeechToTextProvider(client, "dictationStt"); + const session = provider.createSession({ logger: pino({ level: "silent" }) }); + + const connect = session.connect(); + workers[0].emit("exit", null, "SIGABRT"); + workers[0].stderr.emit("data", "dyld: Library not loaded: libsherpa-onnx-c-api.dylib"); + workers[0].emit("close", null, "SIGABRT"); + + await expect(connect).rejects.toThrow( + "Local speech worker exited (signal SIGABRT) while handling session.create (dictationStt). Last stderr: dyld: Library not loaded: libsherpa-onnx-c-api.dylib", + ); + const exitRecord = records.find((record) => record.msg === "Local speech worker exited"); + expect(exitRecord).toMatchObject({ + workerPid: 12345, + signal: "SIGABRT", + stderrTail: "dyld: Library not loaded: libsherpa-onnx-c-api.dylib", + pendingRequests: [expect.objectContaining({ type: "session.create", kind: "dictationStt" })], + }); + }); + + it("does not log intentional shutdowns as worker crashes", async () => { + const { logger, records } = createCapturingLogger(); + const workers: FakeLocalSpeechWorker[] = []; + const client = new LocalSpeechWorkerClient({ + logger, + config: { + modelsDir: "/tmp/models", + voiceSttModel: "parakeet-tdt-0.6b-v2-int8", + dictationSttModel: "parakeet-tdt-0.6b-v2-int8", + voiceTtsModel: "kokoro-en-v0_19", + }, + forkWorker: () => { + const worker = new FakeLocalSpeechWorker(); + workers.push(worker); + return worker; + }, + }); + + const synthesize = client.synthesizeSpeech("hello"); + workers[0].respond(workers[0].sent[0], { + audio: bufferToWorkerBytes(Buffer.from([1])), + format: "pcm;rate=24000", + }); + await synthesize; + + client.shutdown(); + workers[0].emit("close", null, "SIGTERM"); + + expect(records.find((record) => record.msg === "Local speech worker exited")).toBeUndefined(); + expect( + records.find((record) => record.msg === "Local speech worker closed after shutdown"), + ).toMatchObject({ + workerPid: 12345, + signal: "SIGTERM", + }); + }); + it("forwards VAD session events through the shared worker", async () => { const { client, workers } = createClient(); const provider = new WorkerBackedTurnDetectionProvider(client); diff --git a/packages/server/src/server/speech/providers/local/worker-client.ts b/packages/server/src/server/speech/providers/local/worker-client.ts index 88c0d46fb..1f0fa96b9 100644 --- a/packages/server/src/server/speech/providers/local/worker-client.ts +++ b/packages/server/src/server/speech/providers/local/worker-client.ts @@ -28,6 +28,8 @@ import { bufferToWorkerBytes, workerBytesToBuffer } from "./worker-bytes.js"; const DEFAULT_REQUEST_TIMEOUT_MS = 30000; const DEFAULT_IDLE_TTL_MS = 5 * 60 * 1000; const DEFAULT_LOCAL_SAMPLE_RATE = 16000; +const STDERR_TAIL_MAX_CHARS = 8000; +const USER_ERROR_STDERR_MAX_CHARS = 1000; type LocalSpeechWorkerRequestInput = LocalSpeechWorkerRequest extends infer Request ? Request extends LocalSpeechWorkerRequest @@ -38,21 +40,27 @@ type LocalSpeechWorkerRequestInput = LocalSpeechWorkerRequest extends infer Requ interface LocalSpeechWorkerProcess { connected: boolean; killed: boolean; + pid?: number; + stderr?: Readable | null; send(message: LocalSpeechWorkerRequest, callback: (error: Error | null) => void): boolean; disconnect(): void; kill(): boolean; on(event: "message", listener: (message: LocalSpeechWorkerToParentMessage) => void): this; - on(event: "exit", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this; + on(event: "close", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this; } interface PendingRequest { resolve: (value: unknown) => void; reject: (error: Error) => void; timeout: ReturnType; + type: LocalSpeechWorkerRequest["type"]; + summary: Record; + startedAt: number; } interface LocalSpeechWorkerClientOptions { config: LocalSpeechWorkerConfig; + logger: pino.Logger; requestTimeoutMs?: number; idleTtlMs?: number; forkWorker?: () => LocalSpeechWorkerProcess; @@ -90,7 +98,7 @@ function forkLocalSpeechWorker(): LocalSpeechWorkerProcess { env, execArgv: resolveWorkerExecArgv(), serialization: "advanced", - stdio: ["ignore", "ignore", "inherit", "ipc"], + stdio: ["ignore", "ignore", "pipe", "ipc"], }) as LocalSpeechWorkerProcess; } @@ -100,8 +108,75 @@ function isResponse( return message.type === "response"; } +function truncateStart(value: string, maxChars: number): string { + if (value.length <= maxChars) { + return value; + } + return value.slice(value.length - maxChars); +} + +function summarizeWorkerRequest(message: LocalSpeechWorkerRequest): Record { + switch (message.type) { + case "session.create": + return { kind: message.kind, sessionId: message.sessionId }; + case "session.append": + return { sessionId: message.sessionId, audioBytes: message.audio.byteLength }; + case "session.commit": + case "session.clear": + case "session.flush": + case "session.reset": + case "session.close": + return { sessionId: message.sessionId }; + case "stt.transcribe": + return { + model: message.model, + audioBytes: message.audio.byteLength, + format: message.format, + }; + case "tts.synthesize": + return { textLength: message.text.length }; + } +} + +function formatExitStatus(code: number | null, signal: NodeJS.Signals | null): string { + const parts: string[] = []; + if (code !== null) { + parts.push(`code ${code}`); + } + if (signal) { + parts.push(`signal ${signal}`); + } + return parts.length > 0 ? parts.join(", ") : "unknown exit status"; +} + +function formatPendingRequestForMessage(request: Record): string { + const type = typeof request.type === "string" ? request.type : "unknown request"; + const kind = typeof request.kind === "string" ? ` (${request.kind})` : ""; + return `${type}${kind}`; +} + +function buildWorkerExitMessage(params: { + code: number | null; + signal: NodeJS.Signals | null; + pendingRequests: Array>; + stderrTail: string; +}): string { + const pending = + params.pendingRequests.length > 0 + ? ` while handling ${params.pendingRequests + .slice(0, 3) + .map(formatPendingRequestForMessage) + .join(", ")}` + : ""; + const stderr = params.stderrTail + ? ` Last stderr: ${truncateStart(params.stderrTail, USER_ERROR_STDERR_MAX_CHARS)}` + : " Check daemon.log and macOS DiagnosticReports for Paseo Voice crash details."; + return `Local speech worker exited (${formatExitStatus(params.code, params.signal)})${pending}.${stderr}`; +} + export class LocalSpeechWorkerClient { private readonly config: LocalSpeechWorkerConfig; + private readonly logger: pino.Logger; private readonly requestTimeoutMs: number; private readonly idleTtlMs: number; private readonly forkWorker: () => LocalSpeechWorkerProcess; @@ -109,11 +184,16 @@ export class LocalSpeechWorkerClient { private readonly activeSessionIds = new Set(); private readonly sessionEmitters = new Map(); private worker: LocalSpeechWorkerProcess | null = null; + private workerPid: number | null = null; + private stderrTail = ""; + private stderrLineBuffer = ""; private inFlightRequests = 0; private idleTimer: ReturnType | null = null; + private readonly intentionalWorkerCloses = new WeakSet(); constructor(options: LocalSpeechWorkerClientOptions) { this.config = options.config; + this.logger = options.logger.child({ component: "local-speech-worker-client" }); this.requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS; this.idleTtlMs = options.idleTtlMs ?? DEFAULT_IDLE_TTL_MS; this.forkWorker = options.forkWorker ?? forkLocalSpeechWorker; @@ -214,7 +294,9 @@ export class LocalSpeechWorkerClient { this.sessionEmitters.clear(); const worker = this.worker; this.worker = null; + this.workerPid = null; if (worker && !worker.killed) { + this.intentionalWorkerCloses.add(worker); try { worker.disconnect(); } catch { @@ -232,6 +314,7 @@ export class LocalSpeechWorkerClient { const worker = this.ensureWorker(); const requestId = randomUUID(); const message = { ...input, requestId } as LocalSpeechWorkerRequest; + const requestSummary = summarizeWorkerRequest(message); this.inFlightRequests++; this.clearIdleTimer(); @@ -246,6 +329,9 @@ export class LocalSpeechWorkerClient { resolve: (value) => resolve(value as T), reject, timeout, + type: message.type, + summary: requestSummary, + startedAt: Date.now(), }); worker.send(message, (error) => { @@ -271,11 +357,44 @@ export class LocalSpeechWorkerClient { } const worker = this.forkWorker(); this.worker = worker; + this.workerPid = worker.pid ?? null; + this.stderrTail = ""; + this.stderrLineBuffer = ""; + this.logger.info( + { + workerPid: this.workerPid, + modelsDir: this.config.modelsDir, + voiceSttModel: this.config.voiceSttModel, + dictationSttModel: this.config.dictationSttModel, + voiceTtsModel: this.config.voiceTtsModel, + }, + "Local speech worker spawned", + ); + worker.stderr?.on("data", (chunk: Buffer | string) => this.handleWorkerStderr(chunk)); worker.on("message", (message) => this.handleWorkerMessage(message)); - worker.on("exit", () => this.handleWorkerExit()); + worker.on("close", (code, signal) => this.handleWorkerExit(worker, code, signal)); return worker; } + private handleWorkerStderr(chunk: Buffer | string): void { + const text = Buffer.isBuffer(chunk) ? chunk.toString("utf8") : chunk; + this.stderrTail = truncateStart(this.stderrTail + text, STDERR_TAIL_MAX_CHARS); + + this.stderrLineBuffer += text; + const lines = this.stderrLineBuffer.split(/\r?\n/); + this.stderrLineBuffer = lines.pop() ?? ""; + for (const line of lines) { + const trimmed = line.trim(); + if (!trimmed) { + continue; + } + this.logger.warn( + { workerPid: this.workerPid, stderr: trimmed }, + "Local speech worker stderr", + ); + } + } + private handleWorkerMessage(message: LocalSpeechWorkerToParentMessage): void { if (isResponse(message)) { const pending = this.pendingRequests.get(message.requestId); @@ -317,13 +436,84 @@ export class LocalSpeechWorkerClient { } } - private handleWorkerExit(): void { - this.worker = null; + private handleWorkerExit( + worker: LocalSpeechWorkerProcess, + code: number | null, + signal: NodeJS.Signals | null, + ): void { + const wasCurrentWorker = this.worker === worker; + const wasIntentionalClose = this.intentionalWorkerCloses.has(worker); + this.intentionalWorkerCloses.delete(worker); + const workerPid = worker.pid ?? (wasCurrentWorker ? this.workerPid : null); + const pendingRequests = this.describePendingRequests(); + const activeSessionCount = this.activeSessionIds.size; + const stderrTail = this.getStderrTail(); + + if (wasIntentionalClose) { + this.logger.info( + { + workerPid, + code, + signal, + pendingRequests, + activeSessionCount, + stderrTail: stderrTail || null, + }, + "Local speech worker closed after shutdown", + ); + if (wasCurrentWorker) { + this.worker = null; + this.workerPid = null; + } + return; + } + + if (!wasCurrentWorker) { + this.logger.warn( + { + workerPid, + code, + signal, + pendingRequests, + activeSessionCount, + stderrTail: stderrTail || null, + }, + "Stale local speech worker closed", + ); + return; + } + + const error = new Error( + buildWorkerExitMessage({ + code, + signal, + pendingRequests, + stderrTail, + }), + ); + + this.logger.error( + { + err: error, + workerPid, + code, + signal, + pendingRequests, + activeSessionCount, + stderrTail: stderrTail || null, + }, + "Local speech worker exited", + ); + + if (wasCurrentWorker) { + this.worker = null; + this.workerPid = null; + } this.clearIdleTimer(); - this.rejectAllPending(new Error("Local speech worker exited")); + this.rejectAllPending(error); for (const [sessionId, emitter] of this.sessionEmitters) { if (this.activeSessionIds.has(sessionId)) { - emitter.emit("error", new Error("Local speech worker exited")); + this.emitErrorIfObserved(sessionId, emitter, error, workerPid); } } this.activeSessionIds.clear(); @@ -331,6 +521,24 @@ export class LocalSpeechWorkerClient { this.inFlightRequests = 0; } + private describePendingRequests(): Array> { + const now = Date.now(); + return Array.from(this.pendingRequests.entries()).map(([requestId, request]) => + Object.assign( + { + requestId, + type: request.type, + ageMs: Math.max(0, now - request.startedAt), + }, + request.summary, + ), + ); + } + + private getStderrTail(): string { + return truncateStart(this.stderrTail.trim(), STDERR_TAIL_MAX_CHARS); + } + private rejectAllPending(error: Error): void { for (const [requestId, pending] of this.pendingRequests) { clearTimeout(pending.timeout); @@ -344,7 +552,27 @@ export class LocalSpeechWorkerClient { if (!emitter) { return; } - emitter.emit("error", error instanceof Error ? error : new Error(String(error))); + this.emitErrorIfObserved( + sessionId, + emitter, + error instanceof Error ? error : new Error(String(error)), + ); + } + + private emitErrorIfObserved( + sessionId: string, + emitter: EventEmitter, + error: Error, + workerPid: number | null = this.workerPid, + ): void { + if (emitter.listenerCount("error") > 0) { + emitter.emit("error", error); + return; + } + this.logger.warn( + { err: error, workerPid, sessionId }, + "Local speech worker session error had no listener", + ); } private scheduleIdleShutdownIfReady(): void {