diff --git a/docs-site/src/content/docs/reference/architecture.md b/docs-site/src/content/docs/reference/architecture.md index 96a3aec6dc..2bf2ae1e37 100644 --- a/docs-site/src/content/docs/reference/architecture.md +++ b/docs-site/src/content/docs/reference/architecture.md @@ -150,8 +150,12 @@ and derives its routing hint from the actual outgoing model and service tier. Initial upstream quota/model metadata becomes bounded HTTP response headers; later quota updates are attributed to the serving account, not retroactively added to headers already sent. A failure after a WS request was sent does not -trigger an automatic HTTP resend. These mappings do not enable the client-facing -WebSocket setting or change other providers' transport selection. +trigger an automatic HTTP resend. A refusal the backend sends before any output, as one +`error` frame carrying a 4xx `status_code`, is returned as that HTTP status: the frame's +headers are copied except framing and encoding ones, `content-type` is `application/json`, +`cache-control` is `no-store`, and the body is `{"error": ...}` JSON, so the same refresh, +quota and account-rotation handling applies as on the HTTP path. These mappings do not enable +the client-facing WebSocket setting or change other providers' transport selection. Bundled Bun 1.3.14, prereleases, and unverifiable runtime identities use HTTP/SSE. Successful upstream WS responses keep the downstream SSE contract and bypass `tee()` through a bounded eager single-reader relay (4 MiB per raw/enveloped frame and an 8 MiB producer queue). Queue overflow diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index 2f41be02fa..f8f0a86776 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -16,6 +16,50 @@ interface ExchangeOptions { beforeDispatch?: (headers: Headers) => void; } +// The frame's headers describe the upstream's HTTP representation. The body is +// re-encoded as plain JSON below, so the framing and encoding headers would lie. +const REBUILT_REJECTION_DROPPED_HEADERS = new Set([ + "content-encoding", "content-length", "content-type", "transfer-encoding", "connection", "keep-alive", +]); + +/** + * Turn a refused create frame back into the HTTP error it stands for. + * + * When the backend rejects a turn before it starts (an expired token, a usage + * limit), it does not open a response. It sends one `error` frame carrying a + * `status_code`, the error object and the response headers, then closes. Codex's + * own WebSocket client maps that frame to the HTTP error (codex-api + * `responses_websocket.rs`, `map_wrapped_websocket_error_event`). Relayed as an + * SSE `error` event it is lost twice: the relay does not count `error` as a + * terminal and appends an `adapter_eof` incomplete, and Codex's SSE parser has no + * arm for `error`, so the client sees only the `adapter_eof` (#3029) and the + * pre-stream quota, refresh and rotation handlers never see the status. + * + * 4xx only: those refusals come before generation, so answering with the status + * cannot double-generate. 5xx keeps the stream path and its no-resend rule. This + * reaches only the canonical ChatGPT lane: an opt-in `upstreamWebsocket` provider + * commits its response on send, before any frame can be inspected. + */ +function wrappedRejectionResponse(payload: Record): Response | null { + const raw = payload.status_code ?? payload.status; + if (typeof raw !== "number" || !Number.isInteger(raw) || raw < 400 || raw > 499) return null; + const headers = new Headers(); + const source = payload.headers; + if (source && typeof source === "object" && !Array.isArray(source)) { + for (const [name, value] of Object.entries(source as Record)) { + if (typeof value !== "string" || REBUILT_REJECTION_DROPPED_HEADERS.has(name.toLowerCase())) continue; + try { headers.set(name, value); } catch { /* an invalid upstream header name/value is not ours to relay */ } + } + } + headers.set("content-type", "application/json"); + // The body is account-specific and rebuilt here, never a cacheable representation. + headers.set("cache-control", "no-store"); + const error = payload.error && typeof payload.error === "object" && !Array.isArray(payload.error) + ? payload.error + : { type: "upstream_error", message: "Upstream rejected the request" }; + return new Response(JSON.stringify({ error }), { status: raw, headers }); +} + /** The sole SSE exchange state machine for both one-shot and retained sockets. */ export function codexWsExchange(options: ExchangeOptions): Promise { const { session, url, init, prepared, sseFallback, onQuota, beforeDispatch } = options; @@ -171,6 +215,18 @@ export function codexWsExchange(options: ExchangeOptions): Promise { const normalized = normalizeResponsesWsRelayEvent(text); if (!normalized) return; const { type } = normalized; + // Nothing has been committed to the client yet, so a wrapped rejection + // can still become the real HTTP response instead of a 200 stream. + if (!responseCommitted && type === "error") { + const rejection = wrappedRejectionResponse(normalized.payload); + if (rejection) { + terminal = true; + cleanup(); + session.dispose(); + resolve(rejection); + return; + } + } let relayText = normalized.text; let controlFrame = false; if (metadata) { diff --git a/tests/responses/ws-upstream.test.ts b/tests/responses/ws-upstream.test.ts index cfb087a4bb..4aa4de0c14 100644 --- a/tests/responses/ws-upstream.test.ts +++ b/tests/responses/ws-upstream.test.ts @@ -637,9 +637,96 @@ describe("codexWsUpstreamFetch", () => { expect(FakeWebSocket.instances[0].closed).toBe(true); }); + // A refused create frame arrives as one `error` frame with the HTTP status and + // headers, then a close. Codex's SSE parser has no `error` arm, so relaying it + // in-band reaches the client as nothing but the synthesized adapter_eof. + test("answers a wrapped 4xx rejection with its HTTP status instead of a 200 stream", async () => { + const frame = { + type: "error", + error: { type: "usage_limit_reached", message: "The usage limit has been reached", plan_type: "plus", resets_at: 1_800_000_000 }, + status_code: 429, + headers: { + "X-Codex-Primary-Used-Percent": "100", + "X-Codex-Primary-Reset-At": "1800000000", + "Content-Length": "174", + "Content-Type": "application/json", + "Content-Encoding": "gzip", + }, + }; + installFake(ws => { + ws.emit("open", {}); + ws.emit("message", { data: JSON.stringify(frame) }); + ws.emit("close", { code: 1000, reason: "normal" }); + }); + const response = await codexWsUpstreamFetch(CODEX_URL, streamingInit(), (() => { + throw new Error("a sent frame must not be resent over HTTP"); + }) as unknown as typeof fetch); + + expect(response.status).toBe(429); + expect(response.headers.get("content-type")).toBe("application/json"); + expect(response.headers.get("x-codex-primary-used-percent")).toBe("100"); + expect(response.headers.get("x-codex-primary-reset-at")).toBe("1800000000"); + // The body is re-encoded as plain JSON, so the frame's framing and encoding + // headers must not survive, and the account-specific result must not be cached. + expect(response.headers.get("content-length")).toBeNull(); + expect(response.headers.get("content-encoding")).toBeNull(); + expect(response.headers.get("cache-control")).toBe("no-store"); + expect(await response.json()).toEqual({ error: frame.error }); + expect(isCodexWsUpstreamResponse(response)).toBe(false); + const ws = FakeWebSocket.instances[0]!; + expect(ws.sent).toHaveLength(1); + expect(ws.closed).toBe(true); + }); + + test("honors the status field spelling Codex accepts", async () => { + installFake(ws => { + ws.emit("open", {}); + ws.emit("message", { + data: JSON.stringify({ type: "error", status: 401, error: { type: "invalid_token", message: "expired" } }), + }); + }); + const response = await codexWsUpstreamFetch(CODEX_URL, streamingInit(), (() => { + throw new Error("fallback must not run after open"); + }) as unknown as typeof fetch); + expect(response.status).toBe(401); + expect(await response.json()).toEqual({ error: { type: "invalid_token", message: "expired" } }); + }); + + test("keeps a wrapped error on the stream once output has been committed", async () => { + installFake(ws => { + ws.emit("open", {}); + ws.emit("message", { data: JSON.stringify({ type: "response.created", response: { id: "r1" } }) }); + ws.emit("message", { + data: JSON.stringify({ type: "error", status_code: 429, error: { type: "usage_limit_reached", message: "mid-turn" } }), + }); + }); + const response = await codexWsUpstreamFetch(CODEX_URL, streamingInit(), (() => { + throw new Error("fallback must not run after open"); + }) as unknown as typeof fetch); + expect(response.status).toBe(200); + const text = await response.text(); + expect(text).toContain("event: response.created"); + expect(text).toContain("event: error"); + expect(text).toContain("mid-turn"); + }); + + test("leaves a wrapped 5xx on the stream path", async () => { + installFake(ws => { + ws.emit("open", {}); + ws.emit("message", { + data: JSON.stringify({ type: "error", status_code: 502, error: { type: "server_error", message: "bad gateway" } }), + }); + }); + const response = await codexWsUpstreamFetch(CODEX_URL, streamingInit(), (() => { + throw new Error("fallback must not run after open"); + }) as unknown as typeof fetch); + expect(response.status).toBe(200); + expect(await response.text()).toContain("event: error"); + }); + test.each(["error", "response.completed"])("multiline upstream %s JSON remains one valid SSE data value", async type => { const payload = type === "error" - ? { type, status: 400, error: { type: "invalid_request_error", message: "fixture refusal" } } + ? { type, error: { type: "invalid_request_error", message: "fixture refusal" } } : { type, response: { id: "pretty-response", status: "completed", output: [] } }; installFake(ws => { ws.emit("open", {});