From 4a517ee395aba04b35d5cafd2e2fce7ea77465d7 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Mon, 3 Aug 2026 12:35:37 +0800 Subject: [PATCH] fix(deepgram): preserve finalized text on close --- .../realtime-transcription-provider.test.ts | 42 +++++++++++++++-- .../realtime-transcription-provider.ts | 46 +++++++++++++++++++ 2 files changed, 85 insertions(+), 3 deletions(-) diff --git a/extensions/deepgram/realtime-transcription-provider.test.ts b/extensions/deepgram/realtime-transcription-provider.test.ts index 62c02a1bbe0f..0b2bfc9b2e30 100644 --- a/extensions/deepgram/realtime-transcription-provider.test.ts +++ b/extensions/deepgram/realtime-transcription-provider.test.ts @@ -66,6 +66,7 @@ function sendResult( describe("buildDeepgramRealtimeTranscriptionProvider", () => { afterEach(async () => { + vi.useRealTimers(); await cleanup?.(); cleanup = undefined; vi.unstubAllEnvs(); @@ -236,11 +237,12 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { const server = await createDeepgramRealtimeServer({ onRequest: () => undefined, onConnection: (ws) => { - sendResult(ws, { text: "goodbye" }); + sendResult(ws, { text: "good", isFinal: true }); + sendResult(ws, { text: "bye" }); ws.on("message", (data) => { if (JSON.parse(data.toString()).type === "Finalize") { sendResult(ws, { - text: "goodbye", + text: "bye", isFinal: true, fromFinalize: true, }); @@ -256,11 +258,45 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { await session.connect(); session.close(); - await vi.waitFor(() => expect(onTranscript).toHaveBeenCalledWith("goodbye")); + await vi.waitFor(() => expect(onTranscript).toHaveBeenCalledWith("good bye")); expect(onTranscript).toHaveBeenCalledTimes(1); }); + it("flushes finalized text once when finalize produces no result", async () => { + let finalizeRequests = 0; + const server = await createDeepgramRealtimeServer({ + onRequest: () => undefined, + onConnection: (ws) => { + sendResult(ws, { text: "good", isFinal: true }); + sendResult(ws, { text: "bye" }); + ws.on("message", (data) => { + if (JSON.parse(data.toString()).type === "Finalize") { + finalizeRequests += 1; + } + }); + }, + }); + const onPartial = vi.fn(); + const onTranscript = vi.fn(); + const session = buildDeepgramRealtimeTranscriptionProvider().createSession({ + providerConfig: { apiKey: "dummy", baseUrl: server.baseUrl, endpointingMs: 10_000 }, + onPartial, + onTranscript, + }); + + await session.connect(); + await vi.waitFor(() => expect(onPartial).toHaveBeenCalledWith("good bye")); + vi.useFakeTimers(); + session.close(); + session.close(); + await vi.advanceTimersByTimeAsync(5_000); + + expect(finalizeRequests).toBe(1); + expect(onTranscript).toHaveBeenCalledTimes(1); + expect(onTranscript).toHaveBeenCalledWith("good"); + }); + it("does not commit a turn on an utterance-end gap before speech-final", async () => { const server = await createDeepgramRealtimeServer({ onRequest: () => undefined, diff --git a/extensions/deepgram/realtime-transcription-provider.ts b/extensions/deepgram/realtime-transcription-provider.ts index fc0a6930bd47..cef983aec063 100644 --- a/extensions/deepgram/realtime-transcription-provider.ts +++ b/extensions/deepgram/realtime-transcription-provider.ts @@ -63,6 +63,7 @@ const DEEPGRAM_REALTIME_MAX_RECONNECT_ATTEMPTS = 5; const DEEPGRAM_REALTIME_RECONNECT_DELAY_MS = 1000; const DEEPGRAM_REALTIME_MAX_QUEUED_BYTES = 2 * 1024 * 1024; const DEEPGRAM_REALTIME_MAX_RETAINED_TRANSCRIPT_BYTES = 256 * 1024; +const DEEPGRAM_REALTIME_FINALIZE_FALLBACK_MS = DEEPGRAM_REALTIME_CLOSE_TIMEOUT_MS - 100; function readNestedDeepgramConfig(rawConfig: RealtimeTranscriptionProviderConfig) { const raw = readRecord(rawConfig); @@ -175,13 +176,24 @@ function createDeepgramRealtimeTranscriptionSession( let speechStarted = false; let finalizedTranscript = ""; let pendingPartial = ""; + let finalizeRequested = false; + let finalizeFallbackFired = false; + let finalizeFallbackTimer: ReturnType | undefined; const collapseWhitespace = (value: string) => value.replace(/\s+/g, " ").trim(); const joinTranscript = (left: string, right: string) => collapseWhitespace(left && right ? `${left} ${right}` : left || right); + const clearFinalizeFallback = () => { + if (finalizeFallbackTimer) { + clearTimeout(finalizeFallbackTimer); + finalizeFallbackTimer = undefined; + } + }; + const clearTurn = () => { + clearFinalizeFallback(); finalizedTranscript = ""; pendingPartial = ""; speechStarted = false; @@ -217,12 +229,23 @@ function createDeepgramRealtimeTranscriptionSession( } }; + const flushFinalizedTurn = () => { + const full = collapseWhitespace(finalizedTranscript); + clearTurn(); + if (full) { + config.onTranscript?.(full); + } + }; + const handleEvent = ( event: DeepgramRealtimeTranscriptionEvent, transport: RealtimeTranscriptionWebSocketTransport, ) => { switch (event.type) { case "Results": { + if (finalizeFallbackFired) { + return; + } const text = readTranscriptText(event); if (text && !speechStarted) { speechStarted = true; @@ -282,12 +305,35 @@ function createDeepgramRealtimeTranscriptionSession( onOpen: () => { // A reconnect starts a new provider stream. Never merge an old partial // utterance into audio recognized by the replacement connection. + finalizeRequested = false; + finalizeFallbackFired = false; clearTurn(); }, sendAudio: (audio, transport) => { transport.sendBinary(audio); }, onClose: (transport) => { + if (finalizeRequested) { + return; + } + finalizeRequested = true; + if (finalizedTranscript) { + // Finalize may produce no Results event when Deepgram has no buffered + // audio left. Preserve already-finalized text before core force-closes. + finalizeFallbackTimer = setTimeout(() => { + finalizeFallbackTimer = undefined; + finalizeFallbackFired = true; + try { + flushFinalizedTurn(); + } catch (error) { + try { + config.onError?.(error instanceof Error ? error : new Error(String(error))); + } catch { + // Error observers must not turn close fallback into an uncaught timer exception. + } + } + }, DEEPGRAM_REALTIME_FINALIZE_FALLBACK_MS); + } transport.sendJson({ type: "Finalize" }); }, onMessage: (event, transport) => handleEvent(event, transport),