fix(responses): answer a wrapped WebSocket rejection with its HTTP status

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

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<string, unknown>): 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<string, unknown>)) {
+      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<Response> {
   const { session, url, init, prepared, sseFallback, onQuota, beforeDispatch } = options;
@@ -171,6 +215,18 @@ export function codexWsExchange(options: ExchangeOptions): Promise<Response> {
       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) {

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", {});