fix(core): abort stalled provider streams with header and chunk timeouts

The custom fetch on the OpenAI client only set the keep-alive agent. The
SDK's own timeout covers the wait for response headers only, so a stream
that went silent after the headers arrived - a gateway hiccup or a
black-holed connection - never aborted and the CLI spun until the user
interrupted it.

timeoutFetch() wraps the client's fetch with two bounds:

- A 300s response-header timeout, cleared once the headers arrive.
- A 300s per-chunk idle timeout for the response body; every received
  chunk resets it, so a healthy stream of any length is never cut off.

Both timers abort with a descriptive reason ("No response headers after
300s..." / "No data received for 300s while streaming..."), which
surfaces as the request's error message. The caller's own signal is
combined, and user interrupts keep working.

Author: sonderhq

packages/core/src/common/openai-client.ts

@@ -15,6 +15,117 @@ export { resolveOpenAIConnection, DEEPCODE_PLUS_BASE_URL } from "./plus-subscrip
 // keep connections reusable for three minutes after the last request.
 const keepAliveAgent = new Agent({ keepAliveTimeout: 180_000 });
 
+// A stalled provider connection used to hang a request forever: the OpenAI
+// SDK's own timeout only covers the wait for response headers, so a stream
+// that goes silent after the headers arrived (gateway hiccup, black-holed
+// connection) never aborts and the CLI spins until the user interrupts.
+// These defaults bound both waits; they mirror what other coding agents use.
+const STREAM_HEADER_TIMEOUT_MS = 300_000;
+const STREAM_CHUNK_TIMEOUT_MS = 300_000;
+
+type TimeoutFetchOptions = {
+  headerTimeoutMs?: number;
+  chunkTimeoutMs?: number;
+};
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+type FetchLike = (input: any, init?: any) => Promise<Response>;
+
+// Applies a response-header timeout and a per-chunk idle timeout at the fetch
+// layer. The header timer is cleared once headers arrive; after that every
+// received chunk resets the idle timer, so a healthy stream of any length is
+// never cut off — only a silent connection is. The caller's own signal (the
+// SDK's or an abort request) is combined, and our timers abort with a
+// descriptive reason that surfaces as the request's error message.
+export function timeoutFetch(fetchImpl: FetchLike, options: TimeoutFetchOptions = {}): FetchLike {
+  const headerTimeoutMs = options.headerTimeoutMs ?? STREAM_HEADER_TIMEOUT_MS;
+  const chunkTimeoutMs = options.chunkTimeoutMs ?? STREAM_CHUNK_TIMEOUT_MS;
+
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  return async (input: any, init?: any) => {
+    const opts = { ...(init ?? {}) };
+    const headerController = new AbortController();
+    const chunkController = new AbortController();
+    const signals = [...(opts.signal ? [opts.signal] : []), headerController.signal, chunkController.signal];
+    if (signals.length > 1) {
+      opts.signal = AbortSignal.any(signals);
+    } else if (signals.length === 1) {
+      opts.signal = signals[0];
+    }
+
+    const headerTimer = setTimeout(
+      () =>
+        headerController.abort(
+          new Error(`No response headers after ${Math.round(headerTimeoutMs / 1000)}s from the provider.`)
+        ),
+      headerTimeoutMs
+    );
+
+    let response: Response;
+    try {
+      response = await fetchImpl(input, opts);
+    } finally {
+      clearTimeout(headerTimer);
+    }
+    if (!response.body) {
+      return response;
+    }
+    return wrapBodyWithIdleTimeout(response, response.body, chunkController, chunkTimeoutMs);
+  };
+}
+
+function wrapBodyWithIdleTimeout(
+  response: Response,
+  upstream: ReadableStream<Uint8Array>,
+  controller: AbortController,
+  chunkTimeoutMs: number
+): Response {
+  let chunkTimer: ReturnType<typeof setTimeout> | undefined;
+  const clearChunkTimer = () => {
+    if (chunkTimer) {
+      clearTimeout(chunkTimer);
+      chunkTimer = undefined;
+    }
+  };
+  const scheduleChunkAbort = () => {
+    chunkTimer = setTimeout(
+      () =>
+        controller.abort(
+          new Error(`No data received for ${Math.round(chunkTimeoutMs / 1000)}s while streaming from the provider.`)
+        ),
+      chunkTimeoutMs
+    );
+  };
+  const upstreamReader = upstream.getReader();
+  const body = new ReadableStream<Uint8Array>({
+    async pull(downstream) {
+      try {
+        const { done, value } = await upstreamReader.read();
+        if (done) {
+          clearChunkTimer();
+          downstream.close();
+          return;
+        }
+        clearChunkTimer();
+        scheduleChunkAbort();
+        downstream.enqueue(value);
+      } catch (error) {
+        clearChunkTimer();
+        downstream.error(error);
+      }
+    },
+    cancel(reason) {
+      clearChunkTimer();
+      return upstreamReader.cancel(reason);
+    },
+  });
+  return new Response(body, {
+    status: response.status,
+    statusText: response.statusText,
+    headers: response.headers,
+  });
+}
+
 // Module-level cache for the OpenAI client instance.  The client itself is
 // a stateless fetch wrapper, so it is safe to share across calls as long as
 // the apiKey + baseURL stay the same.  Model, thinking-mode and other
@@ -102,7 +213,7 @@ export function createOpenAIClient(
     apiKey: connection.apiKey,
     baseURL: connection.baseURL || undefined,
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    fetch: (url: any, init: any) => undiciFetch(url, { ...init, dispatcher: keepAliveAgent }),
+    fetch: timeoutFetch((url: any, init: any) => undiciFetch(url, { ...init, dispatcher: keepAliveAgent })),
   });
   cachedOpenAIKey = cacheKey;
 

packages/core/src/tests/stream-timeout.test.ts

@@ -0,0 +1,83 @@
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { timeoutFetch } from "../common/openai-client";
+
+function hangingFetch(onSignal?: (signal: AbortSignal) => void) {
+  return (_input: unknown, init?: { signal?: AbortSignal }) =>
+    new Promise<Response>((_resolve, reject) => {
+      const signal = init?.signal;
+      if (!signal) return;
+      onSignal?.(signal);
+      signal.addEventListener("abort", () => reject(signal.reason));
+    });
+}
+
+function stalledStreamFetch(firstChunk: string, onSignal?: (signal: AbortSignal) => void) {
+  return (_input: unknown, init?: { signal?: AbortSignal }) => {
+    const encoder = new TextEncoder();
+    const signal = init?.signal;
+    onSignal?.(signal!);
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(encoder.encode(firstChunk));
+        signal?.addEventListener("abort", () => controller.error(signal.reason));
+      },
+    });
+    return Promise.resolve(new Response(body, { status: 200, headers: { "content-type": "text/event-stream" } }));
+  };
+}
+
+test("timeoutFetch passes a fast response through unchanged", async () => {
+  const fetchImpl = timeoutFetch(async () => new Response('{"ok":true}', { status: 200 }));
+  const res = await fetchImpl("https://api.example.com/v1/chat", { method: "POST" });
+  assert.equal(res.status, 200);
+  assert.equal(await res.text(), '{"ok":true}');
+});
+
+test("timeoutFetch keeps the caller's abort working", async () => {
+  let observed: AbortSignal | undefined;
+  const fetchImpl = timeoutFetch(hangingFetch((signal) => (observed = signal)));
+  const caller = new AbortController();
+  const pending = fetchImpl("https://api.example.com/v1/chat", { signal: caller.signal });
+  caller.abort(new Error("user interrupt"));
+  await assert.rejects(pending, /user interrupt/);
+  assert.ok(observed?.aborted);
+});
+
+test("timeoutFetch aborts when response headers never arrive", async () => {
+  const fetchImpl = timeoutFetch(hangingFetch(), { headerTimeoutMs: 20 });
+  await assert.rejects(fetchImpl("https://api.example.com/v1/chat", { method: "POST" }), /No response headers/);
+});
+
+test("timeoutFetch delivers chunks that keep arriving", async () => {
+  const encoder = new TextEncoder();
+  const fetchImpl = timeoutFetch(
+    async () => {
+      let sent = 0;
+      const body = new ReadableStream<Uint8Array>({
+        pull(controller) {
+          if (sent >= 5) {
+            controller.close();
+            return;
+          }
+          sent += 1;
+          controller.enqueue(encoder.encode(`chunk-${sent}\n\n`));
+        },
+      });
+      return new Response(body, { status: 200 });
+    },
+    { chunkTimeoutMs: 200 }
+  );
+  const res = await fetchImpl("https://api.example.com/v1/chat", { method: "POST" });
+  const text = await res.text();
+  assert.equal(text.split("\n\n").filter(Boolean).length, 5);
+});
+
+test("timeoutFetch aborts a stream that stalls between chunks", async () => {
+  const fetchImpl = timeoutFetch(stalledStreamFetch("data: hello\n\n"), { chunkTimeoutMs: 20 });
+  const res = await fetchImpl("https://api.example.com/v1/chat", { method: "POST" });
+  const reader = res.body!.getReader();
+  const first = await reader.read();
+  assert.equal(new TextDecoder().decode(first.value), "data: hello\n\n");
+  await assert.rejects(reader.read(), /No data received/);
+});