fix(proxy): fan out direct dispatcher streams

open-sse/utils/proxyDispatcher.ts

@@ -43,6 +43,58 @@ type ProxyConfigObject = {
   family?: string;
 };
 
+/**
+ * Direct upstream fan-out dispatcher.
+ *
+ * A single Undici Agent configured with `connections > 1` should be enough in
+ * theory, but real Codex `/backend-api/codex/responses` streams on Node 24 have
+ * still been observed queuing every subsequent same-origin request until the
+ * previous stream emits trailers. Using several one-connection Agents gives
+ * each long SSE stream an independent pool/client and prevents one stream from
+ * monopolizing the effective queue while keeping pipelining disabled.
+ */
+class RoundRobinDispatcher {
+  private readonly dispatchers: Dispatcher[];
+  private nextIndex = 0;
+
+  constructor(dispatchers: Dispatcher[]) {
+    this.dispatchers = dispatchers;
+  }
+
+  dispatch(options: Dispatcher.DispatchOptions, handler: Dispatcher.DispatchHandler): boolean {
+    const dispatcher = this.dispatchers[this.nextIndex % this.dispatchers.length];
+    this.nextIndex = (this.nextIndex + 1) % this.dispatchers.length;
+    return dispatcher.dispatch(options, handler);
+  }
+
+  close(callback?: () => void): Promise<void> | void {
+    const done = Promise.all(this.dispatchers.map((dispatcher) => dispatcher.close())).then(
+      () => undefined
+    );
+    if (callback) {
+      done.then(callback);
+      return;
+    }
+    return done;
+  }
+
+  destroy(
+    errorOrCallback?: Error | null | (() => void),
+    callback?: () => void
+  ): Promise<void> | void {
+    const callbackFn = typeof errorOrCallback === "function" ? errorOrCallback : callback;
+    const error = typeof errorOrCallback === "function" ? null : (errorOrCallback ?? null);
+    const done = Promise.all(this.dispatchers.map((dispatcher) => dispatcher.destroy(error))).then(
+      () => undefined
+    );
+    if (callbackFn) {
+      done.then(callbackFn);
+      return;
+    }
+    return done;
+  }
+}
+
 function getDispatcherCache(): DispatcherCache {
   const globalWithCache = globalThis as GlobalWithDispatcherCache;
   if (!globalWithCache[DISPATCHER_CACHE_KEY]) {
@@ -136,23 +188,35 @@ export function getDefaultDispatcherConnectionLimit(
 
 function getDefaultDispatcherOptions(env: Record<string, string | undefined> = process.env) {
   const options = getDispatcherOptions();
-  // #4580 — On the direct egress path, undici's default pipelining (1) lets a long
-  // SSE stream monopolize the single pooled socket per origin, serializing every
-  // other concurrent request to that same provider. Mirror the proxy fix (#4288):
-  // disable pipelining and keep several connections available. Unlike the proxy
-  // path we KEEP keep-alive — the 1ms TTL there is a cheap-proxy-socket workaround,
-  // not needed (and harmful to perf) for direct connections.
+  // #4580 — On the direct egress path, undici's default pipelining (1) let a long
+  // SSE stream monopolize the single pooled socket per origin. Keep the public
+  // connection-limit option here, but getDefaultDispatcher() fans it out across
+  // independent one-connection Agents; in production traces, one multi-connection
+  // Agent could still queue same-origin Codex streams behind prior trailers.
   return {
     ...options,
     connections: getDefaultDispatcherConnectionLimit(env),
     pipelining: 0,
   };
 }
 
+function createRoundRobinDirectDispatcher(connectionLimit: number): Dispatcher {
+  const baseOptions = getDispatcherOptions();
+  const perAgentOptions = {
+    ...baseOptions,
+    connections: 1,
+    pipelining: 0,
+  };
+  const dispatchers = Array.from({ length: connectionLimit }, () => new Agent(perAgentOptions));
+  return new RoundRobinDispatcher(dispatchers) as unknown as Dispatcher;
+}
+
 export function getDefaultDispatcher(): Dispatcher {
   const globalWithCache = globalThis as GlobalWithDispatcherCache;
   if (!globalWithCache[DEFAULT_DISPATCHER_KEY]) {
-    globalWithCache[DEFAULT_DISPATCHER_KEY] = new Agent(getDefaultDispatcherOptions());
+    globalWithCache[DEFAULT_DISPATCHER_KEY] = createRoundRobinDirectDispatcher(
+      getDefaultDispatcherConnectionLimit()
+    );
   }
   return globalWithCache[DEFAULT_DISPATCHER_KEY];
 }
@@ -415,6 +479,10 @@ export function __getDefaultDispatcherOptionsForTest(
   return getDefaultDispatcherOptions(env);
 }
 
+export function __createRoundRobinDispatcherForTest(dispatchers: Dispatcher[]): Dispatcher {
+  return new RoundRobinDispatcher(dispatchers) as unknown as Dispatcher;
+}
+
 export function createProxyDispatcher(proxyUrl: string): Dispatcher {
   const normalizedUrl = normalizeProxyUrl(proxyUrl, "proxy dispatcher");
   const dispatcherCache = getDispatcherCache();

tests/unit/direct-dispatcher-pipelining-4580.test.ts

@@ -1,6 +1,8 @@
 import { describe, it, afterEach } from "node:test";
 import assert from "node:assert/strict";
+import type { Dispatcher } from "undici";
 import {
+  __createRoundRobinDispatcherForTest,
   __getDefaultDispatcherOptionsForTest,
   __getProxyDispatcherOptionsForTest,
   getDefaultDispatcherConnectionLimit,
@@ -39,7 +41,10 @@ describe("#4580 direct dispatcher options", () => {
   });
 
   it("connection limit honors OMNIROUTE_DIRECT_DISPATCHER_CONNECTIONS", () => {
-    assert.equal(getDefaultDispatcherConnectionLimit({ OMNIROUTE_DIRECT_DISPATCHER_CONNECTIONS: "8" }), 8);
+    assert.equal(
+      getDefaultDispatcherConnectionLimit({ OMNIROUTE_DIRECT_DISPATCHER_CONNECTIONS: "8" }),
+      8
+    );
   });
 
   it("connection limit clamps invalid values to the default", () => {
@@ -49,4 +54,30 @@ describe("#4580 direct dispatcher options", () => {
     );
     assert.equal(getDefaultDispatcherConnectionLimit({}), 32);
   });
+
+  it("fans out direct requests across independent dispatcher pools", () => {
+    const calls: number[] = [];
+    const dispatchers = [0, 1, 2].map(
+      (index) =>
+        ({
+          dispatch() {
+            calls.push(index);
+            return true;
+          },
+          close() {},
+          destroy() {},
+        }) as unknown as Dispatcher
+    );
+    const dispatcher = __createRoundRobinDispatcherForTest(dispatchers);
+    const dispatchOptions = {
+      origin: "https://chatgpt.com",
+      path: "/backend-api/codex/responses",
+      method: "POST",
+    } as unknown as Parameters<Dispatcher["dispatch"]>[0];
+    const handler = {} as Parameters<Dispatcher["dispatch"]>[1];
+
+    for (let i = 0; i < 7; i++) dispatcher.dispatch(dispatchOptions, handler);
+
+    assert.deepEqual(calls, [0, 1, 2, 0, 1, 2, 0]);
+  });
 });