fix(responses): reclaim least recently used idle WebSocket at capacity

docs-site/src/content/docs/guides/codex-integration.md

@@ -427,6 +427,14 @@ provider advertises `supports_websockets = true` only when `"websockets": true`;
 built-in provider may try WebSocket first, and a disabled proxy returns `426` so Codex falls back to
 HTTP/SSE.
 
+The proxy admits up to 128 client-facing Responses WebSocket connections. When a new connection
+finds all 128 slots occupied, it closes the least recently used idle connection to make room. Connections
+with an active response, pending client tool or approval results, a native steering/injection
+session, or buffered output are protected. If no idle connection can be released, the new
+connection receives the existing `503` capacity response. A reclaimed client can reconnect for
+later requests using the existing continuation handling. This happens only under connection
+pressure, needs no setting, and does not change voice connections or upstream connection reuse.
+
 If a canonical ChatGPT forward continuation references expired or missing local replay state,
 opencodex returns `previous_response_not_found` before sending anything upstream. Codex's
 WebSocket client recognizes this error and can reconnect with its full retained context,

scripts/test-layout/layout.json

@@ -722,6 +722,7 @@
     "codex-warmup.test.ts": "codex-integration",
     "codex-web-search-switch.test.ts": "codex-integration",
     "codex-websocket-registry.test.ts": "codex-integration",
+    "websocket-capacity-activity.test.ts": "codex-integration",
     "codex-write-lock.test.ts": "codex-integration",
     "coding-agent-tool-result-images.test.ts": "adapters",
     "cold-spawn-warmup.test.ts": "ci-workflows",
@@ -1919,6 +1920,8 @@
     "winsw.test.ts": "service",
     "workflow-budget.test.ts": "lib",
     "ws-ambiguous-resend.test.ts": "responses", "ws-pre-output-refusal.test.ts": "responses", "responses-pool-overload-failover.test.ts": "responses", "compact-transient-alternate.test.ts": "responses",
+    "ws-capacity-reclamation.test.ts": "responses",
+    "ws-capacity-tool-ownership.test.ts": "responses",
     "ws-endpoint.test.ts": "responses",
     "ws-failure-stage.test.ts": "responses",
     "ws-native-injection.test.ts": "responses",

src/codex/websocket-activity.ts

@@ -0,0 +1,161 @@
+import { createHash } from "node:crypto";
+import { isAuthorizedProviderExecutedCall, type ProviderExecutedCallType } from "../server/responses-undeclared-tool-guard";
+
+// Logical recency avoids clock adjustments and ties between messages in one millisecond.
+let activityOrder = 0;
+const MAX_PENDING_RESULTS = 1024;
+const MAX_RESULT_ID_LENGTH = 1024;
+const resultTypes: Readonly<Record<string, string>> = {
+  function_call: "function_call_output",
+  custom_tool_call: "custom_tool_call_output",
+  local_shell_call: "function_call_output",
+  shell_call: "shell_call_output",
+  computer_call: "computer_call_output",
+  apply_patch_call: "apply_patch_call_output",
+  tool_search_call: "tool_search_output",
+  mcp_approval_request: "mcp_approval_response",
+};
+
+function record(value: unknown): value is Record<string, unknown> {
+  return value !== null && typeof value === "object" && !Array.isArray(value);
+}
+
+function resultKey(type: string, id: string): string {
+  return createHash("sha256").update(JSON.stringify([type, id])).digest("hex");
+}
+
+/** Content-free activity owned by one admitted incoming Responses socket. */
+export class CodexWebSocketActivity {
+  lastUsed = ++activityOrder;
+  closing = false;
+  private turns = 0;
+  private pendingResults = new Set<string>();
+  private returnedResults = new Set<string>();
+  // Hashed identities only; null marks a route-authorized provider-owned call.
+  private argumentCalls = new Map<string, { type: string; result: string | null }>();
+  // Unknown/oversized identities cannot safely prove that client work finished.
+  private untrackedResults = false;
+
+  get idle(): boolean {
+    return !this.closing && this.turns === 0 && this.pendingResults.size === 0 && !this.untrackedResults;
+  }
+
+  touch(): void {
+    this.lastUsed = ++activityOrder;
+  }
+
+  /** Each turn retains its own busy reference, including superseded async cleanup. */
+  beginTurn(input: unknown): () => void {
+    this.turns += 1;
+    this.touch();
+    this.acceptResults(input);
+    // Ordinary creates start a new response even when its upstream omits response.created.
+    // Prior results settle old waits, but cannot suppress new calls reusing their IDs.
+    this.returnedResults.clear();
+    this.argumentCalls.clear();
+    let finished = false;
+    return () => {
+      if (finished) return;
+      finished = true;
+      this.turns -= 1;
+      if (this.turns === 0) this.touch();
+    };
+  }
+
+  /** Called only after admission or native control accepts the client's results. */
+  acceptResults(input: unknown): void {
+    if (this.untrackedResults) return;
+    if (Array.isArray(input)) {
+      for (const item of input) {
+        if (!record(item) || typeof item.type !== "string") continue;
+        const id = item.type === "mcp_approval_response" ? item.approval_request_id : item.call_id;
+        if (typeof id === "string" && id.length <= MAX_RESULT_ID_LENGTH) {
+          // Ordinary Codex shell results use function_call_output; native steering
+          // also accepts its explicit local_shell_call_output continuation stub.
+          const type = item.type === "local_shell_call_output" ? "function_call_output" : item.type;
+          const key = resultKey(type, id);
+          if (this.pendingResults.delete(key)) {
+            if (this.returnedResults.size >= MAX_PENDING_RESULTS) { this.protectUntracked(); return; }
+            else this.returnedResults.add(key);
+          }
+        }
+      }
+    }
+  }
+
+  /** Preserve tool waits even when the response terminal omits its output array. */
+  observePayload(payload: string, providerExecuted?: ReadonlySet<ProviderExecutedCallType>): void {
+    if (this.untrackedResults) return;
+    let frame: unknown;
+    try { frame = JSON.parse(payload); } catch { return; }
+    if (!record(frame)) return;
+    // Native injection can return a result before the same terminal repeats its call.
+    if (frame.type === "response.created") {
+      this.returnedResults.clear();
+      this.argumentCalls.clear();
+    }
+    if (frame.type === "response.output_item.added") this.rememberArgumentCall(frame.item, providerExecuted);
+    if (frame.type === "response.function_call_arguments.done" || frame.type === "response.custom_tool_call_input.done") {
+      const type = frame.type === "response.function_call_arguments.done" ? "function_call" : "custom_tool_call";
+      const call = typeof frame.item_id === "string" && frame.item_id.length <= MAX_RESULT_ID_LENGTH
+        ? this.argumentCalls.get(resultKey("item", frame.item_id)) : undefined;
+      if (!call || call.type !== type) this.protectUntracked();
+      else if (call.result !== null) this.observeResultKey(call.result);
+    }
+    if (frame.type === "response.output_item.done") this.observeOutput(frame.item, providerExecuted);
+    if (["response.completed", "response.incomplete", "response.failed"].includes(String(frame.type))
+      && record(frame.response) && Array.isArray(frame.response.output)) {
+      for (const item of frame.response.output) this.observeOutput(item, providerExecuted);
+    }
+  }
+
+  private rememberArgumentCall(item: unknown, providerExecuted?: ReadonlySet<ProviderExecutedCallType>): void {
+    if (!record(item) || (item.type !== "function_call" && item.type !== "custom_tool_call")) return;
+    if (typeof item.id !== "string" || !item.id.length || item.id.length > MAX_RESULT_ID_LENGTH) return;
+    const key = resultKey("item", item.id);
+    const hosted = providerExecuted && isAuthorizedProviderExecutedCall(item, providerExecuted);
+    if (!hosted && (typeof item.call_id !== "string" || !item.call_id.length || item.call_id.length > MAX_RESULT_ID_LENGTH)) {
+      if (this.argumentCalls.has(key)) this.protectUntracked();
+      return;
+    }
+    const result = hosted ? null : resultKey(resultTypes[item.type]!, item.call_id as string);
+    const previous = this.argumentCalls.get(key);
+    if (previous) {
+      if (previous.type !== item.type || previous.result !== result) this.protectUntracked();
+      return;
+    }
+    if (this.argumentCalls.size >= MAX_PENDING_RESULTS) { this.protectUntracked(); return; }
+    // An unfinished argument stream alone does not prove the client started work.
+    this.argumentCalls.set(key, { type: item.type, result });
+  }
+
+  private observeOutput(item: unknown, providerExecuted?: ReadonlySet<ProviderExecutedCallType>): void {
+    if (this.untrackedResults) return;
+    if (!record(item) || typeof item.type !== "string") return;
+    // Hosted execution may publish both the call and its result in the output stream.
+    this.acceptResults([item]);
+    if (!Object.hasOwn(resultTypes, item.type)) return;
+    if (providerExecuted && isAuthorizedProviderExecutedCall(item, providerExecuted)) return;
+    if (item.type === "tool_search_call" && item.execution !== "client") return;
+    const id = item.type === "mcp_approval_request" ? item.id
+      : item.type === "local_shell_call" || item.type === "tool_search_call" ? item.call_id ?? item.id : item.call_id;
+    if (typeof id !== "string" || !id.length || id.length > MAX_RESULT_ID_LENGTH) {
+      this.protectUntracked();
+      return;
+    }
+    this.observeResultKey(resultKey(resultTypes[item.type]!, id));
+  }
+
+  private observeResultKey(key: string): void {
+    if (this.pendingResults.has(key) || this.returnedResults.has(key)) return;
+    if (this.pendingResults.size >= MAX_PENDING_RESULTS) this.protectUntracked();
+    else this.pendingResults.add(key);
+  }
+
+  private protectUntracked(): void {
+    this.untrackedResults = true;
+    this.pendingResults.clear();
+    this.returnedResults.clear();
+    this.argumentCalls.clear();
+  }
+}

src/codex/websocket-registry.ts

@@ -1,23 +1,47 @@
 import type { ServerWebSocket } from "bun";
 import type { WsData } from "../server/ws-bridge";
 import { createAdmissionGate, type AdmissionMetrics, type AdmissionReservation } from "../lib/admission";
+import { CodexWebSocketActivity } from "./websocket-activity";
 
 export const MAX_TRACKED_CODEX_WEBSOCKETS = 128;
 const websocketGate = createAdmissionGate("codex_websockets", MAX_TRACKED_CODEX_WEBSOCKETS);
+const admittedSockets = new Set<ServerWebSocket<WsData>>();
+
+function reclaimOldestIdleSocket(): void {
+  let oldest: ServerWebSocket<WsData> | undefined;
+  for (const ws of admittedSockets) {
+    const activity = ws.data.responsesActivity;
+    if (!activity?.idle || ws.data.nativeControl || ws.readyState !== 1 || ws.getBufferedAmount() !== 0) continue;
+    if (!oldest || activity.lastUsed < oldest.data.responsesActivity!.lastUsed) oldest = ws;
+  }
+  if (!oldest) return;
+  // Fence late frames before closing. Only the actual close callback releases admission.
+  oldest.data.responsesActivity!.closing = true;
+  try {
+    oldest.close(1001, "Idle Responses connection reclaimed");
+  } catch {
+    // A failed/delayed close never manufactures a free slot or retries another victim.
+    if (oldest.readyState === 1) oldest.data.responsesActivity!.closing = false;
+  }
+}
 
 export function tryReserveCodexWebSocket(): AdmissionReservation<ServerWebSocket<WsData>> | null {
+  if (websocketGate.metrics().active >= MAX_TRACKED_CODEX_WEBSOCKETS) reclaimOldestIdleSocket();
   const gateLease = websocketGate.tryAcquire();
   if (!gateLease) return null;
   let socket: ServerWebSocket<WsData> | undefined;
   let active = true;
   return {
     bind(value) {
-      if (!active) return;
+      if (!active || socket) return;
       socket = value;
+      socket.data.responsesActivity = new CodexWebSocketActivity();
+      admittedSockets.add(socket);
     },
     release() {
       if (!active) return;
       active = false;
+      if (socket) admittedSockets.delete(socket);
       socket = undefined;
       gateLease.release();
     },

src/server/index/live-sideband.ts

@@ -40,8 +40,9 @@ export const MAX_WS_FRAME_BYTES = 50 * 1024 * 1024;
  *
  * What must never happen is neither: an immortal socket plus a destination that silently accepts
  * the orphaned delta. `tests/responses/ws-endpoint.test.ts` holds exactly that pair together.
- * Raising the timeout off 0 is still worth doing for its own reasons (a dead peer holds a socket
- * forever today), and Bun caps the value at 255 seconds, well inside the bound below.
+ * Responses admission separately reclaims its least recently used idle connection under
+ * capacity pressure. That is not a time limit, so expired-reference validation remains required.
+ * Keeping this shared timeout at 0 also preserves quiet live/voice sessions.
  */
 export const WEBSOCKET_IDLE_TIMEOUT_SECONDS = 0;
 /** Ceiling a finite websocket idle timeout must stay under, in seconds. See above. */

src/server/index/websocket-handler.ts

@@ -89,6 +89,7 @@ import {
 import type { ServeOptionsContext } from "./serve-options";
 import type { RequestMetricsRecorder } from "../request-metrics";
 import { resolveInboundBodyLimitBytes } from "../request-decompress";
+import { responseToolOwnership } from "../responses/response-tool-ownership";
 
 /**
  * The WebSocket half of the Bun.serve options, split out of serve-options.ts to keep that file
@@ -184,6 +185,8 @@ export function createWebsocketHandler(
           }
           return;
         }
+        if (ws.data.responsesActivity?.closing) return;
+        ws.data.responsesActivity?.touch();
         const rawBytes = typeof raw === "string" ? Buffer.byteLength(raw) : raw.byteLength;
         if (rawBytes > MAX_WS_FRAME_BYTES) {
           sendJsonFrame(ws, buildWsErrorFrame(413, {
@@ -223,14 +226,18 @@ export function createWebsocketHandler(
             if (frame.type === "response.inject") {
               if (!ws.data.nativeControl?.inject) throw new NativeSteeringError("injection_not_supported", "Native injection is disabled or unavailable on this route.");
               ws.data.nativeControl.inject(frame);
+              ws.data.responsesActivity?.acceptResults(frame.input);
               return;
             }
             if (frame.type === "response.steer") {
               if (!ws.data.nativeControl) throw new NativeSteeringError("steering_not_supported", ws.data.nativeSteeringUnavailable ?? "Native steering transport is unavailable; the route may be unsupported or using HTTP fallback.");
               ws.data.nativeControl.steer(frame);
               return;
             }
-            if (ws.data.nativeControl?.continue(frame)) return;
+            if (ws.data.nativeControl?.continue(frame)) {
+              ws.data.responsesActivity?.acceptResults(frame.input);
+              return;
+            }
           } catch (error) {
             sendJsonFrame(ws, buildWsErrorFrame(400, {
               type: "invalid_request_error",
@@ -294,6 +301,7 @@ export function createWebsocketHandler(
         }
 
         // Only a genuinely admitted turn may receive steering or continuations.
+        const finishActivity = ws.data.responsesActivity?.beginTurn(frame.input);
         ws.data.nativeControl = nativeControl;
         const payload: Record<string, unknown> = { ...frame };
         delete payload.type;
@@ -346,11 +354,16 @@ export function createWebsocketHandler(
                 terminalRecorder = recorder;
               },
             });
+            const observeLog = nativeControl?.relayActive
+              ? createNativeSteeringLogObserver(logCtx, () => recordFirstOutput(logCtx, start))
+              : (payload: string) => inspectResponseLogSsePayload(logCtx, payload);
+            const providerExecuted = responseToolOwnership(response);
             await sendResponseToWebSocket(ws, response, isCurrent, {
               untilEof: nativeControl?.relayActive === true,
-              onSsePayload: nativeControl?.relayActive
-                ? createNativeSteeringLogObserver(logCtx, () => recordFirstOutput(logCtx, start))
-                : payload => inspectResponseLogSsePayload(logCtx, payload),
+              onSsePayload: payload => {
+                ws.data.responsesActivity?.observePayload(payload, providerExecuted);
+                observeLog(payload);
+              },
               onTerminal: status => {
                 terminalRecorder?.(status, logCtx.terminalHttpStatus);
                 finalizeLog(httpStatusForRequestLogTerminal(status, logCtx), {
@@ -392,6 +405,7 @@ export function createWebsocketHandler(
               /* socket already gone or send dropped */
             }
           } finally {
+            finishActivity?.();
             turnAdmissionLease.release();
             if (ws.data.nativeControl === nativeControl) ws.data.nativeControl = undefined;
             if (!logged && turnAbort.signal.aborted) finalizeLog(499);

src/server/lifecycle.ts

@@ -38,6 +38,7 @@ export interface ActiveTurnLease extends AdmissionLease {
   attach(lease: AdmissionLease): void;
   bindAbortSignal(signal: AbortSignal): () => void;
   bindAbortController(ac: AbortController): void;
+  unbindAbortController(ac: AbortController): void;
   beginCodexAccountSelection(): CodexAccountSelectionAdmission;
   isTransferred(): boolean;
 }
@@ -231,6 +232,17 @@ export function tryAdmitTurn(sessionLaneId?: string, clientSignal?: AbortSignal)
       activeTurns.set(ac, lease);
       lease.bindAbortSignal(ac.signal);
     },
+    unbindAbortController(ac) {
+      if (!controllers.delete(ac)) return;
+      if (activeTurns.get(ac) === lease) activeTurns.delete(ac);
+      const listener = abortListeners.get(ac.signal);
+      if (listener) {
+        ac.signal.removeEventListener("abort", listener);
+        abortListeners.delete(ac.signal);
+      }
+      // A completed child relay cannot retire its still-running parent turn.
+      if (controllers.size === 0) lease.release();
+    },
     beginCodexAccountSelection() {
       const mainProfileDraining = nativeMainDrainOwners.size > 0;
       let selectionActive = active && !mainProfileDraining;
@@ -327,7 +339,7 @@ export function unregisterTurn(ac: AbortController): void {
     turnReleaseMisses += 1;
     return;
   }
-  lease.release();
+  lease.unbindAbortController(ac);
 }
 export function isDraining(): boolean { return shutdownDraining || temporaryDrainOwners.size > 0; }
 export function getActiveTurnCount(): number { return turnGate.metrics().active; }

src/server/responses-undeclared-tool-guard.ts

@@ -298,7 +298,7 @@ export function collectProviderExecutedCallTypes(body: unknown): Set<ProviderExe
   return callTypes;
 }
 
-function isAuthorizedProviderExecutedCall(
+export function isAuthorizedProviderExecutedCall(
   item: Record<string, unknown>,
   callTypes: ProviderExecutedCallTypes,
 ): boolean {

src/server/responses/combo-stream-preflight.ts

@@ -4,6 +4,7 @@ import { comboFailureDecision } from "../../combos";
 import { httpStatusFromTerminalError } from "../../lib/errors";
 import type { RequestFailureStage } from "../../lib/request-failure-model";
 import type { RequestLogContext } from "../request-log";
+import { carryResponseToolOwnership } from "./response-tool-ownership";
 import { createSseInspector } from "../relay";
 import { MAX_CLIENT_SSE_FRAME_BYTES } from "../sse-frame-buffer";
 
@@ -198,11 +199,11 @@ function replayBufferedResponse(
       reader.cancel(reason).catch(() => undefined);
     },
   });
-  return new Response(body, {
+  return carryResponseToolOwnership(response, new Response(body, {
     status: response.status,
     statusText: response.statusText,
     headers: response.headers,
-  });
+  }));
 }
 
 function failedTerminalResponse(