From e11c8603f351022f777ccb0fbf89962953ef2432 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Sat, 15 Aug 2026 10:09:58 +0900 Subject: [PATCH] fix(codex): bound warmup response bodies --- src/codex/warmup.ts | 38 +++++++++++++++++++++++++++--------- tests/codex-warmup.test.ts | 40 ++++++++++++++++++++++++++++++++++++++ tests/warmup.test.ts | 34 ++++++++++++++++++++++++++++++++ 3 files changed, 103 insertions(+), 9 deletions(-) diff --git a/src/codex/warmup.ts b/src/codex/warmup.ts index 9278a3d548..9429c5cd19 100644 --- a/src/codex/warmup.ts +++ b/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 { +async function readErrorDetail(res: Response, signal: AbortSignal): Promise { 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; // ChatGPT backend error shape: { error: { message: "..." } } or { detail: "..." } @@ -93,11 +101,16 @@ async function drainWarmupSse(body: ReadableStream): Promise { 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): Promise { } async function tryWarmup(options: CodexWarmupOptions, model: string): Promise { + 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 {}); + try { + void body.cancel().catch(() => {}); + } catch { + // Some custom stream implementations throw synchronously from cancel(). + } } } diff --git a/tests/codex-warmup.test.ts b/tests/codex-warmup.test.ts index b99eab5604..ef6ae7f3fc 100644 --- a/tests/codex-warmup.test.ts +++ b/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 | undefined; + const oversizedBody = new ReadableStream({ + 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(() => {}); + }, + }); + 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" }); + }); }); diff --git a/tests/warmup.test.ts b/tests/warmup.test.ts index ddcd7ca13f..d8ddbe2563 100644 --- a/tests/warmup.test.ts +++ b/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 | undefined; + const errorBody = new ReadableStream({ + 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(() => {}); + }, + }); + 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[] = []; const fetchMock = mock(async (_input: RequestInfo | URL, init?: RequestInit) => {