From 80afb59907ae0a2dc3e017447efc691223d810a2 Mon Sep 17 00:00:00 2001 From: brimob-sowax Date: Thu, 13 Aug 2026 11:42:39 +0700 Subject: [PATCH] fix(qoder): detect billing blocks at stream start, return 403 for failover Peek the first SSE frame in wrapQoderSSE; if statusCodeValue != 200 and the body carries a billing signature (code 112/10605 or pricingUrl), return a synthetic 403 so chatCore marks the connection unavailable and triggers combo/account fallback instead of leaking the error text into chat. wrapQoderSSE becomes async; consumed peek bytes are re-processed in the stream start() seed loop so nothing is dropped. --- open-sse/executors/qoder.js | 98 +++++++++++++++++++-- tests/unit/qoder-billing.test.js | 147 +++++++++++++++++++++++++++++++ tests/unit/qoder.test.js | 14 +-- 3 files changed, 246 insertions(+), 13 deletions(-) create mode 100644 tests/unit/qoder-billing.test.js diff --git a/open-sse/executors/qoder.js b/open-sse/executors/qoder.js index f8691ef5..e52a00df 100644 --- a/open-sse/executors/qoder.js +++ b/open-sse/executors/qoder.js @@ -215,6 +215,52 @@ async function buildQoderRequestBody({ model, body, credentials, log, proxyOptio }; } +/** + * Check if a qoder error message indicates a billing/quota block. + * Signatures: code 112 (quota exhausted), code 10605 (queue throttle), pricingUrl field. + */ +function isBillingBlock(inner) { + if (!inner || typeof inner !== "string") return false; + const lowerMsg = inner.toLowerCase(); + // Match: {"code":"112",...}, {"code":"10605",...}, or pricingUrl field + return /\"code\"\s*:\s*\"(112|10605)\"/.test(inner) || lowerMsg.includes("pricingurl"); +} + +/** + * Peek the first SSE frame to detect billing errors before piping. + * Returns { isBilling, statusVal, message, consumed } — `consumed` is every + * byte read so far (including the peeked line) so the caller can re-process + * it and nothing is dropped from the stream. + */ +async function peekFirstQoderFrame(reader, decoder) { + let consumed = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) return { isBilling: false, consumed, upstreamDone: true }; + + consumed += decoder.decode(value, { stream: true }); + const nl = consumed.indexOf("\n"); + if (nl === -1) continue; // need a full line first + + const line = consumed.slice(0, nl).replace(/\r$/, "").trim(); + if (!line.startsWith("data:")) continue; + + const data = line.slice(5).trimStart(); + if (data === "[DONE]") return { isBilling: false, consumed }; + + let envelope; + try { envelope = JSON.parse(data); } catch { return { isBilling: false, consumed }; } + + const statusVal = typeof envelope.statusCodeValue === "number" ? envelope.statusCodeValue : 200; + const inner = typeof envelope.body === "string" ? envelope.body : ""; + + if (statusVal !== 200 && isBillingBlock(inner)) { + return { isBilling: true, statusVal, message: inner || `qoder billing block (${statusVal})` }; + } + return { isBilling: false, consumed }; + } +} + /** * Wrap the upstream's `{statusCodeValue, body}` SSE envelope into plain * OpenAI SSE chunks the rest of the chatCore pipeline understands. @@ -229,16 +275,34 @@ async function buildQoderRequestBody({ model, body, credentials, log, proxyOptio * [DONE]/error frame (agent keepalive). Non-streaming clients drain via * response.text() which hangs until the socket closes — so on terminal * events we cancel the upstream reader and close our stream immediately. + * + * NEW: Peek first frame to detect billing blocks (code 112/10605/pricingUrl). + * If detected, return 403 response so chatCore marks connection unavailable + * and triggers combo fallback instead of leaking error text into chat. */ -function wrapQoderSSE(response, model) { +async function wrapQoderSSE(response, model) { if (!response.ok || !response.body) return response; const decoder = new TextDecoder(); - const encoder = new TextEncoder(); - let buffer = ""; - let doneEmitted = false; const reader = response.body.getReader(); + // Peek first frame to detect billing block + const peek = await peekFirstQoderFrame(reader, decoder); + if (peek?.isBilling) { + // Billing block detected — return 403 so chatCore fails this connection + await reader.cancel().catch(() => {}); + return new Response( + JSON.stringify({ error: { message: peek.message, code: peek.statusVal } }), + { status: 403, headers: { "Content-Type": "application/json" } } + ); + } + + // Normal flow: re-process every byte the peek consumed, then continue. + let buffer = peek.consumed || ""; + const upstreamDrained = peek.upstreamDone === true; + const encoder = new TextEncoder(); + let doneEmitted = false; + // Process one already-extracted SSE line (no trailing newline). const processLine = (line, controller) => { const trimmed = line.replace(/\r$/, "").trim(); @@ -287,7 +351,28 @@ function wrapQoderSSE(response, model) { // enqueueing would never be re-invoked, hanging consumers like .text(). async start(controller) { try { - while (!doneEmitted) { + // Drain whatever the peek already pulled off the socket first. + let nlSeed; + while ((nlSeed = buffer.indexOf("\n")) !== -1) { + const line = buffer.slice(0, nlSeed); + buffer = buffer.slice(nlSeed + 1); + processLine(line, controller); + if (doneEmitted) { + await reader.cancel().catch(() => {}); + controller.close(); + return; + } + } + if (upstreamDrained) { + // Peek hit end-of-stream: flush any trailing partial line. + buffer += decoder.decode(); + if (buffer.length > 0) { + processLine(buffer, controller); + buffer = ""; + } + } + + while (!doneEmitted && !upstreamDrained) { const { done, value } = await reader.read(); if (done) { buffer += decoder.decode(); @@ -472,7 +557,7 @@ export class QoderExecutor extends BaseExecutor { return { response, url, headers, transformedBody: payload }; } - const wrapped = wrapQoderSSE(response, `qoder/${qoderKey}`); + const wrapped = await wrapQoderSSE(response, `qoder/${qoderKey}`); return { response: wrapped, url, headers, transformedBody: payload }; } @@ -496,4 +581,5 @@ export const __test__ = { normalizeMessages, wrapQoderSSE, buildQoderRequestBody, + isBillingBlock, }; diff --git a/tests/unit/qoder-billing.test.js b/tests/unit/qoder-billing.test.js new file mode 100644 index 00000000..f97a1293 --- /dev/null +++ b/tests/unit/qoder-billing.test.js @@ -0,0 +1,147 @@ +/** + * Unit tests for qoder billing error detection. + * + * Ensures that billing blocks (code 112, 10605, pricingUrl) are detected + * on the first SSE frame and returned as 403 responses so chatCore can + * mark the connection unavailable and trigger combo failover. + */ + +import { describe, it, expect } from "vitest"; +import { __test__ as qoderExecutorInternals } from "../../open-sse/executors/qoder.js"; + +describe("isBillingBlock", () => { + const { isBillingBlock } = qoderExecutorInternals; + + it("detects code 112 (quota exhausted)", () => { + const msg = '{"code":"112","message":"Quota exhausted","pricingUrl":"..."}'; + expect(isBillingBlock(msg)).toBe(true); + }); + + it("detects code 10605 (queue throttle)", () => { + const msg = '{"code":"10605","message":"Queue limit"}'; + expect(isBillingBlock(msg)).toBe(true); + }); + + it("detects pricingUrl field", () => { + const msg = '{"message":"Upgrade required","pricingUrl":"https://..."}'; + expect(isBillingBlock(msg)).toBe(true); + }); + + it("returns false for normal errors without billing markers", () => { + const msg = '{"code":"500","message":"Internal error"}'; + expect(isBillingBlock(msg)).toBe(false); + }); + + it("returns false for empty or non-string input", () => { + expect(isBillingBlock("")).toBe(false); + expect(isBillingBlock(null)).toBe(false); + expect(isBillingBlock(undefined)).toBe(false); + }); +}); + +describe("wrapQoderSSE billing detection", () => { + const { wrapQoderSSE } = qoderExecutorInternals; + + function makeResponse(lines, { status = 200 } = {}) { + const body = new ReadableStream({ + start(controller) { + const encoder = new TextEncoder(); + for (const line of lines) controller.enqueue(encoder.encode(line)); + controller.close(); + }, + }); + return new Response(body, { status }); + } + + it("returns 403 response when first frame is billing block (code 112)", async () => { + const billingEnv = JSON.stringify({ + statusCodeValue: 403, + body: '{"code":"112","message":"Quota exhausted","pricingUrl":"https://qoder.sh/pricing"}', + }); + const upstream = `data: ${billingEnv}\n\n`; + + const wrapped = await wrapQoderSSE(makeResponse([upstream]), "qoder/ultimate"); + + expect(wrapped.status).toBe(403); + expect(wrapped.ok).toBe(false); + const json = await wrapped.json(); + expect(json.error).toBeDefined(); + expect(json.error.message).toContain("112"); + }); + + it("returns 403 response when first frame is billing block (code 10605)", async () => { + const billingEnv = JSON.stringify({ + statusCodeValue: 429, + body: '{"code":"10605","message":"Queue limit"}', + }); + const upstream = `data: ${billingEnv}\n\n`; + + const wrapped = await wrapQoderSSE(makeResponse([upstream]), "qoder/ultimate"); + + expect(wrapped.status).toBe(403); + expect(wrapped.ok).toBe(false); + }); + + it("returns 403 response when first frame has pricingUrl", async () => { + const billingEnv = JSON.stringify({ + statusCodeValue: 402, + body: '{"message":"Payment required","pricingUrl":"https://..."}', + }); + const upstream = `data: ${billingEnv}\n\n`; + + const wrapped = await wrapQoderSSE(makeResponse([upstream]), "qoder/ultimate"); + + expect(wrapped.status).toBe(403); + }); + + it("passes through normal errors (non-billing) as wrapped SSE", async () => { + const errorEnv = JSON.stringify({ + statusCodeValue: 500, + body: "Internal server error", + }); + const upstream = `data: ${errorEnv}\n\n`; + + const wrapped = await wrapQoderSSE(makeResponse([upstream]), "qoder/ultimate"); + + // Normal error: still 200 response, error text in SSE body + expect(wrapped.status).toBe(200); + expect(wrapped.ok).toBe(true); + + const reader = wrapped.body.getReader(); + const decoder = new TextDecoder(); + let buf = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + buf += decoder.decode(value, { stream: true }); + } + buf += decoder.decode(); + + expect(buf).toContain("[qoder error 500"); + expect(buf).toContain("data: [DONE]"); + }); + + it("passes through successful responses unchanged", async () => { + const inner = JSON.stringify({ choices: [{ delta: { content: "hello" } }] }); + const successEnv = JSON.stringify({ statusCodeValue: 200, body: inner }); + const upstream = `data: ${successEnv}\n\n`; + + const wrapped = await wrapQoderSSE(makeResponse([upstream]), "qoder/ultimate"); + + expect(wrapped.status).toBe(200); + expect(wrapped.ok).toBe(true); + + const reader = wrapped.body.getReader(); + const decoder = new TextDecoder(); + let buf = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + buf += decoder.decode(value, { stream: true }); + } + buf += decoder.decode(); + + expect(buf).toContain(`data: ${inner}`); + expect(buf).toContain("data: [DONE]"); + }); +}); diff --git a/tests/unit/qoder.test.js b/tests/unit/qoder.test.js index fc5f20a1..d8fce1ae 100644 --- a/tests/unit/qoder.test.js +++ b/tests/unit/qoder.test.js @@ -401,7 +401,7 @@ describe("wrapQoderSSE", () => { it("forwards an OpenAI envelope chunk and emits [DONE] in flush", async () => { const inner = JSON.stringify({ choices: [{ delta: { content: "hi" } }] }); const upstream = `data: ${JSON.stringify({ statusCodeValue: 200, body: inner })}\n\n`; - const wrapped = wrapQoderSSE(makeResponse([upstream]), "qoder/auto"); + const wrapped = await wrapQoderSSE(makeResponse([upstream]), "qoder/auto"); const out = await drain(wrapped); expect(out).toContain(`data: ${inner}\n\n`); expect(out).toContain("data: [DONE]\n\n"); @@ -413,7 +413,7 @@ describe("wrapQoderSSE", () => { const inner = JSON.stringify({ choices: [{ delta: { content: "tail" } }], finish_reason: "stop" }); // Note: NO trailing \n on the final line. const upstream = `data: ${JSON.stringify({ statusCodeValue: 200, body: inner })}`; - const wrapped = wrapQoderSSE(makeResponse([upstream]), "qoder/auto"); + const wrapped = await wrapQoderSSE(makeResponse([upstream]), "qoder/auto"); const out = await drain(wrapped); expect(out).toContain(`data: ${inner}\n\n`); }); @@ -426,7 +426,7 @@ describe("wrapQoderSSE", () => { const errorEnv = JSON.stringify({ statusCodeValue: 500, body: "boom" }); const validInner = JSON.stringify({ choices: [{ delta: { content: "leak" } }] }); const validEnv = JSON.stringify({ statusCodeValue: 200, body: validInner }); - const wrapped = wrapQoderSSE( + const wrapped = await wrapQoderSSE( makeResponse([`data: ${errorEnv}\n\ndata: ${validEnv}\n\n`]), "qoder/auto", ); @@ -443,7 +443,7 @@ describe("wrapQoderSSE", () => { it("strips embedded newlines from inner body before forwarding", async () => { const innerWithNewlines = '{"choices":[{"delta":{"content":"a\nb"}}]}'; const env = JSON.stringify({ statusCodeValue: 200, body: innerWithNewlines }); - const wrapped = wrapQoderSSE(makeResponse([`data: ${env}\n\n`]), "qoder/auto"); + const wrapped = await wrapQoderSSE(makeResponse([`data: ${env}\n\n`]), "qoder/auto"); const out = await drain(wrapped); // The forwarded data: line should be a single event terminated by \n\n // and contain no internal \n other than the trailing pair. @@ -455,15 +455,15 @@ describe("wrapQoderSSE", () => { it("upstream error envelope produces an error chunk + [DONE]", async () => { const env = JSON.stringify({ statusCodeValue: 503, body: "service unavailable" }); - const wrapped = wrapQoderSSE(makeResponse([`data: ${env}\n\n`]), "qoder/lite"); + const wrapped = await wrapQoderSSE(makeResponse([`data: ${env}\n\n`]), "qoder/lite"); const out = await drain(wrapped); expect(out).toContain("[qoder error 503"); expect(out).toContain("data: [DONE]\n\n"); }); - it("non-ok responses are returned unchanged (no transform)", () => { + it("non-ok responses are returned unchanged (no transform)", async () => { const r = new Response("not ok", { status: 500 }); - const wrapped = wrapQoderSSE(r, "qoder/auto"); + const wrapped = await wrapQoderSSE(r, "qoder/auto"); expect(wrapped).toBe(r); }); });