diff --git a/extensions/mistral/realtime-transcription-provider.test.ts b/extensions/mistral/realtime-transcription-provider.test.ts index 3d421c5bddd5..89a4cbeb8807 100644 --- a/extensions/mistral/realtime-transcription-provider.test.ts +++ b/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(); + }); }); diff --git a/extensions/mistral/realtime-transcription-provider.ts b/extensions/mistral/realtime-transcription-provider.ts index 23241838acfe..8fc7b5175e64 100644 --- a/extensions/mistral/realtime-transcription-provider.ts +++ b/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: } };