fix(transport): apply the fresh-connection policy to a selected provider transport

docs-site/src/content/docs/reference/configuration/server.md

@@ -143,6 +143,28 @@ Compare the diagnostic on the same machine and account under the two network
 modes. A successful TUN test alone does not identify why the service's HTTP proxy
 path failed, and does not establish a general fix.
 
+## Connection reuse for specific upstream hosts
+
+Some upstreams keep a connection open after they have stopped serving it. The next request reuses
+that pooled socket and fails without reaching the provider. `OCX_FRESH_CONNECTION_HOSTS` names the
+hosts that should never reuse a pooled connection, as an environment variable rather than a config
+field, so it can be applied to one machine without editing shared configuration:
+
+```bash
+OCX_FRESH_CONNECTION_HOSTS="api.example.com, relay.example.net" ocx start
+```
+
+The value is a comma-separated list of hostnames. Matching is case-insensitive, covers each named
+host and its subdomains, and ignores leading dots, so `.example.com` and `example.com` both match
+`api.example.com`. Do not include a scheme, port or path. An unset or empty variable leaves the
+default connection behavior unchanged.
+
+A matching send carries `Connection: close` and is dispatched with keep-alive disabled. The
+decision is made against the address actually used on the wire, so it still applies when a provider
+transport rewrites the destination after credential selection. Per-request latency rises slightly
+for those hosts, since each request pays a fresh TCP and TLS handshake; name only the hosts that
+need it.
+
 ## Remote access
 
 The default `127.0.0.1` bind is loopback-only. A non-loopback address such as `0.0.0.0` or a tailnet

src/server/chat-native.ts

@@ -50,7 +50,7 @@ import { enrichOpenCodeZenFreeTierMessage } from "../providers/opencode-zen-rate
 import type { OcxProviderTransport } from "../providers/xai-transport";
 import type { RouteResult } from "../router";
 import type { OcxConfig, OcxProviderConfig } from "../types";
-import { fetchWithHeaderTimeout, providerFetch, safeHostLabel } from "./responses/fetch-helpers";
+import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, sendWithConnectionPolicy } from "./responses/fetch-helpers";
 import { linkAbortSignal } from "./responses";
 import {
   addFinalRequestLog,
@@ -351,9 +351,15 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio
                 if (!headers.has("accept-encoding") && encoding) headers.set("accept-encoding", encoding);
                 if (init.signal?.aborted) throw init.signal.reason;
                 noteProviderAttemptSend(logCtx, route.providerName, activeProvider, logCtx.usageLogInputTokens, transportRecovery ?? recovery);
-                const dispatched = await ((activeProvider as OcxProviderTransport).fetch ?? execute)(request.url, applyUpstreamRecoveryInit({
-                  ...init, method: request.method, headers, body: request.body,
-                }, transportRecovery));
+                // A reselected provider transport is still a physical send: the connection policy
+                // and manual-redirect ownership wrap the selected implementation (#4992).
+                const dispatched = await sendWithConnectionPolicy(
+                  (activeProvider as OcxProviderTransport).fetch ?? execute,
+                  request.url,
+                  applyUpstreamRecoveryInit({
+                    ...init, method: request.method, headers, body: request.body,
+                  }, transportRecovery),
+                );
                 if (!dispatched.ok) await recordKeyAttemptFailure(logCtx, dispatched, init.signal ?? upstream.signal);
                 return dispatched;
               },

src/server/responses/fetch-helpers.ts

@@ -79,6 +79,38 @@ export interface PaceAwareFetch {
 
 export type ProviderFetch = typeof globalThis.fetch & PaceAwareFetch;
 
+/**
+ * Apply the physical-send connection policy to whichever fetch actually performs the send.
+ *
+ * The executor `providerFetch` builds is not the only physical boundary. A `dispatchOverride`
+ * that revalidates credentials re-reads `route.provider.fetch` at send time -- reselection can
+ * install a different provider transport after this wrapper was constructed -- and then calls
+ * that fetch directly instead of the supplied executor. Keeping the policy inside the executor
+ * alone therefore left every provider-scoped transport reusing a pooled socket for a host the
+ * operator had named in `OCX_FRESH_CONNECTION_HOSTS` (#4992). The policy belongs around the
+ * selected fetch so it follows the selection rather than the construction.
+ *
+ * Idempotent on purpose: an override that hands the send back to the supplied executor passes
+ * through here twice, and both passes derive the same headers from the same wire URL.
+ */
+export function sendWithConnectionPolicy(
+  physicalFetch: typeof globalThis.fetch,
+  input: Parameters<typeof globalThis.fetch>[0],
+  init?: RequestInit,
+): Promise<Response> {
+  const headers = new Headers(init?.headers ?? (input instanceof Request ? input.headers : undefined));
+  const fresh = wantsFreshConnection(input);
+  if (fresh) {
+    headers.set("Connection", "close");
+  }
+  return physicalFetch(input, {
+    ...init,
+    headers,
+    redirect: "manual",
+    ...(fresh ? { keepalive: false } : {}),
+  });
+}
+
 export interface ProviderFetchOptions {
   nativeControl?: NativeResponseControl;
   providerName?: string;
@@ -109,19 +141,8 @@ export function providerFetch(
   // Rebuilt dispatches must use the same physical-send boundary as ordinary HTTP sends.
   // Return the original 3xx so the response owner retains its retry/health/relay contract.
   const dispatch = Object.assign(
-    (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
-      const headers = new Headers(init?.headers ?? (input instanceof Request ? input.headers : undefined));
-      const fresh = wantsFreshConnection(input);
-      if (fresh) {
-        headers.set("Connection", "close");
-      }
-      return base(input, {
-        ...init,
-        headers,
-        redirect: "manual",
-        ...(fresh ? { keepalive: false } : {}),
-      });
-    },
+    (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) =>
+      sendWithConnectionPolicy(base, input, init),
     { preconnect },
   ) as typeof globalThis.fetch;
   const httpFetch = Object.assign(

src/server/responses/request-transport.ts

@@ -41,7 +41,7 @@ import {
   resolveCurrentProviderApiKeyTransport,
 } from "../../providers/api-key-selection";
 import { resolveAdapter, resolveWireProtocolOverride } from "../adapter-resolve";
-import { providerFetch } from "./fetch-helpers";
+import { providerFetch, sendWithConnectionPolicy } from "./fetch-helpers";
 import type { ProviderFetchOptions } from "./fetch-helpers";
 import { captureConfigGeneration } from "../../lib/state-store-sweeper";
 import { recordAnthropicAccountQuotaFromHeaders, hasPassiveAccountQuota } from "../../providers/quota";
@@ -405,8 +405,10 @@ export async function prepareResponsesTransport(
             && sentHeaders?.get("authorization") === `Bearer ${snapshot.accessToken}`
             && !sentHeaders?.has("x-api-key");
           // Reselection can choose a provider override instead of the supplied executor.
+          // Either way the send crosses the physical boundary, so the connection policy is
+          // applied around whichever implementation was just selected (#4992).
           commitKeyAttemptSend();
-          const response = await fetchImpl(destination, { ...dispatchInit, redirect: "manual" });
+          const response = await sendWithConnectionPolicy(fetchImpl, destination, { ...dispatchInit, redirect: "manual" });
           if (!response.ok) await recordKeyAttemptFailure(logCtx, response, dispatchInit.signal ?? options.abortSignal);
           // Observe each physical response before retries replace it. The binding belongs to
           // this dispatch, so a manual switch cannot file A's headers against B. Header

structure/transports/responses.md

@@ -45,9 +45,20 @@ modules merely because those imports existed in the pre-split `responses.ts` mon
 
 `OCX_FRESH_CONNECTION_HOSTS` accepts comma-separated hostnames whose outbound HTTP sends bypass
 keep-alive reuse with `Connection: close` and `keepalive: false`; exact hosts and their subdomains
-match case-insensitively. The helper applies this policy at the final executor boundary, after a
-dispatch override has selected or rebuilt the destination, so matching follows the URL sent on the
-wire rather than the URL supplied before credential revalidation.
+match case-insensitively. `sendWithConnectionPolicy` applies the policy around the fetch that
+performs the physical send, after a dispatch override has selected or rebuilt the destination, so
+matching follows the URL sent on the wire rather than the URL supplied before credential
+revalidation.
+
+The wrapped executor alone is not that boundary. An override that revalidates credentials re-reads
+`route.provider.fetch` at send time, because reselection can install a different provider transport
+after the wrapper was built, and then calls that implementation instead of the executor. Both
+production overrides do this -- `oauthDispatch` in `request-transport.ts` and the native Chat
+key-revalidation override in `chat-native.ts` -- so both wrap the selected implementation rather
+than choosing between policy and provider transport. Reporting the executor as the boundary while
+the code let a provider-scoped transport past it is what #4992 recorded, and it is why a
+regression for this policy has to enter through `handleResponses` rather than through a
+hand-written override that cooperates by calling the executor it was handed.
 
 ### Semantic progress ownership
 

tests/responses/fresh-connection-optout.test.ts

@@ -1,6 +1,14 @@
 import { describe, expect, test } from "bun:test";
+import { mkdtempSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
 import { wantsFreshConnection, providerFetch } from "../../src/server/responses/fetch-helpers";
-import type { OcxProviderConfig } from "../../src/types";
+import { saveCredential } from "../../src/oauth/store";
+import { XAI_GROK_CLI_BASE_URL } from "../../src/providers/xai-transport";
+import { handleResponses } from "../../src/server/responses";
+import type { RequestLogContext } from "../../src/server/request-log";
+import type { OcxConfig, OcxProviderConfig } from "../../src/types";
+import { removeTreeWithRetry } from "../helpers/remove-tree";
 
 describe("wantsFreshConnection", () => {
   test("returns false when env is unset or empty", () => {
@@ -246,3 +254,90 @@ describe("providerFetch fresh connection dispatch", () => {
     }
   });
 });
+
+/**
+ * The cases above drive `providerFetch` with an override that cooperates by calling the executor
+ * it was handed. Production's OAuth override does not: it re-reads `route.provider.fetch` at the
+ * send boundary, because credential reselection can install a different provider transport after
+ * the wrapper was built, and calls that implementation directly. Every test above still passed
+ * while the operator's configured host reused a pooled socket on that path (#4992), so the
+ * regression has to enter through `handleResponses` rather than through a hand-written override.
+ *
+ * xAI OAuth is the live consumer: `resolveProviderTransport` installs a provider-scoped fetch and
+ * rewrites the destination to the Grok CLI host, which is what makes it observable here.
+ */
+describe("the OAuth dispatch boundary", () => {
+  test("a provider-scoped transport selected at dispatch receives the fresh-connection policy", async () => {
+    const freshHost = new URL(XAI_GROK_CLI_BASE_URL).hostname;
+    const home = mkdtempSync(join(tmpdir(), "ocx-fresh-connection-oauth-"));
+    const nativeFetch = globalThis.fetch;
+    const previousHosts = process.env.OCX_FRESH_CONNECTION_HOSTS;
+    const previousOpencodexHome = process.env.OPENCODEX_HOME;
+    const previousCodexHome = process.env.CODEX_HOME;
+    process.env.OPENCODEX_HOME = home;
+    process.env.CODEX_HOME = home;
+    process.env.OCX_FRESH_CONNECTION_HOSTS = freshHost;
+    const sends: Array<{ url: string; init?: RequestInit }> = [];
+
+    try {
+      await saveCredential("xai", {
+        access: "xai-access-token",
+        refresh: "xai-refresh-token",
+        expires: Date.now() + 3_600_000,
+        accountId: "xai-acct-1",
+        source: "local-cli",
+      });
+      globalThis.fetch = (async (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
+        const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url;
+        sends.push({ url, init });
+        return Response.json({
+          id: "resp_fresh_connection",
+          status: "completed",
+          output: [],
+          usage: { input_tokens: 10, output_tokens: 5, total_tokens: 15 },
+        });
+      }) as typeof globalThis.fetch;
+
+      const config = {
+        defaultProvider: "xai",
+        providers: {
+          xai: { adapter: "openai-chat", baseUrl: "https://api.x.ai/v1", authMode: "oauth" },
+        },
+      } as OcxConfig;
+      const logCtx: RequestLogContext = { model: "", provider: "" };
+      const response = await handleResponses(
+        new Request("http://localhost/v1/responses", {
+          method: "POST",
+          headers: { "content-type": "application/json" },
+          body: JSON.stringify({ model: "grok-4.6", input: "hello", stream: false }),
+        }),
+        config,
+        logCtx,
+        {},
+      );
+
+      expect(response.status).toBe(200);
+      // Asserted on the host rather than a path, and on the mapped list rather than a filtered
+      // one, so a destination change reports the addresses it observed instead of an empty length.
+      expect(sends.map(send => new URL(send.url).hostname)).toContain(freshHost);
+      const policed = sends.filter(send => new URL(send.url).hostname === freshHost);
+      for (const send of policed) {
+        const headers = new Headers(send.init?.headers);
+        expect(headers.get("Connection")).toBe("close");
+        expect((send.init as { keepalive?: boolean } | undefined)?.keepalive).toBe(false);
+        // And the provider's own implementation still ran: only the xAI wrapper pins this header,
+        // so wrapping the selected fetch did not replace it with the generic executor.
+        expect(headers.get("x-grok-req-id")).toBeTruthy();
+      }
+    } finally {
+      globalThis.fetch = nativeFetch;
+      if (previousHosts === undefined) delete process.env.OCX_FRESH_CONNECTION_HOSTS;
+      else process.env.OCX_FRESH_CONNECTION_HOSTS = previousHosts;
+      if (previousOpencodexHome === undefined) delete process.env.OPENCODEX_HOME;
+      else process.env.OPENCODEX_HOME = previousOpencodexHome;
+      if (previousCodexHome === undefined) delete process.env.CODEX_HOME;
+      else process.env.CODEX_HOME = previousCodexHome;
+      removeTreeWithRetry(home);
+    }
+  });
+});