diff --git a/extensions/deepgram/audio.live.test.ts b/extensions/deepgram/audio.live.test.ts index 66a209f2af70..c6beef41c300 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: "lifemovesprettyfast", }); }, 90_000); }); diff --git a/extensions/deepgram/realtime-transcription-provider.test.ts b/extensions/deepgram/realtime-transcription-provider.test.ts index c194dc28d7ad..c8ef0d2c75a7 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"; @@ -11,9 +12,10 @@ 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 }); + const wss = new WebSocketServer({ noServer: true, maxPayload: 1024 * 1024 }); const clients = new Set(); server.on("upgrade", (request, socket, head) => { @@ -21,6 +23,7 @@ async function createDeepgramRealtimeServer(params: { wss.handleUpgrade(request, socket, head, (ws) => { clients.add(ws); ws.on("close", () => clients.delete(ws)); + params.onConnection?.(ws); }); }); @@ -42,8 +45,38 @@ 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, + }), + ); +} + +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(); await cleanup?.(); cleanup = undefined; vi.unstubAllEnvs(); @@ -139,6 +172,299 @@ 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.has("utterance_end_ms")).toBe(false); 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("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, + 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: "good", isFinal: true }); + sendResult(ws, { text: "bye" }); + ws.on("message", (data) => { + if (parseClientMessage(data).type === "Finalize") { + sendResult(ws, { + text: "bye", + 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("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 (parseClientMessage(data).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, + onConnection: (ws) => { + sendResult(ws, { text: "still", isFinal: true }); + ws.send(JSON.stringify({ type: "UtteranceEnd" })); + sendResult(ws, { text: "speaking", isFinal: true, speechFinal: 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("still speaking")); + session.close(); + + expect(onTranscript).toHaveBeenCalledTimes(1); + }); + + 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({ + onRequest: () => undefined, + onConnection: (ws) => { + connectionCount += 1; + if (connectionCount === 1) { + sendResult(ws, { text: "old" }); + 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("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, + 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 68239d2d62f0..3a2fb38195b5 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_FINALIZE_FALLBACK_MS = DEEPGRAM_REALTIME_CLOSE_TIMEOUT_MS - 100; function readNestedDeepgramConfig(rawConfig: RealtimeTranscriptionProviderConfig) { const raw = readRecord(rawConfig); @@ -169,36 +173,110 @@ function readTranscriptText(event: DeepgramRealtimeTranscriptionEvent): string | function createDeepgramRealtimeTranscriptionSession( config: DeepgramRealtimeTranscriptionSessionConfig, ): RealtimeTranscriptionSession { - let lastTranscript: string | undefined; let speechStarted = false; + let finalizedTranscript = ""; + let pendingPartial = ""; + let finalizeRequested = false; + let finalizeFallbackFired = false; + let finalizeFallbackTimer: ReturnType | undefined; + let openedOnce = false; - const emitTranscript = (text: string) => { - if (text === lastTranscript) { - return; + 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; } - lastTranscript = text; - config.onTranscript?.(text); }; - const handleEvent = (event: DeepgramRealtimeTranscriptionEvent) => { + const clearTurn = () => { + clearFinalizeFallback(); + 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 = () => { + const full = joinTranscript(finalizedTranscript, pendingPartial); + clearTurn(); + if (full) { + config.onTranscript?.(full); + } + }; + + const flushFinalizedTurn = () => { + const full = collapseWhitespace(finalizedTranscript); + clearTurn(); + if (full) { + config.onTranscript?.(full); + } + }; + + const handleEvent = ( + event: DeepgramRealtimeTranscriptionEvent, + transport: RealtimeTranscriptionWebSocketTransport, + ) => { switch (event.type) { case "Results": { - const text = readTranscriptText(event); - if (!text) { + if (finalizeFallbackFired) { return; } - if (!speechStarted) { + const text = readTranscriptText(event); + if (text && !speechStarted) { speechStarted = true; config.onSpeechStart?.(); } - if (event.is_final || event.speech_final) { - emitTranscript(text); - if (event.speech_final) { - speechStarted = false; + if (event.speech_final || event.from_finalize) { + const nextFinalized = text + ? joinTranscript(finalizedTranscript, text) + : finalizedTranscript; + if (!updateTurn(nextFinalized, "", transport)) { + return; } + flushTurn(); return; } - config.onPartial?.(text); + if (!text) { + return; + } + if (event.is_final) { + const nextFinalized = joinTranscript(finalizedTranscript, text); + if (!updateTurn(nextFinalized, "", transport)) { + return; + } + config.onPartial?.(nextFinalized); + } else { + if (!updateTurn(finalizedTranscript, text, transport)) { + return; + } + config.onPartial?.(joinTranscript(finalizedTranscript, text)); + } return; } case "SpeechStarted": @@ -228,13 +306,46 @@ function createDeepgramRealtimeTranscriptionSession( connectClosedBeforeReadyMessage: "Deepgram realtime transcription connection closed before ready", reconnectLimitMessage: "Deepgram realtime transcription reconnect limit reached", + onOpen: () => { + 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; + }, 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: handleEvent, + onMessage: (event, transport) => handleEvent(event, transport), }); }