fix(qoder): unwrap split SSE error envelopes (#13838)

Co-authored-by: Paco Cartones <pacocartones@users.noreply.github.com>

open-sse/executors/qoder.ts

@@ -70,9 +70,29 @@ async function unwrapQoderEnvelope(response: Response): Promise<Response> {
 
   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<Response> {
     );
   }
 
-  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<Response> {
     );
   }
 
-  // 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<Uint8Array>({
     start(controller) {
-      controller.enqueue(value);
+      for (const chunk of peekedChunks) {
+        controller.enqueue(chunk);
+      }
     },
     pull(controller) {
       return reader.read().then(({ done, value }) => {

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;