fix(sse): bound the chatgpt-session stream-open gate with a deadline

The gate withholds the HTTP response until the first event that also counts
as committed output on the buffered path, so both paths agree on failover
status. While it is closed nothing reaches the client — not even a keepalive,
since the first byte commits the 200 — and a turn here runs in a real browser
that can think for a long time, so a client with an idle timeout could hang up
on a perfectly healthy turn.

Race the gate against CHATGPT_SESSION_STREAM_OPEN_TIMEOUT_MS (30s, overridable
per call through the new options argument). A committing event or an error
still wins exactly as before; only when the deadline elapses with neither does
the stream open anyway and keep consuming the same iterator, replaying the
buffered reasoning after the role chunk so nothing is lost or reordered. The
in-flight next() the deadline outran is carried into the stream body instead of
being abandoned, otherwise its event would vanish behind a second next(). The
timer is unref'd and cleared on every exit path.

Every failure that needs a real HTTP status — no browser, missing or expired
credentials, rate limiting, an incompatible route — surfaces within seconds,
far inside the window.

open-sse/executors/chatgpt-session/bridge.ts

@@ -8,6 +8,11 @@
  * (heartbeats, assistant boundaries), commentary-phase text, empty text deltas and reasoning are
  * all non-committing on both paths. Once real content has been emitted the status line is
  * already committed, so a later failure just closes the stream cleanly.
+ *
+ * The gate is bounded: nothing at all can reach the client while it is closed (not even a
+ * keepalive, since the first byte commits the 200), and a turn here runs in a real browser that
+ * may think for a long time. Past `CHATGPT_SESSION_STREAM_OPEN_TIMEOUT_MS` the stream opens
+ * anyway and keeps consuming the same iterator, so a slow-but-healthy turn stays connected.
  */
 
 import { buildErrorBody, sanitizeErrorMessage } from "../../utils/error.ts";
@@ -43,6 +48,30 @@ export type ChatGptSessionStreamOpen =
  */
 const COMMENTARY_PHASE: CodexMessagePhase = "commentary";
 
+/**
+ * How long the stream-open gate waits for the first committing event before opening the stream
+ * anyway.
+ *
+ * The trade-off: while the gate is closed the client sees nothing, so an idle-timeout client can
+ * hang up on a healthy turn that is simply thinking for a long time in the browser. Opening the
+ * stream commits the 200, which costs the ability to answer with a real HTTP status if the turn
+ * dies later — but every failure that needs a real status (no browser, missing or expired
+ * credentials, rate limiting, an incompatible route) surfaces within seconds, far inside this
+ * window. Only a genuinely long-running healthy turn reaches the deadline.
+ *
+ * Callers can override it per request through `options.streamOpenTimeoutMs`; a value that is not
+ * a finite number greater than zero disables the deadline and the gate waits indefinitely.
+ */
+export const CHATGPT_SESSION_STREAM_OPEN_TIMEOUT_MS = 30_000;
+
+/** Race marker for the gate deadline — distinguishable from any `IteratorResult`. */
+const GATE_TIMED_OUT = Symbol("chatgpt-session-gate-timeout");
+
+export interface ChatGptSessionStreamOptions {
+  /** Overrides {@link CHATGPT_SESSION_STREAM_OPEN_TIMEOUT_MS} for this call (tests, tuning). */
+  streamOpenTimeoutMs?: number;
+}
+
 interface OpenAiUsage {
   prompt_tokens: number;
   completion_tokens: number;
@@ -88,7 +117,8 @@ function finishReasonFor(event: AdapterEvent): string {
 
 export async function openChatGptSessionStream(
   events: AsyncIterable<AdapterEvent>,
-  meta: ChatGptSessionResponseMeta
+  meta: ChatGptSessionResponseMeta,
+  options?: ChatGptSessionStreamOptions
 ): Promise<ChatGptSessionStreamOpen> {
   const iterator = events[Symbol.asyncIterator]();
   // Reasoning that arrives while the gate is still closed is real output the client must
@@ -96,19 +126,50 @@ export async function openChatGptSessionStream(
   // absent CONTENT regardless of how much reasoning preceded it.
   const bufferedReasoning: AdapterEvent[] = [];
   let first: AdapterEvent | null = null;
+  // The `iterator.next()` the deadline outran. It is still in flight and will settle with the
+  // event the gate never saw, so the stream body must await THIS promise instead of asking the
+  // iterator for another one — a second `next()` would queue behind it and the first event would
+  // be lost with the abandoned promise.
+  let pendingNext: Promise<IteratorResult<AdapterEvent>> | null = null;
 
-  for (;;) {
-    const next = await iterator.next();
-    if (next.done) break;
-    const event = next.value;
-    if (event.type === "heartbeat" || event.type === "assistant_boundary") continue;
-    if (event.type === "text_delta" && (!event.text || event.phase === COMMENTARY_PHASE)) continue;
-    if (event.type === "thinking_delta") {
-      bufferedReasoning.push(event);
-      continue;
+  const timeoutMs = options?.streamOpenTimeoutMs ?? CHATGPT_SESSION_STREAM_OPEN_TIMEOUT_MS;
+  const bounded = Number.isFinite(timeoutMs) && timeoutMs > 0;
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  const deadline = bounded
+    ? new Promise<typeof GATE_TIMED_OUT>((resolve) => {
+        timer = setTimeout(() => resolve(GATE_TIMED_OUT), timeoutMs);
+        // A gate deadline must never be the reason the process stays alive.
+        (timer as unknown as { unref?: () => void }).unref?.();
+      })
+    : null;
+
+  try {
+    for (;;) {
+      const step = iterator.next();
+      const settled: IteratorResult<AdapterEvent> | typeof GATE_TIMED_OUT = deadline
+        ? await Promise.race([step, deadline])
+        : await step;
+      if (settled === GATE_TIMED_OUT) {
+        pendingNext = step;
+        break;
+      }
+      if (settled.done) break;
+      const event = settled.value;
+      if (event.type === "heartbeat" || event.type === "assistant_boundary") continue;
+      if (event.type === "text_delta" && (!event.text || event.phase === COMMENTARY_PHASE)) {
+        continue;
+      }
+      if (event.type === "thinking_delta") {
+        bufferedReasoning.push(event);
+        continue;
+      }
+      first = event;
+      break;
     }
-    first = event;
-    break;
+  } finally {
+    // Every exit path clears the timer, so a pending deadline can neither fire after the gate
+    // resolved nor hold a handle open.
+    if (timer !== undefined) clearTimeout(timer);
   }
 
   if (first && first.type === "error") {
@@ -181,7 +242,9 @@ export async function openChatGptSessionStream(
             if (!open) break;
           }
           while (open) {
-            const next = await iterator.next();
+            const step = pendingNext ?? iterator.next();
+            pendingNext = null;
+            const next = await step;
             if (next.done) {
               emit(chunk(meta, {}, "stop"));
               break;

tests/unit/chatgpt-session-bridge.test.ts

@@ -2,6 +2,7 @@ import { test } from "node:test";
 import assert from "node:assert/strict";
 
 import {
+  CHATGPT_SESSION_STREAM_OPEN_TIMEOUT_MS,
   buildChatGptSessionCompletion,
   openChatGptSessionStream,
 } from "../../open-sse/executors/chatgpt-session/bridge.ts";
@@ -331,3 +332,109 @@ test("a cooldown-hinted classification reaches the stream-open error verdict", a
   assert.equal(opened.kind, "error");
   assert.equal((opened as { fallbackHint?: string }).fallbackHint, "connection_cooldown");
 });
+
+// I2 — the gate must be bounded: while it is closed the client receives nothing at all (any byte
+// commits the 200), so a healthy turn that thinks for a long time in the browser would otherwise
+// look dead to a client with an idle timeout.
+interface ControllableSource {
+  events: AsyncIterable<AdapterEvent>;
+  push(event: AdapterEvent): void;
+  end(): void;
+}
+
+function controllable(): ControllableSource {
+  const queued: AdapterEvent[] = [];
+  const waiting: Array<(result: IteratorResult<AdapterEvent>) => void> = [];
+  let ended = false;
+  const iterator: AsyncIterator<AdapterEvent> = {
+    next() {
+      const event = queued.shift();
+      if (event) return Promise.resolve({ value: event, done: false });
+      if (ended) return Promise.resolve({ value: undefined, done: true });
+      return new Promise((resolve) => waiting.push(resolve));
+    },
+  };
+  return {
+    events: { [Symbol.asyncIterator]: () => iterator },
+    push(event) {
+      const waiter = waiting.shift();
+      if (waiter) waiter({ value: event, done: false });
+      else queued.push(event);
+    },
+    end() {
+      ended = true;
+      for (const waiter of waiting.splice(0)) waiter({ value: undefined, done: true });
+    },
+  };
+}
+
+test("the default stream-open deadline stays at 30s", () => {
+  assert.equal(CHATGPT_SESSION_STREAM_OPEN_TIMEOUT_MS, 30_000);
+});
+
+test("a silent turn opens the stream once the gate deadline elapses", async () => {
+  const source = controllable();
+  const opened = await openChatGptSessionStream(source.events, META, {
+    streamOpenTimeoutMs: 5,
+  });
+  assert.equal(opened.kind, "stream");
+  const reading = readAll((opened as { stream: ReadableStream<Uint8Array> }).stream);
+  source.push({ type: "heartbeat" });
+  source.push({ type: "text_delta", text: "late answer" });
+  source.push({ type: "done" });
+  source.end();
+  const text = await reading;
+  assert.match(text, /"delta":\{"role":"assistant"\}/);
+  assert.match(text, /"content":"late answer"/);
+  assert.match(text, /"finish_reason":"stop"/);
+  // The role chunk still leads, and heartbeats keep flowing as SSE comments once the gate opened.
+  assert.ok(text.indexOf('"role":"assistant"') < text.indexOf('"content":"late answer"'));
+  assert.match(text, /^: keepalive$/m);
+  assert.ok(text.trimEnd().endsWith("data: [DONE]"));
+});
+
+test("reasoning buffered before the deadline is replayed in order after the gate opens", async () => {
+  const source = controllable();
+  source.push({ type: "thinking_delta", thinking: "first" });
+  source.push({ type: "thinking_delta", thinking: "second" });
+  const opened = await openChatGptSessionStream(source.events, META, {
+    streamOpenTimeoutMs: 5,
+  });
+  assert.equal(opened.kind, "stream");
+  const reading = readAll((opened as { stream: ReadableStream<Uint8Array> }).stream);
+  source.push({ type: "text_delta", text: "answer" });
+  source.push({ type: "done" });
+  source.end();
+  const text = await reading;
+  const roleAt = text.indexOf('"role":"assistant"');
+  const firstAt = text.indexOf('"reasoning_content":"first"');
+  const secondAt = text.indexOf('"reasoning_content":"second"');
+  const answerAt = text.indexOf('"content":"answer"');
+  assert.ok(roleAt >= 0 && firstAt > roleAt && secondAt > firstAt && answerAt > secondAt);
+});
+
+test("an error before the deadline still returns an error verdict, never a timed-out stream", async () => {
+  const source = controllable();
+  const opening = openChatGptSessionStream(source.events, META, {
+    streamOpenTimeoutMs: 1_000,
+  });
+  source.push({ type: "error", message: "ChatGPT reported a usage limit" });
+  const opened = await opening;
+  assert.equal(opened.kind, "error");
+  assert.equal((opened as { status: number }).status, 429);
+});
+
+test("no event is lost to the deadline race", async () => {
+  const source = controllable();
+  const opened = await openChatGptSessionStream(source.events, META, {
+    streamOpenTimeoutMs: 5,
+  });
+  const reading = readAll((opened as { stream: ReadableStream<Uint8Array> }).stream);
+  source.push({ type: "text_delta", text: "one" });
+  source.push({ type: "text_delta", text: "two" });
+  source.push({ type: "done" });
+  source.end();
+  const text = await reading;
+  assert.match(text, /"content":"one"/);
+  assert.match(text, /"content":"two"/);
+});