From 3f457cc77bbc841b1b6a2b41b5662cc0244d41a2 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Fri, 10 Jul 2026 19:39:23 -0300 Subject: [PATCH] fix(codex): surface capacity errors embedded in 200-OK SSE streams (#6710) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(codex): surface capacity errors embedded in 200-OK SSE streams Codex sometimes answers with HTTP 200 and a text/event-stream body whose payload carries a transient error mid-stream (e.g. "Selected model is at capacity...", server_is_overloaded, service_unavailable_error). Because the outer HTTP status was 200, this looked like a successful response to every caller — no retry, no circuit breaker, and no combo/account fallback ever engaged, so a healthy account sat idle while the request silently failed or truncated. Add peekCodexSseTransientError() to open-sse/executors/codex.ts: it peeks the first bytes of a text/event-stream Codex response, pattern-matches the known transient-error signatures, and converts a match into a real 503 Response via errorResponse() (Hard Rule #12 — sanitized, never raw upstream text). A 503 is already a recognized provider-failure status in accountFallback.ts, so combo routing and connection cooldown pick it up automatically. When no error signature is found, the peeked prefix is prepended back onto the remaining upstream body so the passthrough stays byte-identical to the unmodified response. Regression guard: tests/unit/codex-sse-capacity-fallback.test.ts — a model-at-capacity payload and a server_is_overloaded/service_unavailable_error payload both convert to 503; a normal single-chunk SSE stream and one split across multiple network chunks both reassemble byte-for-byte unchanged. Inspired-by: https://github.com/decolua/9router/pull/2452 (sub-bug #3 only — OmniRoute already covers PR #2452's other two sub-bugs: service_tier "fast" normalization and reasoning_effort "max" normalization). Co-authored-by: ryanngit <74137224+ryanngit@users.noreply.github.com> * chore(6710): re-sync onto release tip; CHANGELOG → changelog.d fragment (fragments-first) --------- Co-authored-by: ryanngit <74137224+ryanngit@users.noreply.github.com> --- .../6710-codex-200-sse-capacity-error.md | 1 + open-sse/executors/codex.ts | 162 +++++++++++++++++- .../unit/codex-sse-capacity-fallback.test.ts | 157 +++++++++++++++++ 3 files changed, 319 insertions(+), 1 deletion(-) create mode 100644 changelog.d/fixes/6710-codex-200-sse-capacity-error.md create mode 100644 tests/unit/codex-sse-capacity-fallback.test.ts diff --git a/changelog.d/fixes/6710-codex-200-sse-capacity-error.md b/changelog.d/fixes/6710-codex-200-sse-capacity-error.md new file mode 100644 index 0000000000..e84653d7bd --- /dev/null +++ b/changelog.d/fixes/6710-codex-200-sse-capacity-error.md @@ -0,0 +1 @@ +- **fix(codex):** detect "model at capacity"/overloaded errors embedded in a 200-OK SSE stream and surface them as a real error so account fallback rotates, instead of passing them through as a successful response. (thanks @ryanngit) diff --git a/open-sse/executors/codex.ts b/open-sse/executors/codex.ts index 2fd66b2343..6712c3eb4c 100644 --- a/open-sse/executors/codex.ts +++ b/open-sse/executors/codex.ts @@ -17,7 +17,7 @@ import { CODEX_CHAT_DEFAULT_INSTRUCTIONS, CODEX_DEFAULT_INSTRUCTIONS, } from "../config/codexInstructions.ts"; -import { PROVIDERS } from "../config/constants.ts"; +import { HTTP_STATUS, PROVIDERS } from "../config/constants.ts"; import { getCodexClientVersion, getCodexUserAgent, @@ -34,6 +34,7 @@ import { sanitizeResponsesInputItems } from "../services/responsesInputSanitizer import { normalizeCodexVerbosity } from "../services/codexVerbosity.ts"; import { getThinkingBudgetConfig, ThinkingMode } from "../services/thinkingBudget.ts"; import { CORS_HEADERS } from "../utils/cors.ts"; +import { errorResponse } from "../utils/error.ts"; import { normalizeCodexResponsesInput } from "../utils/responsesInputNormalization.ts"; import * as prl from "../utils/providerRequestLogging.ts"; import { createRequire } from "module"; @@ -550,6 +551,145 @@ export function filterNonstandardCodexSse(response: Response): Response { }); } +// ─── Sub-bug #3 of upstream decolua/9router#2452 (@ryanngit) ───────────────── +// Codex sometimes answers with HTTP 200 and a text/event-stream body whose +// payload carries a transient "model at capacity" / overloaded error mid-stream, +// e.g. { "error": { "message": "Selected model is at capacity..." } }, +// server_is_overloaded, or service_unavailable_error. Left as a 200, this looks +// like a successful response to every caller — no retry, no circuit breaker, no +// combo/account fallback engages (open-sse/services/accountFallback.ts never +// sees a failure status). Peek the first few SSE bytes; when a transient-error +// signature is found, convert the response into a real 503 so account rotation +// kicks in. Otherwise re-assemble the stream from the peeked prefix + the +// remaining upstream body so the passthrough stays byte-identical. +const CODEX_SSE_TRANSIENT_ERROR_PATTERNS = [ + "selected model is at capacity", + "server_is_overloaded", + "service_unavailable_error", +] as const; +// A capacity/overloaded rejection is delivered as the very first SSE event, so a +// small peek window is enough — this bounds how much of a legitimate response we +// buffer before giving up and passing the stream through unchanged. +const CODEX_SSE_PEEK_MAX_BYTES = 8192; + +/** + * Best-effort extraction of the human-readable error message from a peeked SSE + * chunk, so the resulting 503 body carries something more useful than the raw + * pattern that matched. Falls back to the matched pattern when no structured + * `data:` payload could be parsed. + */ +function extractCodexSseErrorMessage(text: string, fallback: string): string { + for (const line of text.split(/\r?\n/)) { + if (!line.startsWith("data:")) continue; + const data = line.slice("data:".length).trim(); + if (!data || data === "[DONE]") continue; + try { + const parsed = JSON.parse(data) as Record; + const directError = parsed.error as Record | undefined; + const nestedError = (parsed.response as Record | undefined)?.error as + | Record + | undefined; + const message = + (typeof directError?.message === "string" && directError.message) || + (typeof nestedError?.message === "string" && nestedError.message) || + (typeof parsed.message === "string" && parsed.message); + if (message) return message; + } catch { + // Non-JSON SSE data line — keep scanning subsequent lines. + } + } + return fallback; +} + +type CodexSseTransientErrorPeek = + | { matched: string; message: string; replacementBody: null } + | { matched: null; message: null; replacementBody: ReadableStream | null }; + +/** + * Peek the first bytes of a Codex SSE response body looking for a transient + * error embedded in an otherwise 200-OK stream. Exported for unit testing. + */ +export async function peekCodexSseTransientError( + response: Response +): Promise { + const contentType = response.headers.get("content-type") || ""; + if (!response.ok || !response.body || !contentType.includes("text/event-stream")) { + return { matched: null, message: null, replacementBody: null }; + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + const chunks: Uint8Array[] = []; + let text = ""; + let matched: string | null = null; + + try { + while (text.length < CODEX_SSE_PEEK_MAX_BYTES) { + const { done, value } = await reader.read(); + if (done) break; + chunks.push(value); + text += decoder.decode(value, { stream: true }); + const lower = text.toLowerCase(); + const hit = CODEX_SSE_TRANSIENT_ERROR_PATTERNS.find((pattern) => lower.includes(pattern)); + if (hit) { + matched = hit; + break; + } + // A real content/completion event this early means the response is + // healthy — stop peeking so we do not needlessly buffer a long stream. + if ( + lower.includes('"type":"response.output_text.delta"') || + lower.includes('"type":"response.completed"') + ) { + break; + } + } + } catch (err) { + console.warn( + `[codex] peekCodexSseTransientError: read error, passing stream through: ${ + err instanceof Error ? err.message : String(err) + }` + ); + } + + if (matched) { + try { + await reader.cancel(); + } catch { + // Upstream socket may already be closing; nothing to clean up. + } + return { matched, message: extractCodexSseErrorMessage(text, matched), replacementBody: null }; + } + + reader.releaseLock(); + + // Re-assemble the stream: peeked prefix chunks, then continue draining the + // same underlying body so bytes downstream of the peek window are untouched. + const upstreamReader = response.body.getReader(); + const replacementBody = new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(chunk); + }, + async pull(controller) { + const { done, value } = await upstreamReader.read(); + if (done) { + controller.close(); + return; + } + controller.enqueue(value); + }, + cancel(reason) { + try { + upstreamReader.cancel(reason); + } catch { + // noop — upstream socket may already be closing. + } + }, + }); + + return { matched: null, message: null, replacementBody }; +} + export function encodeResponseSseEvent(raw: string): { sse: string; terminal: boolean } { let eventType = "message"; let payload = raw; @@ -664,6 +804,26 @@ export class CodexExecutor extends BaseExecutor { (httpResult as { response: Response }).response = filterNonstandardCodexSse(resp); } } + const resp = (httpResult as { response?: Response }).response; + if (resp) { + const peek = await peekCodexSseTransientError(resp); + if (peek.matched) { + input.log?.warn?.( + "RETRY", + `CODEX | 200-OK SSE carried transient error "${peek.matched}" — converting to 503 for account fallback` + ); + (httpResult as { response: Response }).response = errorResponse( + HTTP_STATUS.SERVICE_UNAVAILABLE, + peek.message + ); + } else if (peek.replacementBody) { + (httpResult as { response: Response }).response = new Response(peek.replacementBody, { + status: resp.status, + statusText: resp.statusText, + headers: resp.headers, + }); + } + } return httpResult; } diff --git a/tests/unit/codex-sse-capacity-fallback.test.ts b/tests/unit/codex-sse-capacity-fallback.test.ts new file mode 100644 index 0000000000..2316e1cd02 --- /dev/null +++ b/tests/unit/codex-sse-capacity-fallback.test.ts @@ -0,0 +1,157 @@ +// Sub-bug #3 of upstream decolua/9router#2452 (@ryanngit): Codex sometimes answers +// with HTTP 200 and a text/event-stream body whose payload carries a transient +// "model at capacity" / overloaded error mid-stream. Left unhandled, the 200 +// status makes this look like a successful response — no retry, no circuit +// breaker, no combo/account fallback engages, so the client either hangs or gets +// a truncated stream while a healthy account sits idle. This must be detected and +// converted into a real error Response (503) so accountFallback.ts / combo +// routing rotates to another account. +import test from "node:test"; +import assert from "node:assert/strict"; + +import { CodexExecutor, __setCodexWebSocketTransportForTesting } from "../../open-sse/executors/codex.ts"; + +test.afterEach(() => { + __setCodexWebSocketTransportForTesting(undefined); +}); + +function sseStreamFromChunks(chunks: string[]): ReadableStream { + const encoder = new TextEncoder(); + let i = 0; + return new ReadableStream({ + pull(controller) { + if (i >= chunks.length) { + controller.close(); + return; + } + controller.enqueue(encoder.encode(chunks[i])); + i++; + }, + }); +} + +test("CodexExecutor.execute converts a 200-OK SSE stream carrying a model-at-capacity error into a 503 Response", async () => { + const executor = new CodexExecutor(); + const originalFetch = globalThis.fetch; + + globalThis.fetch = async () => + new Response( + sseStreamFromChunks([ + 'event: error\ndata: {"error":{"message":"Selected model is at capacity. Please try a different model."}}\n\n', + ]), + { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + } + ); + + try { + const result = await executor.execute({ + model: "gpt-5.5", + body: { model: "gpt-5.5", input: [{ role: "user", content: "hello" }] }, + stream: true, + credentials: { accessToken: "codex-token" }, + }); + + assert.notEqual(result.response.status, 200); + assert.equal(result.response.status, 503); + const body = await result.response.json(); + assert.match(body.error.message, /at capacity/i); + // Hard Rule #12: never leak raw stack/paths in the sanitized error body. + assert.equal(body.error.message.includes("at /"), false); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test("CodexExecutor.execute converts server_is_overloaded / service_unavailable_error SSE payloads into a 503 Response", async () => { + const executor = new CodexExecutor(); + const originalFetch = globalThis.fetch; + + globalThis.fetch = async () => + new Response( + sseStreamFromChunks([ + 'event: error\ndata: {"error":{"type":"server_is_overloaded","message":"The server is overloaded. Please retry later."}}\n\n', + ]), + { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + } + ); + + try { + const result = await executor.execute({ + model: "gpt-5.5", + body: { model: "gpt-5.5", input: [{ role: "user", content: "hello" }] }, + stream: true, + credentials: { accessToken: "codex-token" }, + }); + + assert.equal(result.response.status, 503); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test("CodexExecutor.execute reassembles a normal 200-OK SSE stream byte-intact after peeking for transient errors", async () => { + const executor = new CodexExecutor(); + const originalFetch = globalThis.fetch; + + const normalSse = + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"Hello"}\n\n' + + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed"}}\n\n'; + + globalThis.fetch = async () => + new Response(sseStreamFromChunks([normalSse]), { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }); + + try { + const result = await executor.execute({ + model: "gpt-5.5", + body: { model: "gpt-5.5", input: [{ role: "user", content: "hello" }] }, + stream: true, + credentials: { accessToken: "codex-token" }, + }); + + assert.equal(result.response.status, 200); + const text = await result.response.text(); + assert.equal(text, normalSse); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test("CodexExecutor.execute reassembles a normal SSE stream split across multiple network chunks", async () => { + const executor = new CodexExecutor(); + const originalFetch = globalThis.fetch; + + const chunks = [ + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"Hel', + 'lo"}\n\n', + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed"}}\n\n', + ]; + const expected = chunks.join(""); + + globalThis.fetch = async () => + new Response(sseStreamFromChunks(chunks), { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }); + + try { + const result = await executor.execute({ + model: "gpt-5.5", + body: { model: "gpt-5.5", input: [{ role: "user", content: "hello" }] }, + stream: true, + credentials: { accessToken: "codex-token" }, + }); + + assert.equal(result.response.status, 200); + const text = await result.response.text(); + assert.equal(text, expected); + } finally { + globalThis.fetch = originalFetch; + } +});