fix(sse): unwrap Qoder HTTP 200 SSE error envelope so fallback can trigger (#4850)

Integrated into release/v3.8.37 — cherry-picked defining commit onto release tip; CHANGELOG re-merged; tests green.

open-sse/executors/qoder.ts

@@ -14,6 +14,102 @@ import { sanitizeQwenThinkingToolChoice } from "../services/qwenThinking.ts";
 import { buildCosyHeadersForValidation, resolveQoderJobToken } from "../services/qoderCli.ts";
 import { sanitizeErrorMessage } from "../utils/error.ts";
 
+function truncate(text: string, max: number): string {
+  if (text.length <= max) return text;
+  return `${text.slice(0, max)}…`;
+}
+
+/**
+ * Peek at the first SSE event from a Qoder response to detect upstream errors
+ * that Qoder wraps inside an HTTP 200 SSE envelope ({statusCodeValue, body}).
+ * Returns a proper HTTP error Response when found, so downstream fallback
+ * logic (combo routing, account fallback) can trigger. For success, re-creates
+ * the stream with the first chunk prepended so the body passes through
+ * transparently.
+ */
+async function unwrapQoderEnvelope(response: Response): Promise<Response> {
+  if (!response.ok || !response.body) {
+    return response;
+  }
+
+  const reader = response.body.getReader();
+  const decoder = new TextDecoder();
+
+  const { done, value } = await reader.read();
+  if (done) {
+    reader.cancel();
+    return new Response(
+      JSON.stringify({ error: { message: "[qoder] empty response", type: "provider_error" } }),
+      { status: 502, headers: { "Content-Type": "application/json" } }
+    );
+  }
+
+  const text = decoder.decode(value, { stream: true });
+
+  let errorStatus: number | null = null;
+  let errorMsg = "";
+  for (const line of text.split("\n")) {
+    const trimmed = line.trim();
+    if (!trimmed.startsWith("data:")) continue;
+    const jsonStr = trimmed.slice(5).trim();
+    if (jsonStr === "[DONE]") break;
+    try {
+      const envelope = JSON.parse(jsonStr) as Record<string, unknown>;
+      const statusVal =
+        typeof envelope.statusCodeValue === "number" ? envelope.statusCodeValue : 200;
+      if (statusVal !== 200) {
+        errorStatus = statusVal >= 400 ? statusVal : 502;
+        errorMsg =
+          typeof envelope.body === "string" ? envelope.body : `upstream status ${statusVal}`;
+      }
+    } catch {
+      // Malformed JSON — treat as non-error; downstream handling parses it.
+    }
+    break;
+  }
+
+  if (errorStatus) {
+    reader.cancel();
+    const errType =
+      errorStatus === 401 || errorStatus === 403 ? "authentication_error" : "provider_error";
+    return new Response(
+      JSON.stringify({
+        error: {
+          message: `[qoder error ${errorStatus}: ${sanitizeErrorMessage(truncate(errorMsg, 200))}]`,
+          type: errType,
+        },
+      }),
+      { status: errorStatus, headers: { "Content-Type": "application/json" } }
+    );
+  }
+
+  // Re-create the stream with the first chunk prepended so the success body
+  // passes through unchanged.
+  const restStream = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(value);
+    },
+    pull(controller) {
+      return reader.read().then(({ done, value }) => {
+        if (done) {
+          controller.close();
+          return;
+        }
+        controller.enqueue(value);
+      });
+    },
+    cancel() {
+      reader.cancel();
+    },
+  });
+
+  return new Response(restStream, {
+    status: response.status,
+    statusText: response.statusText,
+    headers: new Headers(response.headers),
+  });
+}
+
 function getAuthToken(credentials: ProviderCredentials): string {
   if (typeof credentials.apiKey === "string" && credentials.apiKey.trim()) {
     return credentials.apiKey.trim();
@@ -214,13 +310,12 @@ export class QoderExecutor extends BaseExecutor {
         };
       }
 
-      const newHeaders = new Headers(response.headers);
+      // Qoder wraps upstream errors inside an HTTP 200 SSE envelope
+      // ({statusCodeValue}). Peek at the first event to detect this and return
+      // a proper HTTP error so combo/account fallback logic can trigger.
+      const unwrapped = await unwrapQoderEnvelope(response);
       return {
-        response: new Response(response.body, {
-          status: response.status,
-          statusText: response.statusText,
-          headers: newHeaders,
-        }),
+        response: unwrapped,
         url: endpointUrl,
         headers,
         transformedBody: payload,
@@ -249,3 +344,8 @@ export class QoderExecutor extends BaseExecutor {
 }
 
 export default QoderExecutor;
+
+export const __test__ = {
+  unwrapQoderEnvelope,
+  truncate,
+};

tests/unit/qoder-unwrap-error-envelope.test.ts

@@ -0,0 +1,95 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+
+import { QoderExecutor, __test__ } from "../../open-sse/executors/qoder.ts";
+
+const { unwrapQoderEnvelope } = __test__;
+
+function sseResponse(body: string, status = 200): Response {
+  return new Response(body, {
+    status,
+    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.
+  const wrapped = sseResponse(
+    'data: {"statusCodeValue":429,"body":"rate limit exceeded"}\n\ndata: [DONE]\n\n'
+  );
+
+  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;
+  assert.match(payload.error.message, /qoder error 429/);
+  assert.match(payload.error.message, /rate limit exceeded/);
+});
+
+test("unwrapQoderEnvelope: maps a sub-400 embedded status to 502", async () => {
+  const wrapped = sseResponse('data: {"statusCodeValue":302,"body":"redirect"}\n\n');
+
+  const result = await unwrapQoderEnvelope(wrapped);
+
+  assert.equal(result.status, 502);
+});
+
+test("unwrapQoderEnvelope: classifies embedded 401 as an authentication_error", async () => {
+  const wrapped = sseResponse('data: {"statusCodeValue":401,"body":"invalid token"}\n\n');
+
+  const result = await unwrapQoderEnvelope(wrapped);
+
+  assert.equal(result.status, 401);
+  const payload = (await result.json()) as any;
+  assert.equal(payload.error.type, "authentication_error");
+});
+
+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'
+  );
+
+  const result = await unwrapQoderEnvelope(wrapped);
+
+  assert.equal(result.status, 200);
+  const body = await result.text();
+  // The first chunk must not be swallowed by the peek.
+  assert.match(body, /"content":"O"/);
+  assert.match(body, /"content":"K"/);
+  assert.match(body, /\[DONE\]/);
+});
+
+test("unwrapQoderEnvelope: an empty stream becomes a 502 error", async () => {
+  const result = await unwrapQoderEnvelope(sseResponse(""));
+  assert.equal(result.status, 502);
+});
+
+test("unwrapQoderEnvelope: a non-ok response is returned unchanged", async () => {
+  const errResp = sseResponse("nope", 500);
+  const result = await unwrapQoderEnvelope(errResp);
+  assert.equal(result, errResp);
+});
+
+test("QoderExecutor: stream call surfaces an embedded error envelope as a real HTTP status", async () => {
+  const executor = new QoderExecutor();
+  const originalFetch = globalThis.fetch;
+  globalThis.fetch = async () =>
+    sseResponse('data: {"statusCodeValue":429,"body":"quota exceeded"}\n\ndata: [DONE]\n\n');
+
+  try {
+    const { response } = await executor.execute({
+      model: "qoder-rome-30ba3b",
+      body: { messages: [{ role: "user", content: "hi" }] },
+      stream: true,
+      credentials: { apiKey: "pat_test" },
+    });
+
+    // Before the port this was a 200 — fallback could never trigger.
+    assert.equal(response.status, 429);
+    const payload = (await response.json()) as any;
+    assert.match(payload.error.message, /qoder error 429/);
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});