From 2d3bef3ba23e89d452652b05419eb501a46f4561 Mon Sep 17 00:00:00 2001
From: JUN <bitkyc08@gmail.com>
Date: Mon, 14 Sep 2026 18:54:42 +0900
Subject: [PATCH] feat(server): reserve interactive capacity from a worker
 fan-out (#4546)

Refs #4546. PRD R06/wp5, second slice, completing the module landed in #4612.

A fan-out shares the conversation it serves. Without a reserve, a worker burst takes every
concurrency slot under its own root and the interactive turn that started it waits behind its own
children. runAdmittedHttpTurn now admits each turn against the root workflow as well as the
process-wide turn gate: a request that names a parent thread distinct from its own is treated as
that fan-out and may not take the reserved slots, while a top-level request is the conversation
and may.

The refusal is a local queue-capacity answer, not a synthetic upstream error, and the lease is
released on both the normal and the throwing path so a failed turn cannot leak a slot.
---
 src/server/index.ts | 24 ++++++++++++++++++++++++
 1 file changed, 24 insertions(+)

diff --git a/src/server/index.ts b/src/server/index.ts
index 577e8189f1..8a496d2c84 100644
--- a/src/server/index.ts
+++ b/src/server/index.ts
@@ -134,6 +134,7 @@ import {
   type RequestLogEntry,
 } from "./request-log";
 import { sessionLaneIdFromRequest } from "./request-log-conversation";
+import { admitWorkflowTurn, type WorkflowLane } from "../lib/workflow-budget";
 export {
   addFinalRequestLog,
   filterRequestLogs,
@@ -1292,13 +1293,36 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
   ): Promise<Response> {
     const lease = tryAdmitTurn(sessionLaneIdFromRequest(req.headers));
     if (!lease) return serverBusyResponse(req, "active turns", policy);
+    // A fan-out shares the conversation it serves. Without a reserve, a worker burst takes every
+    // slot under its own root and the interactive turn that started it waits behind its own
+    // children. A request that names a parent is treated as that fan-out; a top-level request is
+    // the conversation and may use the reserved slots.
+    const workflowRootId = req.headers.get("x-codex-parent-thread-id")?.trim() || undefined;
+    const workflowThreadId = req.headers.get("thread-id")?.trim() || undefined;
+    const workflowLane: WorkflowLane = workflowRootId !== undefined
+      && workflowThreadId !== undefined
+      && workflowThreadId !== workflowRootId
+      ? "worker"
+      : "interactive";
+    const workflow = admitWorkflowTurn(workflowRootId, workflowLane, undefined, workflowThreadId);
+    if (workflow && !workflow.admitted) {
+      lease.release();
+      return formatErrorResponse(
+        429,
+        workflow.reason === "workflow-sends-exhausted" ? "workflow_budget_exhausted" : "queue_capacity_exceeded",
+        "This task has reached its concurrent-work limit, so no further upstream request was made. Work already in flight settles as it finishes.",
+      );
+    }
+    const releaseWorkflow = (): void => { if (workflow?.admitted) workflow.lease.release(); };
     let response: Response;
     try {
       response = await work(lease);
     } catch (error) {
+      releaseWorkflow();
       lease.release();
       throw error;
     }
+    releaseWorkflow();
     if (!lease.isTransferred()) {
       lease.release();
     }