Merge pull request #117190 from openclaw/fix/mistral-realtime-transcript-bound-v2

* commit '834af948f4d9086b8c0d9a950150544d487dfe3c':
  fix(mistral): keep terminal error branch lint-clean
  fix(mistral): make realtime errors terminal
  test(mistral): cover realtime transcript overflow
  fix(mistral): bound realtime transcript accumulation

extensions/mistral/realtime-transcription-provider.test.ts

@@ -337,4 +337,106 @@ describe("buildMistralRealtimeTranscriptionProvider", () => {
 
     expect(onPartial.mock.calls.map(([text]) => text)).toEqual(partials);
   });
+
+  it("tracks the in-progress transcript limit as aggregate UTF-8 bytes", async () => {
+    const exactUtf8Limit = "🙂".repeat((256 * 1024) / 4);
+    const splitSurrogatePrefix = "x".repeat(256 * 1024 - 4);
+    const splitSurrogateTranscript = `${splitSurrogatePrefix}🙂`;
+    const baseUrl = await createRealtimeServer(() => {}, [
+      { type: "transcription.text.delta", text: exactUtf8Limit },
+      { type: "transcription.segment", text: "first segment", start: 0, end: 1 },
+      { type: "transcription.text.delta", text: `${splitSurrogatePrefix}\ud83d` },
+      { type: "transcription.text.delta", text: "\ude42" },
+      { type: "transcription.done" },
+    ]);
+    const onError = vi.fn();
+    const onTranscript = vi.fn();
+    const session = buildMistralRealtimeTranscriptionProvider().createSession({
+      providerConfig: { apiKey: "fixture-value", baseUrl },
+      onError,
+      onTranscript,
+    });
+
+    await session.connect();
+    await vi.waitFor(() => {
+      expect(onTranscript.mock.calls.map(([text]) => text)).toEqual([
+        "first segment",
+        splitSurrogateTranscript,
+      ]);
+      expect(session.isConnected()).toBe(false);
+    });
+
+    expect(onError).not.toHaveBeenCalled();
+  });
+
+  it("fails once and ignores late terminal events after 10,000 runaway deltas", async () => {
+    const baseUrl = await createRealtimeServer(() => {}, [
+      ...Array.from({ length: 10_000 }, () => ({
+        type: "transcription.text.delta",
+        text: "x".repeat(32),
+      })),
+      { type: "transcription.segment", text: "late segment", start: 0, end: 1 },
+      { type: "transcription.done", text: "late done" },
+    ]);
+    const onError = vi.fn();
+    const onTranscript = vi.fn();
+    let lastPartialLength = 0;
+    let partialCalls = 0;
+    const session = buildMistralRealtimeTranscriptionProvider().createSession({
+      providerConfig: { apiKey: "fixture-value", baseUrl },
+      onError,
+      onPartial: (partial) => {
+        lastPartialLength = partial.length;
+        partialCalls += 1;
+      },
+      onTranscript,
+    });
+
+    await session.connect();
+    await vi.waitFor(() => {
+      expect(onError).toHaveBeenCalledExactlyOnceWith(
+        expect.objectContaining({
+          message:
+            "Mistral realtime transcription exceeded the 256 KiB in-progress transcript limit",
+        }),
+      );
+      expect(session.isConnected()).toBe(false);
+    });
+
+    expect(partialCalls).toBe(8_192);
+    expect(lastPartialLength).toBe(256 * 1024);
+    expect(onTranscript).not.toHaveBeenCalled();
+  });
+
+  it("makes a ready-state provider error terminal and ignores late events", async () => {
+    const baseUrl = await createRealtimeServer(() => {}, [
+      { type: "transcription.text.delta", text: "draft" },
+      { type: "error", error: { message: "provider failed" } },
+      { type: "transcription.text.delta", text: "x".repeat(256 * 1024 + 1) },
+      { type: "transcription.segment", text: "late segment", start: 0, end: 1 },
+      { type: "transcription.done", text: "late done" },
+    ]);
+    const onError = vi.fn(() => {
+      throw new Error("observer failed");
+    });
+    const onPartial = vi.fn();
+    const onTranscript = vi.fn();
+    const session = buildMistralRealtimeTranscriptionProvider().createSession({
+      providerConfig: { apiKey: "fixture-value", baseUrl },
+      onError,
+      onPartial,
+      onTranscript,
+    });
+
+    await session.connect();
+    await vi.waitFor(() => {
+      expect(onError).toHaveBeenCalledExactlyOnceWith(
+        expect.objectContaining({ message: "provider failed" }),
+      );
+      expect(session.isConnected()).toBe(false);
+    });
+
+    expect(onPartial).toHaveBeenCalledExactlyOnceWith("draft");
+    expect(onTranscript).not.toHaveBeenCalled();
+  });
 });

extensions/mistral/realtime-transcription-provider.ts

@@ -63,6 +63,9 @@ const MISTRAL_REALTIME_MAX_RECONNECT_ATTEMPTS = 5;
 const MISTRAL_REALTIME_RECONNECT_DELAY_MS = 1000;
 const MISTRAL_REALTIME_MAX_QUEUED_BYTES = 2 * 1024 * 1024;
 const MISTRAL_REALTIME_SPEECH_CONTENT = /[\p{L}\p{N}]/u;
+const MISTRAL_REALTIME_MAX_PARTIAL_TRANSCRIPT_BYTES = 256 * 1024;
+const MISTRAL_REALTIME_PARTIAL_TRANSCRIPT_OVERFLOW_MESSAGE =
+  "Mistral realtime transcription exceeded the 256 KiB in-progress transcript limit";
 
 function readNestedMistralConfig(rawConfig: RealtimeTranscriptionProviderConfig) {
   const raw = readRecord(rawConfig);
@@ -162,11 +165,31 @@ function readErrorDetail(event: MistralRealtimeTranscriptionEvent): string {
   return "Mistral realtime transcription error";
 }
 
+function measureTranscriptDeltaBytes(partialText: string, delta: string): number {
+  const previousCodeUnit = partialText.charCodeAt(partialText.length - 1);
+  const nextCodeUnit = delta.charCodeAt(0);
+  const completesSplitSurrogatePair =
+    previousCodeUnit >= 0xd800 &&
+    previousCodeUnit <= 0xdbff &&
+    nextCodeUnit >= 0xdc00 &&
+    nextCodeUnit <= 0xdfff;
+  // Separate UTF-8 measurements encode split surrogates as two replacement
+  // characters (six bytes); the combined transcript encodes one four-byte code point.
+  return Buffer.byteLength(delta, "utf8") - (completesSplitSurrogatePair ? 2 : 0);
+}
+
 function createMistralRealtimeTranscriptionSession(
   config: MistralRealtimeTranscriptionSessionConfig,
 ): RealtimeTranscriptionSession {
   let partialText = "";
+  let partialBytes = 0;
   let hasFinalSegment = false;
+  let terminal = false;
+
+  const clearPartial = () => {
+    partialText = "";
+    partialBytes = 0;
+  };
 
   const emitFinalTranscript = (text: string, source: "segment" | "terminal" | "pending") => {
     if (!text.trim() || (source === "pending" && !MISTRAL_REALTIME_SPEECH_CONTENT.test(text))) {
@@ -176,10 +199,28 @@ function createMistralRealtimeTranscriptionSession(
     config.onTranscript?.(text);
   };
 
+  const failTerminal = (error: Error, transport: RealtimeTranscriptionWebSocketTransport) => {
+    if (terminal) {
+      return;
+    }
+    terminal = true;
+    clearPartial();
+    transport.closeNow();
+    try {
+      config.onError?.(error);
+    } catch {
+      // The terminal provider error already owns the outcome. Do not let an
+      // observer exception re-enter shared error dispatch and emit it twice.
+    }
+  };
+
   const handleEvent = (
     event: MistralRealtimeTranscriptionEvent,
     transport: RealtimeTranscriptionWebSocketTransport,
   ) => {
+    if (terminal) {
+      return;
+    }
     if (event.type === "session.created") {
       transport.sendJson({
         type: "session.update",
@@ -200,30 +241,42 @@ function createMistralRealtimeTranscriptionSession(
     switch (event.type) {
       case "transcription.text.delta":
         if (event.text) {
+          const deltaBytes = measureTranscriptDeltaBytes(partialText, event.text);
+          if (deltaBytes > MISTRAL_REALTIME_MAX_PARTIAL_TRANSCRIPT_BYTES - partialBytes) {
+            failTerminal(
+              new Error(MISTRAL_REALTIME_PARTIAL_TRANSCRIPT_OVERFLOW_MESSAGE),
+              transport,
+            );
+            return;
+          }
           partialText += event.text;
+          partialBytes += deltaBytes;
           config.onPartial?.(partialText);
         }
         return;
       case "transcription.segment":
         if (event.text?.trim()) {
           emitFinalTranscript(event.text, "segment");
-          partialText = "";
+          clearPartial();
         }
         return;
       case "transcription.done": {
+        terminal = true;
         // Final segments already own completed speech; only later buffered
         // speech deltas are new. Punctuation only completes an earlier final.
         const source = hasFinalSegment ? "pending" : "terminal";
         const terminalText =
           source === "pending" ? partialText : event.text?.trim() ? event.text : partialText;
-        emitFinalTranscript(terminalText, source);
-        partialText = "";
-        transport.closeNow();
+        clearPartial();
+        try {
+          emitFinalTranscript(terminalText, source);
+        } finally {
+          transport.closeNow();
+        }
         return;
       }
       case "error":
-        config.onError?.(new Error(readErrorDetail(event)));
-
+        failTerminal(new Error(readErrorDetail(event)), transport);
       default:
     }
   };