diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index efe171ec163..e156f0a346b 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -12,6 +12,7 @@ import { createInferenceSendBudget } from "../inference/context"; import { finalizeOwnedTranslatorBudget, finalizeAccountLease } from "./core-lifetime"; import type { TranslatorBudget } from "../../lib/translator-budget"; import { executeComboResponses } from "./core-combo"; +import { registerTurn, unregisterTurn } from "../lifecycle"; import { runWithCompactionRecovery } from "./compaction-recovery"; /** * Route one `/v1/responses` request through the adapter pipeline: recovery loop, passthrough @@ -67,15 +68,17 @@ export async function handleComboResponses( logCtx: RequestLogContext, options: HandleResponsesOptions & { translatorBudget: TranslatorBudget }, ): Promise { - return executeComboResponses( - req, - rawBody, - comboId, - config, - logCtx, - options, - requestDispatchers, - ); + // Each child relay binds the turn's lease and unbinds it when its stream ends; the lease is + // released once nothing is bound. Hold it for the whole ladder so a refused child's relay + // cannot settle the turn the next target still runs in ("turn already settled"). An accepted + // child's stream is bound before this returns and keeps the lease until it ends. + const ladder = new AbortController(); + registerTurn(ladder, options.turnAdmissionLease); + try { + return await executeComboResponses(req, rawBody, comboId, config, logCtx, options, requestDispatchers); + } finally { + if (options.turnAdmissionLease) unregisterTurn(ladder); + } } const requestDispatchers: ResponsesDispatchers = { handleResponses, handleComboResponses }; export { adapterNeedsForcedContinuation } from "./core-replay"; diff --git a/tests/server/server-combo-zero-output-failover.test.ts b/tests/server/server-combo-zero-output-failover.test.ts index 72afd27a812..47b7b43986a 100644 --- a/tests/server/server-combo-zero-output-failover.test.ts +++ b/tests/server/server-combo-zero-output-failover.test.ts @@ -12,6 +12,7 @@ import { clearComboRecallForTests } from "../../src/server/responses/combo-sessi import { clearKeyCooldowns } from "../../src/providers/key-failover"; import { clearCodexUpstreamHealth } from "../../src/codex/routing"; import { clearRequestLogsForTests, type RequestLogContext } from "../../src/server/request-log"; +import { getActiveTurnCount, tryAdmitTurn } from "../../src/server/lifecycle"; import { clearResponseStateForTests, flushResponseState, @@ -220,6 +221,33 @@ describe("combo zero-output bare Responses error failover", () => { }); }); + test("a refused child relay does not settle the parent turn's admission lease", async () => { + // Each passthrough child binds the parent's lease to its own relay controller. Cancelling the + // refused first stream unbinds that controller; it must not settle the turn the next target + // still runs in, or the second child is aborted on arrival ("turn already settled"). + const sse = (...events: unknown[]) => new Response(events.map(event => `data: ${JSON.stringify(event)}\n\n`).join(""), + { headers: { "content-type": "text/event-stream" } }); + const a = serve(() => sse({ type: "response.created", response: { id: "ra", status: "in_progress" } }, { type: "response.failed", + response: { id: "ra", status: "failed", error: { type: "server_error", code: "upstream_server_error", message: "busy" } } })); + const b = serve(() => sse({ type: "response.completed", response: responsesSuccess("lease backup", "m2") })); + const config = comboConfig({ + a: provider("openai-responses", baseUrl(a), "key-a"), + b: provider("openai-responses", baseUrl(b), "key-b"), + }); + const turnAdmissionLease = tryAdmitTurn(); + if (!turnAdmissionLease) throw new Error("test turn admission unavailable"); + try { + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "combo/free", input: "hello", stream: true }), + }), config, { model: "", provider: "" }, { turnAdmissionLease }); + expect(response.status).toBe(200); + expect(await response.text()).toContain("lease backup"); + expect(getActiveTurnCount()).toBe(0); + } finally { turnAdmissionLease.release(); } + }); + test("undeclared first adapter tool call hops without changing the request catalog", async () => { const requests: Record[] = []; const upstream = (content: string) => serve(async request => {