fix(codex): keep oversized Responses turns off the WS transport

src/server/responses/fetch-helpers.ts

@@ -64,14 +64,26 @@ export function providerFetch(
   options: ProviderFetchOptions = {},
 ): ProviderFetch {
   const base = (provider as OcxProviderConfig & { fetch?: typeof globalThis.fetch }).fetch ?? globalThis.fetch;
+  const preconnect = (...args: Parameters<typeof globalThis.fetch.preconnect>): void => {
+    base.preconnect?.(...args);
+  };
+  const httpFetch = Object.assign(
+    (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) =>
+      base(input, withUpstreamHttpVersion(input, init, provider)),
+    { preconnect },
+  ) as typeof globalThis.fetch;
   // ChatGPT Codex backend: streaming turns ride the responses_websockets
   // transport (measured ~3s faster TTFT than the SSE POST queue); everything
   // else keeps the provider's HTTP fetch. See ws-upstream.ts for the details.
   const unpaced = async (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
     if (typeof input === "string" && init && shouldUseCodexWsUpstream(input, init, runtime)) {
-      return codexWsUpstreamFetch(input, init, base, runtime);
+      // The fallback has to be the same HTTP fetch the non-WS branch would have
+      // used, protocol pin included: a WS turn that falls back is serving the
+      // request over HTTP, and dropping the provider's `upstreamHttpVersion`
+      // there would silently negotiate a transport the operator ruled out.
+      return codexWsUpstreamFetch(input, init, httpFetch, runtime);
     }
-    return base(input, withUpstreamHttpVersion(input, init, provider));
+    return httpFetch(input, init);
   };
   let pacingSlotAcquired = options.pacingSlotAcquired === true;
   const waitForPacing = (signal?: AbortSignal) => {
@@ -87,9 +99,6 @@ export function providerFetch(
     await waitForPacing(init?.signal ?? undefined);
     return unpaced(input, init);
   };
-  const preconnect = (...args: Parameters<typeof globalThis.fetch.preconnect>): void => {
-    base.preconnect?.(...args);
-  };
   return Object.assign(wrapped, {
     preconnect,
     waitForPacing,

src/server/responses/ws-upstream.ts

@@ -28,6 +28,25 @@ const UPGRADE_DEADLINE_MS = 10_000;
 export const MAX_CODEX_WS_FRAME_BYTES = MAX_CLIENT_SSE_FRAME_BYTES;
 export const MAX_CODEX_WS_QUEUE_BYTES = 8 * 1024 * 1024;
 export const MIN_BOUNDED_CODEX_WS_BUN_VERSION = "1.4.0";
+// The backend drops any inbound message of 16 MiB or more: it closes the socket
+// (1009) without a Responses terminal event, which reaches clients as a bare
+// 502 upstream_server_error. Measured against the live endpoint 2026-08-23:
+// 16,777,000 B completed, 16,777,300 B closed in ~1s, every time. The same
+// request body succeeds over HTTP SSE, so the ceiling belongs to this transport
+// alone (see #2426). A full-replay thread reaches it with ~11 pasted
+// screenshots, and then never recovers, because each retry resends the frame.
+export const MAX_CODEX_WS_CREATE_FRAME_BYTES = 16 * 1024 * 1024;
+// Bun frames the payload it is handed, so the send-side budget is the JSON text
+// itself, and nothing is appended between the check and the send. The margin is
+// a conservative cushion, not a computed requirement: it covers RFC 6455 frame
+// overhead in case the backend counts it (14 bytes at this payload size — an
+// 8-byte extended length plus a 4-byte client mask, leaving ~65.5 KiB spare),
+// and it leaves room for a future caller that appends to the frame.
+const CODEX_WS_CREATE_FRAME_MARGIN_BYTES = 64 * 1024;
+export const CODEX_WS_CREATE_FRAME_LIMIT_BYTES =
+  MAX_CODEX_WS_CREATE_FRAME_BYTES - CODEX_WS_CREATE_FRAME_MARGIN_BYTES;
+/** Close code the backend uses for an oversized message (RFC 6455 "message too big"). */
+const WS_CLOSE_MESSAGE_TOO_BIG = 1009;
 
 export type BunRuntimeIdentity = {
   version: string;
@@ -102,6 +121,52 @@ export function shouldUseCodexWsUpstream(
   }
 }
 
+const CLOSED_BEFORE_TERMINAL = "codex websocket closed before a Responses terminal event";
+
+/**
+ * The close code is the only thing that separates "the backend refused this
+ * payload" from "the network dropped", and both used to reach the caller as the
+ * same bare 502. Naming the oversized case here puts that distinction in the
+ * message the client receives.
+ *
+ * It does NOT reach the request log as a typed code. The eager relay turns any
+ * stream error into a generic `upstream_reset` synthetic terminal
+ * (`relay.ts`, `relay-eager.ts`) without feeding that frame back through the
+ * inspector, so `/api/logs` keeps neither this message nor a specific code —
+ * only `streamAborted`. Machine-readable typing would mean changing the error
+ * taxonomy, which is deliberately out of scope for this transport fix.
+ */
+function closedBeforeTerminalMessage(event: unknown): string {
+  const detail = event as { code?: unknown; reason?: unknown } | null | undefined;
+  const code = typeof detail?.code === "number" ? detail.code : null;
+  const reason = typeof detail?.reason === "string" ? detail.reason.trim() : "";
+  if (code === null) return CLOSED_BEFORE_TERMINAL;
+  const suffix = reason ? ` ${code} ${reason}` : ` ${code}`;
+  if (code === WS_CLOSE_MESSAGE_TOO_BIG) {
+    return `codex websocket rejected the request frame as too large (close${suffix});`
+      + ` requests at or above ${MAX_CODEX_WS_CREATE_FRAME_BYTES} bytes must use the HTTP SSE transport`;
+  }
+  return `${CLOSED_BEFORE_TERMINAL} (close${suffix})`;
+}
+
+/**
+ * True when the `response.create` frame is at or above the backend's inbound
+ * message ceiling, so this turn must take the HTTP SSE path instead.
+ *
+ * Sizing a 16 MiB string should not cost a 16 MiB copy. UTF-8 never encodes
+ * below one byte per UTF-16 code unit and never above three, so both tails are
+ * settled from the string length alone; only the narrow band between them pays
+ * for a real byte count, and `Buffer.byteLength` measures without allocating.
+ */
+export function codexWsCreateFrameExceedsLimit(
+  frameText: string,
+  limitBytes: number = CODEX_WS_CREATE_FRAME_LIMIT_BYTES,
+): boolean {
+  if (frameText.length >= limitBytes) return true;
+  if (frameText.length * 3 < limitBytes) return false;
+  return Buffer.byteLength(frameText, "utf8") >= limitBytes;
+}
+
 export function codexWsUpstreamFetch(
   url: string,
   init: RequestInit,
@@ -127,6 +192,14 @@ export function codexWsUpstreamFetch(
     return sseFallback(url, init);
   }
 
+  // Decide before dialing. Once the socket is open the caller already holds a
+  // streaming Response, so the oversized close can only be surfaced as a stream
+  // error — and a resend at that point could double-generate. Measuring the
+  // frame we are about to send keeps the whole failure mode unreachable.
+  if (codexWsCreateFrameExceedsLimit(frameText)) {
+    return sseFallback(url, init);
+  }
+
   const headers: Record<string, string> = {};
   new Headers(init.headers ?? {}).forEach((value, key) => {
     // HTTP-body framing headers do not apply to a WS handshake.
@@ -279,7 +352,7 @@ export function codexWsUpstreamFetch(
       }
     });
 
-    ws.addEventListener("close", () => {
+    ws.addEventListener("close", (event: unknown) => {
       signal?.removeEventListener("abort", onAbort);
       if (!opened) {
         if (settledPreOpen) return;
@@ -297,7 +370,7 @@ export function codexWsUpstreamFetch(
         // here would reach clients with no response.completed/failed at all —
         // relaySseWithFailedTail() only synthesizes a failed terminal when the
         // body read THROWS. Error the stream like a reset TCP socket.
-        try { controller.error(new Error("codex websocket closed before a Responses terminal event")); } catch { /* stream already done */ }
+        try { controller.error(new Error(closedBeforeTerminalMessage(event))); } catch { /* stream already done */ }
       }
     });
 

tests/ws-upstream.test.ts

@@ -5,9 +5,12 @@ import { isEagerRelaySseResponse } from "../src/server/relay";
 import { isWin32EagerRewrite } from "../src/lib/bun-stream-caps";
 import {
   bunSupportsBoundedCodexWsRelay,
+  CODEX_WS_CREATE_FRAME_LIMIT_BYTES,
+  codexWsCreateFrameExceedsLimit,
   codexWsUpstreamFetch as rawCodexWsUpstreamFetch,
   currentBunRuntimeIdentity,
   isCodexWsUpstreamResponse,
+  MAX_CODEX_WS_CREATE_FRAME_BYTES,
   MAX_CODEX_WS_FRAME_BYTES,
   MAX_CODEX_WS_QUEUE_BYTES,
   shouldUseCodexWsUpstream as rawShouldUseCodexWsUpstream,
@@ -672,3 +675,165 @@ describe("codexWsUpstreamFetch", () => {
     expect(FakeWebSocket.instances[0].closed).toBe(true);
   });
 });
+
+describe("codexWsCreateFrameExceedsLimit", () => {
+  test("measures the frame in UTF-8 bytes, not code units", () => {
+    // Two UTF-8 bytes per code unit, so half the limit in "é" is exactly the limit.
+    const halfLimit = CODEX_WS_CREATE_FRAME_LIMIT_BYTES / 2;
+    expect(codexWsCreateFrameExceedsLimit("é".repeat(halfLimit))).toBe(true);
+    expect(codexWsCreateFrameExceedsLimit("é".repeat(halfLimit - 1))).toBe(false);
+  });
+
+  test("holds the boundary at the limit itself", () => {
+    expect(codexWsCreateFrameExceedsLimit("x".repeat(CODEX_WS_CREATE_FRAME_LIMIT_BYTES))).toBe(true);
+    expect(codexWsCreateFrameExceedsLimit("x".repeat(CODEX_WS_CREATE_FRAME_LIMIT_BYTES - 1))).toBe(false);
+  });
+
+  test("keeps a margin under the backend's measured ceiling", () => {
+    // Measured against the live endpoint: 16,777,000 B completed, 16,777,300 B
+    // closed the socket. The gate has to trip below the smaller of those.
+    expect(MAX_CODEX_WS_CREATE_FRAME_BYTES).toBe(16 * 1024 * 1024);
+    expect(CODEX_WS_CREATE_FRAME_LIMIT_BYTES).toBeLessThan(16_777_000);
+  });
+});
+
+describe("oversized Codex create frames", () => {
+  test("takes the HTTP SSE path instead of dialing a socket the backend would close", async () => {
+    installFake(() => { throw new Error("WS must not be dialed for an oversized frame"); });
+    const sentinel = new Response("sse");
+    let fallbackCalls = 0;
+    const fallback = (async () => {
+      fallbackCalls += 1;
+      return sentinel;
+    }) as unknown as typeof fetch;
+
+    const oversized = streamingInit({ padding: "x".repeat(CODEX_WS_CREATE_FRAME_LIMIT_BYTES) });
+    expect(await codexWsUpstreamFetch(CODEX_URL, oversized, fallback)).toBe(sentinel);
+    expect(fallbackCalls).toBe(1);
+    // Nothing was sent, so the SSE resend cannot double-generate a turn.
+    expect(FakeWebSocket.instances).toHaveLength(0);
+  });
+
+  test("keeps the provider's HTTP version pin on the fallback", async () => {
+    installFake(() => { throw new Error("WS must not be dialed for an oversized frame"); });
+    const seen: RequestInit[] = [];
+    const provider = {
+      upstreamHttpVersion: "http1.1",
+      fetch: (async (_input: unknown, init: RequestInit) => {
+        seen.push(init);
+        return new Response("sse");
+      }) as unknown as typeof fetch,
+    } as unknown as OcxProviderConfig;
+    const wrapped = providerFetch(provider, BOUNDED_WS_RUNTIME);
+
+    await wrapped(CODEX_URL, streamingInit({ padding: "x".repeat(CODEX_WS_CREATE_FRAME_LIMIT_BYTES) }));
+
+    expect(FakeWebSocket.instances).toHaveLength(0);
+    // Falling back means serving the turn over HTTP, so the operator's pin has
+    // to survive the transport switch.
+    expect((seen[0] as { protocol?: string }).protocol).toBe("http1.1");
+  });
+
+  test("still uses WS for a frame that fits", async () => {
+    installFake(ws => {
+      ws.emit("open", {});
+      ws.emit("message", { data: JSON.stringify({ type: "response.completed", response: {} }) });
+    });
+    const fallback = (() => {
+      throw new Error("fallback must not run for a frame that fits");
+    }) as unknown as typeof fetch;
+
+    const response = await codexWsUpstreamFetch(CODEX_URL, streamingInit({ padding: "x".repeat(1024) }), fallback);
+    expect(isCodexWsUpstreamResponse(response)).toBe(true);
+    expect(FakeWebSocket.instances).toHaveLength(1);
+  });
+
+  // The unit tests above measure the helper; these two measure the REAL serialized
+  // frame, one byte on each side of the limit. That distinction matters because the
+  // request body is not the frame: `stream` is deleted and `type` is added before
+  // sending, so padding sized against the body would sit at a different offset than
+  // the bytes actually transmitted. An off-by-one lives exactly here and nowhere else.
+  describe("the adjacent-byte transport boundary", () => {
+    // Build padding such that the serialized frame is EXACTLY `target` bytes.
+    function initForFrameBytes(target: number): RequestInit {
+      const probe = frameTextFor(0);
+      const padding = "x".repeat(target - Buffer.byteLength(probe, "utf8"));
+      const init = streamingInit({ padding });
+      const actual = Buffer.byteLength(frameTextFor(padding.length), "utf8");
+      if (actual !== target) throw new Error(`frame sizing is wrong: wanted ${target}, built ${actual}`);
+      return init;
+    }
+
+    function frameTextFor(paddingLength: number): string {
+      const body = JSON.parse(streamingInit({ padding: "x".repeat(paddingLength) }).body as string) as Record<string, unknown>;
+      delete body.stream;
+      return JSON.stringify({ ...body, type: "response.create" });
+    }
+
+    test("a frame one byte under the limit goes over WS", async () => {
+      installFake(ws => {
+        ws.emit("open", {});
+        ws.emit("message", { data: JSON.stringify({ type: "response.completed", response: {} }) });
+      });
+      const fallback = (() => {
+        throw new Error("fallback must not run one byte under the limit");
+      }) as unknown as typeof fetch;
+
+      const response = await codexWsUpstreamFetch(
+        CODEX_URL,
+        initForFrameBytes(CODEX_WS_CREATE_FRAME_LIMIT_BYTES - 1),
+        fallback,
+      );
+      expect(isCodexWsUpstreamResponse(response)).toBe(true);
+      expect(FakeWebSocket.instances).toHaveLength(1);
+      expect(FakeWebSocket.instances[0]!.sent).toHaveLength(1);
+      // Byte length, not code units: the limit is a byte budget, and this
+      // assertion should keep meaning the same thing if the fixture ever
+      // carries non-ASCII text.
+      expect(Buffer.byteLength(FakeWebSocket.instances[0]!.sent[0]!, "utf8"))
+        .toBe(CODEX_WS_CREATE_FRAME_LIMIT_BYTES - 1);
+    });
+
+    test("the very next byte takes SSE without dialing", async () => {
+      installFake(() => { throw new Error("WS must not be dialed at the limit"); });
+      const sentinel = new Response("sse");
+      let fallbackCalls = 0;
+      const fallback = (async () => { fallbackCalls += 1; return sentinel; }) as unknown as typeof fetch;
+
+      const response = await codexWsUpstreamFetch(
+        CODEX_URL,
+        initForFrameBytes(CODEX_WS_CREATE_FRAME_LIMIT_BYTES),
+        fallback,
+      );
+      expect(response).toBe(sentinel);
+      expect(fallbackCalls).toBe(1);
+      expect(FakeWebSocket.instances).toHaveLength(0);
+    });
+  });
+
+  test("names the oversized close instead of reporting a bare drop", async () => {
+    installFake(ws => {
+      ws.emit("open", {});
+      ws.emit("close", { code: 1009, reason: "Message Too Big" });
+    });
+    const response = await codexWsUpstreamFetch(CODEX_URL, streamingInit(), (() => {
+      throw new Error("fallback must not run after open");
+    }) as unknown as typeof fetch);
+
+    await expect(response.text()).rejects.toThrow(
+      /rejected the request frame as too large \(close 1009 Message Too Big\)/,
+    );
+  });
+
+  test("carries the close code for any other pre-terminal drop", async () => {
+    installFake(ws => {
+      ws.emit("open", {});
+      ws.emit("close", { code: 1006 });
+    });
+    const response = await codexWsUpstreamFetch(CODEX_URL, streamingInit(), (() => {
+      throw new Error("fallback must not run after open");
+    }) as unknown as typeof fetch);
+
+    await expect(response.text()).rejects.toThrow("closed before a Responses terminal event (close 1006)");
+  });
+});