fix(deepgram): preserve endpointed voice turns (#117663)

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<Buffer
   }
 }
 
+function convertWavToMulaw8k(wav: Buffer): Buffer {
+  const result = spawnSync(
+    "ffmpeg",
+    [
+      "-hide_banner",
+      "-loglevel",
+      "error",
+      "-i",
+      "pipe:0",
+      "-f",
+      "mulaw",
+      "-ar",
+      "8000",
+      "-ac",
+      "1",
+      "pipe:1",
+    ],
+    { input: wav, maxBuffer: 16 * 1024 * 1024 },
+  );
+  if (result.error) {
+    throw result.error;
+  }
+  if (result.status !== 0) {
+    throw new Error(`ffmpeg failed: ${result.stderr.toString("utf8").trim()}`);
+  }
+  return result.stdout;
+}
+
 describeLive("deepgram live", () => {
   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);
 });

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<void>) | undefined;
 
 async function createDeepgramRealtimeServer(params: {
   onRequest: (url: URL, headers: Record<string, string | string[] | undefined>) => 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<WebSocket>();
 
   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<string, unknown> {
+  const bytes = Buffer.isBuffer(data)
+    ? data
+    : Array.isArray(data)
+      ? Buffer.concat(data)
+      : Buffer.from(data);
+  return JSON.parse(bytes.toString("utf8")) as Record<string, unknown>;
+}
+
 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<void>((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();
+  });
 });

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<typeof setTimeout> | 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),
   });
 }