fix(codex): bound warmup response bodies

src/codex/warmup.ts

@@ -1,5 +1,7 @@
+import { readBoundedResponseBody } from "../lib/bounded-body";
+
 export class CodexWarmupError extends Error {
-  code: "http_status" | "missing_body" | "stream_failed" | "stream_incomplete" | "stream_error" | "invalid_sse" | "no_terminal" | "transport";
+  code: "http_status" | "missing_body" | "stream_failed" | "stream_incomplete" | "stream_error" | "stream_too_large" | "invalid_sse" | "no_terminal" | "transport";
   status?: number;
   /** Upstream error detail extracted from the response body (truncated to 512 chars). */
   upstreamDetail?: string;
@@ -30,12 +32,18 @@ const DEFAULT_MODEL = "gpt-5.4-mini";
 const FALLBACK_MODELS = ["gpt-5.5"];
 const DEFAULT_TIMEOUT_MS = 30_000;
 const MAX_ERROR_BODY_BYTES = 2048;
+const MAX_WARMUP_STREAM_BYTES = 1024 * 1024;
 
 /** Read the first MAX_ERROR_BODY_BYTES of a response body and extract an error message. */
-async function readErrorDetail(res: Response): Promise<string | undefined> {
+async function readErrorDetail(res: Response, signal: AbortSignal): Promise<string | undefined> {
   try {
-    const text = await res.text();
-    const trimmed = text.slice(0, MAX_ERROR_BODY_BYTES);
+    const body = await readBoundedResponseBody(res, {
+      signal,
+      maxBytes: MAX_ERROR_BODY_BYTES,
+      fatalUtf8: true,
+    });
+    if (!body.displaySafe) return undefined;
+    const trimmed = body.text;
     try {
       const json = JSON.parse(trimmed) as Record<string, unknown>;
       // ChatGPT backend error shape: { error: { message: "..." } } or { detail: "..." }
@@ -93,11 +101,16 @@ async function drainWarmupSse(body: ReadableStream<Uint8Array>): Promise<void> {
   const reader = body.getReader();
   const decoder = new TextDecoder();
   let buffer = "";
+  let bytesRead = 0;
 
   try {
     for (;;) {
       const { done, value } = await reader.read();
       if (done) break;
+      if (value.byteLength > MAX_WARMUP_STREAM_BYTES - bytesRead) {
+        throw new CodexWarmupError("stream_too_large", "Codex warmup stream exceeded the size limit");
+      }
+      bytesRead += value.byteLength;
       buffer += decoder.decode(value, { stream: true });
 
       for (;;) {
@@ -131,8 +144,10 @@ async function drainWarmupSse(body: ReadableStream<Uint8Array>): Promise<void> {
 }
 
 async function tryWarmup(options: CodexWarmupOptions, model: string): Promise<void> {
+  let signal: AbortSignal;
   let res: Response;
   try {
+    signal = AbortSignal.timeout(options.timeoutMs ?? DEFAULT_TIMEOUT_MS);
     res = await fetch(CODEX_RESPONSES_URL, {
       method: "POST",
       headers: {
@@ -147,25 +162,30 @@ async function tryWarmup(options: CodexWarmupOptions, model: string): Promise<vo
         stream: true,
         store: false,
       }),
-      signal: AbortSignal.timeout(options.timeoutMs ?? DEFAULT_TIMEOUT_MS),
+      signal,
     });
   } catch (err) {
     throw new CodexWarmupError("transport", "Codex warmup request failed", { cause: err });
   }
 
   if (!res.ok) {
-    const upstreamDetail = await readErrorDetail(res);
+    const upstreamDetail = await readErrorDetail(res, signal);
     throw new CodexWarmupError("http_status", "Codex warmup was rejected", {
       status: res.status,
       upstreamDetail,
     });
   }
-  if (!res.body) throw new CodexWarmupError("missing_body");
+  const body = res.body;
+  if (!body) throw new CodexWarmupError("missing_body");
 
   try {
-    await drainWarmupSse(res.body);
+    await drainWarmupSse(body);
   } finally {
-    await res.body?.cancel().catch(() => {});
+    try {
+      void body.cancel().catch(() => {});
+    } catch {
+      // Some custom stream implementations throw synchronously from cancel().
+    }
   }
 }
 

tests/codex-warmup.test.ts

@@ -60,6 +60,41 @@ describe("codex warmup", () => {
       .rejects.toMatchObject({ name: "CodexWarmupError", code: "invalid_sse" });
   });
 
+  test("rejects an oversized unterminated SSE stream without waiting for cancellation", async () => {
+    let cancelled = false;
+    let closeTimer: ReturnType<typeof setTimeout> | undefined;
+    const oversizedBody = new ReadableStream<Uint8Array>({
+      start(controller) {
+        const chunk = new Uint8Array(256 * 1024).fill(65);
+        for (let index = 0; index < 5; index += 1) controller.enqueue(chunk);
+        closeTimer = setTimeout(() => controller.close(), 50);
+      },
+      cancel() {
+        cancelled = true;
+        if (closeTimer !== undefined) clearTimeout(closeTimer);
+        return new Promise<void>(() => {});
+      },
+    });
+    globalThis.fetch = (async () => new Response(oversizedBody, { status: 200 })) as typeof fetch;
+
+    await expect(warmCodexAccount({ accessToken: "a", chatgptAccountId: "c" }))
+      .rejects.toMatchObject({ name: "CodexWarmupError", code: "stream_too_large" });
+    expect(cancelled).toBe(true);
+  });
+
+  test("accepts a completed SSE stream at the exact byte limit", async () => {
+    const encoder = new TextEncoder();
+    const terminal = 'data: {"type":"response.completed"}\n\n';
+    const terminalBytes = encoder.encode(terminal).byteLength;
+    const fillerBytes = 1024 * 1024 - terminalBytes;
+    const filler = `:${"x".repeat(fillerBytes - 3)}\n\n`;
+    const stream = `${filler}${terminal}`;
+    expect(encoder.encode(stream).byteLength).toBe(1024 * 1024);
+    globalThis.fetch = (async () => sseResponse(stream)) as typeof fetch;
+
+    await expect(warmCodexAccount({ accessToken: "a", chatgptAccountId: "c" })).resolves.toBeUndefined();
+  });
+
   test("rejects EOF before success terminal", async () => {
     globalThis.fetch = (async () => sseResponse('event: response.created\ndata: {"type":"response.created"}\n\n')) as typeof fetch;
     await expect(warmCodexAccount({ accessToken: "a", chatgptAccountId: "c" }))
@@ -80,4 +115,9 @@ describe("codex warmup", () => {
       expect((err as Error).message).not.toContain("revoked");
     }
   });
+
+  test("classifies invalid timeout options as transport failures", async () => {
+    await expect(warmCodexAccount({ accessToken: "a", chatgptAccountId: "c", timeoutMs: -1 }))
+      .rejects.toMatchObject({ name: "CodexWarmupError", code: "transport" });
+  });
 });

tests/warmup.test.ts

@@ -64,6 +64,40 @@ describe("codex warmup improvements", () => {
     }
   });
 
+  test("warmCodexAccount discards oversized error details and cancels without waiting", async () => {
+    const encoder = new TextEncoder();
+    const detail = JSON.stringify({ detail: "must not surface" });
+    const firstChunk = encoder.encode(`${detail}${" ".repeat(1024 - detail.length)}`);
+    const paddingChunk = encoder.encode(" ".repeat(1024));
+    let cancelled = false;
+    let closeTimer: ReturnType<typeof setTimeout> | undefined;
+    const errorBody = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(firstChunk);
+        controller.enqueue(paddingChunk);
+        controller.enqueue(paddingChunk);
+        closeTimer = setTimeout(() => controller.close(), 50);
+      },
+      cancel() {
+        cancelled = true;
+        if (closeTimer !== undefined) clearTimeout(closeTimer);
+        return new Promise<void>(() => {});
+      },
+    });
+    globalThis.fetch = mock(async () => new Response(errorBody, { status: 401 })) as unknown as typeof fetch;
+
+    try {
+      await warmCodexAccount({ accessToken: "access-test", chatgptAccountId: "acct-test" });
+      throw new Error("expected warmup to reject");
+    } catch (err) {
+      expect(err).toBeInstanceOf(CodexWarmupError);
+      expect((err as CodexWarmupError).code).toBe("http_status");
+      expect((err as CodexWarmupError).upstreamDetail).toBeUndefined();
+      expect(codexWarmupFailureReason(err)).toBe("http_status:401");
+    }
+    expect(cancelled).toBe(true);
+  });
+
   test("warmCodexAccount retries FALLBACK_MODELS when the default model returns 400", async () => {
     const parsedBodies: Record<string, unknown>[] = [];
     const fetchMock = mock(async (_input: RequestInfo | URL, init?: RequestInit) => {