From f093bf06fda405769366c59e8a85b42b103363d6 Mon Sep 17 00:00:00 2001 From: olddonkey Date: Sun, 23 Aug 2026 23:09:28 -0700 Subject: [PATCH 1/2] fix(codex): keep oversized Responses turns off the WS transport The Codex backend closes the socket on any inbound message of 16 MiB or more without sending a Responses terminal event, which reached clients as a bare 502 upstream_server_error. Because the wrapper only fell back to SSE when the *upgrade* failed, a thread that crossed the ceiling could never recover: every retry resent the same oversized frame. Measured against the live endpoint on 2026-08-23: 16,777,000 B completed, 16,777,300 B closed the socket in ~1s, reproducibly. The same body still succeeds over HTTP SSE, so the limit belongs to this transport alone. Size the `response.create` frame before dialing and take the SSE path when it does not fit. Deciding before the socket opens is what keeps the resend safe -- after open the caller already holds a streaming Response, and a retry there could double-generate the turn. Two supporting changes: - Carry the WS close code and reason into the stream error. A 1009 was previously indistinguishable from a network drop, and nothing in usage.jsonl or /api/logs recorded the real cause. - Apply the provider's `upstreamHttpVersion` pin to the SSE fallback. The fallback is a routine path now, and serving a turn over HTTP while silently dropping the operator's protocol pin is wrong. Co-Authored-By: Claude Fable 5 --- src/server/responses/fetch-helpers.ts | 19 +++-- src/server/responses/ws-upstream.ts | 67 ++++++++++++++++- tests/ws-upstream.test.ts | 102 ++++++++++++++++++++++++++ 3 files changed, 181 insertions(+), 7 deletions(-) diff --git a/src/server/responses/fetch-helpers.ts b/src/server/responses/fetch-helpers.ts index eb82200cd0..efc455797a 100644 --- a/src/server/responses/fetch-helpers.ts +++ b/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): void => { + base.preconnect?.(...args); + }; + const httpFetch = Object.assign( + (input: Parameters[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[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): void => { - base.preconnect?.(...args); - }; return Object.assign(wrapped, { preconnect, waitForPacing, diff --git a/src/server/responses/ws-upstream.ts b/src/server/responses/ws-upstream.ts index d17236d98c..9651c9ab98 100644 --- a/src/server/responses/ws-upstream.ts +++ b/src/server/responses/ws-upstream.ts @@ -28,6 +28,22 @@ 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. The margin only absorbs the difference between the measurement here +// and what a future caller might append after it. +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 +118,45 @@ 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 logs as the + * same bare 502. Naming the oversized case here is what makes it diagnosable + * without reading the client's own rollout files. + */ +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 +182,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 = {}; new Headers(init.headers ?? {}).forEach((value, key) => { // HTTP-body framing headers do not apply to a WS handshake. @@ -279,7 +342,7 @@ export function codexWsUpstreamFetch( } }); - ws.addEventListener("close", () => { + ws.addEventListener("close", (event: unknown) => { signal?.removeEventListener("abort", onAbort); if (!opened) { if (settledPreOpen) return; @@ -297,7 +360,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 */ } } }); diff --git a/tests/ws-upstream.test.ts b/tests/ws-upstream.test.ts index 2a6c747970..6063e9db27 100644 --- a/tests/ws-upstream.test.ts +++ b/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,102 @@ 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); + }); + + 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)"); + }); +}); From 566a4714d7bcce3b4d29d2df1a3e430db0f41853 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Tue, 25 Aug 2026 05:58:09 +0900 Subject: [PATCH 2/2] test(codex): pin the transport boundary at the adjacent byte MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sizing helper already had unit tests, but nothing proved the real serialized frame routes correctly one byte on each side of the limit. That gap matters because the request body is not the frame: `stream` is deleted and `type` is added before sending, so padding sized against the body sits eleven bytes away from what is actually transmitted. An off-by-one would live exactly there and pass every existing test. These two build padding so the serialized frame is exactly limit-1 and exactly limit, then assert the whole path: one socket and one send of the expected byte length under, zero sockets and one SSE call at it. Flipping the gate from >= to > fails the second one, so it catches a real off-by-one at the transport level rather than only in the helper. Two comment corrections while here. The close-code comment claimed the named 1009 message makes the failure diagnosable from the logs; it does not. The eager relay turns any stream error into a generic `upstream_reset` synthetic terminal without feeding it back through the inspector, so `/api/logs` retains only `streamAborted`. The message reaches the client and stops there, and the comment now says so rather than promising observability the code does not deliver. The margin comment described 64 KiB as absorbing a future append. There is no append. It is a conservative cushion, and the useful thing to record is what it actually covers: RFC 6455 framing is 14 bytes at this payload size — an 8-byte extended length plus a 4-byte client mask — so even a backend counting frame headers has ~65.5 KiB of room. Tests: 59 pass, 1 skip across ws-upstream, sse-failed-tail, and upstream-http-version. tsc --noEmit clean. --- src/server/responses/ws-upstream.ts | 20 ++++++--- tests/ws-upstream.test.ts | 63 +++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 5 deletions(-) diff --git a/src/server/responses/ws-upstream.ts b/src/server/responses/ws-upstream.ts index 9651c9ab98..275eff6dfb 100644 --- a/src/server/responses/ws-upstream.ts +++ b/src/server/responses/ws-upstream.ts @@ -37,8 +37,11 @@ export const MIN_BOUNDED_CODEX_WS_BUN_VERSION = "1.4.0"; // 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. The margin only absorbs the difference between the measurement here -// and what a future caller might append after it. +// 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; @@ -122,9 +125,16 @@ const CLOSED_BEFORE_TERMINAL = "codex websocket closed before a Responses termin /** * The close code is the only thing that separates "the backend refused this - * payload" from "the network dropped", and both used to reach the logs as the - * same bare 502. Naming the oversized case here is what makes it diagnosable - * without reading the client's own rollout files. + * 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; diff --git a/tests/ws-upstream.test.ts b/tests/ws-upstream.test.ts index 6063e9db27..2d7eba9a38 100644 --- a/tests/ws-upstream.test.ts +++ b/tests/ws-upstream.test.ts @@ -748,6 +748,69 @@ describe("oversized Codex create frames", () => { 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; + 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", {});