fix(network): bound direct-path response-start timeout

.env.example

@@ -1283,6 +1283,14 @@ CURSOR_USER_AGENT="Cursor/3.4"
 # FETCH_BODY_TIMEOUT_MS=600000       # Time to receive full response body
 # FETCH_CONNECT_TIMEOUT_MS=30000     # TCP connection establishment (default: 30s)
 # FETCH_KEEPALIVE_TIMEOUT_MS=4000    # Keep-alive socket idle timeout (default: 4s)
+# OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS=30000  # Bounded response-start window per direct
+#                                     # (no-proxy) attempt (#10214). A silently-dropped
+#                                     # pooled keep-alive socket surfaces no transport
+#                                     # error, so without this bound a direct request can
+#                                     # stall until undici's headersTimeout (600s) or the
+#                                     # caller's deadline; on expiry the request retries
+#                                     # once on a fresh no-keep-alive socket. 0 disables
+#                                     # the bound (default: 30000 = 30s).
 
 # Default timeout (ms) for src/shared/utils/fetchTimeout.ts. Acts as the
 # fallback when FETCH_TIMEOUT_MS is unset. Default: 120000 (2 min).

changelog.d/fixes/10528-direct-dispatcher-response-start-timeout.md

@@ -0,0 +1 @@
+- **fix(network):** direct (no-proxy) egress now bounds each attempt's response-start window (default 30s, `OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS`) and retries once on a fresh no-keep-alive socket, so a silently-dropped pooled keep-alive connection can no longer stall direct providers (opencode-go, command-code) until a service restart ([#10214](https://github.com/diegosouzapw/OmniRoute/issues/10214))

docs/reference/ENVIRONMENT.md

@@ -698,6 +698,7 @@ REQUEST_TIMEOUT_MS (global override)
 | `OMNIROUTE_AGENT_GOAL_STREAM_RECOVERY`    | `true`               | Enable early stream recovery automatically for detected `/goal` agent runs. Set `false`/`0`/`off` to disable the goal-specific opt-in. This can only ADD recovery on top of the operator default — it never overrides an explicit `STREAM_RECOVERY_ENABLED`/DB settings opt-out. |
 | `OMNIROUTE_CODEX_DROP_NONSTANDARD_EVENTS` | _(off)_              | Strip non-standard `codex.*` SSE events (e.g. `codex.rate_limits`) that break the OpenAI SDK's `responses.stream()` with a 502. Set `true`/`1`/`yes` to enable. |
 | `FETCH_HEADERS_TIMEOUT_MS`                | = `FETCH_TIMEOUT_MS` | Time to receive response headers.                                                                                                                               |
+| `OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS`    | `30000` (30s)         | Maximum response-start wait (ms) for each direct no-proxy attempt. A timeout retries once on a fresh socket; set `0` to disable the bound and retain the previous behavior. |
 | `FETCH_BODY_TIMEOUT_MS`                   | = `FETCH_TIMEOUT_MS` | Time to receive the full response body.                                                                                                                         |
 | `FETCH_CONNECT_TIMEOUT_MS`                | `30000`              | TCP connection establishment timeout.                                                                                                                           |
 | `FETCH_KEEPALIVE_TIMEOUT_MS`              | `4000`               | Keep-alive socket idle timeout.                                                                                                                                 |

open-sse/utils/directResponseStartTimeout.ts

@@ -0,0 +1,77 @@
+type DirectFetchOptions = RequestInit & { dispatcher?: unknown };
+type DirectFetch = (
+  input: RequestInfo | URL,
+  options: DirectFetchOptions
+) => Promise<Response>;
+
+const DEFAULT_DIRECT_HEADERS_TIMEOUT_MS = 30_000;
+const DIRECT_RESPONSE_START_TIMEOUT_CODE = "DIRECT_RESPONSE_START_TIMEOUT";
+
+export function resolveDirectHeadersTimeoutMs(
+  env: Record<string, string | undefined> = process.env
+): number {
+  const raw = env.OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS;
+  if (raw == null || raw.trim() === "") return DEFAULT_DIRECT_HEADERS_TIMEOUT_MS;
+  const parsed = Number(raw);
+  return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : 0;
+}
+
+function createDirectResponseStartTimeout(timeoutMs: number): Error & { code: string } {
+  const err = new Error(
+    `Direct response did not start within ${timeoutMs}ms — retrying on a fresh socket`
+  ) as Error & { code: string };
+  err.name = "TimeoutError";
+  err.code = DIRECT_RESPONSE_START_TIMEOUT_CODE;
+  return err;
+}
+
+export function isDirectResponseStartTimeout(err: unknown): boolean {
+  return (
+    !!err &&
+    typeof err === "object" &&
+    "code" in err &&
+    err.code === DIRECT_RESPONSE_START_TIMEOUT_CODE
+  );
+}
+
+function mergeAbortSignals(
+  primary: AbortSignal | null | undefined,
+  secondary: AbortSignal
+): AbortSignal {
+  if (!primary) return secondary;
+  if (primary.aborted) return primary;
+  const controller = new AbortController();
+  const onPrimaryAbort = () => controller.abort(primary.reason);
+  const onSecondaryAbort = () => controller.abort(secondary.reason);
+  const cleanup = () => {
+    primary.removeEventListener("abort", onPrimaryAbort);
+    secondary.removeEventListener("abort", onSecondaryAbort);
+  };
+  primary.addEventListener("abort", onPrimaryAbort, { once: true });
+  secondary.addEventListener("abort", onSecondaryAbort, { once: true });
+  controller.signal.addEventListener("abort", cleanup, { once: true });
+  return controller.signal;
+}
+
+export async function directFetchWithBoundedResponseStart(
+  input: RequestInfo | URL,
+  options: DirectFetchOptions,
+  fetchImpl: DirectFetch,
+  timeoutMs: number
+): Promise<Response> {
+  if (!timeoutMs || timeoutMs <= 0) return fetchImpl(input, options);
+  const attemptController = new AbortController();
+  const timer = setTimeout(
+    () => attemptController.abort(createDirectResponseStartTimeout(timeoutMs)),
+    timeoutMs
+  );
+  timer.unref?.();
+  try {
+    return await fetchImpl(input, {
+      ...options,
+      signal: mergeAbortSignals(options.signal, attemptController.signal),
+    });
+  } finally {
+    clearTimeout(timer);
+  }
+}

open-sse/utils/proxyFetch.ts

@@ -19,6 +19,11 @@ import {
   isControlPlaneProxyDirectFallbackEnabled,
   isFeatureFlagEnabled,
 } from "@/shared/utils/featureFlags";
+import {
+  directFetchWithBoundedResponseStart,
+  isDirectResponseStartTimeout,
+  resolveDirectHeadersTimeoutMs,
+} from "./directResponseStartTimeout.ts";
 
 // #9100: relay egress (Vercel / Deno / Cloudflare edge functions) used to go
 // through bare `originalFetch` — NO connection pooling, NO timeout, NO retry.
@@ -154,7 +159,6 @@ type TlsFingerprintStore = {
   provider?: string | null;
   sessionScope?: string;
 };
-
 /**
  * #5217 (Gap-secondary): a mutable sink that records the proxy actually applied
  * by `runWithProxyContext` for the in-flight request. Executors that pin their
@@ -802,15 +806,7 @@ async function patchedFetch(
         (deps.nativeFetch as FetchWithDispatcher | undefined) ?? originalFetchWithDispatcher;
       return _nativeFetch(input, options);
     }
-    // Direct connection (no proxy) — use undici with custom dispatcher for timeout control.
+    // Direct undici path: bound response-start, fresh-socket retry, and body guard.
-    // Falls back to original native fetch if dispatcher initialization fails (#1054).
-    // Retries once on transient dispatcher errors before falling back (fix: proxyfetch-undici-retry).
-    //
-    // Non-replayable body guard: if the body is stream-like (ReadableStream/Blob)
-    // or the input is a Request that carries a body, the first dispatcher attempt
-    // owns that body. Retrying or falling back to native fetch would replay a
-    // consumed/locked body and can mask the original transport error with
-    // "Response body object should not be disturbed or locked".
     const hasNonReplayableBody = requestHasNonReplayableBody(input, options);
     const maxAttempts = hasNonReplayableBody ? 1 : 2;
     const _undiciDirect =
@@ -818,32 +814,44 @@ async function patchedFetch(
     const _nativeFallback =
       (deps.nativeFetch as FetchWithDispatcher | undefined) ?? originalFetchWithDispatcher;
     let lastDispatcherError: unknown = null;
+    const directHeadersTimeoutMs = resolveDirectHeadersTimeoutMs();
+    let targetHostForLogs = "";
+    try {
+      targetHostForLogs = new URL(targetUrl).host;
+    } catch {
+      // ignore — logging is best-effort
+    }
     for (let attempt = 0; attempt < maxAttempts; attempt++) {
       try {
-        return await _undiciDirect(input, {
+        return await directFetchWithBoundedResponseStart(
-          ...options,
+          input,
-          // #4252: first attempt uses the pooled keep-alive dispatcher; a retry
+          {
-          // (after a transient socket error) uses the no-keep-alive dispatcher so
+            ...options,
-          // it opens a FRESH socket instead of grabbing another stale pooled one
+            dispatcher: attempt === 0 ? getDefaultDispatcher() : getRetryDispatcher(),
-          // — the burst pattern was the retry re-hitting a dead pooled socket and
+          },
-          // then falling through to native fetch (which also pools) → 502.
+          _undiciDirect,
-          dispatcher: attempt === 0 ? getDefaultDispatcher() : getRetryDispatcher(),
+          directHeadersTimeoutMs
-        });
+        );
       } catch (dispatcherError) {
+        if (isDirectResponseStartTimeout(dispatcherError)) {
+          if (attempt === 0 && maxAttempts > 1) {
+            console.warn(
+              `[ProxyFetch] Direct response-start timeout (${directHeadersTimeoutMs}ms) on pooled dispatcher — retrying on fresh no-keep-alive dispatcher: ${targetHostForLogs}`
+            );
+            lastDispatcherError = dispatcherError;
+            continue;
+          }
+          throw dispatcherError;
+        }
         const msg =
           dispatcherError instanceof Error ? dispatcherError.message : String(dispatcherError);
-        // CAUTION: Do NOT fallback to native fetch if the error is a version mismatch (invalid onRequestStart)
-        // because the native fetch will definitely fail with the undici v8 dispatcher.
         if (msg.includes("onRequestStart")) {
           console.error(
             `[ProxyFetch] Fatal version mismatch: Dispatcher (v8) vs Fetch (v6/native). Hardware upgrade or SOCKS5 config isolation required. Error: ${msg}`
           );
           throw dispatcherError;
         }
-        // Only retry/fallback for connection/dispatcher errors, not HTTP errors.
+        // Retry/fallback only for connection errors, never HTTP errors.
-        // Prefer the .code property when available (more stable across undici
-        // versions than message-string matching); fall back to substring match
-        // for errors that lack a structured code.
         tagProxyUnreachable(dispatcherError);
         const errCode = (dispatcherError as { code?: unknown })?.code;
         if (
@@ -854,10 +862,7 @@ async function patchedFetch(
           msg.includes("UND_ERR")
         ) {
           if (attempt === 0 && maxAttempts > 1) {
-            // First failure — retry once after a short backoff before giving up.
+            // Retry after a short fixed backoff on a fresh socket.
-            // Delay is OMNIROUTE_RETRY_BACKOFF_MS (default 10ms): a fixed backoff
-            // beats random jitter here because the retry opens a fresh socket, so
-            // jitter was pure added latency with no herd benefit.
             lastDispatcherError = dispatcherError;
             await new Promise((r) => setTimeout(r, RETRY_BACKOFF_MS));
             continue;
@@ -873,7 +878,7 @@ async function patchedFetch(
             throw tagProxyUnreachable(dispatcherError);
           }
 
-          // All attempts exhausted — try proxy fallback before native fetch
+          // Exhausted attempts: try proxy fallback before native fetch.
           if (
             !tlsDirectFallback &&
             source === "direct" &&
@@ -899,20 +904,14 @@ async function patchedFetch(
               }
             }
           }
-          // Preserve original phrase intact for monitoring: "Undici dispatcher failed, falling back to native fetch"
+          // Preserve the original monitoring phrase and append the transport cause.
-          // #4252: append the flattened err.cause (code/syscall/errno/address) — the bare
-          // "fetch failed" message hides what actually broke, making bursts undiagnosable.
           console.warn(
             `[ProxyFetch] Undici dispatcher failed, falling back to native fetch (after retry): ${describeFetchCause(dispatcherError)}`
           );
           try {
             return await _nativeFallback(input, options);
           } catch (nativeError) {
-            // #4252: both the undici dispatcher AND native fetch failed. Surface BOTH
+            // Surface both dispatcher and native causes immediately.
-            // causes (server log) and tag the propagated error so the combo executor sees
-            // a diagnosable failure IMMEDIATELY instead of a bare "fetch failed" — the
-            // latter left jobs sitting until the 30s semaphore queue timeout, which then
-            // tripped the circuit breaker.
             const detail = `dispatcher=[${describeFetchCause(dispatcherError)}] native=[${describeFetchCause(nativeError)}]`;
             console.warn(`[ProxyFetch] native fetch fallback ALSO failed: ${detail}`);
             if (nativeError instanceof Error) {

tests/unit/proxyfetch-direct-response-start-timeout-10214.test.ts

@@ -0,0 +1,143 @@
+/**
+ * #10214 — Direct (no-proxy) requests stall on a silently-dropped pooled
+ * keep-alive socket until the caller's deadline or a service restart.
+ *
+ * The default direct dispatcher pools keep-alive sockets for up to
+ * `fetchKeepAliveTimeoutMs` (4 s). A socket that silently drops (half-open, no
+ * RST) surfaces NO transport error — undici's headersTimeout (600 s default) is
+ * the only guard, so the existing fresh-socket retry (which fires on
+ * UND_ERR/ECONNRESET/fetch-failed) never triggers. Observed live: opencode-go
+ * and command-code stall 100% of routed requests until `systemctl restart`.
+ *
+ * The fix bounds the response-start window per direct attempt
+ * (`OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS`, default 30 s) and retries once on the
+ * fresh no-keep-alive dispatcher (a brand-new socket) when the pooled attempt
+ * times out — converting the zombie-socket stall into a clean failover.
+ */
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { proxyFetch } from "../../open-sse/utils/proxyFetch.ts";
+import { getDefaultDispatcher, getRetryDispatcher } from "../../open-sse/utils/proxyDispatcher.ts";
+
+const DIRECT_RESPONSE_START_TIMEOUT_CODE = "DIRECT_RESPONSE_START_TIMEOUT";
+
+/** Simulates a silent half-open pooled socket: the request never resolves, but
+ *  observes the abort signal like real undici does (rejects with the reason). */
+function hangingFetch(capture: {
+  calls: number;
+  dispatchers: unknown[];
+}): (input: RequestInfo | URL, init?: RequestInit) => Promise<Response> {
+  return (input, init) => {
+    capture.calls++;
+    capture.dispatchers.push((init as { dispatcher?: unknown } | undefined)?.dispatcher);
+    return new Promise<Response>((_, reject) => {
+      const signal = init?.signal;
+      signal?.addEventListener(
+        "abort",
+        () =>
+          reject(signal.reason instanceof Error ? signal.reason : new Error(String(signal.reason))),
+        { once: true }
+      );
+      // never resolve — the upstream accepted the connection but sends nothing
+    });
+  };
+}
+
+function withFastTimeout<T>(fn: () => Promise<T>): Promise<T> {
+  process.env.OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS = "50";
+  return fn().finally(() => {
+    delete process.env.OMNIROUTE_DIRECT_HEADERS_TIMEOUT_MS;
+  });
+}
+
+test("#10214 a response-start timeout on the pooled attempt retries on the FRESH no-keep-alive dispatcher", async () => {
+  const capture = { calls: 0, dispatchers: [] as unknown[] };
+
+  const mockUndici = async (input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
+    capture.calls++;
+    capture.dispatchers.push((init as { dispatcher?: unknown } | undefined)?.dispatcher);
+    if (capture.calls === 1) {
+      // First attempt hits a silently-dead pooled socket — hang, no error.
+      return new Promise<Response>((_, reject) => {
+        init?.signal?.addEventListener("abort", () => reject(init.signal!.reason), { once: true });
+      });
+    }
+    return new Response("ok", { status: 200 });
+  };
+  const mockNative = async (): Promise<Response> =>
+    new Response("native-should-not-fire", { status: 200 });
+
+  const res = await withFastTimeout(() =>
+    proxyFetch(
+      "https://opencode.ai/zen/go/v1/chat/completions",
+      { method: "POST" },
+      { undiciFetch: mockUndici, nativeFetch: mockNative }
+    )
+  );
+
+  assert.equal(capture.calls, 2, "pooled attempt times out and must retry once");
+  assert.equal(await res.text(), "ok");
+  // The regression guard: attempt 0 used the pooled keep-alive dispatcher; the
+  // retry used the fresh no-keep-alive dispatcher — a DIFFERENT instance, so the
+  // retry opens a brand-new socket that cannot be the zombie.
+  assert.equal(
+    capture.dispatchers[0],
+    getDefaultDispatcher(),
+    "first attempt must use the pooled default dispatcher"
+  );
+  assert.equal(
+    capture.dispatchers[1],
+    getRetryDispatcher(),
+    "timeout retry must use the fresh no-keep-alive dispatcher"
+  );
+  assert.notEqual(capture.dispatchers[0], capture.dispatchers[1]);
+});
+
+test("#10214 when the fresh-dispatcher retry also stalls, the timeout surfaces (no native fallback)", async () => {
+  const capture = { calls: 0, dispatchers: [] as unknown[] };
+  const mockUndici = hangingFetch(capture);
+  const mockNative = async (): Promise<Response> =>
+    new Response("native-should-not-fire", { status: 200 });
+
+  await assert.rejects(
+    withFastTimeout(() =>
+      proxyFetch(
+        "https://opencode.ai/zen/go/v1/chat/completions",
+        { method: "POST" },
+        { undiciFetch: mockUndici, nativeFetch: mockNative }
+      )
+    ),
+    (err: unknown) => {
+      assert.equal(
+        (err as { code?: unknown }).code,
+        DIRECT_RESPONSE_START_TIMEOUT_CODE,
+        "final failure must be the classified direct response-start timeout"
+      );
+      return true;
+    }
+  );
+  assert.equal(capture.calls, 2, "both attempts must have been made");
+  assert.equal(capture.dispatchers[0], getDefaultDispatcher());
+  assert.equal(capture.dispatchers[1], getRetryDispatcher());
+});
+
+test("#10214 a healthy fast response is untouched by the bound (single attempt, no retry)", async () => {
+  const capture = { calls: 0, dispatchers: [] as unknown[] };
+  const mockUndici = async (input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
+    capture.calls++;
+    capture.dispatchers.push((init as { dispatcher?: unknown } | undefined)?.dispatcher);
+    return new Response("ok", { status: 200 });
+  };
+
+  const res = await withFastTimeout(() =>
+    proxyFetch(
+      "https://opencode.ai/zen/go/v1/chat/completions",
+      { method: "POST" },
+      { undiciFetch: mockUndici }
+    )
+  );
+
+  assert.equal(capture.calls, 1, "healthy request must not retry");
+  assert.equal(capture.dispatchers[0], getDefaultDispatcher());
+  assert.equal(await res.text(), "ok");
+});