From 1ea87603c0c8a3395210ad161d919998a908629e Mon Sep 17 00:00:00 2001 From: Paco Cartones <253313177+pacocartones@users.noreply.github.com> Date: Thu, 17 Sep 2026 22:17:33 +0200 Subject: [PATCH] fix(qoder): unwrap split SSE error envelopes (#13838) Co-authored-by: Paco Cartones --- open-sse/executors/qoder.ts | 36 ++++++++++--- .../unit/qoder-unwrap-error-envelope.test.ts | 50 +++++++++++++++++-- 2 files changed, 75 insertions(+), 11 deletions(-) diff --git a/open-sse/executors/qoder.ts b/open-sse/executors/qoder.ts index 1b19289a1a..42af382b53 100644 --- a/open-sse/executors/qoder.ts +++ b/open-sse/executors/qoder.ts @@ -70,9 +70,29 @@ async function unwrapQoderEnvelope(response: Response): Promise { const reader = response.body.getReader(); const decoder = new TextDecoder(); + const peekedChunks: Uint8Array[] = []; + let peekedText = ""; + let peekedBytes = 0; + let reachedEnd = false; - const { done, value } = await reader.read(); - if (done) { + while (peekedBytes < 64 * 1024) { + const { done, value } = await reader.read(); + if (done) { + reachedEnd = true; + peekedText += decoder.decode(); + break; + } + + peekedChunks.push(value); + peekedBytes += value.byteLength; + peekedText += decoder.decode(value, { stream: true }); + + if (peekedText.includes("\n\n") || peekedText.includes("\r\n\r\n")) { + break; + } + } + + if (peekedChunks.length === 0 && reachedEnd) { reader.cancel(); return new Response( JSON.stringify({ error: { message: "[qoder] empty response", type: "provider_error" } }), @@ -80,11 +100,9 @@ async function unwrapQoderEnvelope(response: Response): Promise { ); } - const text = decoder.decode(value, { stream: true }); - let errorStatus: number | null = null; let errorMsg = ""; - for (const line of text.split("\n")) { + for (const line of peekedText.split("\n")) { const trimmed = line.trim(); if (!trimmed.startsWith("data:")) continue; const jsonStr = trimmed.slice(5).trim(); @@ -119,11 +137,13 @@ async function unwrapQoderEnvelope(response: Response): Promise { ); } - // Re-create the stream with the first chunk prepended so the success body - // passes through unchanged. + // Re-create the stream with every peeked chunk prepended so the success body + // passes through byte-for-byte unchanged. const restStream = new ReadableStream({ start(controller) { - controller.enqueue(value); + for (const chunk of peekedChunks) { + controller.enqueue(chunk); + } }, pull(controller) { return reader.read().then(({ done, value }) => { diff --git a/tests/unit/qoder-unwrap-error-envelope.test.ts b/tests/unit/qoder-unwrap-error-envelope.test.ts index dff9fcd4f9..abede69eb4 100644 --- a/tests/unit/qoder-unwrap-error-envelope.test.ts +++ b/tests/unit/qoder-unwrap-error-envelope.test.ts @@ -5,6 +5,10 @@ import { QoderExecutor, __test__ } from "../../open-sse/executors/qoder.ts"; const { unwrapQoderEnvelope } = __test__; +type ErrorPayload = { + error: { message: string; type?: string }; +}; + function sseResponse(body: string, status = 200): Response { return new Response(body, { status, @@ -12,6 +16,18 @@ function sseResponse(body: string, status = 200): Response { }); } +function chunkedSseResponse(chunks: Uint8Array[]): Response { + return new Response( + new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(chunk); + controller.close(); + }, + }), + { headers: { "Content-Type": "text/event-stream" } } + ); +} + test("unwrapQoderEnvelope: surfaces an embedded non-200 statusCodeValue as a real HTTP error", async () => { // Qoder wraps an upstream 429 inside a 200 SSE envelope. Before the fix this // passed straight through as a 200, so combo/account fallback never fired. @@ -22,7 +38,7 @@ test("unwrapQoderEnvelope: surfaces an embedded non-200 statusCodeValue as a rea const result = await unwrapQoderEnvelope(wrapped); assert.equal(result.status, 429, "embedded 429 must become a real HTTP 429"); - const payload = (await result.json()) as any; + const payload = (await result.json()) as ErrorPayload; assert.match(payload.error.message, /qoder error 429/); assert.match(payload.error.message, /rate limit exceeded/); }); @@ -41,10 +57,26 @@ test("unwrapQoderEnvelope: classifies embedded 401 as an authentication_error", const result = await unwrapQoderEnvelope(wrapped); assert.equal(result.status, 401); - const payload = (await result.json()) as any; + const payload = (await result.json()) as ErrorPayload; assert.equal(payload.error.type, "authentication_error"); }); +test("unwrapQoderEnvelope: detects an error event split at every byte boundary", async () => { + const encoded = new TextEncoder().encode( + 'data: {"statusCodeValue":429,"body":"quota 🚫 exceeded"}\n\ndata: [DONE]\n\n' + ); + + for (let offset = 1; offset < encoded.length; offset += 1) { + const result = await unwrapQoderEnvelope( + chunkedSseResponse([encoded.slice(0, offset), encoded.slice(offset)]) + ); + + assert.equal(result.status, 429, `split at byte ${offset}`); + const payload = (await result.json()) as ErrorPayload; + assert.match(payload.error.message, /quota 🚫 exceeded/); + } +}); + test("unwrapQoderEnvelope: passes a successful stream through with the first chunk intact", async () => { const wrapped = sseResponse( 'data: {"choices":[{"delta":{"content":"O"}}]}\n\ndata: {"choices":[{"delta":{"content":"K"}}]}\n\ndata: [DONE]\n\n' @@ -60,6 +92,18 @@ test("unwrapQoderEnvelope: passes a successful stream through with the first chu assert.match(body, /\[DONE\]/); }); +test("unwrapQoderEnvelope: preserves every byte while peeking across multiple chunks", async () => { + const encoded = new TextEncoder().encode( + 'data: {"choices":[{"delta":{"content":"O🚀K"}}]}\n\ndata: [DONE]\n\n' + ); + const result = await unwrapQoderEnvelope( + chunkedSseResponse([encoded.slice(0, 11), encoded.slice(11, 44), encoded.slice(44)]) + ); + + assert.equal(result.status, 200); + assert.deepEqual(new Uint8Array(await result.arrayBuffer()), encoded); +}); + test("unwrapQoderEnvelope: an empty stream becomes a 502 error", async () => { const result = await unwrapQoderEnvelope(sseResponse("")); assert.equal(result.status, 502); @@ -87,7 +131,7 @@ test("QoderExecutor: stream call surfaces an embedded error envelope as a real H // Before the port this was a 200 — fallback could never trigger. assert.equal(response.status, 429); - const payload = (await response.json()) as any; + const payload = (await response.json()) as ErrorPayload; assert.match(payload.error.message, /qoder error 429/); } finally { globalThis.fetch = originalFetch;