From ac55b201ba8b2b41ba0186621b1fa199e948e96c Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Fri, 26 Jun 2026 00:14:55 -0300 Subject: [PATCH] fix(sse): unwrap Qoder HTTP 200 SSE error envelope so fallback can trigger (#4850) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Integrated into release/v3.8.37 — cherry-picked defining commit onto release tip; CHANGELOG re-merged; tests green. --- open-sse/executors/qoder.ts | 112 +++++++++++++++++- .../unit/qoder-unwrap-error-envelope.test.ts | 95 +++++++++++++++ 2 files changed, 201 insertions(+), 6 deletions(-) create mode 100644 tests/unit/qoder-unwrap-error-envelope.test.ts diff --git a/open-sse/executors/qoder.ts b/open-sse/executors/qoder.ts index 8e251d94ec..f52faad70c 100644 --- a/open-sse/executors/qoder.ts +++ b/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 { + 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; + 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({ + 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, +}; diff --git a/tests/unit/qoder-unwrap-error-envelope.test.ts b/tests/unit/qoder-unwrap-error-envelope.test.ts new file mode 100644 index 0000000000..dff9fcd4f9 --- /dev/null +++ b/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; + } +});