fix(codex): bound warmup response bodies

src/codex/auth-api.ts

@@ -98,7 +98,7 @@ import {
 } from "./main-account-cache";
 export { clearMainAccountInfoCache } from "./main-account-cache";
 import { maskEmail } from "../lib/privacy";
-import { CodexWarmupError, codexWarmupFailureReason, warmCodexAccount } from "./warmup";
+import { codexWarmupFailureReason, warmCodexAccount } from "./warmup";
 export { maskEmail } from "../lib/privacy";
 import type { CodexAccount, CodexAccountCredentials, OcxConfig } from "../types";
 import type { CatalogDisposition } from "./convergence-types";
@@ -396,13 +396,10 @@ async function verifyCodexAccountWarmup(
     return { ok: true, validatedAt: Date.now() };
   } catch (err) {
     const reason = codexWarmupFailureReason(err);
-    const upstream = err instanceof CodexWarmupError ? err.upstreamDetail : undefined;
     return {
       ok: false,
       response: jsonResponse({
-        error: upstream
-          ? `Codex account warmup failed: ${upstream}`
-          : "Codex account warmup failed. Reauthenticate the account and try again.",
+        error: "Codex account warmup failed. Reauthenticate the account and try again.",
         code: "codex_warmup_failed",
         reason,
         accountId,

src/codex/warmup.ts

@@ -1,19 +1,18 @@
+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;
 
   constructor(
     code: CodexWarmupError["code"],
     message = "Codex warmup failed",
-    options: { status?: number; cause?: unknown; upstreamDetail?: string } = {},
+    options: { status?: number; cause?: unknown } = {},
   ) {
     super(message);
     this.name = "CodexWarmupError";
     this.code = code;
     this.status = options.status;
-    this.upstreamDetail = options.upstreamDetail;
     if (options.cause !== undefined) this.cause = options.cause;
   }
 }
@@ -29,37 +28,31 @@ const CODEX_RESPONSES_URL = "https://chatgpt.com/backend-api/codex/responses";
 const DEFAULT_MODEL = "gpt-5.4-mini";
 const FALLBACK_MODELS = ["gpt-5.5"];
 const DEFAULT_TIMEOUT_MS = 30_000;
+const MAX_TIMEOUT_MS = 0x7fff_ffff;
 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> {
+/** Bound and release an upstream error body without exposing provider-controlled text. */
+async function drainErrorBody(res: Response, signal: AbortSignal): Promise<void> {
   try {
-    const text = await res.text();
-    const trimmed = text.slice(0, MAX_ERROR_BODY_BYTES);
-    try {
-      const json = JSON.parse(trimmed) as Record<string, unknown>;
-      // ChatGPT backend error shape: { error: { message: "..." } } or { detail: "..." }
-      const nested = json.error;
-      if (nested && typeof nested === "object" && typeof (nested as Record<string, unknown>).message === "string") {
-        return ((nested as Record<string, unknown>).message as string).slice(0, 512);
-      }
-      if (typeof json.detail === "string") return json.detail.slice(0, 512);
-      if (typeof json.error === "string") return (json.error as string).slice(0, 512);
-      if (typeof json.message === "string") return json.message.slice(0, 512);
-    } catch {
-      // Non-JSON response body may contain sensitive data (tokens, credentials).
-      // Only surface structured error messages, never raw text.
+    await readBoundedResponseBody(res, {
+      signal,
+      maxBytes: MAX_ERROR_BODY_BYTES,
+      fatalUtf8: true,
+    });
+  } catch (error) {
+    if (signal.aborted) {
+      throw new CodexWarmupError("transport", "Codex warmup request failed", {
+        cause: error,
+      });
     }
-    return undefined;
-  } catch {
-    return undefined;
+    // The bounded reader owns cancellation for oversized, invalid, or stalled bodies.
   }
 }
 
 function safeWarmupReason(err: unknown): string {
   if (err instanceof CodexWarmupError) {
-    const base = err.status ? `${err.code}:${err.status}` : err.code;
-    return err.upstreamDetail ? `${base} — ${err.upstreamDetail}` : base;
+    return err.status ? `${err.code}:${err.status}` : err.code;
   }
   return "transport";
 }
@@ -89,83 +82,196 @@ function parseSseFrame(frame: string): unknown | null {
   }
 }
 
-async function drainWarmupSse(body: ReadableStream<Uint8Array>): Promise<void> {
+async function drainWarmupSse(body: ReadableStream<Uint8Array>, signal: AbortSignal): Promise<void> {
   const reader = body.getReader();
   const decoder = new TextDecoder();
-  let buffer = "";
+  let buffer = new Uint8Array(Math.min(MAX_WARMUP_STREAM_BYTES, 64 * 1024));
+  let bufferedBytes = 0;
+  let scanOffset = 0;
+  let bytesRead = 0;
+  const abortError = () => new CodexWarmupError("transport", "Codex warmup request failed", {
+    cause: signal.reason,
+  });
+  const cancelReader = () => {
+    try {
+      void reader.cancel(signal.reason).catch(() => {});
+    } catch {
+      // Some custom stream implementations throw synchronously from cancel().
+    }
+  };
+  // Fetch implementations usually error the response body when their signal is
+  // aborted, but a ReadableStream is not intrinsically coupled to that signal.
+  // Race only the currently pending read against a removable abort listener;
+  // Bun 1.3 can leave read() parked until a custom source's cancel promise
+  // settles, while a shared never-settled race promise would retain one handler
+  // per chunk. Cancellation remains best-effort and is never awaited.
+  const readWithSignal = (): Promise<Awaited<ReturnType<typeof reader.read>>> => {
+    if (signal.aborted) {
+      cancelReader();
+      return Promise.reject(abortError());
+    }
+    const read = reader.read();
+    void read.catch(() => {});
+    return new Promise((resolve, reject) => {
+      let settled = false;
+      const finish = (action: () => void) => {
+        if (settled) return;
+        settled = true;
+        signal.removeEventListener("abort", onAbort);
+        action();
+      };
+      const onAbort = () => {
+        cancelReader();
+        finish(() => reject(abortError()));
+      };
+      signal.addEventListener("abort", onAbort, { once: true });
+      if (signal.aborted) {
+        onAbort();
+        return;
+      }
+      read.then(
+        result => finish(() => resolve(result)),
+        error => finish(() => reject(error)),
+      );
+    });
+  };
+  const ensureCapacity = (requiredBytes: number) => {
+    if (requiredBytes <= buffer.byteLength) return;
+    const grown = new Uint8Array(Math.min(
+      MAX_WARMUP_STREAM_BYTES,
+      Math.max(requiredBytes, buffer.byteLength * 2),
+    ));
+    grown.set(buffer.subarray(0, bufferedBytes));
+    buffer = grown;
+  };
+  const findFrameDelimiter = (start: number): { index: number; length: 2 | 3 | 4 } | undefined => {
+    for (let index = start; index < bufferedBytes - 1; index += 1) {
+      const firstLength = buffer[index] === 10
+        ? 1
+        : buffer[index] === 13 && buffer[index + 1] === 10 ? 2 : 0;
+      if (firstLength === 0) continue;
+      const secondStart = index + firstLength;
+      const secondLength = buffer[secondStart] === 10
+        ? 1
+        : buffer[secondStart] === 13 && buffer[secondStart + 1] === 10 ? 2 : 0;
+      if (secondLength > 0) return { index, length: (firstLength + secondLength) as 2 | 3 | 4 };
+    }
+    return undefined;
+  };
+  const acceptFrame = (frame: Uint8Array): boolean => {
+    const parsed = parseSseFrame(decoder.decode(frame));
+    const type = eventTypeFromData(parsed);
+    if (type === "response.completed") return true;
+    if (type === "response.failed") throw new CodexWarmupError("stream_failed");
+    if (type === "response.incomplete") throw new CodexWarmupError("stream_incomplete");
+    if (type === "error") throw new CodexWarmupError("stream_error");
+    return false;
+  };
 
   try {
+    if (signal.aborted) throw abortError();
     for (;;) {
-      const { done, value } = await reader.read();
+      const { done, value } = await readWithSignal();
+      if (signal.aborted) throw abortError();
       if (done) break;
-      buffer += decoder.decode(value, { stream: true });
+      if (value.byteLength > MAX_WARMUP_STREAM_BYTES - bytesRead) {
+        throw new CodexWarmupError("stream_too_large", "Codex warmup stream exceeded the size limit");
+      }
+      bytesRead += value.byteLength;
+      ensureCapacity(bufferedBytes + value.byteLength);
+      buffer.set(value, bufferedBytes);
+      bufferedBytes += value.byteLength;
 
+      let consumedBytes = 0;
       for (;;) {
-        const frameEnd = buffer.search(/\r?\n\r?\n/);
-        if (frameEnd < 0) break;
-        const frame = buffer.slice(0, frameEnd);
-        const delimiterLength = buffer[frameEnd] === "\r" ? 4 : 2;
-        buffer = buffer.slice(frameEnd + delimiterLength);
-        const parsed = parseSseFrame(frame);
-        const type = eventTypeFromData(parsed);
-        if (type === "response.completed") return;
-        if (type === "response.failed") throw new CodexWarmupError("stream_failed");
-        if (type === "response.incomplete") throw new CodexWarmupError("stream_incomplete");
-        if (type === "error") throw new CodexWarmupError("stream_error");
+        const delimiter = findFrameDelimiter(scanOffset);
+        if (!delimiter) {
+          // A delimiter can start at most three bytes before the next chunk.
+          scanOffset = Math.max(consumedBytes, bufferedBytes - 3);
+          break;
+        }
+        if (acceptFrame(buffer.subarray(consumedBytes, delimiter.index))) return;
+        consumedBytes = delimiter.index + delimiter.length;
+        scanOffset = consumedBytes;
+      }
+      if (consumedBytes > 0) {
+        buffer.copyWithin(0, consumedBytes, bufferedBytes);
+        bufferedBytes -= consumedBytes;
+        scanOffset = Math.max(0, scanOffset - consumedBytes);
       }
     }
 
-    if (buffer.trim()) {
-      const parsed = parseSseFrame(buffer);
-      const type = eventTypeFromData(parsed);
-      if (type === "response.completed") return;
-      if (type === "response.failed") throw new CodexWarmupError("stream_failed");
-      if (type === "response.incomplete") throw new CodexWarmupError("stream_incomplete");
-      if (type === "error") throw new CodexWarmupError("stream_error");
-    }
+    if (bufferedBytes > 0 && acceptFrame(buffer.subarray(0, bufferedBytes))) return;
 
     throw new CodexWarmupError("no_terminal", "Codex warmup ended before completion");
   } finally {
-    reader.releaseLock();
+    try {
+      reader.releaseLock();
+    } catch {
+      // A hostile cancel promise may keep the final read locked after timeout.
+    }
   }
 }
 
 async function tryWarmup(options: CodexWarmupOptions, model: string): Promise<void> {
-  let res: Response;
-  try {
-    res = await fetch(CODEX_RESPONSES_URL, {
-      method: "POST",
-      headers: {
-        Authorization: `Bearer ${options.accessToken}`,
-        "ChatGPT-Account-Id": options.chatgptAccountId,
-        "Content-Type": "application/json",
-      },
-      body: JSON.stringify({
-        model,
-        instructions: "Reply with OK.",
-        input: [{ type: "message", role: "user", content: [{ type: "input_text", text: "hi" }] }],
-        stream: true,
-        store: false,
-      }),
-      signal: AbortSignal.timeout(options.timeoutMs ?? DEFAULT_TIMEOUT_MS),
-    });
-  } catch (err) {
-    throw new CodexWarmupError("transport", "Codex warmup request failed", { cause: err });
-  }
-
-  if (!res.ok) {
-    const upstreamDetail = await readErrorDetail(res);
-    throw new CodexWarmupError("http_status", "Codex warmup was rejected", {
-      status: res.status,
-      upstreamDetail,
+  const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS;
+  if (!Number.isSafeInteger(timeoutMs) || timeoutMs < 0 || timeoutMs > MAX_TIMEOUT_MS) {
+    throw new CodexWarmupError("transport", "Codex warmup request failed", {
+      cause: new RangeError("Codex warmup timeout is outside the supported range"),
     });
   }
-  if (!res.body) throw new CodexWarmupError("missing_body");
+  // Bun 1.3 can leave AbortSignal.timeout() dormant while a custom response
+  // stream has a pending read. A ref'ed timer and explicit controller make the
+  // same deadline cover response headers and the full success/error body.
+  const deadline = new AbortController();
+  const signal = deadline.signal;
+  const timer = setTimeout(() => {
+    deadline.abort(new DOMException("Codex warmup timed out", "TimeoutError"));
+  }, timeoutMs);
 
   try {
-    await drainWarmupSse(res.body);
+    let res: Response;
+    try {
+      res = await fetch(CODEX_RESPONSES_URL, {
+        method: "POST",
+        headers: {
+          Authorization: `Bearer ${options.accessToken}`,
+          "ChatGPT-Account-Id": options.chatgptAccountId,
+          "Content-Type": "application/json",
+        },
+        body: JSON.stringify({
+          model,
+          instructions: "Reply with OK.",
+          input: [{ type: "message", role: "user", content: [{ type: "input_text", text: "hi" }] }],
+          stream: true,
+          store: false,
+        }),
+        signal,
+      });
+    } catch (err) {
+      throw new CodexWarmupError("transport", "Codex warmup request failed", { cause: err });
+    }
+
+    if (!res.ok) {
+      await drainErrorBody(res, signal);
+      throw new CodexWarmupError("http_status", "Codex warmup was rejected", {
+        status: res.status,
+      });
+    }
+    const body = res.body;
+    if (!body) throw new CodexWarmupError("missing_body");
+
+    try {
+      await drainWarmupSse(body, signal);
+    } finally {
+      try {
+        void body.cancel().catch(() => {});
+      } catch {
+        // Some custom stream implementations throw synchronously from cancel().
+      }
+    }
   } finally {
-    await res.body?.cancel().catch(() => {});
+    clearTimeout(timer);
   }
 }