fix(server): bound Responses body-read timeout

src/server/index.ts

@@ -1186,7 +1186,6 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
       }
 
       if (url.pathname === "/v1/responses" && req.method === "POST") {
-        disableResponsesRequestTimeout(req, requestServer);
         if (isDraining()) {
           return drainingResponse(req, policy);
         }
@@ -1215,6 +1214,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         return runAdmittedHttpTurn(req, policy, async turnAdmissionLease => {
           const response = await handleResponses(req, config, logCtx, {
             turnAdmissionLease,
+            onRequestBodyRead: () => disableResponsesRequestTimeout(req, requestServer),
             abortSignal: req.signal,
             onFirstOutput: () => recordFirstOutput(logCtx, start),
             onNativePassthroughTerminal: status => {

src/server/responses/core.ts

@@ -719,6 +719,8 @@ export interface ConsumedComboFailure {
 
 export interface HandleResponsesOptions {
   turnAdmissionLease?: AdmissionLease;
+  /** Called at most once after the complete client body is read and accepted for dispatch. */
+  onRequestBodyRead?: () => void;
   forceEmptyResponseId?: boolean;
   abortSignal?: AbortSignal;
   /** One-shot TTFT callback: first non-empty model output observed (WP4). */
@@ -1495,11 +1497,20 @@ async function handleResponsesInner(
   try {
     body = await readJsonRequestBody(req, translatorBudget);
   } catch (err) {
+    if (options.abortSignal?.aborted || req.signal.aborted) {
+      return clientCancelledResponse();
+    }
     return decodeRequestErrorResponse(err, "responses");
   }
   const comboId = !options.comboAttempt ? comboIdFromRawBody(body, config) : null;
   if (comboId && Object.hasOwn(config.combos ?? {}, comboId)) {
-    return handleComboResponses(req, body, comboId, config, logCtx, options);
+    options.onRequestBodyRead?.();
+    return handleComboResponses(req, body, comboId, config, logCtx, {
+      ...options,
+      // The original request body was accepted above. Combo children are synthetic
+      // replays and must not repeat the caller-owned timeout transition.
+      onRequestBodyRead: undefined,
+    });
   }
   let unreadableEncryptedAgentTask = hasUnreadableEncryptedAgentTask(
     (body as { input?: unknown } | undefined)?.input,
@@ -1555,6 +1566,7 @@ async function handleResponsesInner(
     }
     return formatErrorResponse(400, "invalid_request_error", err instanceof Error ? err.message : String(err));
   }
+  options.onRequestBodyRead?.();
   const responseStateOptions = (force = false): { force?: boolean; clientThreadId?: string } => ({
     ...(force ? { force: true } : {}),
     ...(parsed._clientThreadId ? { clientThreadId: parsed._clientThreadId } : {}),

src/server/responses/policy-fallback.ts

@@ -114,6 +114,17 @@ export async function handleResponsesWithPolicyFallback(
   deps: PolicyFallbackDeps = {},
 ): Promise<Response> {
   const runCore = deps.runCore ?? handleResponsesCore;
+  let requestBodyReadNotified = false;
+  const coreOptions: CoreOptions = options.onRequestBodyRead
+    ? {
+      ...options,
+      onRequestBodyRead: () => {
+        if (requestBodyReadNotified) return;
+        requestBodyReadNotified = true;
+        options.onRequestBodyRead?.();
+      },
+    }
+    : options;
   let rawBody: Record<string, unknown> | null = null;
   try {
     const parsed = await readJsonRequestBody(req.clone());
@@ -122,7 +133,7 @@ export async function handleResponsesWithPolicyFallback(
     // Core owns the client-facing parse/decompression error.
   }
 
-  let response = await runCore(req, config, logCtx, options);
+  let response = await runCore(req, config, logCtx, coreOptions);
   const initialTrace = logCtx.routeDecision;
   const initialRequestedModel = logCtx.requestedModel;
   if (!rawBody || !isPolicyDecision(initialTrace)) return response;
@@ -140,7 +151,7 @@ export async function handleResponsesWithPolicyFallback(
     finishFailedPolicyAttempt(logCtx, response.status);
     const retryRequest = requestWithCandidate(req, rawBody, next);
     try {
-      response = await runCore(retryRequest, config, logCtx, options);
+      response = await runCore(retryRequest, config, logCtx, coreOptions);
     } finally {
       logCtx.requestedModel = initialRequestedModel;
       logCtx.routeDecision = initialTrace;

tests/routing-policy-fallback.test.ts

@@ -58,9 +58,15 @@ describe("policy candidate fallback", () => {
     const logCtx = { requestedModel: "policy/daily", routeDecision: trace, attempts: [] } as unknown as RequestLogContext;
     const seenModels: string[] = [];
     const seenTerminalCodes: Array<string | undefined> = [];
+    let bodyAcceptedCount = 0;
 
-    const response = await handleResponsesWithPolicyFallback(request(), {} as OcxConfig, logCtx, {}, {
-      runCore: async (req, _config, childLog) => {
+    const response = await handleResponsesWithPolicyFallback(request(), {} as OcxConfig, logCtx, {
+      onRequestBodyRead: () => {
+        bodyAcceptedCount += 1;
+      },
+    }, {
+      runCore: async (req, _config, childLog, options) => {
+        options.onRequestBodyRead?.();
         const body = await req.json() as { model: string };
         seenModels.push(body.model);
         seenTerminalCodes.push(childLog.terminalErrorCode);
@@ -82,6 +88,7 @@ describe("policy candidate fallback", () => {
     });
 
     expect(response.status).toBe(200);
+    expect(bodyAcceptedCount).toBe(1);
     expect(seenModels).toEqual(["policy/daily", "provider-b/model-b"]);
     expect(seenTerminalCodes).toEqual([undefined, undefined]);
     expect(logCtx.requestedModel).toBe("policy/daily");

tests/server-auth.test.ts

@@ -36,6 +36,7 @@ import {
 } from "../src/server";
 import { clearRequestLogsForTests, getRequestLogEntries } from "../src/server/request-log";
 import { handleManagementAPI } from "../src/server/management-api";
+import { handleResponses } from "../src/server/responses";
 import type { OcxConfig } from "../src/types";
 import { fakeChatGptJwt } from "./helpers/fake-chatgpt-jwt";
 import { installIsolatedCodexHome, type IsolatedCodexHome } from "./helpers/isolated-codex-home";
@@ -352,6 +353,132 @@ describe("server local API auth", () => {
     })).toBe(false);
   });
 
+  test("responses handler keeps the request timeout until the body is fully accepted", async () => {
+    let controller!: ReadableStreamDefaultController<Uint8Array>;
+    const body = new ReadableStream<Uint8Array>({
+      start(value) {
+        controller = value;
+      },
+    });
+    const cfg = config();
+    cfg.defaultProvider = "fixture";
+    cfg.providers = {
+      fixture: { ...cfg.providers.openai!, disabled: true },
+    };
+    const req = new Request("http://localhost/v1/responses", {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body,
+    });
+    let accepted = false;
+    const responsePromise = handleResponses(req, cfg, {
+      model: "unknown",
+      provider: "unknown",
+    }, {
+      onRequestBodyRead: () => {
+        accepted = true;
+      },
+    });
+
+    controller.enqueue(new TextEncoder().encode('{"model":"fixture/gpt-test","input":"hello"'));
+    await Bun.sleep(10);
+    expect(accepted).toBe(false);
+
+    controller.enqueue(new TextEncoder().encode("}"));
+    controller.close();
+    const response = await responsePromise;
+    expect(accepted).toBe(true);
+    expect(response.status).toBe(404);
+  });
+
+  test("responses handler classifies an aborted pending body as client cancellation", async () => {
+    let bodyController!: ReadableStreamDefaultController<Uint8Array>;
+    const body = new ReadableStream<Uint8Array>({
+      start(value) {
+        bodyController = value;
+      },
+    });
+    const abortController = new AbortController();
+    const req = new Request("http://localhost/v1/responses", {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body,
+      signal: abortController.signal,
+    });
+    let accepted = false;
+    const responsePromise = handleResponses(req, config(), {
+      model: "unknown",
+      provider: "unknown",
+    }, {
+      abortSignal: abortController.signal,
+      onRequestBodyRead: () => {
+        accepted = true;
+      },
+    });
+
+    bodyController.enqueue(new TextEncoder().encode('{"model":"openai/gpt-test","input":"hello"'));
+    await Bun.sleep(10);
+    expect(accepted).toBe(false);
+
+    abortController.abort();
+    const response = await responsePromise;
+    expect(response.status).toBe(499);
+    expect(accepted).toBe(false);
+  });
+
+  test("responses handler accepts a combo body exactly once across failover children", async () => {
+    const upstreamModels: string[] = [];
+    const upstream = Bun.serve({
+      hostname: "127.0.0.1",
+      port: 0,
+      async fetch(request) {
+        const body = await request.json() as { model?: string };
+        upstreamModels.push(body.model ?? "missing");
+        return Response.json({ error: { message: "rate limited; try the next combo target" } }, {
+          status: 429,
+          headers: { "retry-after": "1" },
+        });
+      },
+    });
+    const baseUrl = `${upstream.url.toString().replace(/\/$/, "")}/v1`;
+    const cfg: OcxConfig = {
+      port: 0,
+      defaultProvider: "first",
+      providers: {
+        first: { adapter: "openai-responses", baseUrl, apiKey: "first-key", allowPrivateNetwork: true },
+        second: { adapter: "openai-responses", baseUrl, apiKey: "second-key", allowPrivateNetwork: true },
+      },
+      combos: {
+        request_timeout: {
+          strategy: "failover",
+          targets: [
+            { provider: "first", model: "first-model" },
+            { provider: "second", model: "second-model" },
+          ],
+        },
+      },
+    };
+    let acceptedCount = 0;
+
+    try {
+      const response = await handleResponses(new Request("http://localhost/v1/responses", {
+        method: "POST",
+        headers: { "content-type": "application/json" },
+        body: JSON.stringify({ model: "combo/request_timeout", input: "hello", stream: false }),
+      }), cfg, { model: "unknown", provider: "unknown" }, {
+        onRequestBodyRead: () => {
+          acceptedCount += 1;
+        },
+      });
+
+      expect(response.status).toBe(429);
+      expect(acceptedCount).toBe(1);
+      expect(upstreamModels).toEqual(["first-model", "second-model"]);
+    } finally {
+      await upstream.stop(true);
+    }
+  });
+
   test("loopback hostnames do not require opencodex API auth", () => {
     expect(isLoopbackHostname(undefined)).toBe(true);
     expect(isLoopbackHostname("")).toBe(true);