merge(mistral): adapt transcript bound to final ownership

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

@@ -0,0 +1,178 @@
+// Mistral lifecycle tests cover bounded transcript accumulation and terminal events.
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { buildMistralRealtimeTranscriptionProvider } from "./realtime-transcription-provider.js";
+
+const { FakeWebSocket } = vi.hoisted(() => {
+  type Listener = (...args: unknown[]) => void;
+
+  class MockWebSocket {
+    static readonly OPEN = 1;
+    static readonly CLOSED = 3;
+    static instances: MockWebSocket[] = [];
+
+    binaryType = "nodebuffer";
+    closeCalls = 0;
+    readonly listeners = new Map<string, Listener[]>();
+    readyState = 0;
+    sent: string[] = [];
+
+    constructor() {
+      MockWebSocket.instances.push(this);
+    }
+
+    on(event: string, listener: Listener): this {
+      const listeners = this.listeners.get(event) ?? [];
+      listeners.push(listener);
+      this.listeners.set(event, listeners);
+      return this;
+    }
+
+    emit(event: string, ...args: unknown[]): void {
+      for (const listener of this.listeners.get(event) ?? []) {
+        listener(...args);
+      }
+    }
+
+    send(payload: string): void {
+      this.sent.push(payload);
+    }
+
+    close(code?: number, reason?: string): void {
+      this.closeCalls += 1;
+      if (this.readyState === MockWebSocket.CLOSED) {
+        return;
+      }
+      this.readyState = MockWebSocket.CLOSED;
+      this.emit("close", code ?? 1000, Buffer.from(reason ?? ""));
+    }
+  }
+
+  return { FakeWebSocket: MockWebSocket };
+});
+
+vi.mock("ws", () => ({
+  default: FakeWebSocket,
+}));
+
+type FakeWebSocketInstance = InstanceType<typeof FakeWebSocket>;
+
+function emitEvent(socket: FakeWebSocketInstance, event: unknown): void {
+  socket.emit("message", Buffer.from(JSON.stringify(event)));
+}
+
+async function connectSession(callbacks: {
+  onError?: (error: Error) => void;
+  onPartial?: (partial: string) => void;
+  onTranscript?: (transcript: string) => void;
+}) {
+  const session = buildMistralRealtimeTranscriptionProvider().createSession({
+    providerConfig: {
+      apiKey: "fixture-value",
+      baseUrl: "ws://mistral.test",
+    },
+    ...callbacks,
+  });
+  const connecting = session.connect();
+  let socket: FakeWebSocketInstance | undefined;
+  await vi.waitFor(() => {
+    socket = FakeWebSocket.instances[0];
+    if (!socket) {
+      throw new Error("expected session to create a websocket");
+    }
+  });
+  if (!socket) {
+    throw new Error("expected session to create a websocket");
+  }
+  socket.readyState = FakeWebSocket.OPEN;
+  socket.emit("open");
+  emitEvent(socket, { type: "session.created" });
+  await connecting;
+  return { session, socket };
+}
+
+describe("Mistral realtime transcription lifecycle", () => {
+  beforeEach(() => {
+    FakeWebSocket.instances = [];
+  });
+
+  it("preserves partial, segment, and done transcript semantics", async () => {
+    const errors: string[] = [];
+    const partials: string[] = [];
+    const transcripts: string[] = [];
+    const { session, socket } = await connectSession({
+      onError: (error) => errors.push(error.message),
+      onPartial: (partial) => partials.push(partial),
+      onTranscript: (transcript) => transcripts.push(transcript),
+    });
+
+    emitEvent(socket, { type: "transcription.text.delta", text: "hel" });
+    emitEvent(socket, { type: "transcription.text.delta", text: "lo" });
+    emitEvent(socket, { type: "transcription.segment", text: "hello final" });
+    emitEvent(socket, { type: "transcription.text.delta", text: "next" });
+    emitEvent(socket, {
+      type: "transcription.done",
+      text: "provider full transcript remains ignored",
+    });
+
+    expect(partials).toEqual(["hel", "hello", "next"]);
+    expect(transcripts).toEqual(["hello final", "next"]);
+    expect(errors).toEqual([]);
+    expect(socket.closeCalls).toBe(1);
+    expect(session.isConnected()).toBe(false);
+  });
+
+  it("tracks the in-progress transcript limit as aggregate UTF-8 bytes", async () => {
+    const errors: string[] = [];
+    const transcripts: string[] = [];
+    const { socket } = await connectSession({
+      onError: (error) => errors.push(error.message),
+      onTranscript: (transcript) => transcripts.push(transcript),
+    });
+    const exactUtf8Limit = "🙂".repeat((256 * 1024) / 4);
+    const splitSurrogatePrefix = "x".repeat(256 * 1024 - 4);
+    const splitSurrogateTranscript = `${splitSurrogatePrefix}🙂`;
+
+    emitEvent(socket, { type: "transcription.text.delta", text: exactUtf8Limit });
+    emitEvent(socket, { type: "transcription.segment", text: "first segment" });
+    emitEvent(socket, {
+      type: "transcription.text.delta",
+      text: `${splitSurrogatePrefix}\ud83d`,
+    });
+    emitEvent(socket, { type: "transcription.text.delta", text: "\ude42" });
+    emitEvent(socket, { type: "transcription.done" });
+
+    expect(errors).toEqual([]);
+    expect(transcripts).toEqual(["first segment", splitSurrogateTranscript]);
+    expect(socket.closeCalls).toBe(1);
+  });
+
+  it("fails once and ignores late terminal events after 10,000 runaway deltas", async () => {
+    const errors: string[] = [];
+    const transcripts: string[] = [];
+    let lastPartialLength = 0;
+    let partialCalls = 0;
+    const { session, socket } = await connectSession({
+      onError: (error) => errors.push(error.message),
+      onPartial: (partial) => {
+        lastPartialLength = partial.length;
+        partialCalls += 1;
+      },
+      onTranscript: (transcript) => transcripts.push(transcript),
+    });
+
+    for (let index = 0; index < 10_000; index += 1) {
+      emitEvent(socket, { type: "transcription.text.delta", text: "x".repeat(32) });
+    }
+    emitEvent(socket, { type: "transcription.segment", text: "late segment" });
+    emitEvent(socket, { type: "transcription.done", text: "late done" });
+
+    expect(errors).toEqual([
+      "Mistral realtime transcription exceeded the 256 KiB in-progress transcript limit",
+    ]);
+    expect(partialCalls).toBe(8_192);
+    expect(lastPartialLength).toBe(256 * 1024);
+    expect(transcripts).toEqual([]);
+    expect(socket.closeCalls).toBe(1);
+    expect(session.isConnected()).toBe(false);
+  });
+});

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 failPartialOverflow = (transport: RealtimeTranscriptionWebSocketTransport) => {
+    if (terminal) {
+      return;
+    }
+    terminal = true;
+    clearPartial();
+    transport.closeNow();
+    try {
+      config.onError?.(new Error(MISTRAL_REALTIME_PARTIAL_TRANSCRIPT_OVERFLOW_MESSAGE));
+    } 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,25 +241,35 @@ 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) {
+            failPartialOverflow(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":