diff --git a/src/server/index.ts b/src/server/index.ts index 099991d272..0c5bf7ae59 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -1186,7 +1186,6 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server { const response = await handleResponses(req, config, logCtx, { turnAdmissionLease, + onRequestBodyRead: () => disableResponsesRequestTimeout(req, requestServer), abortSignal: req.signal, onFirstOutput: () => recordFirstOutput(logCtx, start), onNativePassthroughTerminal: status => { diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index ca928e21cf..cc8b38d900 100644 --- a/src/server/responses/core.ts +++ b/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 } : {}), diff --git a/src/server/responses/policy-fallback.ts b/src/server/responses/policy-fallback.ts index 7e9cbd3c74..05a1627d81 100644 --- a/src/server/responses/policy-fallback.ts +++ b/src/server/responses/policy-fallback.ts @@ -114,6 +114,17 @@ export async function handleResponsesWithPolicyFallback( deps: PolicyFallbackDeps = {}, ): Promise { 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 | 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; diff --git a/tests/routing-policy-fallback.test.ts b/tests/routing-policy-fallback.test.ts index 8ba1193e1b..cc0c511999 100644 --- a/tests/routing-policy-fallback.test.ts +++ b/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 = []; + 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"); diff --git a/tests/server-auth.test.ts b/tests/server-auth.test.ts index 1a5f78fc7a..4bc5ac0b0f 100644 --- a/tests/server-auth.test.ts +++ b/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; + const body = new ReadableStream({ + 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; + const body = new ReadableStream({ + 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);