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..275eff6dfb 100644 --- a/src/server/responses/ws-upstream.ts +++ b/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 = {}; 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 */ } } }); diff --git a/tests/ws-upstream.test.ts b/tests/ws-upstream.test.ts index 2a6c747970..2d7eba9a38 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,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; + 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)"); + }); +});