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