fix(workflow-budget): reserve physical sends atomically

docs-site/src/content/docs/reference/configuration/server.md

@@ -95,7 +95,7 @@ OpenCodex bounds how much one task may fan out. A task is the root conversation,
 | Field | Default | Allowed | Meaning |
 | --- | --- | --- | --- |
 | `maxConcurrentChildren` | `8` | 2–64 | Turns one task may run at the same time. One slot is always reserved for the task's own conversation, so subagents may use one fewer. |
-| `maxPhysicalSends` | `256` | 1–10000 | Upstream sends, retries included, after which the task's new requests are refused for the rest of the 10-minute window. The count is checked when a request is admitted, so retries inside a request that was already admitted, and requests admitted concurrently, can take the total past the ceiling. |
+| `maxPhysicalSends` | `256` | 1–10000 | Upstream send admissions, retries included, per 10-minute window. Every physical admission checks and claims a slot atomically, so concurrent requests and retries share the same ceiling. An unused dispatch permit consumes no slot; a claimed physical start remains counted even if it later fails. |
 | `maxDistinctChildren` | `64` | 1–1024 | Distinct subagent threads one task may start inside the same window. |
 
 The defaults suit most installs. Raise them when a workflow that legitimately runs many subagents in parallel keeps hitting a ceiling; lower them to stop a runaway fan-out sooner:

scripts/test-layout/layout-responses.json

@@ -8,5 +8,6 @@
   "passthrough-headers.test.ts": "responses",
   "passthrough-abort.test.ts": "responses",
   "current-turn-tool-catalog-direct.test.ts": "responses",
-  "current-turn-tool-catalog-namespace.test.ts": "responses"
+  "current-turn-tool-catalog-namespace.test.ts": "responses",
+  "responses-workflow-send-ceiling.test.ts": "responses"
 }

src/lib/request-execution-budget.ts

@@ -140,6 +140,10 @@ export type DispatchDecision =
  * send rather than merely describe one.
  */
 export interface RequestSendObserver {
+  /** Atomically claim the root workflow count at physical admission, before inference I/O. */
+  beforePhysicalSend?(): boolean;
+  /** Record an already-sent legacy receipt without booking its durable spend twice. */
+  recordWorkflowSend?(): void;
   /**
    * Book one physical send. False refuses the dispatch before the budget charges it.
    *
@@ -149,7 +153,7 @@ export interface RequestSendObserver {
    * would cross a ceiling never joins the total, the total stays just under, and the ceiling
    * never fires for any later request either.
    */
-  charge(options?: { alreadySent?: boolean; deferDispatch?: boolean; onReserved?: (proof: SpendReservationProof) => void; targetKey?: string }): boolean;
+  charge(options?: { alreadySent?: boolean; deferDispatch?: boolean; workflowAlreadyCounted?: boolean; onReserved?: (proof: SpendReservationProof) => void; targetKey?: string }): boolean;
   /** Confirm the exact reservation once its dispatch is known. */
   readonly enforced?: boolean;
   readonly policyStarted?: boolean;
@@ -172,10 +176,12 @@ export interface RequestExecutionBudget extends TransientSendBudget {
   readonly physicalStarted?: number;
   readonly physicalLimit?: number;
   readonly spendEnforced?: boolean;
+  readonly workflowCountObserved?: boolean;
   readonly spendPolicyStarted?: boolean;
   readonly spendAdmissionPolicy?: SpendAdmissionPolicy;
   startRequest?(target?: SpendScopes & { targetKey?: string }): void;
   claimPhysicalSend?(): number | undefined;
+  claimWorkflowSend?(): boolean;
   beginSpendProducer?(): { close(): void } | undefined;
   readonly logicalRequestId: string;
   readonly policyVersion: string;
@@ -264,6 +270,10 @@ interface SharedSendLedger {
 }
 
 const sharedSendLedgers = new WeakMap<RequestExecutionBudget, SharedSendLedger>();
+/** Whether physical admission owns this budget's workflow accounting. */
+export function hasWorkflowSendObserver(budget: TransientSendBudget): boolean {
+  return (budget as RequestExecutionBudget).workflowCountObserved === true;
+}
 const dispatchSpendProofs = new WeakMap<SingleUseDispatchPermit, {
   owner: SharedSendLedger;
   claim(): SpendReservationProof | undefined;
@@ -281,14 +291,20 @@ export function reportDispatchSends(
   budget: TransientSendBudget,
   sends: number,
   permit?: SingleUseDispatchPermit,
+  workflowAlreadyCounted = false,
 ): void {
   const count = Number.isFinite(sends) ? Math.max(0, Math.trunc(sends)) : 0;
   if (count === 0) return;
   const receipt = permit && dispatchSpendProofs.get(permit);
   const prepaid = receipt && receipt.owner === sharedSendLedgers.get(budget as RequestExecutionBudget)
     && receipt.report() ? 1 : 0;
   // Unnamed, foreign, released or already-reported permits cannot consume another receipt.
-  budget.used += count - prepaid;
+  const counter = sharedSendLedgers.get(budget as RequestExecutionBudget);
+  if (!workflowAlreadyCounted && prepaid > 0) counter?.observer?.recordWorkflowSend?.();
+  if (workflowAlreadyCounted && counter && counter.observer?.enforced !== true) {
+    counter.spent += count - prepaid;
+    for (let index = prepaid; index < count; index++) counter.observer?.charge({ alreadySent: true, workflowAlreadyCounted: true });
+  } else budget.used += count - prepaid;
 }
 
 /** A helper owns its starts and can never consume another helper's receipts. */
@@ -319,7 +335,16 @@ export function createPhysicalSendReporter(
   let currentOrdinal: number | undefined;
   const report: PhysicalSendReporter = (sends) => {
     if (closed) return;
-    if (!producer) { if (observer?.enforced !== true) { reportDispatchSends(budget, sends, permit); telemetry?.(sends); } return; }
+    if (!producer) {
+      if (observer?.enforced === true) return;
+      const count = Number.isFinite(sends) ? Math.max(0, Math.trunc(sends)) : 0;
+      const admitted = Math.min(count, started - reported);
+      if (admitted > 0) reportDispatchSends(budget, admitted, permit, true);
+      if (count > admitted) reportDispatchSends(budget, count - admitted, permit);
+      reported += admitted;
+      telemetry?.(count);
+      return;
+    }
     const count = Math.min(Math.max(0, Math.trunc(sends)), started - reported);
     try { producer.report(count); }
     finally {
@@ -335,7 +360,11 @@ export function createPhysicalSendReporter(
     const selected = target();
     if ((budget as RequestExecutionBudget).startRequest) (budget as RequestExecutionBudget).startRequest!(selected);
     else observer?.startRequest?.(selected);
-    if (observer?.enforced !== true) return true;
+    if (observer?.enforced !== true) {
+      if (observer?.beforePhysicalSend?.() === false) return false;
+      started++;
+      return true;
+    }
     producer ??= observer.beginReporter?.();
     const receipt = permit && dispatchSpendProofs.get(permit);
     const proof = receipt?.owner === counter ? receipt?.claim() : undefined;
@@ -409,6 +438,7 @@ function createRequestExecutionBudgetWithLedger(
     get physicalStarted(): number { return counter.physicalStarted; },
     get physicalLimit(): number | undefined { return counter.physicalLimit; },
     get spendEnforced(): boolean { return observer?.enforced === true; },
+    get workflowCountObserved(): boolean { return observer?.beforePhysicalSend !== undefined; },
     get spendPolicyStarted(): boolean { return observer?.policyStarted === true; },
     get spendAdmissionPolicy(): SpendAdmissionPolicy | undefined { return observer?.spendAdmissionPolicy; },
     startRequest(target) {
@@ -418,10 +448,14 @@ function createRequestExecutionBudgetWithLedger(
       }
     },
     beginSpendProducer() { return observer?.enforced ? observer.beginReporter?.() : undefined; },
+    claimWorkflowSend(): boolean {
+      budget.startRequest?.();
+      return observer?.beforePhysicalSend?.() ?? true;
+    },
     claimPhysicalSend(): number | undefined {
       budget.startRequest?.();
       if (observer?.enforced !== true || counter.physicalLimit === undefined) return undefined;
-      if (counter.physicalStarted >= counter.physicalLimit) return undefined;
+      if (counter.physicalStarted >= counter.physicalLimit || !budget.claimWorkflowSend!()) return undefined;
       return ++counter.physicalStarted;
     },
     get used(): number { return counter.spent; },
@@ -512,7 +546,7 @@ function createRequestExecutionBudgetWithLedger(
       let executing = false;
       let physicalOrdinal: number | undefined;
       const startPhysical = (): boolean => {
-        if (observer?.enforced !== true) return true;
+        if (observer?.enforced !== true) return budget.claimWorkflowSend!();
         const seed = observer.ensureSeed?.({ targetKey: intent.targetKey }, spendProof);
         if (!seed) return false;
         const ordinal = budget.claimPhysicalSend?.();

src/lib/workflow-budget.ts

@@ -671,6 +671,28 @@ export function chargeWorkflowSends(
   state.lastSeenMs = now;
 }
 
+/**
+ * Atomically reserve one physical send under a root's live count ceiling.
+ *
+ * JavaScript runs this read-and-increment without an await, so concurrent request and retry
+ * ladders cannot all observe the same remaining slot. Call only at physical admission, after
+ * unused dispatch permits can be released; a claimed physical start remains counted.
+ */
+export function reserveWorkflowSend(
+  rootId: string | undefined,
+  policy: WorkflowBudgetPolicy = DEFAULT_WORKFLOW_BUDGET_POLICY,
+  now: number = Date.now(),
+): boolean {
+  if (!rootId) return true;
+  const state = roots.get(rootId);
+  if (!state) return true;
+  if (windowedSends(state, now) >= policy.maxPhysicalSends) return false;
+  state.sends += 1;
+  recordWindowedSends(state, now, 1);
+  state.lastSeenMs = now;
+  return true;
+}
+
 /**
  * Settle a send's reservation with the usage the response actually reported. Idempotent
  * per send id -- a second call returns false and books nothing. When the usage frame was

src/server/chat-native.ts

@@ -471,9 +471,9 @@ export async function runNativeChatAttempt(
         throw new SendBudgetExhaustedError(errorDestinationLabel(url));
       }
       physicalSends += 1;
-      reportDispatchSends(sendBudget, 1, execution.comboDispatchPermit);
+      reportDispatchSends(sendBudget, 1, execution.comboDispatchPermit, true);
       initialSendStarted = true;
-    } else if (!spendTracker?.charge()) throw new NativeChatSpendRefusal();
+    } else if (!spendTracker?.charge({ workflowAlreadyCounted: true })) throw new NativeChatSpendRefusal();
   };
 
   let responseOwnsDistributedKey = true;
@@ -704,7 +704,8 @@ export async function runNativeChatAttempt(
       return failLocal(429, sendError.message, SEND_BUDGET_EXHAUSTED_CODE, SEND_BUDGET_EXHAUSTED_CODE);
     }
     if (sendError instanceof NativeChatSpendRefusal) {
-      const refusal = workflowRefusalResponse("workflow-spend-exhausted", logCtx);
+      const refusal = workflowRefusalResponse(logCtx.errorCode === "workflow_sends_exhausted"
+        ? "workflow-sends-exhausted" : "workflow-spend-exhausted", logCtx);
       finishLog(429);
       return refusal;
     }

src/server/messages-native.ts

@@ -639,7 +639,7 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
                 if (spendTracker.refusals) throw new NativeMessagesSpendRefusal();
                 throw new SendBudgetExhaustedError();
               }
-              if (!physicalBudget.spendEnforced && !spendTracker.charge()) throw new NativeMessagesSpendRefusal();
+              if (!physicalBudget.spendEnforced && !spendTracker.charge({ workflowAlreadyCounted: true })) throw new NativeMessagesSpendRefusal();
               noteProviderAttemptSend(logCtx, route.providerName, activeProvider, logCtx.usageLogInputTokens, transportRecovery ?? recovery);
               // Last synchronous ownership reservation before the physical fetch. Retain this
               // exact incarnation across its await; a returned response can never reserve one.
@@ -844,7 +844,8 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
     }
     if (sendError instanceof SendBudgetExhaustedError) return fail(429, sendError.message, "send_budget_exhausted", "send_budget_exhausted");
     if (sendError instanceof NativeMessagesSpendRefusal) {
-      const refusal = workflowRefusalResponse("workflow-spend-exhausted", logCtx);
+      const refusal = workflowRefusalResponse(logCtx.errorCode === "workflow_sends_exhausted"
+        ? "workflow-sends-exhausted" : "workflow-spend-exhausted", logCtx);
       finishLog(429);
       return refusal;
     }