From eb175db8ed9db1323989c8776efabc8b14c6a578 Mon Sep 17 00:00:00 2001 From: Jefferson Felizardo Date: Tue, 23 Jun 2026 23:07:51 -0300 Subject: [PATCH] fix(codex): drop non-standard codex.* events that break responses.stream (env-gated, #4602) (#4715) Integrated into release/v3.8.36 --- .env.example | 6 + config/quality/file-size-baseline.json | 2 +- docs/reference/ENVIRONMENT.md | 1 + open-sse/executors/codex.ts | 103 ++++++++++++++++-- .../codex-drop-nonstandard-events.test.ts | 51 +++++++++ 5 files changed, 152 insertions(+), 11 deletions(-) create mode 100644 tests/unit/codex-drop-nonstandard-events.test.ts diff --git a/.env.example b/.env.example index 8fe017ec21..7a2cdc237e 100644 --- a/.env.example +++ b/.env.example @@ -830,6 +830,12 @@ GEMINI_CLI_USER_AGENT="google-api-nodejs-client/10.3.0" # CODEX_USER_AGENT string. Used by: open-sse/config/codexClient.ts. # CODEX_CLIENT_VERSION=0.132.0 +# Kill-switch to strip non-standard `codex.*` SSE events (e.g. codex.rate_limits) +# from the Codex Responses stream. These frames break the OpenAI SDK's +# responses.stream() with a 502 "Controller is already closed". Off by default; +# set to true/1/yes to enable. Used by: open-sse/executors/codex.ts. +# OMNIROUTE_CODEX_DROP_NONSTANDARD_EVENTS=true + # ═══════════════════════════════════════════════════════════════════════════════ # 13. CLI FINGERPRINT COMPATIBILITY (Anti-Detection) # ═══════════════════════════════════════════════════════════════════════════════ diff --git a/config/quality/file-size-baseline.json b/config/quality/file-size-baseline.json index 3878badf28..75622029d5 100644 --- a/config/quality/file-size-baseline.json +++ b/config/quality/file-size-baseline.json @@ -116,7 +116,7 @@ "open-sse/executors/base.ts": 1414, "open-sse/executors/chatgpt-web.ts": 2870, "open-sse/executors/claude-web.ts": 1057, - "open-sse/executors/codex.ts": 1449, + "open-sse/executors/codex.ts": 1528, "open-sse/executors/cursor.ts": 1453, "open-sse/executors/deepseek-web.ts": 1148, "_rebaseline_2026_06_22_4644_deepseek_web_tools": "PR #4644 (BugsBag/robust deepseek-web tool-call parsing): open-sse/executors/deepseek-web.ts 1117->1125 (+8). The new agentic tool-call path emits surrounding text + reasoning before tool_calls and swaps to the dedicated deepseekWebTools.ts parser; the +8 lines are cohesive wiring at the existing transformSSE chokepoint (the parser itself lives in the new deepseekWebTools.ts file, already under cap). The PR's own fast-gate (PR->release) does not run check:file-size, so this surfaced only at release reconcile. Covered by tests/unit/deepseek-web-tools-variants.test.ts + deepseek-web-tools-execute.test.ts.", diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index 24c46f9851..266c7fe90a 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -576,6 +576,7 @@ REQUEST_TIMEOUT_MS (global override) | `FETCH_TIMEOUT_MS` | `600000` | Total HTTP request timeout for upstream provider calls. | | `STREAM_IDLE_TIMEOUT_MS` | `600000` | Max silence between SSE chunks before aborting. Extended-thinking models rarely pause >90s. | | `STREAM_READINESS_TIMEOUT_MS` | `80000` | Time to receive the first non-ping SSE event. Inherits `REQUEST_TIMEOUT_MS` when set. | +| `OMNIROUTE_CODEX_DROP_NONSTANDARD_EVENTS` | _(off)_ | Strip non-standard `codex.*` SSE events (e.g. `codex.rate_limits`) that break the OpenAI SDK's `responses.stream()` with a 502. Set `true`/`1`/`yes` to enable. | | `FETCH_HEADERS_TIMEOUT_MS` | = `FETCH_TIMEOUT_MS` | Time to receive response headers. | | `FETCH_BODY_TIMEOUT_MS` | = `FETCH_TIMEOUT_MS` | Time to receive the full response body. | | `FETCH_CONNECT_TIMEOUT_MS` | `30000` | TCP connection establishment timeout. | diff --git a/open-sse/executors/codex.ts b/open-sse/executors/codex.ts index 03c2769beb..ffa7b0d81d 100644 --- a/open-sse/executors/codex.ts +++ b/open-sse/executors/codex.ts @@ -754,6 +754,57 @@ function toCodexResponseFailedEvent(parsed: Record): Record { + const match = /^event:\s*(.+)$/m.exec(block); + return !!match && match[1].trim().startsWith("codex."); + }; + const transform = new TransformStream({ + transform(chunk, controller) { + buffer += decoder.decode(chunk, { stream: true }); + let sep: number; + while ((sep = buffer.indexOf("\n\n")) !== -1) { + const block = buffer.slice(0, sep + 2); + buffer = buffer.slice(sep + 2); + if (!dropBlock(block)) controller.enqueue(encoder.encode(block)); + } + }, + flush(controller) { + if (buffer && !dropBlock(buffer)) controller.enqueue(encoder.encode(buffer)); + }, + }); + return new Response(response.body.pipeThrough(transform), { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }); +} + export function encodeResponseSseEvent(raw: string): { sse: string; terminal: boolean } { let eventType = "message"; let payload = raw; @@ -775,6 +826,27 @@ export function encodeResponseSseEvent(raw: string): { sse: string; terminal: bo // Keep message as the generic SSE event for non-JSON upstream payloads. } + // Env-gated: drop non-standard `codex.*` events (notably `codex.rate_limits`) + // before they reach the client. They are NOT part of the OpenAI Responses API + // and break strict consumers: the OpenAI SDK's responses.stream() chokes on + // the unknown event type / empty data and tears the stream down, surfacing as + // "Invalid state: Controller is already closed". The earlier empty-payload + // check below never caught codex.rate_limits — over WS the frame carries a + // non-empty JSON payload (`{"type":"codex.rate_limits", ...}`), so + // `!payload.trim()` is false. Match by event type instead. Opt-in via + // OMNIROUTE_CODEX_DROP_NONSTANDARD_EVENTS (the HTTP transport is handled + // separately by filterNonstandardCodexSse, since super.execute forwards the + // upstream stream verbatim and never runs this function). + if (eventType.startsWith("codex.") && codexDropNonstandardEvents()) { + return { sse: "", terminal }; + } + + // Drop frames whose raw payload is empty (defensive; non-JSON / blank upstream + // chunks). Frames that carry a payload are preserved. + if (!payload.trim()) { + return { sse: "", terminal }; + } + return { sse: `event: ${eventType}\ndata: ${payload}\n\n`, terminal }; } @@ -840,7 +912,14 @@ export class CodexExecutor extends BaseExecutor { const nextInput = { ...input, credentials }; if (!isCodexResponsesWebSocketRequired(nextInput.model, nextInput.credentials)) { - return super.execute(nextInput); + const httpResult = await super.execute(nextInput); + if (codexDropNonstandardEvents()) { + const resp = (httpResult as { response?: Response }).response; + if (resp?.body) { + (httpResult as { response: Response }).response = filterNonstandardCodexSse(resp); + } + } + return httpResult; } const url = CODEX_RESPONSES_WS_URL; @@ -973,15 +1052,19 @@ export class CodexExecutor extends BaseExecutor { : Buffer.from(event.data as Buffer).toString("utf8"); const sseEvent = encodeResponseSseEvent(raw); if (closed) return; - try { - controller.enqueue(encoder.encode(sseEvent.sse)); - } catch { - finishStream({ - reason: "downstream_closed", - emitDone: false, - closeController: false, - }); - return; + // Filtered events (codex.* / empty payload) return an empty `sse` — + // skip them so no empty frame reaches the client. + if (sseEvent.sse) { + try { + controller.enqueue(encoder.encode(sseEvent.sse)); + } catch { + finishStream({ + reason: "downstream_closed", + emitDone: false, + closeController: false, + }); + return; + } } if (sseEvent.terminal) { finishStream({ reason: "terminal_event" }); diff --git a/tests/unit/codex-drop-nonstandard-events.test.ts b/tests/unit/codex-drop-nonstandard-events.test.ts new file mode 100644 index 0000000000..8db3190f6e --- /dev/null +++ b/tests/unit/codex-drop-nonstandard-events.test.ts @@ -0,0 +1,51 @@ +// Regression guard for #4715: the Codex HTTP transport forwards the upstream SSE +// stream verbatim, including the non-standard `event: codex.rate_limits` frame +// (no `data:` line). That frame breaks the OpenAI SDK's responses.stream() with +// HTTP 502 "Controller is already closed". filterNonstandardCodexSse() strips +// every `codex.*` event block from the byte stream while preserving standard ones. +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; +import { filterNonstandardCodexSse } from "../../open-sse/executors/codex.ts"; + +function sseResponse(body: string): Response { + return new Response(body, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); +} + +async function readAll(res: Response): Promise { + return await res.text(); +} + +describe("filterNonstandardCodexSse (#4715)", () => { + it("drops codex.* event blocks but keeps standard response.* events", async () => { + const stream = + "event: response.created\ndata: {\"type\":\"response.created\"}\n\n" + + "event: codex.rate_limits\n\n" + + "event: response.output_text.delta\ndata: {\"delta\":\"hi\"}\n\n" + + "event: response.completed\ndata: {\"type\":\"response.completed\"}\n\n"; + const out = await readAll(filterNonstandardCodexSse(sseResponse(stream))); + assert.ok(!out.includes("codex.rate_limits"), "codex.* frame must be stripped"); + assert.ok(out.includes("response.created"), "standard events preserved"); + assert.ok(out.includes("response.output_text.delta"), "standard delta preserved"); + assert.ok(out.includes("response.completed"), "terminal event preserved"); + }); + + it("passes through non-SSE responses untouched", async () => { + const json = new Response("{\"ok\":true}", { + status: 200, + headers: { "content-type": "application/json" }, + }); + const out = filterNonstandardCodexSse(json); + assert.equal(await out.text(), "{\"ok\":true}"); + }); + + it("drops a trailing codex.* block with no double-newline terminator (flush path)", async () => { + const stream = + "event: response.created\ndata: {}\n\n" + "event: codex.token_count\ndata: {}"; + const out = await readAll(filterNonstandardCodexSse(sseResponse(stream))); + assert.ok(out.includes("response.created")); + assert.ok(!out.includes("codex.token_count")); + }); +});