fix(codex): handle fast tier and capacity SSE (#2452)

- map service_tier=fast to upstream priority; drop unsupported tiers
- normalize reasoning effort max to xhigh (codex-only)
- convert 200-SSE model-capacity errors into 503 so account fallback rotates
- keep normal SSE output intact after peeking

Co-authored-by: Cursor <cursoragent@cursor.com>

open-sse/executors/codex.js

@@ -8,13 +8,21 @@ import {
 import { normalizeResponsesInput } from "../translator/formats/responsesApi.js";
 import { fetchImageAsBase64 } from "../translator/concerns/image.js";
 import { getModelUpstreamId } from "../config/providerModels.js";
-import { DEFAULT_RETRY_CONFIG, resolveRetryEntry } from "../config/runtimeConfig.js";
+import { DEFAULT_RETRY_CONFIG, HTTP_STATUS, resolveRetryEntry } from "../config/runtimeConfig.js";
 import { dbg } from "../utils/debugLog.js";
 import { resolveSessionId } from "../utils/sessionManager.js";
 
-// SSE error patterns inside 200-OK body that should trigger retry as if 503
+// SSE error patterns inside 200-OK bodies. Some retry same account first; capacity rotates accounts.
-const CODEX_SSE_OVERLOADED_PATTERNS = ["server_is_overloaded", "service_unavailable_error"];
+const CODEX_SSE_RETRY_PATTERNS = ["server_is_overloaded", "service_unavailable_error"];
-const CODEX_SSE_PEEK_BYTES = 4096;
+const CODEX_SSE_ACCOUNT_FALLBACK_PATTERNS = ["selected model is at capacity", "model_at_capacity"];
+const CODEX_SSE_USER_OUTPUT_PATTERNS = [
+  "event: response.output_text.delta",
+  "event: response.function_call_arguments.delta",
+  '"type":"response.output_text.delta"',
+  '"type":"response.function_call_arguments.delta"',
+];
+const CODEX_SSE_PEEK_BYTES = 256 * 1024;
+const CODEX_MODEL_CAPACITY_MESSAGE = "Selected model is at capacity. Please try a different model.";
 
 // Server-generated item id prefixes that Codex /responses cannot resolve when store=false
 const SERVER_ID_PATTERN = /^(rs|fc|resp|msg)_/;
@@ -116,6 +124,62 @@ function resolveCacheSessionId(body, credentials) {
   });
 }
 
+function normalizeReasoningEffort(value) {
+  return value === "max" ? "xhigh" : value;
+}
+
+function findNestedMessage(value, depth = 0) {
+  if (!value || depth > 6 || typeof value === "string") return null;
+  if (Array.isArray(value)) {
+    for (const item of value) {
+      const found = findNestedMessage(item, depth + 1);
+      if (found) return found;
+    }
+    return null;
+  }
+  if (typeof value !== "object") return null;
+  if (typeof value.message === "string" && value.message.trim()) return value.message;
+  if (typeof value.error?.message === "string" && value.error.message.trim()) return value.error.message;
+  if (typeof value.response?.error?.message === "string" && value.response.error.message.trim()) return value.response.error.message;
+  for (const child of Object.values(value)) {
+    const found = findNestedMessage(child, depth + 1);
+    if (found) return found;
+  }
+  return null;
+}
+
+function extractSseErrorMessage(text, fallback) {
+  const exact = text?.match(/Selected model is at capacity\. Please try a different model\./i)?.[0];
+  if (exact) return exact;
+
+  for (const line of String(text || "").split(/\r?\n/)) {
+    if (!line.startsWith("data:")) continue;
+    const data = line.slice(5).trim();
+    if (!data || data === "[DONE]") continue;
+    try {
+      const message = findNestedMessage(JSON.parse(data));
+      if (message) return message;
+    } catch {
+      // Ignore non-JSON SSE data lines.
+    }
+  }
+
+  return fallback || CODEX_MODEL_CAPACITY_MESSAGE;
+}
+
+function codexSseErrorResponse(status, message) {
+  return new Response(JSON.stringify({
+    error: {
+      message,
+      type: status >= 500 ? "server_error" : "invalid_request_error",
+      code: status === HTTP_STATUS.SERVICE_UNAVAILABLE ? "service_unavailable" : "upstream_error",
+    }
+  }), {
+    status,
+    headers: { "Content-Type": "application/json" },
+  });
+}
+
 /**
  * Codex Executor - handles OpenAI Codex API (Responses API format)
  * Automatically injects default instructions if missing
@@ -136,9 +200,9 @@ export class CodexExecutor extends BaseExecutor {
     // Identify client type to Codex backend (matches official codex CLI)
     if (!headers["originator"]) headers["originator"] = "codex_cli_rs";
     // Workspace binding header — improves account scope + cache affinity
-    const workspaceId = credentials?.providerSpecificData?.workspaceId;
+    const workspaceId = credentials?.providerSpecificData?.workspaceId || credentials?.providerSpecificData?.chatgptAccountId;
-    if (typeof workspaceId === "string" && workspaceId && !headers["chatgpt-account-id"]) {
+    if (typeof workspaceId === "string" && workspaceId && !headers["ChatGPT-Account-ID"]) {
-      headers["chatgpt-account-id"] = workspaceId;
+      headers["ChatGPT-Account-ID"] = workspaceId;
     }
     return headers;
   }
@@ -198,7 +262,7 @@ export class CodexExecutor extends BaseExecutor {
     let attempt = 0;
     while (true) {
       const result = await super.execute(args);
-      const peek = await this._peekSseOverloaded(result.response);
+      const peek = await this._peekSseTransientError(result.response);
       if (!peek.matched) {
         // Replace body with re-assembled stream (prefix bytes already read + rest)
         if (peek.replacementBody) {
@@ -210,48 +274,57 @@ export class CodexExecutor extends BaseExecutor {
         }
         return result;
       }
+      if (peek.accountFallback) {
+        args.log?.warn?.("RETRY", `CODEX | SSE account fallback "${peek.message}"`);
+        result.response = codexSseErrorResponse(HTTP_STATUS.SERVICE_UNAVAILABLE, peek.message || CODEX_MODEL_CAPACITY_MESSAGE);
+        return result;
+      }
       if (attempt >= attempts) {
         args.log?.warn?.("RETRY", `CODEX | SSE overloaded "${peek.matched}" — retries exhausted (${attempt}/${attempts})`);
-        // Out of retries → return with replacement body so client gets the error
+        result.response = codexSseErrorResponse(HTTP_STATUS.SERVICE_UNAVAILABLE, peek.message || peek.matched);
-        if (peek.replacementBody) {
-          result.response = new Response(peek.replacementBody, {
-            status: result.response.status,
-            statusText: result.response.statusText,
-            headers: result.response.headers,
-          });
-        }
         return result;
       }
       attempt++;
       args.log?.debug?.("RETRY", `CODEX | SSE "${peek.matched}" retry ${attempt}/${attempts} after ${delayMs / 1000}s`);
       dbg("CODEX", `SSE overloaded "${peek.matched}" → retry ${attempt}/${attempts} in ${delayMs}ms`);
-      try { await result.response.body?.cancel?.(); } catch { /* noop */ }
       await new Promise(r => setTimeout(r, delayMs));
     }
   }
 
-  // Peek first N bytes of SSE body to detect upstream "overloaded" errors.
+  // Peek first N bytes of SSE body to detect upstream transient errors.
-  // Returns { matched: string|null, replacementBody: ReadableStream|null }.
+  // Returns { matched: string|null, message: string|null, accountFallback: boolean, replacementBody: ReadableStream|null }.
-  // Caller MUST use replacementBody (original body has been read).
+  // Caller must use replacementBody when no error matched (original body has been read).
-  async _peekSseOverloaded(response) {
+  async _peekSseTransientError(response) {
-    if (!response || !response.ok || !response.body) return { matched: null, replacementBody: null };
+    if (!response || !response.ok || !response.body) return { matched: null, message: null, accountFallback: false, replacementBody: null };
     const reader = response.body.getReader();
     const decoder = new TextDecoder();
     const chunks = [];
     let text = "";
     let matched = null;
+    let accountFallback = false;
     try {
       while (text.length < CODEX_SSE_PEEK_BYTES) {
         const { done, value } = await reader.read();
         if (done) break;
         chunks.push(value);
         text += decoder.decode(value, { stream: true });
-        const hit = CODEX_SSE_OVERLOADED_PATTERNS.find(p => text.includes(p));
+        const lowerText = text.toLowerCase();
-        if (hit) { matched = hit; break; }
+        const accountHit = CODEX_SSE_ACCOUNT_FALLBACK_PATTERNS.find(p => lowerText.includes(p));
+        if (accountHit) { matched = accountHit; accountFallback = true; break; }
+        const retryHit = CODEX_SSE_RETRY_PATTERNS.find(p => lowerText.includes(p));
+        if (retryHit) { matched = retryHit; break; }
+        if (CODEX_SSE_USER_OUTPUT_PATTERNS.some(p => lowerText.includes(p))) break;
       }
     } catch (e) {
       dbg("CODEX", `peek read error: ${e.message}`);
     }
+
+    if (matched) {
+      try { await reader.cancel(); } catch { /* noop */ }
+      try { reader.releaseLock(); } catch { /* noop */ }
+      return { matched, message: extractSseErrorMessage(text, matched), accountFallback, replacementBody: null };
+    }
+
     reader.releaseLock();
 
     // Re-assemble stream: prefix chunks + remaining upstream body
@@ -273,7 +346,7 @@ export class CodexExecutor extends BaseExecutor {
         try { upstreamReader?.cancel(reason); } catch { /* noop */ }
       },
     });
-    return { matched, replacementBody };
+    return { matched: null, message: null, accountFallback: false, replacementBody };
   }
 
   // Parse Codex usage_limit_reached to extract precise resetsAtMs; fallback to default otherwise
@@ -347,7 +420,7 @@ export class CodexExecutor extends BaseExecutor {
 
     // Extract thinking level from model name suffix
     // e.g., gpt-5.3-codex-high → high, gpt-5.3-codex → medium (default)
-    const effortLevels = ['none', 'low', 'medium', 'high', 'xhigh'];
+    const effortLevels = ['none', 'minimal', 'low', 'medium', 'high', 'xhigh'];
     let modelEffort = null;
     for (const level of effortLevels) {
       if (body.model.endsWith(`-${level}`)) {
@@ -360,10 +433,11 @@ export class CodexExecutor extends BaseExecutor {
 
     // Priority: explicit reasoning.effort > reasoning_effort param > model suffix > default (medium)
     if (!body.reasoning) {
-      const effort = body.reasoning_effort || modelEffort || 'low';
+      const effort = normalizeReasoningEffort(body.reasoning_effort || modelEffort || 'low');
       body.reasoning = { effort, summary: "auto" };
-    } else if (!body.reasoning.summary) {
+    } else {
-      body.reasoning.summary = "auto";
+      body.reasoning.effort = normalizeReasoningEffort(body.reasoning.effort);
+      if (!body.reasoning.summary) body.reasoning.summary = "auto";
     }
     delete body.reasoning_effort;
 
@@ -391,6 +465,9 @@ export class CodexExecutor extends BaseExecutor {
     delete body.safety_identifier; // Droid CLI sends this but Codex doesn't support it
     delete body.previous_response_id; // store=false → backend can't resolve previous resp; avoid 404
 
+    if (body.service_tier === "fast") body.service_tier = "priority";
+    if (body.service_tier && body.service_tier !== "priority") delete body.service_tier;
+
     // Final allowlist filter — strip any unknown field that could trigger upstream "routing_unsupported"
     for (const k of Object.keys(body)) {
       if (!RESPONSES_API_ALLOWLIST.has(k)) delete body[k];

tests/unit/codex-fast-capacity.test.js

@@ -0,0 +1,71 @@
+import { describe, expect, it } from "vitest";
+import { CodexExecutor } from "../../open-sse/executors/codex.js";
+
+function streamFromText(text) {
+  const encoder = new TextEncoder();
+  return new ReadableStream({
+    start(controller) {
+      controller.enqueue(encoder.encode(text));
+      controller.close();
+    },
+  });
+}
+
+describe("Codex fast tier and capacity handling", () => {
+  it("maps Codex fast tier to priority and max reasoning to xhigh", () => {
+    const executor = new CodexExecutor();
+    const body = executor.transformRequest("gpt-5.5", {
+      model: "gpt-5.5",
+      input: "hi",
+      reasoning_effort: "max",
+      service_tier: "fast",
+    }, true, {});
+
+    expect(body.service_tier).toBe("priority");
+    expect(body.reasoning.effort).toBe("xhigh");
+  });
+
+  it("uses ChatGPT workspace header fallback", () => {
+    const executor = new CodexExecutor();
+    const headers = executor.buildHeaders({
+      accessToken: "token",
+      connectionId: "conn_1",
+      providerSpecificData: { chatgptAccountId: "acct_1" },
+    });
+
+    expect(headers["ChatGPT-Account-ID"]).toBe("acct_1");
+  });
+
+  it("classifies 200-SSE model capacity as account fallback", async () => {
+    const executor = new CodexExecutor();
+    const response = new Response(streamFromText([
+      "event: error",
+      'data: {"error":{"message":"Selected model is at capacity. Please try a different model."}}',
+      "",
+    ].join("\n")), {
+      status: 200,
+      headers: { "Content-Type": "text/event-stream" },
+    });
+
+    const peek = await executor._peekSseTransientError(response);
+    expect(peek.accountFallback).toBe(true);
+    expect(peek.message).toBe("Selected model is at capacity. Please try a different model.");
+  });
+
+  it("reassembles normal SSE after peeking", async () => {
+    const executor = new CodexExecutor();
+    const text = [
+      "event: response.output_text.delta",
+      'data: {"type":"response.output_text.delta","delta":"OK"}',
+      "",
+    ].join("\n");
+    const response = new Response(streamFromText(text), {
+      status: 200,
+      headers: { "Content-Type": "text/event-stream" },
+    });
+
+    const peek = await executor._peekSseTransientError(response);
+    expect(peek.matched).toBeNull();
+    await expect(new Response(peek.replacementBody).text()).resolves.toBe(text);
+  });
+});