pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-08-03 22:02:08 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(codex): drop non-standard codex.* events that break responses.stream (env-gated, #4602) (#4715)

Browse Source 
Integrated into release/v3.8.36
This commit is contained in:
Jefferson Felizardo
2026-06-23 23:07:51 -03:00
committed by GitHub
parent 4d61198e1a
commit eb175db8ed
5 changed files with 152 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.env.example
View File

@@ -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)
# ═══════════════════════════════════════════════════════════════════════════════

2
config/quality/file-size-baseline.json
View File

@@ -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.",

1
docs/reference/ENVIRONMENT.md
View File

@@ -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. |

103
open-sse/executors/codex.ts
View File

@@ -754,6 +754,57 @@ function toCodexResponseFailedEvent(parsed: Record<string, unknown>): Record<str
};
}
// Env-gated kill-switch: drop ALL non-standard `codex.*` SSE events (notably
// `codex.rate_limits`) from the Responses stream. These events are NOT part of
// the OpenAI Responses API — strict clients (e.g. the OpenAI SDK's
// `responses.stream()`) choke on the unknown event type / empty data field and
// tear the stream down, surfacing as "Invalid state: Controller is already
// closed". Opt-in so the default still forwards them for clients that want them.
function codexDropNonstandardEvents(): boolean {
const v = process.env.OMNIROUTE_CODEX_DROP_NONSTANDARD_EVENTS;
return v === "true" || v === "1" || v === "yes";
}
// SSE block filter for the HTTP Responses path (super.execute). The HTTP
// transport forwards the upstream stream verbatim — including the non-standard
// `event: codex.rate_limits` frame (no data line) — so the WS-only filter in
// encodeResponseSseEvent never runs for it. When the kill-switch is on, strip
// every `codex.*` event block from the byte stream before it reaches the client.
// Exported for unit testing (#4715). Strips `codex.*` SSE event blocks from a
// streaming Response when the OMNIROUTE_CODEX_DROP_NONSTANDARD_EVENTS kill-switch is on.
export function filterNonstandardCodexSse(response: Response): Response {
const contentType = response.headers.get("content-type") || "";
if (!response.body || !contentType.includes("text/event-stream")) {
return response;
}
const decoder = new TextDecoder();
const encoder = new TextEncoder();
let buffer = "";
const dropBlock = (block: string): boolean => {
const match = /^event:\s*(.+)$/m.exec(block);
return !!match && match[1].trim().startsWith("codex.");
};
const transform = new TransformStream<Uint8Array, Uint8Array>({
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" });

51
tests/unit/codex-drop-nonstandard-events.test.ts Normal file
View File

@@ -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<string> {
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"));
});
});