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

@@ -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,
 };

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]");
+  });
+});

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);
   });
 });