From 1aac7ff6e251c08d7080e4d99d1a665bb58daee4 Mon Sep 17 00:00:00 2001 From: "H. H." <16421338+Weiming-Hu@users.noreply.github.com> Date: Sat, 1 Aug 2026 18:20:07 -0400 Subject: [PATCH 01/11] changes so that agent actually waits --- .../realtime-transcription-provider.ts | 69 +++++++++++++++++-- 1 file changed, 62 insertions(+), 7 deletions(-) diff --git a/extensions/deepgram/realtime-transcription-provider.ts b/extensions/deepgram/realtime-transcription-provider.ts index 68239d2d62f0..86f932f5fbe7 100644 --- a/extensions/deepgram/realtime-transcription-provider.ts +++ b/extensions/deepgram/realtime-transcription-provider.ts @@ -172,6 +172,22 @@ function createDeepgramRealtimeTranscriptionSession( let lastTranscript: string | undefined; let speechStarted = false; + // Deepgram emits `is_final: true` at internal phrase boundaries mid-utterance, + // independently of the `endpointing` silence window that gates `speech_final`. + // Ending the turn on `is_final` therefore cuts the speaker off during natural + // pauses and ignores the configured endpointing entirely. Instead we buffer the + // finalized segments and only end the turn when the speaker actually goes quiet: + // either Deepgram sends `speech_final`, or a client-side silence timer expires. + // The timer is the reliable signal on noisy telephony audio, where comfort noise + // means `speech_final` may never arrive. + let finalizedSegments: string[] = []; + let pendingPartial = ""; + let flushTimer: ReturnType | null = null; + const silenceFlushMs = + typeof config.endpointingMs === "number" && config.endpointingMs > 0 + ? config.endpointingMs + : DEEPGRAM_REALTIME_DEFAULT_ENDPOINTING_MS; + const emitTranscript = (text: string) => { if (text === lastTranscript) { return; @@ -180,10 +196,45 @@ function createDeepgramRealtimeTranscriptionSession( config.onTranscript?.(text); }; + const clearFlushTimer = () => { + if (flushTimer) { + clearTimeout(flushTimer); + flushTimer = null; + } + }; + + const collapseWhitespace = (value: string) => value.replace(/\s+/g, " ").trim(); + + const flushTurn = () => { + clearFlushTimer(); + const full = collapseWhitespace([...finalizedSegments, pendingPartial].join(" ")); + finalizedSegments = []; + pendingPartial = ""; + speechStarted = false; + if (full) { + emitTranscript(full); + } + }; + + const scheduleFlush = () => { + clearFlushTimer(); + flushTimer = setTimeout(flushTurn, silenceFlushMs); + flushTimer.unref?.(); + }; + const handleEvent = (event: DeepgramRealtimeTranscriptionEvent) => { switch (event.type) { case "Results": { const text = readTranscriptText(event); + // `speech_final` means the speaker has been silent for `endpointing` ms: + // end the turn immediately, appending any final words on this event. + if (event.speech_final) { + if (text) { + finalizedSegments.push(text); + } + flushTurn(); + return; + } if (!text) { return; } @@ -191,14 +242,17 @@ function createDeepgramRealtimeTranscriptionSession( speechStarted = true; config.onSpeechStart?.(); } - if (event.is_final || event.speech_final) { - emitTranscript(text); - if (event.speech_final) { - speechStarted = false; - } - return; + // Buffer finalized segments and interim words instead of ending the turn, + // then (re)arm the silence timer so a mid-sentence pause no longer cuts in. + if (event.is_final) { + finalizedSegments.push(text); + pendingPartial = ""; + config.onPartial?.(collapseWhitespace(finalizedSegments.join(" "))); + } else { + pendingPartial = text; + config.onPartial?.(collapseWhitespace([...finalizedSegments, text].join(" "))); } - config.onPartial?.(text); + scheduleFlush(); return; } case "SpeechStarted": @@ -232,6 +286,7 @@ function createDeepgramRealtimeTranscriptionSession( transport.sendBinary(audio); }, onClose: (transport) => { + clearFlushTimer(); transport.sendJson({ type: "Finalize" }); }, onMessage: handleEvent, From e9a5e188910b3131a56c398fd515578ee228faa4 Mon Sep 17 00:00:00 2001 From: "H. H." <16421338+Weiming-Hu@users.noreply.github.com> Date: Sat, 1 Aug 2026 18:20:07 -0400 Subject: [PATCH 02/11] clean up comments --- .../deepgram/realtime-transcription-provider.ts | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/extensions/deepgram/realtime-transcription-provider.ts b/extensions/deepgram/realtime-transcription-provider.ts index 86f932f5fbe7..71181b3aa0e0 100644 --- a/extensions/deepgram/realtime-transcription-provider.ts +++ b/extensions/deepgram/realtime-transcription-provider.ts @@ -172,14 +172,6 @@ function createDeepgramRealtimeTranscriptionSession( let lastTranscript: string | undefined; let speechStarted = false; - // Deepgram emits `is_final: true` at internal phrase boundaries mid-utterance, - // independently of the `endpointing` silence window that gates `speech_final`. - // Ending the turn on `is_final` therefore cuts the speaker off during natural - // pauses and ignores the configured endpointing entirely. Instead we buffer the - // finalized segments and only end the turn when the speaker actually goes quiet: - // either Deepgram sends `speech_final`, or a client-side silence timer expires. - // The timer is the reliable signal on noisy telephony audio, where comfort noise - // means `speech_final` may never arrive. let finalizedSegments: string[] = []; let pendingPartial = ""; let flushTimer: ReturnType | null = null; @@ -226,8 +218,6 @@ function createDeepgramRealtimeTranscriptionSession( switch (event.type) { case "Results": { const text = readTranscriptText(event); - // `speech_final` means the speaker has been silent for `endpointing` ms: - // end the turn immediately, appending any final words on this event. if (event.speech_final) { if (text) { finalizedSegments.push(text); @@ -242,8 +232,6 @@ function createDeepgramRealtimeTranscriptionSession( speechStarted = true; config.onSpeechStart?.(); } - // Buffer finalized segments and interim words instead of ending the turn, - // then (re)arm the silence timer so a mid-sentence pause no longer cuts in. if (event.is_final) { finalizedSegments.push(text); pendingPartial = ""; From b33b72181a5f560745c9ee7fd88cfa88134c114a Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Mon, 3 Aug 2026 09:48:47 +0800 Subject: [PATCH 03/11] fix(deepgram): bound realtime utterance lifecycle --- .../realtime-transcription-provider.test.ts | 189 ++++++++++++++++++ .../realtime-transcription-provider.ts | 99 ++++++--- 2 files changed, 257 insertions(+), 31 deletions(-) diff --git a/extensions/deepgram/realtime-transcription-provider.test.ts b/extensions/deepgram/realtime-transcription-provider.test.ts index c194dc28d7ad..9e6044fe8aeb 100644 --- a/extensions/deepgram/realtime-transcription-provider.test.ts +++ b/extensions/deepgram/realtime-transcription-provider.test.ts @@ -11,6 +11,7 @@ let cleanup: (() => Promise) | undefined; async function createDeepgramRealtimeServer(params: { onRequest: (url: URL, headers: Record) => void; + onConnection?: (ws: WebSocket) => void; }) { const server = createServer(); const wss = new WebSocketServer({ noServer: true }); @@ -21,6 +22,7 @@ async function createDeepgramRealtimeServer(params: { wss.handleUpgrade(request, socket, head, (ws) => { clients.add(ws); ws.on("close", () => clients.delete(ws)); + params.onConnection?.(ws); }); }); @@ -42,6 +44,26 @@ async function createDeepgramRealtimeServer(params: { return { baseUrl: `http://127.0.0.1:${port}/deepgram/v1` }; } +function sendResult( + ws: WebSocket, + params: { + text: string; + isFinal?: boolean; + speechFinal?: boolean; + fromFinalize?: boolean; + }, +) { + ws.send( + JSON.stringify({ + type: "Results", + channel: { alternatives: [{ transcript: params.text }] }, + is_final: params.isFinal ?? false, + speech_final: params.speechFinal ?? false, + from_finalize: params.fromFinalize ?? false, + }), + ); +} + describe("buildDeepgramRealtimeTranscriptionProvider", () => { afterEach(async () => { await cleanup?.(); @@ -141,4 +163,171 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { expect(requests[0]?.url.searchParams.get("model")).toBe("nova-3"); expect(requests[0]?.headers.authorization).toBe("Token dummy"); }); + + it("buffers finalized segments until the utterance is complete", async () => { + const server = await createDeepgramRealtimeServer({ + onRequest: () => undefined, + onConnection: (ws) => { + sendResult(ws, { text: "hello", isFinal: true }); + sendResult(ws, { text: "world", isFinal: true, speechFinal: true }); + }, + }); + const onPartial = vi.fn(); + const onTranscript = vi.fn(); + const session = buildDeepgramRealtimeTranscriptionProvider().createSession({ + providerConfig: { apiKey: "dummy", baseUrl: server.baseUrl, endpointingMs: 1000 }, + onPartial, + onTranscript, + }); + + await session.connect(); + await vi.waitFor(() => expect(onTranscript).toHaveBeenCalledWith("hello world")); + session.close(); + + expect(onPartial).toHaveBeenCalledWith("hello"); + expect(onTranscript).toHaveBeenCalledTimes(1); + }); + + it("replaces the provisional tail with the text-bearing speech-final result", async () => { + const server = await createDeepgramRealtimeServer({ + onRequest: () => undefined, + onConnection: (ws) => { + sendResult(ws, { text: "hello" }); + sendResult(ws, { text: "hello", isFinal: true, speechFinal: true }); + }, + }); + const onTranscript = vi.fn(); + const session = buildDeepgramRealtimeTranscriptionProvider().createSession({ + providerConfig: { apiKey: "dummy", baseUrl: server.baseUrl, endpointingMs: 1000 }, + onTranscript, + }); + + await session.connect(); + await vi.waitFor(() => expect(onTranscript).toHaveBeenCalledWith("hello")); + session.close(); + + expect(onTranscript).toHaveBeenCalledTimes(1); + }); + + it("preserves identical transcripts from consecutive utterances", async () => { + const server = await createDeepgramRealtimeServer({ + onRequest: () => undefined, + onConnection: (ws) => { + sendResult(ws, { text: "yes", isFinal: true, speechFinal: true }); + sendResult(ws, { text: "yes", isFinal: true, speechFinal: true }); + }, + }); + const onTranscript = vi.fn(); + const session = buildDeepgramRealtimeTranscriptionProvider().createSession({ + providerConfig: { apiKey: "dummy", baseUrl: server.baseUrl, endpointingMs: 1000 }, + onTranscript, + }); + + await session.connect(); + await vi.waitFor(() => expect(onTranscript).toHaveBeenCalledTimes(2)); + session.close(); + + expect(onTranscript.mock.calls).toEqual([["yes"], ["yes"]]); + }); + + it("flushes finalized text returned after a client finalize request", async () => { + const server = await createDeepgramRealtimeServer({ + onRequest: () => undefined, + onConnection: (ws) => { + sendResult(ws, { text: "goodbye" }); + ws.on("message", (data) => { + if (JSON.parse(data.toString()).type === "Finalize") { + sendResult(ws, { + text: "goodbye", + isFinal: true, + fromFinalize: true, + }); + } + }); + }, + }); + const onTranscript = vi.fn(); + const session = buildDeepgramRealtimeTranscriptionProvider().createSession({ + providerConfig: { apiKey: "dummy", baseUrl: server.baseUrl, endpointingMs: 10_000 }, + onTranscript, + }); + + await session.connect(); + session.close(); + await vi.waitFor(() => expect(onTranscript).toHaveBeenCalledWith("goodbye")); + + expect(onTranscript).toHaveBeenCalledTimes(1); + }); + + it("flushes a finalized segment after the endpointing fallback delay", async () => { + const server = await createDeepgramRealtimeServer({ + onRequest: () => undefined, + onConnection: (ws) => { + sendResult(ws, { text: "fallback", isFinal: true }); + }, + }); + const onTranscript = vi.fn(); + const session = buildDeepgramRealtimeTranscriptionProvider().createSession({ + providerConfig: { apiKey: "dummy", baseUrl: server.baseUrl, endpointingMs: 25 }, + onTranscript, + }); + + await session.connect(); + await vi.waitFor(() => expect(onTranscript).toHaveBeenCalledWith("fallback")); + session.close(); + }); + + it("does not merge an interrupted turn into a reconnected provider stream", async () => { + let connectionCount = 0; + const server = await createDeepgramRealtimeServer({ + onRequest: () => undefined, + onConnection: (ws) => { + connectionCount += 1; + if (connectionCount === 1) { + sendResult(ws, { text: "old", isFinal: true }); + ws.close(); + return; + } + sendResult(ws, { text: "new", isFinal: true, speechFinal: true }); + }, + }); + const onTranscript = vi.fn(); + const session = buildDeepgramRealtimeTranscriptionProvider().createSession({ + providerConfig: { apiKey: "dummy", baseUrl: server.baseUrl, endpointingMs: 10_000 }, + onTranscript, + }); + + await session.connect(); + await vi.waitFor(() => expect(onTranscript).toHaveBeenCalledWith("new"), { + timeout: 3000, + }); + session.close(); + + expect(onTranscript).toHaveBeenCalledTimes(1); + }); + + it("terminates instead of retaining an oversized utterance", async () => { + const server = await createDeepgramRealtimeServer({ + onRequest: () => undefined, + onConnection: (ws) => { + sendResult(ws, { text: "x".repeat(256 * 1024), isFinal: true }); + sendResult(ws, { text: "y" }); + }, + }); + const onError = vi.fn(); + const session = buildDeepgramRealtimeTranscriptionProvider().createSession({ + providerConfig: { apiKey: "dummy", baseUrl: server.baseUrl, endpointingMs: 1000 }, + onError, + }); + + await session.connect(); + await vi.waitFor(() => + expect(onError).toHaveBeenCalledWith( + expect.objectContaining({ + message: expect.stringContaining("retained transcript exceeded"), + }), + ), + ); + session.close(); + }); }); diff --git a/extensions/deepgram/realtime-transcription-provider.ts b/extensions/deepgram/realtime-transcription-provider.ts index 71181b3aa0e0..ee8b7819fc07 100644 --- a/extensions/deepgram/realtime-transcription-provider.ts +++ b/extensions/deepgram/realtime-transcription-provider.ts @@ -5,6 +5,7 @@ import { type RealtimeTranscriptionProviderPlugin, type RealtimeTranscriptionSession, type RealtimeTranscriptionSessionCreateRequest, + type RealtimeTranscriptionWebSocketTransport, } from "openclaw/plugin-sdk/realtime-transcription"; import { normalizeResolvedSecretInputString } from "openclaw/plugin-sdk/secret-input"; import { @@ -48,6 +49,7 @@ type DeepgramRealtimeTranscriptionEvent = { }; is_final?: boolean; speech_final?: boolean; + from_finalize?: boolean; error?: unknown; message?: string; }; @@ -60,6 +62,8 @@ const DEEPGRAM_REALTIME_CLOSE_TIMEOUT_MS = 5_000; 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_ENDPOINTING_FALLBACK_MARGIN_MS = 250; function readNestedDeepgramConfig(rawConfig: RealtimeTranscriptionProviderConfig) { const raw = readRecord(rawConfig); @@ -169,24 +173,15 @@ function readTranscriptText(event: DeepgramRealtimeTranscriptionEvent): string | function createDeepgramRealtimeTranscriptionSession( config: DeepgramRealtimeTranscriptionSessionConfig, ): RealtimeTranscriptionSession { - let lastTranscript: string | undefined; let speechStarted = false; - - let finalizedSegments: string[] = []; + let finalizedTranscript = ""; let pendingPartial = ""; let flushTimer: ReturnType | null = null; const silenceFlushMs = - typeof config.endpointingMs === "number" && config.endpointingMs > 0 + (typeof config.endpointingMs === "number" && config.endpointingMs > 0 ? config.endpointingMs - : DEEPGRAM_REALTIME_DEFAULT_ENDPOINTING_MS; - - const emitTranscript = (text: string) => { - if (text === lastTranscript) { - return; - } - lastTranscript = text; - config.onTranscript?.(text); - }; + : DEEPGRAM_REALTIME_DEFAULT_ENDPOINTING_MS) + + DEEPGRAM_REALTIME_ENDPOINTING_FALLBACK_MARGIN_MS; const clearFlushTimer = () => { if (flushTimer) { @@ -197,14 +192,44 @@ function createDeepgramRealtimeTranscriptionSession( const collapseWhitespace = (value: string) => value.replace(/\s+/g, " ").trim(); - const flushTurn = () => { + const joinTranscript = (left: string, right: string) => + collapseWhitespace(left && right ? `${left} ${right}` : left || right); + + const clearTurn = () => { clearFlushTimer(); - const full = collapseWhitespace([...finalizedSegments, pendingPartial].join(" ")); - finalizedSegments = []; + finalizedTranscript = ""; pendingPartial = ""; speechStarted = false; + }; + + const updateTurn = ( + nextFinalized: string, + nextPartial: string, + transport: RealtimeTranscriptionWebSocketTransport, + ) => { + const retainedBytes = + Buffer.byteLength(nextFinalized, "utf8") + Buffer.byteLength(nextPartial, "utf8"); + if (retainedBytes > DEEPGRAM_REALTIME_MAX_RETAINED_TRANSCRIPT_BYTES) { + clearTurn(); + config.onError?.( + new Error( + `Deepgram realtime retained transcript exceeded ${DEEPGRAM_REALTIME_MAX_RETAINED_TRANSCRIPT_BYTES} bytes`, + ), + ); + transport.closeNow(); + return false; + } + finalizedTranscript = nextFinalized; + pendingPartial = nextPartial; + return true; + }; + + const flushTurn = () => { + clearFlushTimer(); + const full = joinTranscript(finalizedTranscript, pendingPartial); + clearTurn(); if (full) { - emitTranscript(full); + config.onTranscript?.(full); } }; @@ -214,13 +239,20 @@ function createDeepgramRealtimeTranscriptionSession( flushTimer.unref?.(); }; - const handleEvent = (event: DeepgramRealtimeTranscriptionEvent) => { + const handleEvent = ( + event: DeepgramRealtimeTranscriptionEvent, + transport: RealtimeTranscriptionWebSocketTransport, + ) => { switch (event.type) { case "Results": { const text = readTranscriptText(event); - if (event.speech_final) { - if (text) { - finalizedSegments.push(text); + if (text && !speechStarted) { + speechStarted = true; + config.onSpeechStart?.(); + } + if (event.speech_final || event.from_finalize) { + if (text && !updateTurn(joinTranscript(finalizedTranscript, text), "", transport)) { + return; } flushTurn(); return; @@ -228,17 +260,17 @@ function createDeepgramRealtimeTranscriptionSession( if (!text) { return; } - if (!speechStarted) { - speechStarted = true; - config.onSpeechStart?.(); - } if (event.is_final) { - finalizedSegments.push(text); - pendingPartial = ""; - config.onPartial?.(collapseWhitespace(finalizedSegments.join(" "))); + const nextFinalized = joinTranscript(finalizedTranscript, text); + if (!updateTurn(nextFinalized, "", transport)) { + return; + } + config.onPartial?.(nextFinalized); } else { - pendingPartial = text; - config.onPartial?.(collapseWhitespace([...finalizedSegments, text].join(" "))); + if (!updateTurn(finalizedTranscript, text, transport)) { + return; + } + config.onPartial?.(joinTranscript(finalizedTranscript, text)); } scheduleFlush(); return; @@ -270,6 +302,11 @@ function createDeepgramRealtimeTranscriptionSession( connectClosedBeforeReadyMessage: "Deepgram realtime transcription connection closed before ready", reconnectLimitMessage: "Deepgram realtime transcription reconnect limit reached", + onOpen: () => { + // A reconnect starts a new provider stream. Never merge an old partial + // utterance into audio recognized by the replacement connection. + clearTurn(); + }, sendAudio: (audio, transport) => { transport.sendBinary(audio); }, @@ -277,7 +314,7 @@ function createDeepgramRealtimeTranscriptionSession( clearFlushTimer(); transport.sendJson({ type: "Finalize" }); }, - onMessage: handleEvent, + onMessage: (event, transport) => handleEvent(event, transport), }); } From 509dee1150a73ad7e5c4dd35018bfefe8a4addc9 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Mon, 3 Aug 2026 11:07:00 +0800 Subject: [PATCH 04/11] test(deepgram): remove live speech dependency --- extensions/deepgram/audio.live.test.ts | 47 ++++++++++++++++++-------- 1 file changed, 32 insertions(+), 15 deletions(-) diff --git a/extensions/deepgram/audio.live.test.ts b/extensions/deepgram/audio.live.test.ts index 66a209f2af70..fb1423226d08 100644 --- a/extensions/deepgram/audio.live.test.ts +++ b/extensions/deepgram/audio.live.test.ts @@ -1,15 +1,12 @@ // Deepgram tests cover audio plugin behavior. -import { - runRealtimeSttLiveTest, - synthesizeElevenLabsLiveSpeech, -} from "openclaw/plugin-sdk/provider-test-contracts"; +import { spawnSync } from "node:child_process"; +import { runRealtimeSttLiveTest } from "openclaw/plugin-sdk/provider-test-contracts"; import { isLiveTestEnabled } from "openclaw/plugin-sdk/test-live"; import { describe, expect, it } from "vitest"; import { transcribeDeepgramAudio } from "./audio.js"; import { buildDeepgramRealtimeTranscriptionProvider } from "./realtime-transcription-provider.js"; const DEEPGRAM_KEY = process.env.DEEPGRAM_API_KEY ?? ""; -const ELEVENLABS_KEY = process.env.ELEVENLABS_API_KEY ?? ""; const DEEPGRAM_MODEL = process.env.DEEPGRAM_MODEL?.trim() || "nova-3"; const DEEPGRAM_BASE_URL = process.env.DEEPGRAM_BASE_URL?.trim(); const SAMPLE_URL = @@ -34,6 +31,34 @@ async function fetchSampleBuffer(url: string, timeoutMs: number): Promise { it("transcribes sample audio", async () => { const buffer = await fetchSampleBuffer(SAMPLE_URL, 15000); @@ -50,17 +75,8 @@ describeLive("deepgram live", () => { }, 30000); it("streams realtime STT through the registered transcription provider", async () => { - if (!ELEVENLABS_KEY) { - throw new Error("ELEVENLABS_API_KEY required to synthesize live realtime STT input"); - } const provider = buildDeepgramRealtimeTranscriptionProvider(); - const phrase = "Testing OpenClaw Deepgram realtime transcription integration OK."; - const speech = await synthesizeElevenLabsLiveSpeech({ - text: phrase, - apiKey: ELEVENLABS_KEY, - outputFormat: "ulaw_8000", - timeoutMs: 30_000, - }); + const speech = convertWavToMulaw8k(await fetchSampleBuffer(SAMPLE_URL, 15_000)); expect(speech.byteLength).toBeGreaterThan(0); await runRealtimeSttLiveTest({ @@ -71,6 +87,7 @@ describeLive("deepgram live", () => { endpointingMs: 500, }, audio: Buffer.concat([Buffer.alloc(4000, 0xff), speech, Buffer.alloc(8000, 0xff)]), + expectedNormalizedText: "life moves pretty fast", }); }, 90_000); }); From 03a36570e70e2654cc9d7d6860c47288c77a1b95 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Mon, 3 Aug 2026 11:15:15 +0800 Subject: [PATCH 05/11] test(deepgram): bound websocket fixture frames --- extensions/deepgram/realtime-transcription-provider.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/extensions/deepgram/realtime-transcription-provider.test.ts b/extensions/deepgram/realtime-transcription-provider.test.ts index 9e6044fe8aeb..4d9844f97848 100644 --- a/extensions/deepgram/realtime-transcription-provider.test.ts +++ b/extensions/deepgram/realtime-transcription-provider.test.ts @@ -14,7 +14,7 @@ async function createDeepgramRealtimeServer(params: { onConnection?: (ws: WebSocket) => void; }) { const server = createServer(); - const wss = new WebSocketServer({ noServer: true }); + const wss = new WebSocketServer({ noServer: true, maxPayload: 1024 * 1024 }); const clients = new Set(); server.on("upgrade", (request, socket, head) => { From dd6941a473b986b092b4adcbc47d89c9dc4184b1 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Mon, 3 Aug 2026 11:59:50 +0800 Subject: [PATCH 06/11] fix(deepgram): use provider utterance boundaries --- .../realtime-transcription-provider.test.ts | 32 +++++++++++++++- .../realtime-transcription-provider.ts | 37 +++++++------------ 2 files changed, 44 insertions(+), 25 deletions(-) diff --git a/extensions/deepgram/realtime-transcription-provider.test.ts b/extensions/deepgram/realtime-transcription-provider.test.ts index 4d9844f97848..92f8b14a6a0a 100644 --- a/extensions/deepgram/realtime-transcription-provider.test.ts +++ b/extensions/deepgram/realtime-transcription-provider.test.ts @@ -161,6 +161,8 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { expect(requests).toHaveLength(1); expect(requests[0]?.url.pathname).toBe("/deepgram/v1/listen"); expect(requests[0]?.url.searchParams.get("model")).toBe("nova-3"); + expect(requests[0]?.url.searchParams.get("endpointing")).toBe("800"); + expect(requests[0]?.url.searchParams.get("utterance_end_ms")).toBe("1000"); expect(requests[0]?.headers.authorization).toBe("Token dummy"); }); @@ -259,11 +261,12 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { expect(onTranscript).toHaveBeenCalledTimes(1); }); - it("flushes a finalized segment after the endpointing fallback delay", async () => { + it("flushes a finalized segment on the provider utterance-end event", async () => { const server = await createDeepgramRealtimeServer({ onRequest: () => undefined, onConnection: (ws) => { sendResult(ws, { text: "fallback", isFinal: true }); + ws.send(JSON.stringify({ type: "UtteranceEnd" })); }, }); const onTranscript = vi.fn(); @@ -277,6 +280,33 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { session.close(); }); + it("does not infer silence from a gap between provisional results", async () => { + let socket: WebSocket | undefined; + const server = await createDeepgramRealtimeServer({ + onRequest: () => undefined, + onConnection: (ws) => { + socket = ws; + sendResult(ws, { text: "still speaking" }); + }, + }); + const onPartial = vi.fn(); + const onTranscript = vi.fn(); + const session = buildDeepgramRealtimeTranscriptionProvider().createSession({ + providerConfig: { apiKey: "dummy", baseUrl: server.baseUrl, endpointingMs: 25 }, + onPartial, + onTranscript, + }); + + await session.connect(); + await vi.waitFor(() => expect(onPartial).toHaveBeenCalledWith("still speaking")); + await new Promise((resolve) => setTimeout(resolve, 350)); + expect(onTranscript).not.toHaveBeenCalled(); + + sendResult(socket!, { text: "continuous speech", isFinal: true, speechFinal: true }); + await vi.waitFor(() => expect(onTranscript).toHaveBeenCalledWith("continuous speech")); + session.close(); + }); + it("does not merge an interrupted turn into a reconnected provider stream", async () => { let connectionCount = 0; const server = await createDeepgramRealtimeServer({ diff --git a/extensions/deepgram/realtime-transcription-provider.ts b/extensions/deepgram/realtime-transcription-provider.ts index ee8b7819fc07..0079ade1cd59 100644 --- a/extensions/deepgram/realtime-transcription-provider.ts +++ b/extensions/deepgram/realtime-transcription-provider.ts @@ -63,7 +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_ENDPOINTING_FALLBACK_MARGIN_MS = 250; +const DEEPGRAM_REALTIME_MIN_UTTERANCE_END_MS = 1000; function readNestedDeepgramConfig(rawConfig: RealtimeTranscriptionProviderConfig) { const raw = readRecord(rawConfig); @@ -131,6 +131,14 @@ function toDeepgramRealtimeWsUrl(config: DeepgramRealtimeTranscriptionSessionCon url.searchParams.set("channels", "1"); url.searchParams.set("interim_results", String(config.interimResults)); url.searchParams.set("endpointing", String(config.endpointingMs)); + if (config.interimResults) { + // Deepgram derives UtteranceEnd from word timings and rejects values below + // one second. Unlike a client timer, background noise does not rearm it. + url.searchParams.set( + "utterance_end_ms", + String(Math.max(config.endpointingMs, DEEPGRAM_REALTIME_MIN_UTTERANCE_END_MS)), + ); + } if (config.language) { url.searchParams.set("language", config.language); } @@ -176,19 +184,6 @@ function createDeepgramRealtimeTranscriptionSession( let speechStarted = false; let finalizedTranscript = ""; let pendingPartial = ""; - let flushTimer: ReturnType | null = null; - const silenceFlushMs = - (typeof config.endpointingMs === "number" && config.endpointingMs > 0 - ? config.endpointingMs - : DEEPGRAM_REALTIME_DEFAULT_ENDPOINTING_MS) + - DEEPGRAM_REALTIME_ENDPOINTING_FALLBACK_MARGIN_MS; - - const clearFlushTimer = () => { - if (flushTimer) { - clearTimeout(flushTimer); - flushTimer = null; - } - }; const collapseWhitespace = (value: string) => value.replace(/\s+/g, " ").trim(); @@ -196,7 +191,6 @@ function createDeepgramRealtimeTranscriptionSession( collapseWhitespace(left && right ? `${left} ${right}` : left || right); const clearTurn = () => { - clearFlushTimer(); finalizedTranscript = ""; pendingPartial = ""; speechStarted = false; @@ -225,7 +219,6 @@ function createDeepgramRealtimeTranscriptionSession( }; const flushTurn = () => { - clearFlushTimer(); const full = joinTranscript(finalizedTranscript, pendingPartial); clearTurn(); if (full) { @@ -233,12 +226,6 @@ function createDeepgramRealtimeTranscriptionSession( } }; - const scheduleFlush = () => { - clearFlushTimer(); - flushTimer = setTimeout(flushTurn, silenceFlushMs); - flushTimer.unref?.(); - }; - const handleEvent = ( event: DeepgramRealtimeTranscriptionEvent, transport: RealtimeTranscriptionWebSocketTransport, @@ -272,7 +259,10 @@ function createDeepgramRealtimeTranscriptionSession( } config.onPartial?.(joinTranscript(finalizedTranscript, text)); } - scheduleFlush(); + return; + } + case "UtteranceEnd": { + flushTurn(); return; } case "SpeechStarted": @@ -311,7 +301,6 @@ function createDeepgramRealtimeTranscriptionSession( transport.sendBinary(audio); }, onClose: (transport) => { - clearFlushTimer(); transport.sendJson({ type: "Finalize" }); }, onMessage: (event, transport) => handleEvent(event, transport), From 73b0312e525126200e1ac83c632166112eed16f7 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Mon, 3 Aug 2026 12:16:07 +0800 Subject: [PATCH 07/11] fix(deepgram): keep endpointing turn-authoritative --- .../realtime-transcription-provider.test.ts | 11 +++++++---- .../deepgram/realtime-transcription-provider.ts | 13 ------------- 2 files changed, 7 insertions(+), 17 deletions(-) diff --git a/extensions/deepgram/realtime-transcription-provider.test.ts b/extensions/deepgram/realtime-transcription-provider.test.ts index 92f8b14a6a0a..62c02a1bbe0f 100644 --- a/extensions/deepgram/realtime-transcription-provider.test.ts +++ b/extensions/deepgram/realtime-transcription-provider.test.ts @@ -162,7 +162,7 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { expect(requests[0]?.url.pathname).toBe("/deepgram/v1/listen"); expect(requests[0]?.url.searchParams.get("model")).toBe("nova-3"); expect(requests[0]?.url.searchParams.get("endpointing")).toBe("800"); - expect(requests[0]?.url.searchParams.get("utterance_end_ms")).toBe("1000"); + expect(requests[0]?.url.searchParams.has("utterance_end_ms")).toBe(false); expect(requests[0]?.headers.authorization).toBe("Token dummy"); }); @@ -261,12 +261,13 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { expect(onTranscript).toHaveBeenCalledTimes(1); }); - it("flushes a finalized segment on the provider utterance-end event", async () => { + it("does not commit a turn on an utterance-end gap before speech-final", async () => { const server = await createDeepgramRealtimeServer({ onRequest: () => undefined, onConnection: (ws) => { - sendResult(ws, { text: "fallback", isFinal: true }); + sendResult(ws, { text: "still", isFinal: true }); ws.send(JSON.stringify({ type: "UtteranceEnd" })); + sendResult(ws, { text: "speaking", isFinal: true, speechFinal: true }); }, }); const onTranscript = vi.fn(); @@ -276,8 +277,10 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { }); await session.connect(); - await vi.waitFor(() => expect(onTranscript).toHaveBeenCalledWith("fallback")); + await vi.waitFor(() => expect(onTranscript).toHaveBeenCalledWith("still speaking")); session.close(); + + expect(onTranscript).toHaveBeenCalledTimes(1); }); it("does not infer silence from a gap between provisional results", async () => { diff --git a/extensions/deepgram/realtime-transcription-provider.ts b/extensions/deepgram/realtime-transcription-provider.ts index 0079ade1cd59..fc0a6930bd47 100644 --- a/extensions/deepgram/realtime-transcription-provider.ts +++ b/extensions/deepgram/realtime-transcription-provider.ts @@ -63,7 +63,6 @@ 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_MIN_UTTERANCE_END_MS = 1000; function readNestedDeepgramConfig(rawConfig: RealtimeTranscriptionProviderConfig) { const raw = readRecord(rawConfig); @@ -131,14 +130,6 @@ function toDeepgramRealtimeWsUrl(config: DeepgramRealtimeTranscriptionSessionCon url.searchParams.set("channels", "1"); url.searchParams.set("interim_results", String(config.interimResults)); url.searchParams.set("endpointing", String(config.endpointingMs)); - if (config.interimResults) { - // Deepgram derives UtteranceEnd from word timings and rejects values below - // one second. Unlike a client timer, background noise does not rearm it. - url.searchParams.set( - "utterance_end_ms", - String(Math.max(config.endpointingMs, DEEPGRAM_REALTIME_MIN_UTTERANCE_END_MS)), - ); - } if (config.language) { url.searchParams.set("language", config.language); } @@ -261,10 +252,6 @@ function createDeepgramRealtimeTranscriptionSession( } return; } - case "UtteranceEnd": { - flushTurn(); - return; - } case "SpeechStarted": speechStarted = true; config.onSpeechStart?.(); From 4a517ee395aba04b35d5cafd2e2fce7ea77465d7 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Mon, 3 Aug 2026 12:35:37 +0800 Subject: [PATCH 08/11] 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), From 81ae848c5b81fac0ad74f4ab4757b7eeb6115145 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Mon, 3 Aug 2026 12:56:59 +0800 Subject: [PATCH 09/11] test(deepgram): satisfy realtime fixture lint --- .../realtime-transcription-provider.test.ts | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/extensions/deepgram/realtime-transcription-provider.test.ts b/extensions/deepgram/realtime-transcription-provider.test.ts index 0b2bfc9b2e30..b94c76d74798 100644 --- a/extensions/deepgram/realtime-transcription-provider.test.ts +++ b/extensions/deepgram/realtime-transcription-provider.test.ts @@ -4,6 +4,7 @@ import type { AddressInfo } from "node:net"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { afterEach, describe, expect, it, vi } from "vitest"; import type WebSocket from "ws"; +import type { RawData } from "ws"; import { WebSocketServer } from "ws"; import { buildDeepgramRealtimeTranscriptionProvider } from "./realtime-transcription-provider.js"; @@ -64,6 +65,15 @@ function sendResult( ); } +function parseClientMessage(data: RawData): Record { + const bytes = Buffer.isBuffer(data) + ? data + : Array.isArray(data) + ? Buffer.concat(data) + : Buffer.from(data); + return JSON.parse(bytes.toString("utf8")) as Record; +} + describe("buildDeepgramRealtimeTranscriptionProvider", () => { afterEach(async () => { vi.useRealTimers(); @@ -240,7 +250,7 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { sendResult(ws, { text: "good", isFinal: true }); sendResult(ws, { text: "bye" }); ws.on("message", (data) => { - if (JSON.parse(data.toString()).type === "Finalize") { + if (parseClientMessage(data).type === "Finalize") { sendResult(ws, { text: "bye", isFinal: true, @@ -271,7 +281,7 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { sendResult(ws, { text: "good", isFinal: true }); sendResult(ws, { text: "bye" }); ws.on("message", (data) => { - if (JSON.parse(data.toString()).type === "Finalize") { + if (parseClientMessage(data).type === "Finalize") { finalizeRequests += 1; } }); @@ -338,7 +348,9 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { await session.connect(); await vi.waitFor(() => expect(onPartial).toHaveBeenCalledWith("still speaking")); - await new Promise((resolve) => setTimeout(resolve, 350)); + await new Promise((resolve) => { + setTimeout(resolve, 350); + }); expect(onTranscript).not.toHaveBeenCalled(); sendResult(socket!, { text: "continuous speech", isFinal: true, speechFinal: true }); From a74fcc5271412233b4251a4161c18b6603b2cbf1 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Mon, 3 Aug 2026 17:31:53 +0800 Subject: [PATCH 10/11] test(deepgram): match normalized live transcript --- extensions/deepgram/audio.live.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/extensions/deepgram/audio.live.test.ts b/extensions/deepgram/audio.live.test.ts index fb1423226d08..c6beef41c300 100644 --- a/extensions/deepgram/audio.live.test.ts +++ b/extensions/deepgram/audio.live.test.ts @@ -87,7 +87,7 @@ describeLive("deepgram live", () => { endpointingMs: 500, }, audio: Buffer.concat([Buffer.alloc(4000, 0xff), speech, Buffer.alloc(8000, 0xff)]), - expectedNormalizedText: "life moves pretty fast", + expectedNormalizedText: "lifemovesprettyfast", }); }, 90_000); }); From 5254e84d7d2ad890204658ea4938960342af75f8 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Mon, 3 Aug 2026 17:56:45 +0800 Subject: [PATCH 11/11] fix(deepgram): preserve transcript integrity --- .../realtime-transcription-provider.test.ts | 58 ++++++++++++++++++- .../realtime-transcription-provider.ts | 17 ++++-- 2 files changed, 70 insertions(+), 5 deletions(-) diff --git a/extensions/deepgram/realtime-transcription-provider.test.ts b/extensions/deepgram/realtime-transcription-provider.test.ts index b94c76d74798..c8ef0d2c75a7 100644 --- a/extensions/deepgram/realtime-transcription-provider.test.ts +++ b/extensions/deepgram/realtime-transcription-provider.test.ts @@ -222,6 +222,33 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { expect(onTranscript).toHaveBeenCalledTimes(1); }); + it("does not promote a rejected provisional tail on an empty speech-final result", async () => { + const server = await createDeepgramRealtimeServer({ + onRequest: () => undefined, + onConnection: (ws) => { + sendResult(ws, { text: "delete everything" }); + sendResult(ws, { text: "", isFinal: true, speechFinal: true }); + ws.send(JSON.stringify({ type: "SpeechStarted" })); + }, + }); + const onPartial = vi.fn(); + const onSpeechStart = vi.fn(); + const onTranscript = vi.fn(); + const session = buildDeepgramRealtimeTranscriptionProvider().createSession({ + providerConfig: { apiKey: "dummy", baseUrl: server.baseUrl, endpointingMs: 1000 }, + onPartial, + onSpeechStart, + onTranscript, + }); + + await session.connect(); + await vi.waitFor(() => expect(onSpeechStart).toHaveBeenCalledTimes(2)); + session.close(); + + expect(onPartial).toHaveBeenCalledWith("delete everything"); + expect(onTranscript).not.toHaveBeenCalled(); + }); + it("preserves identical transcripts from consecutive utterances", async () => { const server = await createDeepgramRealtimeServer({ onRequest: () => undefined, @@ -365,7 +392,7 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { onConnection: (ws) => { connectionCount += 1; if (connectionCount === 1) { - sendResult(ws, { text: "old", isFinal: true }); + sendResult(ws, { text: "old" }); ws.close(); return; } @@ -387,6 +414,35 @@ describe("buildDeepgramRealtimeTranscriptionProvider", () => { expect(onTranscript).toHaveBeenCalledTimes(1); }); + it("preserves finalized speech as a separate turn when the provider reconnects", async () => { + let connectionCount = 0; + const server = await createDeepgramRealtimeServer({ + onRequest: () => undefined, + onConnection: (ws) => { + connectionCount += 1; + if (connectionCount === 1) { + sendResult(ws, { text: "old", isFinal: true }); + ws.close(); + return; + } + sendResult(ws, { text: "new", isFinal: true, speechFinal: true }); + }, + }); + const onTranscript = vi.fn(); + const session = buildDeepgramRealtimeTranscriptionProvider().createSession({ + providerConfig: { apiKey: "dummy", baseUrl: server.baseUrl, endpointingMs: 10_000 }, + onTranscript, + }); + + await session.connect(); + await vi.waitFor(() => expect(onTranscript).toHaveBeenCalledTimes(2), { + timeout: 3000, + }); + session.close(); + + expect(onTranscript.mock.calls).toEqual([["old"], ["new"]]); + }); + it("terminates instead of retaining an oversized utterance", async () => { const server = await createDeepgramRealtimeServer({ onRequest: () => undefined, diff --git a/extensions/deepgram/realtime-transcription-provider.ts b/extensions/deepgram/realtime-transcription-provider.ts index cef983aec063..3a2fb38195b5 100644 --- a/extensions/deepgram/realtime-transcription-provider.ts +++ b/extensions/deepgram/realtime-transcription-provider.ts @@ -179,6 +179,7 @@ function createDeepgramRealtimeTranscriptionSession( let finalizeRequested = false; let finalizeFallbackFired = false; let finalizeFallbackTimer: ReturnType | undefined; + let openedOnce = false; const collapseWhitespace = (value: string) => value.replace(/\s+/g, " ").trim(); @@ -252,7 +253,10 @@ function createDeepgramRealtimeTranscriptionSession( config.onSpeechStart?.(); } if (event.speech_final || event.from_finalize) { - if (text && !updateTurn(joinTranscript(finalizedTranscript, text), "", transport)) { + const nextFinalized = text + ? joinTranscript(finalizedTranscript, text) + : finalizedTranscript; + if (!updateTurn(nextFinalized, "", transport)) { return; } flushTurn(); @@ -303,11 +307,16 @@ function createDeepgramRealtimeTranscriptionSession( "Deepgram realtime transcription connection closed before ready", reconnectLimitMessage: "Deepgram realtime transcription reconnect limit reached", onOpen: () => { - // A reconnect starts a new provider stream. Never merge an old partial - // utterance into audio recognized by the replacement connection. + if (openedOnce) { + // The replacement stream cannot replay confirmed text from the old + // connection. Emit it as an interrupted turn, but discard its partial tail. + flushFinalizedTurn(); + } else { + openedOnce = true; + clearTurn(); + } finalizeRequested = false; finalizeFallbackFired = false; - clearTurn(); }, sendAudio: (audio, transport) => { transport.sendBinary(audio);