diff --git a/changelog.d/fixes/12737-codex-websocket-premature-close.md b/changelog.d/fixes/12737-codex-websocket-premature-close.md new file mode 100644 index 0000000000..3b615ea220 --- /dev/null +++ b/changelog.d/fixes/12737-codex-websocket-premature-close.md @@ -0,0 +1 @@ +- **fix(codex):** the Codex WebSocket transport now emits a terminal `response.failed` (code `upstream_websocket_closed`) when the upstream socket closes before a terminal response event, instead of ending the client stream as if it had completed normally — preventing silent output truncation and allowing fallback/retry to trigger ([#12737](https://github.com/diegosouzapw/OmniRoute/pull/12737)). diff --git a/config/quality/file-size-baseline.json b/config/quality/file-size-baseline.json index 62b2085c42..c8568605db 100644 --- a/config/quality/file-size-baseline.json +++ b/config/quality/file-size-baseline.json @@ -249,6 +249,7 @@ "_rebaseline_2026_09_15_13748_13749_merged_test_growth_basereds": "Base-red drain (#12732): two security fixes merged on 2026-09-15 each grew a frozen test file with their own regression coverage, and PR-mode check:file-size does not relax testFrozen against the base, so every PR into release/v3.8.51 went red on file-size. Recorded against the merged state: tests/unit/image-generation-handler.test.ts 2133->2235 (#13748 public-only guard on client-supplied image URLs); tests/unit/batch_api.test.ts 1345->1348 (#13749 API-key ownership on files and batches). No cap is raised beyond the merged LOC.", "_rebaseline_2026_08_24_video_bridge_fu01_fu03_fu04_result_cache_tests": "PRs #11362 (FU-01 cache hardening) + #11382 (FU-03 visual dedup policy identity) + #11383 (FU-04 focused analysis mode) own test growth: videoBridgeResultCache.test.ts <1000->1040, +40 (sum of three stacked PRs boarded together in the same merge-batch, each adding its own cache-identity assertions on the shared result-cache seam). Owner pre-authorized rebaseline for legitimate PR growth (2026-08-19 directive).", "_rebaseline_basered_codebuddy_cn": "Base-red fix (#4664 CodeBuddy CN): oauth-providers-config.test.ts 867->870 (+3) to align the EXPECTED provider list/config with the codebuddy-cn provider that #4664 added to the registry without updating this test (it asserts 'exactly once').", + "_rebaseline_2026_09_04_12737_codex_ws_premature_close_tests": "PR #12737 own test growth: executor-codex.test.ts 1465->1620 (+155, entirely this PR's diff — regression coverage for the premature WebSocket close fix: emits terminal response.failed with code upstream_websocket_closed when the socket closes before any terminal event, and proves no second terminal event fires after a normal post-response.completed close).", "_rebaseline_pr4613_compatible_provider_groups": "Reconcile #4613 already-merged growth: providers-page-utils.test.ts 1004->1052 (+48, buildCompatibleProviderGroups partition unit test). Fast-gate PR->release does not run check:file-size, so this surfaced post-merge.", "tests/integration/chat-pipeline.test.ts": 1740, "tests/unit/account-fallback-service.test.ts": 2056, @@ -258,7 +259,7 @@ "tests/unit/chatgpt-web.test.ts": 4911, "tests/unit/combo-routing-engine.test.ts": 3625, "tests/unit/db-migration-runner.test.ts": 1509, - "tests/unit/executor-codex.test.ts": 1465, + "tests/unit/executor-codex.test.ts": 1620, "tests/unit/executor-default-base.test.ts": 1632, "tests/unit/grok-web.test.ts": 2985, "tests/unit/image-generation-handler.test.ts": 2235, @@ -460,10 +461,11 @@ "_rebaseline_2026_09_16_13448_adaptive_effort_targetformat_gate": "#13448 rework: open-sse/handlers/chatCore.ts 6142->6156 (+14, PR's own growth: the X-OmniRoute-Effort header capture near THINKING_MARKER_HEADER and the wireAdaptiveEffort(translatedBody, {...}) call site right after applyDefaultReasoningEffort, plus this rework's +1 targetFormat argument at that same call). The targetFormat gate itself (ctx.targetFormat !== FORMATS.OPENAI short-circuit) lives in the non-frozen open-sse/handlers/chatCore/adaptiveEffortWiring.ts leaf, not here -- irreducible call-site wiring at the existing post-translation reasoning-normalization chokepoint. Covered by tests/unit/adaptive-effort-wiring.test.ts (targetFormat gate, red-on-tip) and tests/unit/adaptive-effort-model-default-13448.test.ts.", "_rebaseline_pr1043_minimax_tts": "Upstream port decolua/9router#1043 (toanalien) own growth: audioSpeech.ts 965->1061 (+96). Adds MiniMax T2A v2 TTS dispatch (handleMinimaxSpeech + hexToBytes helper) — provider entry was already in audioRegistry (format: minimax-tts) but no handler existed, falling through to the OpenAI-compatible default that fails (T2A has custom shape + hex-encoded audio + base_resp envelope). New branch sits next to the other inline provider branches (xiaomi-mimo, coqui, tortoise, aws-polly) — extracting would just create indirection. Covered by tests/unit/minimax-tts-1043.test.ts (3 tests, GREEN: success, base_resp error, invalid-hex).", "_rebaseline_pr4592_exclude_exhausted_auto": "Reconcile #4592 already-merged growth: combo.ts 2991->3036 (+45, terminal-status quota-cutoff exclusion in buildAutoCandidates + opt-in gate). Fast-gate PR->release does not run check:file-size.", + "_rebaseline_2026_09_04_12737_codex_ws_premature_close": "PR #12737 own growth, re-measured after merging release/v3.8.51: open-sse/executors/codex.ts 1528->1530 (+2 over the frozen cap; the PR adds +13 lines and the tip had 11 lines of headroom), now also logging the failure and the WS close code/reason via review follow-up. The ws.onclose handler now fails the stream (failController with code upstream_websocket_closed) when the socket closes before any terminal response event, instead of finishStream(upstream_closed) silently truncating output as a completed stream. The +8 is the guard + routing through the existing failController at the existing onclose chokepoint — not extractable without hiding the close-handler semantics. Covered by the two new regression tests in tests/unit/executor-codex.test.ts (premature close emits exactly one response.failed; normal close after response.completed emits no second terminal event).", "open-sse/executors/antigravity.ts": 1665, "open-sse/executors/base.ts": 1753, "open-sse/executors/chatgpt-web.ts": 5056, - "open-sse/executors/codex.ts": 1528, + "open-sse/executors/codex.ts": 1530, "open-sse/executors/cursor.ts": 1847, "open-sse/executors/muse-spark-web.ts": 1405, "open-sse/handlers/chatCore.ts": 6287, diff --git a/open-sse/executors/codex.ts b/open-sse/executors/codex.ts index 82bd7931de..2c9d98d25e 100644 --- a/open-sse/executors/codex.ts +++ b/open-sse/executors/codex.ts @@ -79,7 +79,7 @@ type WreqWebSocket = { close: (code?: number, reason?: string) => void; onmessage: ((event: { data: unknown }) => void) | null; onerror: ((event: { message?: string }) => void) | null; - onclose: (() => void) | null; + onclose: ((event?: { code?: number; reason?: string }) => void) | null; }; type WebsocketFn = (url: string, opts?: Record) => Promise; type ResponsesMessageInput = { role?: unknown; phase?: unknown; content?: unknown }; @@ -973,8 +973,9 @@ export class CodexExecutor extends BaseExecutor { } }; - const failController = (code: string, _message: string) => { + const failController = (code: string, message: string) => { if (closed) return; + nextInput.log?.warn?.("CODEX", `WebSocket stream failed (${code}): ${message}`); const controller = streamController; const payload = JSON.stringify({ type: "response.failed", @@ -987,6 +988,7 @@ export class CodexExecutor extends BaseExecutor { try { controller?.enqueue(encoder.encode(`event: response.failed\ndata: ${payload}\n\n`)); } catch { + console.warn("[codex] failController: failed to enqueue response.failed"); // Downstream closed before the failure could be delivered. } finishStream({ reason: "upstream_failed" }); @@ -1043,8 +1045,19 @@ export class CodexExecutor extends BaseExecutor { event.message || "Codex upstream WebSocket error" ); }; - ws.onclose = () => { - finishStream({ reason: "upstream_closed", closeSocket: false }); + ws.onclose = (event) => { + // A close after a terminal event already finished the stream — no-op. + // A close before any terminal event means the upstream died mid-response: + // emit a terminal response.failed instead of ending the client stream as + // if it completed normally (silent truncation). + if (closed) return; + const closeDetail = event + ? ` (code ${event.code ?? "unknown"}${event.reason ? `: ${event.reason}` : ""})` + : ""; + failController( + "upstream_websocket_closed", + `Codex upstream WebSocket closed before a terminal response event${closeDetail}` + ); }; if (!closed) { await prl.captureCurrentProviderBody(url, headers, bodyString, nextInput.log); diff --git a/open-sse/utils/codexPublicError.ts b/open-sse/utils/codexPublicError.ts index 6a8c92209c..65093b0740 100644 --- a/open-sse/utils/codexPublicError.ts +++ b/open-sse/utils/codexPublicError.ts @@ -76,6 +76,7 @@ const CODEX_PUBLIC_ERROR_RULES = new Map([ "upstream_websocket_connect_failed", { type: "provider_error", allowsStatus: exactStatuses(502) }, ], + ["upstream_websocket_closed", { type: "provider_error", allowsStatus: exactStatuses(502) }], ["upstream_websocket_error", { type: "provider_error", allowsStatus: exactStatuses(502) }], ["usage_limit_reached", { type: "rate_limit_error", allowsStatus: exactStatuses(429) }], ]); diff --git a/tests/unit/executor-codex.test.ts b/tests/unit/executor-codex.test.ts index b795e9932f..f1b4a6e4bb 100644 --- a/tests/unit/executor-codex.test.ts +++ b/tests/unit/executor-codex.test.ts @@ -1030,6 +1030,161 @@ test("CodexExecutor.execute captures the exact websocket request body before sen assert.equal(sentBody.model, "gpt-5.5"); }); +test("CodexExecutor.execute emits response.failed when websocket closes before a terminal event", async () => { + const executor = new CodexExecutor(); + const ws: MockCodexWebSocket = { + send() { + queueMicrotask(() => { + ws.onmessage?.({ + data: JSON.stringify({ + type: "response.output_text.delta", + delta: "partial output", + }), + }); + ws.onclose?.(); + }); + }, + close() {}, + onmessage: null, + onerror: null, + onclose: null, + }; + __setCodexWebSocketTransportForTesting(async () => ws); + + const result = await executor.execute({ + model: "gpt-5.5-xhigh", + body: { model: "gpt-5.5-xhigh", input: [{ role: "user", content: "hello" }] }, + stream: true, + credentials: { + accessToken: "codex-token", + providerSpecificData: { codexTransport: "websocket" }, + }, + }); + const body = await result.response.text(); + + assert.match(body, /event: response\.failed/); + const terminalEvents = body.match(/event: response\.(?:completed|failed|incomplete)/g) ?? []; + assert.deepEqual(terminalEvents, ["event: response.failed"]); + + const dataLine = body.split("\n").find((line) => line.includes('"upstream_websocket_closed"')); + assert.ok(dataLine); + const payload = JSON.parse(dataLine.slice("data: ".length)); + assert.equal(payload.type, "response.failed"); + assert.equal(payload.response.status, "failed"); + assert.equal(payload.response.error.code, "upstream_websocket_closed"); +}); + +test("CodexExecutor.execute does not emit a second terminal event after normal websocket close", async () => { + const executor = new CodexExecutor(); + const ws: MockCodexWebSocket = { + send() { + queueMicrotask(() => { + ws.onmessage?.({ + data: JSON.stringify({ + type: "response.completed", + response: { id: "resp_complete", status: "completed" }, + }), + }); + ws.onclose?.(); + }); + }, + close() {}, + onmessage: null, + onerror: null, + onclose: null, + }; + __setCodexWebSocketTransportForTesting(async () => ws); + + const result = await executor.execute({ + model: "gpt-5.5-xhigh", + body: { model: "gpt-5.5-xhigh", input: [{ role: "user", content: "hello" }] }, + stream: true, + credentials: { + accessToken: "codex-token", + providerSpecificData: { codexTransport: "websocket" }, + }, + }); + const body = await result.response.text(); + + const terminalEvents = body.match(/event: response\.(?:completed|failed|incomplete)/g) ?? []; + assert.deepEqual(terminalEvents, ["event: response.completed"]); + assert.doesNotMatch(body, /upstream_websocket_closed/); +}); + +test("CodexExecutor.execute emits a single response.failed when onerror precedes onclose", async () => { + const executor = new CodexExecutor(); + const ws: MockCodexWebSocket = { + send() { + queueMicrotask(() => { + ws.onmessage?.({ + data: JSON.stringify({ + type: "response.output_text.delta", + delta: "partial output", + }), + }); + // Real WebSocket implementations fire onerror before onclose on an + // abnormal close — the closed latch must keep this to one terminal event. + ws.onerror?.({ message: "socket hang up" }); + ws.onclose?.({ code: 1006 }); + }); + }, + close() {}, + onmessage: null, + onerror: null, + onclose: null, + }; + __setCodexWebSocketTransportForTesting(async () => ws); + + const result = await executor.execute({ + model: "gpt-5.5-xhigh", + body: { model: "gpt-5.5-xhigh", input: [{ role: "user", content: "hello" }] }, + stream: true, + credentials: { + accessToken: "codex-token", + providerSpecificData: { codexTransport: "websocket" }, + }, + }); + const body = await result.response.text(); + + const terminalEvents = body.match(/event: response\.(?:completed|failed|incomplete)/g) ?? []; + assert.deepEqual(terminalEvents, ["event: response.failed"]); + // The first failure wins: onerror fired before onclose, so the emitted code is + // upstream_websocket_error, not upstream_websocket_closed. + assert.match(body, /upstream_websocket_error/); + assert.doesNotMatch(body, /upstream_websocket_closed/); +}); + +test("CodexExecutor.execute emits response.failed when websocket closes with no prior events", async () => { + const executor = new CodexExecutor(); + const ws: MockCodexWebSocket = { + send() { + queueMicrotask(() => { + ws.onclose?.({ code: 1006, reason: "abnormal closure" }); + }); + }, + close() {}, + onmessage: null, + onerror: null, + onclose: null, + }; + __setCodexWebSocketTransportForTesting(async () => ws); + + const result = await executor.execute({ + model: "gpt-5.5-xhigh", + body: { model: "gpt-5.5-xhigh", input: [{ role: "user", content: "hello" }] }, + stream: true, + credentials: { + accessToken: "codex-token", + providerSpecificData: { codexTransport: "websocket" }, + }, + }); + const body = await result.response.text(); + + const terminalEvents = body.match(/event: response\.(?:completed|failed|incomplete)/g) ?? []; + assert.deepEqual(terminalEvents, ["event: response.failed"]); + assert.match(body, /upstream_websocket_closed/); +}); + test("CodexExecutor.execute adds CLI-like session identity headers without changing response flow", async () => { const executor = new CodexExecutor(); const originalFetch = globalThis.fetch;