fix(sse): bound Codex SSE peek read with per-read timeout (#8020) (#8043)

peekCodexSseTransientError() ran before chatCore's normal
readiness/idle-timeout pipeline and read the first SSE chunk with a
bare reader.read() — no timeout wrapper. A 200 text/event-stream body
that never emitted a byte hung for ~15min (901399ms observed) before
the platform killed the connection and surfaced a generic 502.

Wrap the peek loop's read and the re-assembled passthrough body's
pull() in readStreamChunkWithTimeout, bounded PER READ (not a total
deadline) so a long-but-alive reasoning stream keeps resetting the
window on every chunk it emits. On timeout the reader is cancelled and
the request now fails fast with a 504 instead of hanging.

New small module open-sse/executors/codex/bodyTimeout.ts holds the
wrapping helpers to keep codex.ts within its frozen size baseline.

changelog.d/fixes/8020-codex-peek-read-timeout.md

@@ -0,0 +1 @@
+- fix(sse): bound the Codex SSE peek/passthrough body reads with a per-read timeout so a silently stalled upstream body settles in FETCH_BODY_TIMEOUT_MS instead of hanging ~15min and returning a generic 502 (#8020)

open-sse/executors/codex.ts

@@ -17,7 +17,8 @@ import {
   CODEX_CHAT_DEFAULT_INSTRUCTIONS,
   CODEX_DEFAULT_INSTRUCTIONS,
 } from "../config/codexInstructions.ts";
-import { HTTP_STATUS, PROVIDERS } from "../config/constants.ts";
+import { FETCH_BODY_TIMEOUT_MS, HTTP_STATUS, PROVIDERS } from "../config/constants.ts";
+import { readCodexPeekChunk, buildCodexTimeoutSafePassthroughBody } from "./codex/bodyTimeout.ts";
 import {
   getCodexClientVersion,
   getCodexUserAgent,
@@ -675,15 +676,23 @@ function extractCodexSseErrorMessage(text: string, fallback: string): string {
 }
 
 type CodexSseTransientErrorPeek =
-  | { matched: string; message: string; replacementBody: null }
-  | { matched: null; message: null; replacementBody: ReadableStream<Uint8Array> | null };
+  | { matched: string; message: string; replacementBody: null; timedOut?: false }
+  | {
+      matched: null;
+      message: null;
+      replacementBody: ReadableStream<Uint8Array> | null;
+      timedOut?: boolean;
+    };
 
 /**
  * Peek the first bytes of a Codex SSE response body looking for a transient
  * error embedded in an otherwise 200-OK stream. Exported for unit testing.
+ * `timeoutMs` bounds EACH individual read (#8020) — defaults to
+ * FETCH_BODY_TIMEOUT_MS; overridable so tests can settle fast/deterministically.
  */
 export async function peekCodexSseTransientError(
-  response: Response
+  response: Response,
+  timeoutMs: number = FETCH_BODY_TIMEOUT_MS
 ): Promise<CodexSseTransientErrorPeek> {
   const contentType = response.headers.get("content-type") || "";
   // #7536: check content-type BEFORE touching `response.body`. On the wreq-js
@@ -706,8 +715,12 @@ export async function peekCodexSseTransientError(
 
   try {
     while (text.length < CODEX_SSE_PEEK_MAX_BYTES) {
-      const { done, value } = await reader.read();
+      const { done, value, timedOut } = await readCodexPeekChunk(reader, timeoutMs);
+      if (timedOut) {
+        return { matched: null, message: null, replacementBody: null, timedOut: true };
+      }
       if (done) break;
+      if (!value) continue;
       chunks.push(value);
       text += decoder.decode(value, { stream: true });
       const lower = text.toLowerCase();
@@ -749,26 +762,7 @@ export async function peekCodexSseTransientError(
   // undici (every non-stream Codex request 502'd, then got mis-classified as a
   // 60s rate limit). Keep the original reader; never touch response.body again.
   const upstreamReader = reader;
-  const replacementBody = new ReadableStream<Uint8Array>({
-    start(controller) {
-      for (const chunk of chunks) controller.enqueue(chunk);
-    },
-    async pull(controller) {
-      const { done, value } = await upstreamReader.read();
-      if (done) {
-        controller.close();
-        return;
-      }
-      controller.enqueue(value);
-    },
-    cancel(reason) {
-      try {
-        upstreamReader.cancel(reason).catch(() => {});
-      } catch {
-        // noop — upstream socket may already be closing.
-      }
-    },
-  });
+  const replacementBody = buildCodexTimeoutSafePassthroughBody(chunks, upstreamReader, timeoutMs);
 
   return { matched: null, message: null, replacementBody };
 }
@@ -905,6 +899,17 @@ export class CodexExecutor extends BaseExecutor {
             HTTP_STATUS.SERVICE_UNAVAILABLE,
             peek.message
           );
+        } else if (peek.timedOut) {
+          // #8020: the peek's first-chunk read never returned (upstream body went
+          // silent). Convert to a bounded 504 instead of letting the caller hang.
+          input.log?.warn?.(
+            "TIMEOUT",
+            "CODEX | 200-OK SSE peek read timed out — upstream body stalled, returning 504"
+          );
+          (httpResult as { response: Response }).response = errorResponse(
+            HTTP_STATUS.GATEWAY_TIMEOUT,
+            "Upstream Codex SSE body read timed out"
+          );
         } else if (peek.replacementBody) {
           (httpResult as { response: Response }).response = new Response(peek.replacementBody, {
             status: resp.status,

open-sse/executors/codex/bodyTimeout.ts

@@ -0,0 +1,88 @@
+/**
+ * Per-read timeout helpers for the Codex SSE peek/passthrough body reads (#8020).
+ *
+ * `peekCodexSseTransientError()` in ../codex.ts reads the first bytes of a Codex
+ * SSE response body BEFORE the response reaches chatCore's normal readiness/idle
+ * pipeline, so a 200 text/event-stream whose body never emits a byte bypassed
+ * FETCH_BODY_TIMEOUT_MS / STREAM_IDLE_TIMEOUT_MS entirely and hung on a bare
+ * `reader.read()` for ~15 minutes before the platform killed the connection as a
+ * generic 502. These helpers wrap every read (the peek loop AND the re-assembled
+ * passthrough body's pull()) in `readStreamChunkWithTimeout`, PER READ rather than
+ * against a single total-request deadline, so a long-but-alive reasoning stream
+ * that keeps emitting chunks never trips the timeout — only a stream that goes
+ * silent for `timeoutMs` does.
+ */
+import { readStreamChunkWithTimeout } from "../../handlers/chatCore/upstreamTimeouts.ts";
+
+function isBodyTimeoutError(err: unknown): boolean {
+  return err instanceof Error && err.name === "BodyTimeoutError";
+}
+
+async function cancelReaderSafely(reader: ReadableStreamDefaultReader<Uint8Array>): Promise<void> {
+  try {
+    await reader.cancel();
+  } catch {
+    // Upstream socket may already be closing; nothing to clean up.
+  }
+}
+
+/**
+ * Reads the next peek-loop chunk under `timeoutMs`. On a `BodyTimeoutError` the
+ * upstream reader is cancelled (releasing the socket) and `timedOut: true` is
+ * returned instead of throwing, so the caller can short-circuit straight to a
+ * bounded error response rather than falling through to an unbounded passthrough.
+ */
+export async function readCodexPeekChunk(
+  reader: ReadableStreamDefaultReader<Uint8Array>,
+  timeoutMs: number
+): Promise<{ done: boolean; value?: Uint8Array; timedOut: boolean }> {
+  try {
+    const { done, value } = await readStreamChunkWithTimeout(reader, timeoutMs);
+    return { done, value, timedOut: false };
+  } catch (err) {
+    if (isBodyTimeoutError(err)) {
+      await cancelReaderSafely(reader);
+      return { done: true, timedOut: true };
+    }
+    throw err;
+  }
+}
+
+/**
+ * Builds the re-assembled Codex SSE body (peeked prefix chunks + continued drain
+ * of the same reader), with every subsequent read bounded by `timeoutMs`. A
+ * timeout on the passthrough cancels the upstream reader and errors the stream
+ * controller instead of hanging the client connection forever.
+ */
+export function buildCodexTimeoutSafePassthroughBody(
+  chunks: Uint8Array[],
+  upstreamReader: ReadableStreamDefaultReader<Uint8Array>,
+  timeoutMs: number
+): ReadableStream<Uint8Array> {
+  return new ReadableStream<Uint8Array>({
+    start(controller) {
+      for (const chunk of chunks) controller.enqueue(chunk);
+    },
+    async pull(controller) {
+      try {
+        const { done, value } = await readStreamChunkWithTimeout(upstreamReader, timeoutMs);
+        if (done) {
+          controller.close();
+          return;
+        }
+        if (!value) return;
+        controller.enqueue(value);
+      } catch (err) {
+        await cancelReaderSafely(upstreamReader);
+        controller.error(err);
+      }
+    },
+    cancel(reason) {
+      try {
+        upstreamReader.cancel(reason).catch(() => {});
+      } catch {
+        // noop — upstream socket may already be closing.
+      }
+    },
+  });
+}

tests/unit/bug-8020-codex-peek-body-timeout.test.ts

@@ -0,0 +1,71 @@
+// #8020: `peekCodexSseTransientError()`'s first-chunk read (open-sse/executors/codex.ts) had
+// NO timeout wrapper — a 200 text/event-stream response whose body never emits a byte hung on a
+// bare `reader.read()` for ~15 minutes (901399ms observed) before the platform killed the
+// connection and surfaced a generic 502. This runs BEFORE chatCore's normal readiness/idle-timeout
+// pipeline takes over, so FETCH_BODY_TIMEOUT_MS / STREAM_IDLE_TIMEOUT_MS never applied to it.
+//
+// Fix: every read in the peek loop and the re-assembled passthrough body is now bounded by a
+// PER-READ timeout (open-sse/executors/codex/bodyTimeout.ts), so a stalled body settles fast
+// instead of hanging, while a long-but-alive stream that keeps emitting chunks never trips it.
+import test from "node:test";
+import assert from "node:assert/strict";
+
+import { peekCodexSseTransientError } from "../../open-sse/executors/codex.ts";
+
+// Small, explicit override — never depends on the ~120s/600s production default, so this test
+// settles fast and deterministically regardless of env configuration.
+const TEST_TIMEOUT_MS = 200;
+
+function stuckSseResponse(): Response {
+  const stuck = new ReadableStream<Uint8Array>({
+    pull() {
+      // Never enqueue and never close — simulates an upstream body that goes silent.
+    },
+  });
+  return new Response(stuck, {
+    status: 200,
+    headers: { "content-type": "text/event-stream" },
+  });
+}
+
+test(
+  "peekCodexSseTransientError does not hang forever on a silently stuck SSE body (#8020)",
+  { timeout: 5000 },
+  async () => {
+    const start = Date.now();
+    const peek = await peekCodexSseTransientError(stuckSseResponse(), TEST_TIMEOUT_MS);
+    const elapsed = Date.now() - start;
+
+    assert.ok(
+      elapsed < 2000,
+      `expected peek to settle within 2000ms via a body timeout, took ${elapsed}ms`
+    );
+    assert.equal(peek.timedOut, true, "expected the peek to report timedOut on a stalled body");
+    assert.equal(peek.matched, null);
+    assert.equal(peek.replacementBody, null);
+  }
+);
+
+test(
+  "peekCodexSseTransientError still detects a transient error when the body responds promptly",
+  async () => {
+    const encoder = new TextEncoder();
+    const response = new Response(
+      new ReadableStream<Uint8Array>({
+        pull(controller) {
+          controller.enqueue(
+            encoder.encode(
+              'event: error\ndata: {"error":{"message":"Selected model is at capacity."}}\n\n'
+            )
+          );
+          controller.close();
+        },
+      }),
+      { status: 200, headers: { "content-type": "text/event-stream" } }
+    );
+
+    const peek = await peekCodexSseTransientError(response, TEST_TIMEOUT_MS);
+    assert.equal(peek.timedOut ?? false, false);
+    assert.match(peek.matched ?? "", /capacity/);
+  }
+);