From 5fd1590105b386608713931c5d849f9c497d1844 Mon Sep 17 00:00:00 2001 From: JUN Date: Tue, 15 Sep 2026 04:18:58 +0900 Subject: [PATCH 1/4] fix(lib): bound the root workflow ceilings by a window instead of a lifetime (#4546) state.sends only grew and state.children was a Set only ever added to, so with the root id being the caller thread the cap became a session expiry: a Codex session that reached 256 sends was refused for the rest of the process even after hours idle, curable only by restarting the proxy. The cap was written against a burst, and a burst is a rate. Sends now go into a bounded twelve-slot ring and distinct children into a last-seen map pruned on read, both measured over a ten-minute window. maxConcurrentChildren is untouched because it is already instantaneous. A count inside a window is never larger than the lifetime count, so no install sees a new refusal; that is asserted rather than argued. Verification posture: local suite, typecheck, install and build NOT run by explicit instruction. Hosted CI at the exact final head is the only proof. Pushed with --no-verify. --- src/lib/workflow-budget.ts | 160 +++++++++++++++++++++++++++--- tests/lib/workflow-budget.test.ts | 102 +++++++++++++++++++ 2 files changed, 247 insertions(+), 15 deletions(-) diff --git a/src/lib/workflow-budget.ts b/src/lib/workflow-budget.ts index 8ca435352c..657c3be241 100644 --- a/src/lib/workflow-budget.ts +++ b/src/lib/workflow-budget.ts @@ -29,10 +29,25 @@ import { export interface WorkflowBudgetPolicy { /** Children admitted concurrently under one root. */ readonly maxConcurrentChildren: number; - /** Physical model sends charged to one root across its whole life. */ + /** Physical model sends charged to one root INSIDE {@link WorkflowBudgetPolicy.windowMs}. */ readonly maxPhysicalSends: number; - /** Distinct children one root may ever create. */ + /** Distinct children one root may have inside the same window. */ readonly maxDistinctChildren: number; + /** + * The interval both counts are measured over. + * + * These were lifetime totals, and a lifetime total is the wrong instrument. The cap was + * written against a fan-out that sends once per child seven hundred times, which is a RATE; + * a running total cannot tell that from an ordinary session spread over an afternoon and + * refuses both. Because the root id is the caller thread, for Codex that made the ceiling a + * session expiry: a session reaching it was refused for the rest of the process even after + * going idle for hours, and the only cure was restarting the proxy. + * + * Omitted means {@link WORKFLOW_DEFAULT_WINDOW_MS}. A count inside a window is never larger + * than the same count over a lifetime, so windowing can only ever admit more for identical + * traffic -- no install sees a refusal it would not have seen before. + */ + readonly windowMs?: number; /** * Concurrency slots a fan-out may never take. An interactive turn arriving into a saturated * root still gets admitted; without this a worker burst starves the conversation it serves. @@ -47,14 +62,85 @@ export interface WorkflowBudgetPolicy { readonly maxTrackedRoots: number; } +/** + * Ten minutes. Long enough that the burst this ceiling was written against -- seven hundred + * sends in a minute -- is still refused several times over, and short enough that an ordinary + * session, which averages far less than a send every two seconds, never approaches it. + */ +export const WORKFLOW_DEFAULT_WINDOW_MS = 10 * 60_000; + export const DEFAULT_WORKFLOW_BUDGET_POLICY: WorkflowBudgetPolicy = { maxConcurrentChildren: 8, maxPhysicalSends: 256, maxDistinctChildren: 64, interactiveReserve: 1, maxTrackedRoots: 512, + windowMs: WORKFLOW_DEFAULT_WINDOW_MS, }; +/** Fixed ring size. Ten minutes over twelve slots gives fifty-second granularity. */ +const WORKFLOW_WINDOW_SLOTS = 12; + +function workflowWindowMs(policy: WorkflowBudgetPolicy): number { + const declared = policy.windowMs; + return declared !== undefined && Number.isFinite(declared) && declared > 0 + ? declared + : WORKFLOW_DEFAULT_WINDOW_MS; +} + +function windowSlotMs(policy: WorkflowBudgetPolicy): number { + return Math.max(1, Math.ceil(workflowWindowMs(policy) / WORKFLOW_WINDOW_SLOTS)); +} + +/** + * Add sends to the ring, resetting a slot whose turn has come round again. + * + * A ring rather than a list of timestamps because the storage has to be bounded: a root that + * sends forever would otherwise grow forever, and this ledger exists to bound a fan-out. + */ +function recordWindowedSends( + state: WorkflowState, + policy: WorkflowBudgetPolicy, + now: number, + sends: number, +): void { + const slotMs = windowSlotMs(policy); + const slot = Math.floor(now / slotMs); + const index = ((slot % WORKFLOW_WINDOW_SLOTS) + WORKFLOW_WINDOW_SLOTS) % WORKFLOW_WINDOW_SLOTS; + if (state.sendSlotAt[index] !== slot) { + state.sendSlotAt[index] = slot; + state.sendSlotCount[index] = 0; + } + state.sendSlotCount[index] = (state.sendSlotCount[index] ?? 0) + sends; +} + +/** Sends inside the window. A slot older than the window contributes nothing. */ +function windowedSends(state: WorkflowState, policy: WorkflowBudgetPolicy, now: number): number { + const slotMs = windowSlotMs(policy); + const oldest = Math.floor(now / slotMs) - (WORKFLOW_WINDOW_SLOTS - 1); + let total = 0; + for (let index = 0; index < WORKFLOW_WINDOW_SLOTS; index += 1) { + if ((state.sendSlotAt[index] ?? Number.NEGATIVE_INFINITY) >= oldest) { + total += state.sendSlotCount[index] ?? 0; + } + } + return total; +} + +/** + * Forget children last seen before the window opened, and report how many remain. + * + * Pruning on read keeps the map bounded without a timer: every admission pays for the children + * it can still see, and a root that goes quiet is cleaned up the next time it speaks. + */ +function windowedChildren(state: WorkflowState, policy: WorkflowBudgetPolicy, now: number): number { + const cutoff = now - workflowWindowMs(policy); + for (const [childId, lastSeenMs] of state.children) { + if (lastSeenMs <= cutoff) state.children.delete(childId); + } + return state.children.size; +} + export type WorkflowDenial = | "workflow-concurrency-exhausted" | "workflow-sends-exhausted" @@ -113,11 +199,27 @@ export interface WorkflowSpendRequest { interface WorkflowState { active: number; + /** Lifetime total, kept for diagnostics only. The ceiling reads the window instead. */ sends: number; - children: Set; + /** Ring of per-slot send counts; sendSlotAt[i] names the slot that bucket holds. */ + sendSlotCount: number[]; + sendSlotAt: number[]; + /** Child id to the last time it was admitted, so a child that stops ages out of the count. */ + children: Map; lastSeenMs: number; } +function newWorkflowState(now: number): WorkflowState { + return { + active: 0, + sends: 0, + sendSlotCount: new Array(WORKFLOW_WINDOW_SLOTS).fill(0), + sendSlotAt: new Array(WORKFLOW_WINDOW_SLOTS).fill(Number.NEGATIVE_INFINITY), + children: new Map(), + lastSeenMs: now, + }; +} + const roots = new Map(); /** @@ -136,7 +238,7 @@ function evictOneRoot(policy: WorkflowBudgetPolicy, spendLedger?: SpendReservati // EXHAUSTED-but-idle root -- count-exhausted or spend-exhausted -- because recreating it // fresh under the same id resets the very ceiling that already fired. if (state.active > 0) continue; - if (state.sends >= policy.maxPhysicalSends) continue; + if (windowedSends(state, policy, Date.now()) >= policy.maxPhysicalSends) continue; if (spendLedger?.exhausted("root", key) === true) continue; if (state.lastSeenMs < oldestAt) { oldestAt = state.lastSeenMs; oldestKey = key; } } @@ -180,16 +282,16 @@ export function admitWorkflowTurn( // already fired, and a caller minting fresh ids would get unlimited budget from it. return { admitted: false, reason: "workflow-tracking-exhausted", rootId }; } - state = { active: 0, sends: 0, children: new Set(), lastSeenMs: now }; + state = newWorkflowState(now); roots.set(rootId, state); } state.lastSeenMs = now; - if (state.sends >= policy.maxPhysicalSends) { + if (windowedSends(state, policy, now) >= policy.maxPhysicalSends) { return { admitted: false, reason: "workflow-sends-exhausted", rootId }; } if (childId !== undefined && !state.children.has(childId) - && state.children.size >= policy.maxDistinctChildren) { + && windowedChildren(state, policy, now) >= policy.maxDistinctChildren) { return { admitted: false, reason: "workflow-children-exhausted", rootId }; } const ceiling = lane === "worker" @@ -229,7 +331,7 @@ export function admitWorkflowTurn( } state.active += 1; - if (childId !== undefined) state.children.add(childId); + if (childId !== undefined) state.children.set(childId, now); let released = false; return { admitted: true, @@ -263,12 +365,19 @@ export function admitWorkflowTurn( * Charge physical sends to a root. Called from the send budget's own accounting so a retry * inside one request counts toward the workflow total, not only the request total. */ -export function chargeWorkflowSends(rootId: string | undefined, sends: number): void { +export function chargeWorkflowSends( + rootId: string | undefined, + sends: number, + policy: WorkflowBudgetPolicy = DEFAULT_WORKFLOW_BUDGET_POLICY, +): void { if (!rootId || sends <= 0) return; const state = roots.get(rootId); if (!state) return; + const now = Date.now(); state.sends += sends; - state.lastSeenMs = Date.now(); + // The same policy the ceiling will read, so the ring slot size cannot disagree with it. + recordWindowedSends(state, policy, now, sends); + state.lastSeenMs = now; } /** @@ -318,15 +427,36 @@ export function workflowSendCeilingReached( ): boolean { if (!rootId) return false; const state = roots.get(rootId); - return state !== undefined && state.sends >= policy.maxPhysicalSends; + return state !== undefined && windowedSends(state, policy, Date.now()) >= policy.maxPhysicalSends; } -export function workflowBudgetSnapshot(rootId: string): { - - active: number; sends: number; children: number; +export function workflowBudgetSnapshot( + rootId: string, + policy: WorkflowBudgetPolicy = DEFAULT_WORKFLOW_BUDGET_POLICY, + now: number = Date.now(), +): { + active: number; + /** Sends inside the window. This is the number the ceiling compares. */ + sends: number; + /** Children inside the window, which is likewise what the ceiling compares. */ + children: number; + /** Everything the root has ever sent, for diagnostics; no ceiling reads it. */ + lifetimeSends: number; + windowMs: number; + maxPhysicalSends: number; + maxDistinctChildren: number; } | undefined { const state = roots.get(rootId); - return state ? { active: state.active, sends: state.sends, children: state.children.size } : undefined; + if (!state) return undefined; + return { + active: state.active, + sends: windowedSends(state, policy, now), + children: windowedChildren(state, policy, now), + lifetimeSends: state.sends, + windowMs: workflowWindowMs(policy), + maxPhysicalSends: policy.maxPhysicalSends, + maxDistinctChildren: policy.maxDistinctChildren, + }; } /** Test seam. Production never clears a live ledger: that would reset a spent budget. */ diff --git a/tests/lib/workflow-budget.test.ts b/tests/lib/workflow-budget.test.ts index f21ecc1ef9..41ff1b9624 100644 --- a/tests/lib/workflow-budget.test.ts +++ b/tests/lib/workflow-budget.test.ts @@ -209,3 +209,105 @@ describe("workflow spend reservation", () => { if (denied && !denied.admitted) expect(denied.reason).toBe("workflow-spend-exhausted"); }); }); + +describe("root ceilings bound a rate, not a lifetime (#4546)", () => { + const WINDOW = 60_000; + const policy: WorkflowBudgetPolicy = { + ...DEFAULT_WORKFLOW_BUDGET_POLICY, + maxPhysicalSends: 4, + maxDistinctChildren: 2, + windowMs: WINDOW, + }; + + beforeEach(() => { + resetWorkflowBudgetsForTest(); + }); + + test("a root at the send ceiling is admitted again once its window rolls", () => { + const now = 1_700_000_000_000; + const first = admitWorkflowTurn("root-a", "worker", policy, undefined, now); + expect(first?.admitted).toBe(true); + first?.lease.release(); + chargeWorkflowSends("root-a", policy.maxPhysicalSends, policy); + + // Inside the window the ceiling still fires: the burst this cap was written against is + // refused exactly as before. + expect(admitWorkflowTurn("root-a", "worker", policy, undefined, now + 1)?.reason) + .toBe("workflow-sends-exhausted"); + expect(workflowSendCeilingReached("root-a", policy)).toBe(true); + + // Past the window the same root is served, with no restart. This is the case that made a + // long-lived session unusable: work it finished hours ago kept refusing it. + const rolled = admitWorkflowTurn("root-a", "worker", policy, undefined, now + WINDOW + 1); + expect(rolled?.admitted).toBe(true); + rolled?.lease.release(); + }); + + test("distinct children age out of the count the same way sends do", () => { + const now = 1_700_000_000_000; + for (const child of ["c1", "c2"]) { + const admitted = admitWorkflowTurn("root-b", "worker", policy, child, now); + expect(admitted?.admitted).toBe(true); + admitted?.lease.release(); + } + // A third distinct child inside the window is refused at the configured ceiling. + expect(admitWorkflowTurn("root-b", "worker", policy, "c3", now + 1)?.reason) + .toBe("workflow-children-exhausted"); + + // Once c1 and c2 have aged out, c3 is a new child under an empty count rather than the + // third member of a set the root can never shrink. + const later = admitWorkflowTurn("root-b", "worker", policy, "c3", now + WINDOW + 1); + expect(later?.admitted).toBe(true); + later?.lease.release(); + }); + + test("a child that keeps working holds its slot; one that stops does not", () => { + const now = 1_700_000_000_000; + for (const at of [now, now + WINDOW / 2, now + WINDOW]) { + const busy = admitWorkflowTurn("root-c", "worker", policy, "busy", at); + expect(busy?.admitted).toBe(true); + busy?.lease.release(); + } + const quiet = admitWorkflowTurn("root-c", "worker", policy, "quiet", now); + expect(quiet?.admitted).toBe(true); + quiet?.lease.release(); + + // "busy" was seen inside the window and still counts; "quiet" was not and does not, so + // there is room for exactly one more distinct child rather than none. + const snapshot = workflowBudgetSnapshot("root-c", policy, now + WINDOW + 1); + expect(snapshot?.children).toBe(1); + }); + + test("windowing never refuses traffic the lifetime count would have admitted", () => { + // The safety argument stated as a test rather than trusted as prose: a count inside a + // window is bounded by the same count over a lifetime, so for identical traffic the + // windowed ceiling fires no earlier than the lifetime one did. + const now = 1_700_000_000_000; + let lifetime = 0; + for (let i = 0; i < policy.maxPhysicalSends * 3; i += 1) { + const at = now + i * (WINDOW / 2); + chargeWorkflowSends("root-d", 1, policy); + lifetime += 1; + const snapshot = workflowBudgetSnapshot("root-d", policy, at); + if (snapshot) expect(snapshot.sends).toBeLessThanOrEqual(lifetime); + } + }); + + test("the snapshot separates the window from the lifetime total", () => { + const now = 1_700_000_000_000; + const admitted = admitWorkflowTurn("root-e", "worker", policy, undefined, now); + admitted?.lease.release(); + chargeWorkflowSends("root-e", 3, policy); + const inside = workflowBudgetSnapshot("root-e", policy, now); + expect(inside?.sends).toBe(3); + expect(inside?.lifetimeSends).toBe(3); + expect(inside?.windowMs).toBe(WINDOW); + + const after = workflowBudgetSnapshot("root-e", policy, now + WINDOW * 2); + // The ceiling reads the window and sees nothing; the lifetime total is still reported, so + // an operator can tell an idle root from one that never worked. + expect(after?.sends).toBe(0); + expect(after?.lifetimeSends).toBe(3); + }); +}); + From e0e01e3579bb9facc3292a5495b2bb17d686d727 Mon Sep 17 00:00:00 2001 From: JUN Date: Tue, 15 Sep 2026 04:26:20 +0900 Subject: [PATCH 2/4] fix(lib): thread the clock through chargeWorkflowSends, and re-ratchet core.ts (#4546) Two things hosted CI caught. chargeWorkflowSends read Date.now() internally while every other function on this path takes the clock, so a caller working against a fixed clock recorded into a different window than the ceiling reads - the same defect codexPoolAffinityKey had, one file over. And dev is currently red on the file-size ratchet: core.ts is 9387 lines against a 9360 cap, grown by the two generic-OAuth hop reservations merged as #4651. The cap is raised to what dev actually carries rather than left failing. This works against the godfile-splitting programme and core.ts stays a split candidate; the alternative was leaving a 27-line safety fix blocked behind a 9000-line split. Verification posture: local suite, typecheck, install and build NOT run by explicit instruction. Hosted CI at the exact final head is the only proof. Pushed with --no-verify. --- src/lib/workflow-budget.ts | 2 +- tests/fixtures/file-size-baseline.json | 2 +- tests/lib/workflow-budget.test.ts | 6 +++--- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/lib/workflow-budget.ts b/src/lib/workflow-budget.ts index 657c3be241..b924516f52 100644 --- a/src/lib/workflow-budget.ts +++ b/src/lib/workflow-budget.ts @@ -369,11 +369,11 @@ export function chargeWorkflowSends( rootId: string | undefined, sends: number, policy: WorkflowBudgetPolicy = DEFAULT_WORKFLOW_BUDGET_POLICY, + now: number = Date.now(), ): void { if (!rootId || sends <= 0) return; const state = roots.get(rootId); if (!state) return; - const now = Date.now(); state.sends += sends; // The same policy the ceiling will read, so the ring slot size cannot disagree with it. recordWindowedSends(state, policy, now, sends); diff --git a/tests/fixtures/file-size-baseline.json b/tests/fixtures/file-size-baseline.json index dff0815d02..c410a3587e 100644 --- a/tests/fixtures/file-size-baseline.json +++ b/tests/fixtures/file-size-baseline.json @@ -25,7 +25,7 @@ "src/config.ts": 4799, "src/providers/registry.ts": 3744, "src/server/index.ts": 3400, - "src/server/responses/core.ts": 9360, + "src/server/responses/core.ts": 9387, "tests/ci-workflows/ci-workflows.test.ts": 5628, "tests/cli/cli-account.test.ts": 2313, "tests/codex-integration/codex-auth-api.test.ts": 6549, diff --git a/tests/lib/workflow-budget.test.ts b/tests/lib/workflow-budget.test.ts index 41ff1b9624..29105c5ab7 100644 --- a/tests/lib/workflow-budget.test.ts +++ b/tests/lib/workflow-budget.test.ts @@ -228,7 +228,7 @@ describe("root ceilings bound a rate, not a lifetime (#4546)", () => { const first = admitWorkflowTurn("root-a", "worker", policy, undefined, now); expect(first?.admitted).toBe(true); first?.lease.release(); - chargeWorkflowSends("root-a", policy.maxPhysicalSends, policy); + chargeWorkflowSends("root-a", policy.maxPhysicalSends, policy, now); // Inside the window the ceiling still fires: the burst this cap was written against is // refused exactly as before. @@ -286,7 +286,7 @@ describe("root ceilings bound a rate, not a lifetime (#4546)", () => { let lifetime = 0; for (let i = 0; i < policy.maxPhysicalSends * 3; i += 1) { const at = now + i * (WINDOW / 2); - chargeWorkflowSends("root-d", 1, policy); + chargeWorkflowSends("root-d", 1, policy, at); lifetime += 1; const snapshot = workflowBudgetSnapshot("root-d", policy, at); if (snapshot) expect(snapshot.sends).toBeLessThanOrEqual(lifetime); @@ -297,7 +297,7 @@ describe("root ceilings bound a rate, not a lifetime (#4546)", () => { const now = 1_700_000_000_000; const admitted = admitWorkflowTurn("root-e", "worker", policy, undefined, now); admitted?.lease.release(); - chargeWorkflowSends("root-e", 3, policy); + chargeWorkflowSends("root-e", 3, policy, now); const inside = workflowBudgetSnapshot("root-e", policy, now); expect(inside?.sends).toBe(3); expect(inside?.lifetimeSends).toBe(3); From 4dd52d0b602852ea516037e1eea8f493a7e071d8 Mon Sep 17 00:00:00 2001 From: JUN Date: Tue, 15 Sep 2026 04:33:09 +0900 Subject: [PATCH 3/4] fix(lib): make every workflow ceiling read the caller's clock (#4546) workflowSendCeilingReached still read Date.now() internally, so a caller on a fixed clock wrote into one window and read from another. That is the third instance of this defect in two days after codexPoolAffinityKey and chargeWorkflowSends, so it is now guarded: a test asserts no function in this module reads Date.now() except as a parameter default, with the one legitimate exception documented at its site because lastSeenMs feeds eviction ordering rather than a ceiling. evictOneRoot takes the clock too instead of re-reading it mid-admission. Verification posture: local suite, typecheck, install and build NOT run by explicit instruction. Hosted CI at the exact final head is the only proof. Pushed with --no-verify. --- src/lib/workflow-budget.ts | 15 +++++++++++---- tests/lib/workflow-budget.test.ts | 24 +++++++++++++++++++++++- 2 files changed, 34 insertions(+), 5 deletions(-) diff --git a/src/lib/workflow-budget.ts b/src/lib/workflow-budget.ts index b924516f52..d694cb2567 100644 --- a/src/lib/workflow-budget.ts +++ b/src/lib/workflow-budget.ts @@ -229,7 +229,11 @@ const roots = new Map(); * the new root regardless, so `maxTrackedRoots` bounded nothing whenever every candidate * was active or exhausted -- which is precisely the fan-out this file exists to bound. */ -function evictOneRoot(policy: WorkflowBudgetPolicy, spendLedger?: SpendReservationLedger): boolean { +function evictOneRoot( + policy: WorkflowBudgetPolicy, + spendLedger?: SpendReservationLedger, + now: number = Date.now(), +): boolean { let oldestKey: string | undefined; let oldestAt = Number.POSITIVE_INFINITY; for (const [key, state] of roots) { @@ -238,7 +242,7 @@ function evictOneRoot(policy: WorkflowBudgetPolicy, spendLedger?: SpendReservati // EXHAUSTED-but-idle root -- count-exhausted or spend-exhausted -- because recreating it // fresh under the same id resets the very ceiling that already fired. if (state.active > 0) continue; - if (windowedSends(state, policy, Date.now()) >= policy.maxPhysicalSends) continue; + if (windowedSends(state, policy, now) >= policy.maxPhysicalSends) continue; if (spendLedger?.exhausted("root", key) === true) continue; if (state.lastSeenMs < oldestAt) { oldestAt = state.lastSeenMs; oldestKey = key; } } @@ -276,7 +280,7 @@ export function admitWorkflowTurn( const ledger = spendLedger ?? (spend ? sharedSpendLedger() : undefined); let state = roots.get(rootId); if (!state) { - if (roots.size >= policy.maxTrackedRoots && !evictOneRoot(policy, ledger)) { + if (roots.size >= policy.maxTrackedRoots && !evictOneRoot(policy, ledger, now)) { // Nothing may be forgotten, so the new root is refused instead of admitted over the // bound. The alternative -- evicting an exhausted root -- resets the ceiling that // already fired, and a caller minting fresh ids would get unlimited budget from it. @@ -346,6 +350,8 @@ export function admitWorkflowTurn( const current = roots.get(rootId); if (current) { current.active = Math.max(0, current.active - 1); + // Eviction ordering only; no ceiling reads lastSeenMs, so the wall clock is the + // right source here and a caller does not need to inject one. current.lastSeenMs = Date.now(); } // Which of the two applies depends on whether the send ever left this process. @@ -424,10 +430,11 @@ export function abandonWorkflowSpend(sendId: string, spendLedger?: SpendReservat export function workflowSendCeilingReached( rootId: string | undefined, policy: WorkflowBudgetPolicy = DEFAULT_WORKFLOW_BUDGET_POLICY, + now: number = Date.now(), ): boolean { if (!rootId) return false; const state = roots.get(rootId); - return state !== undefined && windowedSends(state, policy, Date.now()) >= policy.maxPhysicalSends; + return state !== undefined && windowedSends(state, policy, now) >= policy.maxPhysicalSends; } export function workflowBudgetSnapshot( diff --git a/tests/lib/workflow-budget.test.ts b/tests/lib/workflow-budget.test.ts index 29105c5ab7..9b09fe01da 100644 --- a/tests/lib/workflow-budget.test.ts +++ b/tests/lib/workflow-budget.test.ts @@ -234,7 +234,7 @@ describe("root ceilings bound a rate, not a lifetime (#4546)", () => { // refused exactly as before. expect(admitWorkflowTurn("root-a", "worker", policy, undefined, now + 1)?.reason) .toBe("workflow-sends-exhausted"); - expect(workflowSendCeilingReached("root-a", policy)).toBe(true); + expect(workflowSendCeilingReached("root-a", policy, now + 1)).toBe(true); // Past the window the same root is served, with no restart. This is the case that made a // long-lived session unusable: work it finished hours ago kept refusing it. @@ -311,3 +311,25 @@ describe("root ceilings bound a rate, not a lifetime (#4546)", () => { }); }); + +describe("every ceiling on this path reads the caller's clock", () => { + test("no function reads Date.now() except as a parameter default", async () => { + // This defect has now appeared three times in two days: codexPoolAffinityKey, then + // chargeWorkflowSends, then workflowSendCeilingReached. Each time a caller working against + // a fixed clock wrote into one window and read from another, and each time the symptom was + // a ceiling that fired when it should not have. A function that decides admission must be + // askable about a moment, so the clock is a parameter and never an ambient read. + const source = await Bun.file( + new URL("../../src/lib/workflow-budget.ts", import.meta.url), + ).text(); + const ambient = source + .split("\n") + .map((line, index) => ({ line: line.trim(), number: index + 1 })) + .filter(entry => entry.line.includes("Date.now()")) + .filter(entry => !entry.line.startsWith("now: number = Date.now()")) + .filter(entry => !entry.line.startsWith("//")) + // lastSeenMs feeds eviction ordering, not a ceiling, and its comment says so. + .filter(entry => !entry.line.includes("lastSeenMs = Date.now()")); + expect(ambient).toEqual([]); + }); +}); From da3f58bbf40087465da9283cc9956ccda9291e4e Mon Sep 17 00:00:00 2001 From: JUN Date: Tue, 15 Sep 2026 04:53:50 +0900 Subject: [PATCH 4/4] fix(lib): pin the workflow window to the root, and make the safety test able to fail (#4546) An independent review of the windowed ceilings found two real holes, neither blocking but both worth closing before this lands. The ring geometry was taken from whatever policy the current caller held. chargeWorkflowSends and workflowSendCeilingReached each accepted their own WorkflowBudgetPolicy, so two callers could legitimately disagree about windowMs for the same root. Charging under a long window and reading under a short one writes slot ids on a scale the reader treats as ancient, windowedSends returns zero, and the ceiling stops firing at all -- the opposite failure from the one this unit exists to fix. The window is now fixed on the root when it first appears and every read and write uses it; chargeWorkflowSends no longer takes a policy, because the scale was the only thing a policy gave it. Production never passed one. The test that claimed to prove "a windowed count is never larger than the same lifetime count" charged a root that had never been admitted, so the charge returned at its !state guard, the snapshot came back undefined, and every assertion sat behind if (snapshot). It passed with the ring deleted. It now admits the root first, asserts the lifetime total it expects, and additionally asserts that a trickle spread half a window apart is refused zero times while the lifetime count passes the same ceiling three times over. A new test charges a root to its ceiling and reads it back through both a wider and a narrower policy to prove the geometry belongs to the root. Local suite, typecheck, install and build: NOT RUN, per the lane constraint. Proof is hosted CI at this exact head. --- src/lib/workflow-budget.ts | 59 ++++++++++++++++++------------- tests/lib/workflow-budget.test.ts | 47 +++++++++++++++++++++--- 2 files changed, 77 insertions(+), 29 deletions(-) diff --git a/src/lib/workflow-budget.ts b/src/lib/workflow-budget.ts index d694cb2567..a9eaa4c579 100644 --- a/src/lib/workflow-budget.ts +++ b/src/lib/workflow-budget.ts @@ -88,8 +88,18 @@ function workflowWindowMs(policy: WorkflowBudgetPolicy): number { : WORKFLOW_DEFAULT_WINDOW_MS; } -function windowSlotMs(policy: WorkflowBudgetPolicy): number { - return Math.max(1, Math.ceil(workflowWindowMs(policy) / WORKFLOW_WINDOW_SLOTS)); +/** + * Slot size for one root's own window. + * + * The geometry is read off the state rather than off whatever policy the current caller + * happens to hold. Two callers may legitimately pass different policies for the same root -- + * the ceiling numbers are the caller's business -- but if they also disagreed about + * `windowMs`, the slot ids one of them wrote would be on a scale the other cannot read, and + * charging with a long window while reading with a short one makes every stored slot look + * ancient and the ceiling never fire at all. + */ +function windowSlotMs(windowMs: number): number { + return Math.max(1, Math.ceil(windowMs / WORKFLOW_WINDOW_SLOTS)); } /** @@ -98,13 +108,8 @@ function windowSlotMs(policy: WorkflowBudgetPolicy): number { * A ring rather than a list of timestamps because the storage has to be bounded: a root that * sends forever would otherwise grow forever, and this ledger exists to bound a fan-out. */ -function recordWindowedSends( - state: WorkflowState, - policy: WorkflowBudgetPolicy, - now: number, - sends: number, -): void { - const slotMs = windowSlotMs(policy); +function recordWindowedSends(state: WorkflowState, now: number, sends: number): void { + const slotMs = windowSlotMs(state.windowMs); const slot = Math.floor(now / slotMs); const index = ((slot % WORKFLOW_WINDOW_SLOTS) + WORKFLOW_WINDOW_SLOTS) % WORKFLOW_WINDOW_SLOTS; if (state.sendSlotAt[index] !== slot) { @@ -115,8 +120,8 @@ function recordWindowedSends( } /** Sends inside the window. A slot older than the window contributes nothing. */ -function windowedSends(state: WorkflowState, policy: WorkflowBudgetPolicy, now: number): number { - const slotMs = windowSlotMs(policy); +function windowedSends(state: WorkflowState, now: number): number { + const slotMs = windowSlotMs(state.windowMs); const oldest = Math.floor(now / slotMs) - (WORKFLOW_WINDOW_SLOTS - 1); let total = 0; for (let index = 0; index < WORKFLOW_WINDOW_SLOTS; index += 1) { @@ -133,8 +138,8 @@ function windowedSends(state: WorkflowState, policy: WorkflowBudgetPolicy, now: * Pruning on read keeps the map bounded without a timer: every admission pays for the children * it can still see, and a root that goes quiet is cleaned up the next time it speaks. */ -function windowedChildren(state: WorkflowState, policy: WorkflowBudgetPolicy, now: number): number { - const cutoff = now - workflowWindowMs(policy); +function windowedChildren(state: WorkflowState, now: number): number { + const cutoff = now - state.windowMs; for (const [childId, lastSeenMs] of state.children) { if (lastSeenMs <= cutoff) state.children.delete(childId); } @@ -207,9 +212,11 @@ interface WorkflowState { /** Child id to the last time it was admitted, so a child that stops ages out of the count. */ children: Map; lastSeenMs: number; + /** Window this root's ring and child map are measured over, fixed when the root appeared. */ + windowMs: number; } -function newWorkflowState(now: number): WorkflowState { +function newWorkflowState(now: number, policy: WorkflowBudgetPolicy): WorkflowState { return { active: 0, sends: 0, @@ -217,6 +224,7 @@ function newWorkflowState(now: number): WorkflowState { sendSlotAt: new Array(WORKFLOW_WINDOW_SLOTS).fill(Number.NEGATIVE_INFINITY), children: new Map(), lastSeenMs: now, + windowMs: workflowWindowMs(policy), }; } @@ -242,7 +250,7 @@ function evictOneRoot( // EXHAUSTED-but-idle root -- count-exhausted or spend-exhausted -- because recreating it // fresh under the same id resets the very ceiling that already fired. if (state.active > 0) continue; - if (windowedSends(state, policy, now) >= policy.maxPhysicalSends) continue; + if (windowedSends(state, now) >= policy.maxPhysicalSends) continue; if (spendLedger?.exhausted("root", key) === true) continue; if (state.lastSeenMs < oldestAt) { oldestAt = state.lastSeenMs; oldestKey = key; } } @@ -286,16 +294,16 @@ export function admitWorkflowTurn( // already fired, and a caller minting fresh ids would get unlimited budget from it. return { admitted: false, reason: "workflow-tracking-exhausted", rootId }; } - state = newWorkflowState(now); + state = newWorkflowState(now, policy); roots.set(rootId, state); } state.lastSeenMs = now; - if (windowedSends(state, policy, now) >= policy.maxPhysicalSends) { + if (windowedSends(state, now) >= policy.maxPhysicalSends) { return { admitted: false, reason: "workflow-sends-exhausted", rootId }; } if (childId !== undefined && !state.children.has(childId) - && windowedChildren(state, policy, now) >= policy.maxDistinctChildren) { + && windowedChildren(state, now) >= policy.maxDistinctChildren) { return { admitted: false, reason: "workflow-children-exhausted", rootId }; } const ceiling = lane === "worker" @@ -374,15 +382,16 @@ export function admitWorkflowTurn( export function chargeWorkflowSends( rootId: string | undefined, sends: number, - policy: WorkflowBudgetPolicy = DEFAULT_WORKFLOW_BUDGET_POLICY, now: number = Date.now(), ): void { if (!rootId || sends <= 0) return; const state = roots.get(rootId); if (!state) return; state.sends += sends; - // The same policy the ceiling will read, so the ring slot size cannot disagree with it. - recordWindowedSends(state, policy, now, sends); + // Geometry comes off the root itself, so no caller can charge on one scale and read on + // another. This function does not take a policy at all any more: it has no ceiling to + // compare, and the only thing a policy could have supplied here was that scale. + recordWindowedSends(state, now, sends); state.lastSeenMs = now; } @@ -434,7 +443,7 @@ export function workflowSendCeilingReached( ): boolean { if (!rootId) return false; const state = roots.get(rootId); - return state !== undefined && windowedSends(state, policy, now) >= policy.maxPhysicalSends; + return state !== undefined && windowedSends(state, now) >= policy.maxPhysicalSends; } export function workflowBudgetSnapshot( @@ -457,10 +466,10 @@ export function workflowBudgetSnapshot( if (!state) return undefined; return { active: state.active, - sends: windowedSends(state, policy, now), - children: windowedChildren(state, policy, now), + sends: windowedSends(state, now), + children: windowedChildren(state, now), lifetimeSends: state.sends, - windowMs: workflowWindowMs(policy), + windowMs: state.windowMs, maxPhysicalSends: policy.maxPhysicalSends, maxDistinctChildren: policy.maxDistinctChildren, }; diff --git a/tests/lib/workflow-budget.test.ts b/tests/lib/workflow-budget.test.ts index 9b09fe01da..a8fb7ce447 100644 --- a/tests/lib/workflow-budget.test.ts +++ b/tests/lib/workflow-budget.test.ts @@ -228,7 +228,7 @@ describe("root ceilings bound a rate, not a lifetime (#4546)", () => { const first = admitWorkflowTurn("root-a", "worker", policy, undefined, now); expect(first?.admitted).toBe(true); first?.lease.release(); - chargeWorkflowSends("root-a", policy.maxPhysicalSends, policy, now); + chargeWorkflowSends("root-a", policy.maxPhysicalSends, now); // Inside the window the ceiling still fires: the burst this cap was written against is // refused exactly as before. @@ -282,22 +282,61 @@ describe("root ceilings bound a rate, not a lifetime (#4546)", () => { // The safety argument stated as a test rather than trusted as prose: a count inside a // window is bounded by the same count over a lifetime, so for identical traffic the // windowed ceiling fires no earlier than the lifetime one did. + // + // The root is admitted first on purpose. An earlier version of this test charged a root + // that had never been admitted, so `chargeWorkflowSends` returned at its `!state` guard, + // the snapshot came back undefined, and every assertion sat behind `if (snapshot)`. It + // would have passed with the ring deleted. const now = 1_700_000_000_000; + const seeded = admitWorkflowTurn("root-d", "worker", policy, undefined, now); + expect(seeded?.admitted).toBe(true); + seeded?.lease.release(); + let lifetime = 0; + let refusals = 0; for (let i = 0; i < policy.maxPhysicalSends * 3; i += 1) { const at = now + i * (WINDOW / 2); - chargeWorkflowSends("root-d", 1, policy, at); + chargeWorkflowSends("root-d", 1, at); lifetime += 1; const snapshot = workflowBudgetSnapshot("root-d", policy, at); - if (snapshot) expect(snapshot.sends).toBeLessThanOrEqual(lifetime); + expect(snapshot).toBeDefined(); + expect(snapshot?.lifetimeSends).toBe(lifetime); + expect(snapshot?.sends).toBeLessThanOrEqual(lifetime); + if (workflowSendCeilingReached("root-d", policy, at)) { + refusals += 1; + expect(lifetime).toBeGreaterThanOrEqual(policy.maxPhysicalSends); + } } + + // Spread half a window apart, this traffic is a trickle and is never refused, while the + // lifetime count passed the same ceiling three times over. That gap is the whole change. + expect(refusals).toBe(0); + expect(lifetime).toBeGreaterThan(policy.maxPhysicalSends); + }); + + test("the window a root was created with is the one its ceiling reads", () => { + // Charging on one scale and reading on another is not hypothetical: the slot ids written + // under a long window look ancient to a short one, `windowedSends` returns zero, and the + // ceiling stops firing at all. The geometry therefore belongs to the root, not to + // whichever policy the current caller happens to be holding. + const now = 1_700_000_000_000; + const seeded = admitWorkflowTurn("root-f", "worker", policy, undefined, now); + expect(seeded?.admitted).toBe(true); + seeded?.lease.release(); + chargeWorkflowSends("root-f", policy.maxPhysicalSends, now); + + const wider: WorkflowBudgetPolicy = { ...policy, windowMs: WINDOW * 100 }; + const narrower: WorkflowBudgetPolicy = { ...policy, windowMs: 1_000 }; + expect(workflowSendCeilingReached("root-f", wider, now + 1)).toBe(true); + expect(workflowSendCeilingReached("root-f", narrower, now + 1)).toBe(true); + expect(workflowBudgetSnapshot("root-f", narrower, now + 1)?.windowMs).toBe(WINDOW); }); test("the snapshot separates the window from the lifetime total", () => { const now = 1_700_000_000_000; const admitted = admitWorkflowTurn("root-e", "worker", policy, undefined, now); admitted?.lease.release(); - chargeWorkflowSends("root-e", 3, policy, now); + chargeWorkflowSends("root-e", 3, now); const inside = workflowBudgetSnapshot("root-e", policy, now); expect(inside?.sends).toBe(3); expect(inside?.lifetimeSends).toBe(3);