fix(codex): fail the stream when the websocket closes before a terminal event (#12737)

* fix(codex): fail the stream when the websocket closes before a terminal event

* docs(changelog): add fragment for Codex websocket premature-close fix

* chore(quality): rebaseline file-size baselines for #12737 test and executor growth

* fix(codex): log websocket failures and harden premature-close tests

Review follow-up: failController now logs the failure (code + message)
via nextInput.log, and onclose surfaces the WS close code/reason in the
log line (the public payload stays sanitized through the allowlist).
Adds regression tests for the onerror-before-onclose sequence and a
close with zero prior events.

* chore(quality): re-measure the #12737 codex.ts file-size rebaseline after the release merge

The PR's own annotation was written against base 1505 and the frozen cap was
already at 1528 on release/v3.8.51 (which grew the file independently). With
this PR's +13 lines the merged file measures 1530, so the frozen entry moves
1528->1530 (+2 of genuine PR growth; the remaining 11 lines fit the headroom
the tip already had). No other frozen entry touched.

---------

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

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

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,

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<string, unknown>) => Promise<WreqWebSocket>;
 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);

open-sse/utils/codexPublicError.ts

@@ -76,6 +76,7 @@ const CODEX_PUBLIC_ERROR_RULES = new Map<string, CodexPublicErrorRule>([
     "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) }],
 ]);

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;