From 1340e7d7e6e22a1f7cd6fba67374c434d84dd7bb Mon Sep 17 00:00:00 2001 From: JUN Date: Tue, 15 Sep 2026 05:31:40 +0900 Subject: [PATCH 1/5] fix(server): name the ceiling that refused, and let an operator see and clear one root (#4546) The workflow budget could refuse a task and leave nothing behind to explain it. Two specific gaps, both measured against the code rather than assumed. The refusal never reached the request log at all. runAdmittedHttpTurn returns before it calls work(), and every addFinalRequestLog in that file is inside work, so there was no row and no context to mark. And the ceiling name never reached the client either: classifyError rewrites every 429 to rate_limit_error / rate_limit_exceeded, so the body was shaped exactly like a provider rate limit and the workflow_budget_exhausted type argument was discarded on the way out. All four count denials also shared one sentence about a "concurrent-work limit", which was true of exactly one of them -- a task that had hit the SEND ceiling was told to wait for turns to finish, and waiting never helped because nothing was running. Each denial now has its own sentence naming its ceiling and saying this proxy decided it without contacting anyone. The wire status and type are unchanged on purpose, since altering them changes how every client retries, so the machine-readable name rides alongside on x-opencodex-local-refusal. Nothing upstream sets that header, which is what makes its presence conclusive. Both call sites now go through one src/server/workflow-refusal.ts instead of two inline blocks that had drifted apart; where a log context exists, the row is marked synthetic through the same helper #4639 introduced. A refusal that parsed no body, chose no model and contacted no provider is not a usage row, and forcing one would put a fabricated model and provider into usage.jsonl. It goes instead into a bounded ring of recent budget events inside the budget itself, recorded at every refusal return site in admitWorkflowTurn -- including the spend denials, which are decided inside the ledger branch and never surface to the caller that formats the response. GET /api/workflow-budget reads the tracked roots or one root, and POST /api/workflow-budget/clear clears exactly one. The clear is bounded in a specific way: it moves the windowed send ring and the child map and nothing else. active belongs to turns still in flight, and zeroing it would let their releases drive the count negative and hand out slots already taken. The spend ledger is money an operator did not ask to forgive, and a count ceiling is not a licence to reset it. The lifetime send total survives too, so clearing a ceiling cannot launder the record of what the root actually did. A test drives that last one directly: after a clear, an exhausted token budget still refuses. Both routes are declared deferred-verb in the route registry. They are owed CLI verbs and the ledger is process memory, so unlike the Lab routes there is no local projection the CLI could read instead. Local suite, typecheck, install and build: NOT RUN, per the standing instruction. Hosted CI at the exact head is the only proof. Pushed --no-verify. --- .../030_wfc_diff_plan.md | 102 ++++++++ scripts/test-layout/layout.json | 1 + src/lib/workflow-budget.ts | 220 ++++++++++++++++-- src/server/index.ts | 9 +- src/server/management-api.ts | 12 + src/server/management/route-registry.ts | 3 + .../management/workflow-budget-routes.ts | 133 +++++++++++ src/server/responses/core.ts | 9 +- src/server/workflow-refusal.ts | 45 ++++ tests/fixtures/test-layout-expected.json | 1 + tests/lib/workflow-budget.test.ts | 146 ++++++++++++ .../management-workflow-budget-routes.test.ts | 108 +++++++++ 12 files changed, 764 insertions(+), 25 deletions(-) create mode 100644 devlog/_plan/260915_workflow_budget_window/030_wfc_diff_plan.md create mode 100644 src/server/management/workflow-budget-routes.ts create mode 100644 src/server/workflow-refusal.ts create mode 100644 tests/server/management-workflow-budget-routes.test.ts diff --git a/devlog/_plan/260915_workflow_budget_window/030_wfc_diff_plan.md b/devlog/_plan/260915_workflow_budget_window/030_wfc_diff_plan.md new file mode 100644 index 0000000000..7b4617b716 --- /dev/null +++ b/devlog/_plan/260915_workflow_budget_window/030_wfc_diff_plan.md @@ -0,0 +1,102 @@ +# 030 — wfc: the diff + +Written after reading the code rather than from the sketch in 020, because two of +that sketch's assumptions turned out to be wrong. + +## What the investigation changed + +**The refusal never reaches the request log at all.** `runAdmittedHttpTurn` +returns `formatErrorResponse(429, ...)` before it calls `work()`, and every +`addFinalRequestLog` call in that file is inside `work`. There is no +`logCtx` at that point and nothing to mark. 020 assumed the row existed and +only lacked a field. + +**The ceiling name does not survive onto the wire.** `formatErrorResponse` +runs `classifyError`, which rewrites any 429 to +`{ type: "rate_limit_error", code: "rate_limit_exceeded" }`. The +`workflow_budget_exhausted` string passed at the call site is discarded. So the +body an operator sees today is byte-identical in shape to a provider rate limit, +and the message is the only field that can carry anything. + +That also means the message is wrong for three of the four denials: all four +reasons emit one sentence about a "concurrent-work limit", and only +`workflow-concurrency-exhausted` is actually that. + +## The change + +**Name the ceiling where the operator will read it.** Each `WorkflowDenial` +gets its own sentence, stating which ceiling fired and that this proxy made the +decision without contacting a provider. The wire status and type stay exactly as +they are — a client's retry behaviour must not change — so a response header +`x-opencodex-local-refusal: ` carries the machine-readable name +alongside. An upstream 429 never sets it, which is the distinction 020 asked for. + +**Record the refusal where the request log cannot go.** A refusal that parsed no +body, chose no model and contacted no provider is not a usage row, and forcing +one would put a fabricated model and provider into `usage.jsonl`. Instead +`src/lib/workflow-budget.ts` keeps a bounded ring of recent budget events — +every refusal and every operator clear, with the root, the ceiling, the counts at +the time and a timestamp. Every entry in it is by construction a local decision, +which is a stronger guarantee than a flag on a shared row. + +Where a `logCtx` does exist — the pre-dispatch ceiling check in +`src/server/responses/core.ts` — the refusal additionally goes through +`markLocalRequestLogRefusal`, the same helper #4639 introduced, so the row that +does get written says `terminalSource: "synthetic"`. + +**Expose and clear.** A new management module serves +`GET /api/workflow-budget` (tracked roots, or one root with `?root=`, plus the +recent events) and `POST /api/workflow-budget/clear` with `{ "root": "" }`. + +The clear is bounded in a specific way: it resets the windowed send ring and the +child map, and it touches neither `active` nor the spend ledger. Clearing a +*count* ceiling must not clear *spend* — a token budget the operator did not ask +to forgive, and an active lease count that belongs to turns still in flight. +The clear is written into the same event ring, so it is on the record next to the +refusals it answers. + +## Files + +- `src/lib/workflow-budget.ts` — `workflowDenialSummary`, a bounded event ring + (`recordWorkflowBudgetEvent`, `listWorkflowBudgetEvents`), + `listTrackedWorkflowRoots`, and `clearWorkflowBudgetForRoot`. +- `src/server/index.ts` — per-reason message, the local-refusal header, and the + event record in `runAdmittedHttpTurn`. +- `src/server/responses/core.ts` — the same for the pre-dispatch ceiling check, + plus `markLocalRequestLogRefusal` where the log context exists. +- `src/server/management/workflow-budget-routes.ts` — the two endpoints. +- `src/server/management-api.ts` — lazy `OnDemand` wrapper and dispatch entry. +- `src/server/management/route-registry.ts` — the two inventory entries. +- `tests/lib/workflow-budget.test.ts` — clear is scoped and recorded; the event + ring is bounded. +- `tests/server/management-workflow-budget-routes.test.ts` — both endpoints, and + that a data-plane key cannot reach the clear. +- `scripts/test-layout/layout.json` and + `tests/fixtures/test-layout-expected.json` — the new test file, registered in + both as the layout guard requires. + +## Acceptance + +1. Each of the four denials produces a message naming its own ceiling, and the + response carries `x-opencodex-local-refusal` with the machine-readable name. +2. Every refusal and every clear lands in the bounded event ring; the + `core.ts` path additionally marks its request-log row synthetic. +3. `GET /api/workflow-budget` reads one root, and + `POST /api/workflow-budget/clear` clears exactly that root. +4. The clear leaves `active` and the spend ledger untouched, is recorded, and is + refused for a caller that only holds a data-plane key. + +## Owed, and not in this work-phase + +`GET /api/workflow-budget` and `POST /api/workflow-budget/clear` are owed CLI +verbs. An operator looking at a 429 is usually already in a terminal, and the +ledger is process memory, so unlike the Lab routes there is no local SQLite +projection the CLI could read instead — the verb has to be an HTTP call. Both are +declared `deferred-verb` in the route registry against this document, which is +what keeps them out of the undeclared-route ratchet without pretending the gap +does not exist. + +## Verification posture + +Local suite, typecheck, install and build: NOT RUN, by standing instruction. +Hosted CI at the exact final head is the only proof. Pushed with `--no-verify`. diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index abe1e8db67..72508aaf6f 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -888,6 +888,7 @@ "management-origin-tls.test.ts": "server", "management-provider-validation.test.ts": "server", "management-route-registry.test.ts": "server", + "management-workflow-budget-routes.test.ts": "server", "memory-watchdog.test.ts": "server", "meta-model-api-provider.test.ts": "providers", "meta-muse-oauth.test.ts": "providers", diff --git a/src/lib/workflow-budget.ts b/src/lib/workflow-budget.ts index a9eaa4c579..125890d0fa 100644 --- a/src/lib/workflow-budget.ts +++ b/src/lib/workflow-budget.ts @@ -162,6 +162,122 @@ export type WorkflowDenial = /** The reservation could not be made durable, and a configured ceiling requires it. */ | "workflow-spend-undurable"; +/** + * The sentence an operator reads, plus the machine-readable name of the ceiling that fired. + * + * All four count denials used to share one sentence about a "concurrent-work limit", which was + * accurate for exactly one of them. Worse, the wire cannot carry the distinction on its own: + * `classifyError` rewrites every 429 to `rate_limit_error` / `rate_limit_exceeded`, so the body + * of a refusal this proxy made is shaped exactly like a provider rate limit. Each sentence + * therefore says which ceiling fired AND that no provider was contacted, because that is the + * first thing an operator needs and the only place left to put it. + */ +export function workflowDenialSummary(reason: WorkflowDenial): { code: string; message: string } { + switch (reason) { + case "workflow-sends-exhausted": + return { + code: "workflow_sends_exhausted", + message: "This proxy refused the request locally: the task reached its send ceiling for" + + " the current window, so no provider was contacted. The window rolls forward on its" + + " own; work already in flight settles as it finishes.", + }; + case "workflow-children-exhausted": + return { + code: "workflow_children_exhausted", + message: "This proxy refused the request locally: the task reached its ceiling on" + + " distinct child threads for the current window, so no provider was contacted." + + " A child that goes quiet ages out of the count.", + }; + case "workflow-concurrency-exhausted": + return { + code: "workflow_concurrency_exhausted", + message: "This proxy refused the request locally: the task has no free concurrency slot," + + " so no provider was contacted. Slots are released as the turns holding them finish.", + }; + case "workflow-spend-exhausted": + return { + code: "workflow_spend_exhausted", + message: "This proxy refused the request locally: the task reached a configured token" + + " ceiling, so no provider was contacted.", + }; + case "workflow-tracking-exhausted": + return { + code: "workflow_tracking_exhausted", + message: "This proxy refused the request locally: it is already tracking as many tasks as" + + " it may, and every one of them is busy or over its own ceiling, so no provider was" + + " contacted.", + }; + case "workflow-send-replayed": + return { + code: "workflow_send_replayed", + message: "This proxy refused the request locally: this send was already reserved once, and" + + " a repeat buys no second dispatch.", + }; + case "workflow-spend-undurable": + return { + code: "workflow_spend_undurable", + message: "This proxy refused the request locally: the token reservation could not be made" + + " durable and a configured ceiling requires it, so no provider was contacted.", + }; + } +} + +/** + * Response header naming the ceiling that refused, on a refusal this proxy made itself. + * + * It exists because the body cannot carry it: `classifyError` rewrites every 429 to + * `rate_limit_error` / `rate_limit_exceeded`, so a local refusal and a provider rate limit are + * byte-identical in shape. Changing that classification would change how every client retries, + * so the name goes beside the body instead. No upstream sets this header, which is precisely + * what makes its presence conclusive. + */ +export const WORKFLOW_LOCAL_REFUSAL_HEADER = "x-opencodex-local-refusal"; + +export type WorkflowBudgetEventKind = "refused" | "cleared"; + +export interface WorkflowBudgetEvent { + readonly at: number; + readonly kind: WorkflowBudgetEventKind; + readonly rootId: string; + /** The ceiling that fired. Present for `refused`, absent for `cleared`. */ + readonly reason?: WorkflowDenial; + /** Windowed sends at the moment of the event. */ + readonly sends: number; + /** Windowed distinct children at the moment of the event. */ + readonly children: number; +} + +/** + * How many events are kept. Small on purpose: this is an operator's recent-history view, not an + * audit log, and it lives in the same process memory the ceilings do. + */ +export const WORKFLOW_EVENT_CAPACITY = 64; + +const budgetEvents: WorkflowBudgetEvent[] = []; + +/** + * Record a local budget decision. + * + * This exists because the refusal has nowhere else to go. The HTTP admission check runs before + * the body is parsed, so there is no model, no provider and no request-log context to attach to; + * writing a usage row there would mean inventing both. Every entry here is by construction a + * decision this proxy made without contacting anyone, which is a stronger statement than a flag + * on a row shared with upstream results. + */ +function recordBudgetEvent(event: WorkflowBudgetEvent): void { + budgetEvents.push(event); + while (budgetEvents.length > WORKFLOW_EVENT_CAPACITY) budgetEvents.shift(); +} + +/** Newest first. `limit` is clamped to what is actually kept. */ +export function listWorkflowBudgetEvents(limit: number = WORKFLOW_EVENT_CAPACITY): WorkflowBudgetEvent[] { + const wanted = Number.isFinite(limit) && limit > 0 + ? Math.min(Math.floor(limit), WORKFLOW_EVENT_CAPACITY) + : 0; + if (wanted === 0) return []; + return budgetEvents.slice(-wanted).reverse(); +} + export type WorkflowLane = "interactive" | "worker"; export interface WorkflowAdmission { @@ -287,12 +403,28 @@ export function admitWorkflowTurn( // still see spend-exhausted entries. With neither, no token tracking is in play. const ledger = spendLedger ?? (spend ? sharedSpendLedger() : undefined); let state = roots.get(rootId); + // Every refusal below goes on the record through this one seam. Recording at each return + // site instead of at the HTTP caller is what makes the record complete: the spend denials + // are decided inside the ledger branch and never surface as a distinct reason to the caller + // that formats the response. + const refuse = (reason: WorkflowDenial, spendScope?: SpendScope): WorkflowDecision => { + const current = roots.get(rootId); + recordBudgetEvent({ + at: now, + kind: "refused", + rootId, + reason, + sends: current ? windowedSends(current, now) : 0, + children: current ? windowedChildren(current, now) : 0, + }); + return { admitted: false, reason, rootId, ...(spendScope ? { spendScope } : {}) }; + }; if (!state) { 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. - return { admitted: false, reason: "workflow-tracking-exhausted", rootId }; + return refuse("workflow-tracking-exhausted"); } state = newWorkflowState(now, policy); roots.set(rootId, state); @@ -300,17 +432,17 @@ export function admitWorkflowTurn( state.lastSeenMs = now; if (windowedSends(state, now) >= policy.maxPhysicalSends) { - return { admitted: false, reason: "workflow-sends-exhausted", rootId }; + return refuse("workflow-sends-exhausted"); } if (childId !== undefined && !state.children.has(childId) && windowedChildren(state, now) >= policy.maxDistinctChildren) { - return { admitted: false, reason: "workflow-children-exhausted", rootId }; + return refuse("workflow-children-exhausted"); } const ceiling = lane === "worker" ? Math.max(0, policy.maxConcurrentChildren - policy.interactiveReserve) : policy.maxConcurrentChildren; if (state.active >= ceiling) { - return { admitted: false, reason: "workflow-concurrency-exhausted", rootId }; + return refuse("workflow-concurrency-exhausted"); } if (spend && ledger) { @@ -333,12 +465,10 @@ export function admitWorkflowTurn( : denial.reason === "tracking-capacity-exhausted" ? "workflow-tracking-exhausted" : "workflow-spend-exhausted"; - return { - admitted: false, + return refuse( reason, - rootId, - spendScope: denial.reason === "spend-limit-exceeded" ? denial.scope : undefined, - }; + denial.reason === "spend-limit-exceeded" ? denial.scope : undefined, + ); } } @@ -446,11 +576,7 @@ export function workflowSendCeilingReached( return state !== undefined && windowedSends(state, now) >= policy.maxPhysicalSends; } -export function workflowBudgetSnapshot( - rootId: string, - policy: WorkflowBudgetPolicy = DEFAULT_WORKFLOW_BUDGET_POLICY, - now: number = Date.now(), -): { +export interface WorkflowBudgetSnapshot { active: number; /** Sends inside the window. This is the number the ceiling compares. */ sends: number; @@ -461,7 +587,13 @@ export function workflowBudgetSnapshot( windowMs: number; maxPhysicalSends: number; maxDistinctChildren: number; -} | undefined { +} + +export function workflowBudgetSnapshot( + rootId: string, + policy: WorkflowBudgetPolicy = DEFAULT_WORKFLOW_BUDGET_POLICY, + now: number = Date.now(), +): WorkflowBudgetSnapshot | undefined { const state = roots.get(rootId); if (!state) return undefined; return { @@ -475,7 +607,65 @@ export function workflowBudgetSnapshot( }; } +/** + * Roots this process is currently tracking, most recently active first. + * + * Bounded by `limit` because `maxTrackedRoots` is 512 and an operator asking what is going on + * wants the busy end of that, not a dump. + */ +export function listTrackedWorkflowRoots( + limit = 64, + policy: WorkflowBudgetPolicy = DEFAULT_WORKFLOW_BUDGET_POLICY, + now: number = Date.now(), +): Array<{ rootId: string } & WorkflowBudgetSnapshot> { + const wanted = Number.isFinite(limit) && limit > 0 ? Math.floor(limit) : 0; + if (wanted === 0) return []; + return [...roots.entries()] + .sort((left, right) => right[1].lastSeenMs - left[1].lastSeenMs) + .slice(0, wanted) + .flatMap(([rootId]) => { + const snapshot = workflowBudgetSnapshot(rootId, policy, now); + return snapshot ? [{ rootId, ...snapshot }] : []; + }); +} + +/** + * Clear ONE root's windowed count ceilings, and report what they were. + * + * Three things are deliberately left alone. `active` belongs to turns still in flight, and + * zeroing it would let their releases drive the count negative and hand out concurrency slots + * that are already taken. The spend ledger is a token budget an operator did not ask to + * forgive, and a count ceiling is not a licence to reset it. `sends` -- the lifetime total -- + * survives too, so the record of what this root actually did cannot be laundered by clearing + * it; only the ceilings move. + * + * Returns the snapshot taken immediately before the clear, so the caller can put on the record + * what it forgave, or `undefined` when the root is not tracked at all. + */ +export function clearWorkflowBudgetForRoot( + rootId: string, + policy: WorkflowBudgetPolicy = DEFAULT_WORKFLOW_BUDGET_POLICY, + now: number = Date.now(), +): WorkflowBudgetSnapshot | undefined { + const state = roots.get(rootId); + if (!state) return undefined; + const before = workflowBudgetSnapshot(rootId, policy, now); + state.sendSlotCount.fill(0); + state.sendSlotAt.fill(Number.NEGATIVE_INFINITY); + state.children.clear(); + state.lastSeenMs = now; + recordBudgetEvent({ + at: now, + kind: "cleared", + rootId, + sends: before?.sends ?? 0, + children: before?.children ?? 0, + }); + return before; +} + /** Test seam. Production never clears a live ledger: that would reset a spent budget. */ export function resetWorkflowBudgetsForTest(): void { roots.clear(); + budgetEvents.length = 0; } diff --git a/src/server/index.ts b/src/server/index.ts index 8a496d2c84..042a9744bb 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -135,6 +135,7 @@ import { } from "./request-log"; import { sessionLaneIdFromRequest } from "./request-log-conversation"; import { admitWorkflowTurn, type WorkflowLane } from "../lib/workflow-budget"; +import { workflowRefusalResponse } from "./workflow-refusal"; export { addFinalRequestLog, filterRequestLogs, @@ -1307,11 +1308,9 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server { if (workflow?.admitted) workflow.lease.release(); }; let response: Response; diff --git a/src/server/management-api.ts b/src/server/management-api.ts index 4dc2a16c77..bf1ccd7d14 100644 --- a/src/server/management-api.ts +++ b/src/server/management-api.ts @@ -144,6 +144,17 @@ async function handleQuotaResetRoutesOnDemand(ctx: ManagementContext): Promise { + if (!pathInManagementNamespace(ctx.url.pathname, "/api/workflow-budget", true)) return null; + const { handleWorkflowBudgetRoutes } = await import("./management/workflow-budget-routes"); + return handleWorkflowBudgetRoutes(ctx); +} + async function handleGrokCouponRoutesOnDemand(ctx: ManagementContext): Promise { if (!pathInManagementNamespace(ctx.url.pathname, "/api/grok/reset-coupons", true)) return null; const { handleGrokCouponRoutes } = await import("./management/grok-coupon-routes"); @@ -263,6 +274,7 @@ export async function handleManagementAPI( ?? (await handleLogsUsageRoutes(ctx)) ?? (await handleRequestHistoryRoutes(ctx)) ?? (await handleQuotaResetRoutesOnDemand(ctx)) + ?? (await handleWorkflowBudgetRoutesOnDemand(ctx)) ?? (await handleGrokCouponRoutesOnDemand(ctx)) ?? (await handleRoutingAnalyticsRoutes(ctx)) ?? (await handleRoutingProfileRoutesOnDemand(ctx)) diff --git a/src/server/management/route-registry.ts b/src/server/management/route-registry.ts index 2f77f0fe44..bb47a39b04 100644 --- a/src/server/management/route-registry.ts +++ b/src/server/management/route-registry.ts @@ -308,6 +308,9 @@ export const MANAGEMENT_ROUTES: readonly ManagementRoute[] = [ { method: "PUT", path: "/api/provider-context-caps", module: "server/management/provider-routes", mutates: true }, // server/management/quota-reset-routes { method: "GET", path: "/api/quota-resets", module: "server/management/quota-reset-routes", mutates: false, mechanism: "negated-guard" }, + // server/management/workflow-budget-routes + { method: "GET", path: "/api/workflow-budget", module: "server/management/workflow-budget-routes", mutates: false, exempt: { reason: "deferred-verb", why: "Reading a root's live budget is owed a CLI verb -- an operator staring at a 429 is usually already in a terminal -- but the ledger is process memory with no local transport to read it through, so the verb has to be an HTTP call the CLI does not yet make.", owner: "260915_workflow_budget_window wfc", ownerDoc: "devlog/_plan/260915_workflow_budget_window/030_wfc_diff_plan.md" } }, + { method: "POST", path: "/api/workflow-budget/clear", module: "server/management/workflow-budget-routes", mutates: true, exempt: { reason: "deferred-verb", why: "Clearing one root is owed the same verb as the read above and for the same reason. It is deliberately not shipped as a verb in this work-phase: the read comes first, because an operator who cannot see which ceiling fired has no basis for deciding to forgive it.", owner: "260915_workflow_budget_window wfc", ownerDoc: "devlog/_plan/260915_workflow_budget_window/030_wfc_diff_plan.md" } }, // server/management/request-history-routes { method: "GET", path: "/api/request-history", module: "server/management/request-history-routes", mutates: false }, // server/management/routing-analytics-routes diff --git a/src/server/management/workflow-budget-routes.ts b/src/server/management/workflow-budget-routes.ts new file mode 100644 index 0000000000..4cdebd4572 --- /dev/null +++ b/src/server/management/workflow-budget-routes.ts @@ -0,0 +1,133 @@ +/** + * Operator view of the in-memory workflow-budget ledger, plus a targeted clear. + * + * Loaded on demand from src/server/management-api.ts, which is the FOURTH entry in the protected + * set of tests/core-lab-boundary.test.ts — added precisely because eagerly importing handlers + * there put ~70 modules on every dashboard request. A static import here would make this + * subsystem the next instance of that bug. + * + * Authentication is inherited: every /api route passes through requireManagementAuth before the + * chain runs, so these handlers add no auth code of their own. The GET spends no user identity. + * The POST clears one root's windowed count ceilings; it does not spend identity, and the + * underlying ledger leaves in-flight concurrency and the token spend record untouched. + */ + +import { jsonResponse } from "../auth-cors"; +import type { OcxConfig } from "../../types"; +import type { ManagementContext } from "./context"; +import { readManagementJsonBodyOr } from "./body"; +import { + clearWorkflowBudgetForRoot, + listTrackedWorkflowRoots, + listWorkflowBudgetEvents, + workflowBudgetSnapshot, + WORKFLOW_EVENT_CAPACITY, +} from "../../lib/workflow-budget"; + +const DEFAULT_LIMIT = 20; +const MAX_LIMIT = WORKFLOW_EVENT_CAPACITY; +const MAX_ROOT_ID_LENGTH = 200; + +/** + * A root id is an opaque caller-thread token, not a path. Without a length cap a client + * could POST a multi-megabyte string that we would then store as a map key and echo back + * in events; 200 is well above any thread id we have seen and small enough to put in a URL. + */ +function parseRootId(raw: unknown): string | null { + if (typeof raw !== "string") return null; + const trimmed = raw.trim(); + if (!trimmed || trimmed.length > MAX_ROOT_ID_LENGTH) return null; + return trimmed; +} + +function invalidRootResponse(req: Request, config: OcxConfig): Response { + return jsonResponse( + { error: { code: "invalid_root", message: "root must be a non-empty string of at most 200 characters" } }, + 400, + req, + config, + ); +} + +function parseLimitParam(rawLimit: string | null): { ok: true; limit: number } | { ok: false } { + if (rawLimit !== null && !/^\d+$/.test(rawLimit)) return { ok: false }; + return { + ok: true, + limit: rawLimit === null ? DEFAULT_LIMIT : Math.min(MAX_LIMIT, Number.parseInt(rawLimit, 10)), + }; +} + +export async function handleWorkflowBudgetRoutes(ctx: ManagementContext): Promise { + const { url, req, config } = ctx; + + if (url.pathname === "/api/workflow-budget") { + if (req.method !== "GET") return null; + + const parsedLimit = parseLimitParam(url.searchParams.get("limit")); + if (!parsedLimit.ok) { + return jsonResponse( + { error: { code: "invalid_limit", message: "limit must be a non-negative integer" } }, + 400, + req, + config, + ); + } + const { limit } = parsedLimit; + + const rawRoot = url.searchParams.get("root"); + if (rawRoot !== null) { + const rootId = parseRootId(rawRoot); + if (rootId === null) return invalidRootResponse(req, config); + const snapshot = workflowBudgetSnapshot(rootId); + // An unknown id is a 200 with `root: null`, not a 404: the operator asked what this + // process currently holds for that token, and "nothing" is a legitimate answer. POST + // /clear is the opposite — claiming to forgive a ceiling that was never tracked would + // report a success that did not happen. + return jsonResponse( + { + root: snapshot ? { rootId, ...snapshot } : null, + events: listWorkflowBudgetEvents(WORKFLOW_EVENT_CAPACITY) + .filter((event) => event.rootId === rootId) + .slice(0, limit), + }, + 200, + req, + config, + ); + } + + return jsonResponse( + { + roots: listTrackedWorkflowRoots(limit), + events: listWorkflowBudgetEvents(limit), + }, + 200, + req, + config, + ); + } + + if (url.pathname === "/api/workflow-budget/clear") { + if (req.method !== "POST") return null; + + const body = await readManagementJsonBodyOr(req, {}); + const rawRoot = body && typeof body === "object" && !Array.isArray(body) + ? (body as { root?: unknown }).root + : undefined; + const rootId = parseRootId(rawRoot); + if (rootId === null) return invalidRootResponse(req, config); + + const before = clearWorkflowBudgetForRoot(rootId); + if (!before) { + return jsonResponse( + { error: { code: "unknown_root", message: "root is not currently tracked" } }, + 404, + req, + config, + ); + } + return jsonResponse({ cleared: true, root: rootId, before }, 200, req, config); + } + + return null; +} diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 355d5bbd2b..bac95dcfc8 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -243,6 +243,7 @@ import { chargeWorkflowSends, workflowSendCeilingReached, } from "../../lib/workflow-budget"; +import { workflowRefusalResponse } from "../workflow-refusal"; import { ForwardAdmissionCredentialError, hasForwardableCodexBearer, @@ -5355,11 +5356,9 @@ async function handleResponsesInner( // laundering this ceiling exists to stop. The client is told the task needs a new grant // rather than being given a synthetic upstream error. if (workflowSendCeilingReached(workflowRootId)) { - return formatErrorResponse( - 429, - "workflow_budget_exhausted", - "This task has used its whole send budget, so no further upstream request was made. Requests already in flight settle as they finish.", - ); + // A log context exists here, unlike at HTTP admission, so the row this request writes is + // marked synthetic rather than reading as a request that vanished with zero sends. + return workflowRefusalResponse("workflow-sends-exhausted", logCtx); } // No floor. Math.max(1, ...) meant an exhausted request still funded one send on every // recovery leg, so a bounded per-leg allowance never became a bounded per-request one. diff --git a/src/server/workflow-refusal.ts b/src/server/workflow-refusal.ts new file mode 100644 index 0000000000..2df72039fe --- /dev/null +++ b/src/server/workflow-refusal.ts @@ -0,0 +1,45 @@ +/** + * The one place that knows how this proxy refuses a turn on its own workflow budget. + * + * It is a module rather than two inline blocks because the two call sites -- the HTTP admission + * check in `src/server/index.ts` and the pre-dispatch ceiling check in + * `src/server/responses/core.ts` -- had drifted into saying different things about the same + * refusal, and because the non-obvious part below has to be stated once and not twice. + */ +import { formatErrorResponse } from "../bridge"; +import { markLocalRequestLogRefusal, type RequestLogContext } from "./request-log"; +import { + WORKFLOW_LOCAL_REFUSAL_HEADER, + workflowDenialSummary, + type WorkflowDenial, +} from "../lib/workflow-budget"; + +/** + * Build the 429 for a refusal this proxy made itself. + * + * The status and type arguments below do not reach the client: `classifyError` rewrites every + * 429 to `rate_limit_error` / `rate_limit_exceeded`, so the body is shaped exactly like a + * provider rate limit. That is a deliberate wire contract -- changing it would change how every + * client retries -- which leaves two places to carry the truth. The message names the ceiling + * that fired and says no provider was contacted, and the header carries the machine-readable + * name. Nothing upstream sets that header, so its presence is conclusive. + * + * When a request-log context exists, the row is additionally marked synthetic through the helper + * #4639 introduced for the same problem. The HTTP admission check has no context to pass -- it + * runs before the body is parsed, so there is no model and no provider yet -- which is why that + * argument is optional rather than required. + */ +export function workflowRefusalResponse( + reason: WorkflowDenial, + logCtx?: RequestLogContext, +): Response { + const summary = workflowDenialSummary(reason); + if (logCtx) markLocalRequestLogRefusal(logCtx, summary.code); + const refusal = formatErrorResponse( + 429, + reason === "workflow-sends-exhausted" ? "workflow_budget_exhausted" : "queue_capacity_exceeded", + summary.message, + ); + refusal.headers.set(WORKFLOW_LOCAL_REFUSAL_HEADER, summary.code); + return refusal; +} diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index b4f994c4aa..7e50fb16b5 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -716,6 +716,7 @@ "management-origin-tls.test.ts": "server", "management-provider-validation.test.ts": "server", "management-route-registry.test.ts": "server", + "management-workflow-budget-routes.test.ts": "server", "memory-watchdog.test.ts": "server", "meta-model-api-provider.test.ts": "providers", "meta-muse-oauth.test.ts": "providers", diff --git a/tests/lib/workflow-budget.test.ts b/tests/lib/workflow-budget.test.ts index a8fb7ce447..e87fd6108d 100644 --- a/tests/lib/workflow-budget.test.ts +++ b/tests/lib/workflow-budget.test.ts @@ -7,13 +7,22 @@ import { import { admitWorkflowTurn, chargeWorkflowSends, + clearWorkflowBudgetForRoot, DEFAULT_WORKFLOW_BUDGET_POLICY, + listTrackedWorkflowRoots, + listWorkflowBudgetEvents, resetWorkflowBudgetsForTest, settleWorkflowSpend, workflowBudgetSnapshot, + workflowDenialSummary, workflowSendCeilingReached, + WORKFLOW_EVENT_CAPACITY, + WORKFLOW_LOCAL_REFUSAL_HEADER, + type WorkflowDenial, type WorkflowBudgetPolicy, } from "../../src/lib/workflow-budget"; +import { workflowRefusalResponse } from "../../src/server/workflow-refusal"; +import type { RequestLogContext } from "../../src/server/request-log"; const memoryJournal = (): SpendJournal & { lines: string[] } => { const lines: string[] = []; @@ -372,3 +381,140 @@ describe("every ceiling on this path reads the caller's clock", () => { expect(ambient).toEqual([]); }); }); + +describe("a refusal an operator can read, name and clear (#4546)", () => { + const ALL_DENIALS: WorkflowDenial[] = [ + "workflow-concurrency-exhausted", + "workflow-sends-exhausted", + "workflow-children-exhausted", + "workflow-spend-exhausted", + "workflow-tracking-exhausted", + "workflow-send-replayed", + "workflow-spend-undurable", + ]; + + beforeEach(() => { + resetWorkflowBudgetsForTest(); + }); + + test("each ceiling gets its own sentence rather than one shared with the others", () => { + // The bug this replaces: all four count denials emitted one sentence about a + // "concurrent-work limit", so an operator who had hit the SEND ceiling was told to wait for + // turns to finish. Waiting never helped, because no turn was running. + const messages = ALL_DENIALS.map(reason => workflowDenialSummary(reason).message); + expect(new Set(messages).size).toBe(ALL_DENIALS.length); + for (const message of messages) { + // Every one of them has to say whose decision this was; that is the half an operator + // cannot recover from the wire, since the body is shaped like a provider rate limit. + expect(message).toContain("This proxy refused the request locally"); + } + expect(workflowDenialSummary("workflow-sends-exhausted").message).toContain("send ceiling"); + expect(workflowDenialSummary("workflow-children-exhausted").message).toContain("child threads"); + }); + + test("the response carries the machine-readable ceiling name a 429 body cannot", () => { + const refusal = workflowRefusalResponse("workflow-children-exhausted"); + expect(refusal.status).toBe(429); + expect(refusal.headers.get(WORKFLOW_LOCAL_REFUSAL_HEADER)).toBe("workflow_children_exhausted"); + }); + + test("a refusal with a log context marks its row synthetic", () => { + const logCtx = { model: "m", provider: "p" } as RequestLogContext; + workflowRefusalResponse("workflow-sends-exhausted", logCtx); + expect(logCtx.terminalSource).toBe("synthetic"); + expect(logCtx.localTerminalReason).toBe("workflow_sends_exhausted"); + }); + + test("every refusal lands on the record with the counts that caused it", () => { + const now = 1_700_000_000_000; + const policy: WorkflowBudgetPolicy = { ...DEFAULT_WORKFLOW_BUDGET_POLICY, maxPhysicalSends: 2 }; + const seeded = admitWorkflowTurn("root-r", "worker", policy, undefined, now); + seeded?.lease.release(); + chargeWorkflowSends("root-r", 2, now); + const denied = admitWorkflowTurn("root-r", "worker", policy, undefined, now + 1); + expect(denied?.admitted).toBe(false); + + const [latest] = listWorkflowBudgetEvents(4); + expect(latest?.kind).toBe("refused"); + expect(latest?.rootId).toBe("root-r"); + expect(latest?.reason).toBe("workflow-sends-exhausted"); + expect(latest?.sends).toBe(2); + }); + + test("the event record is bounded", () => { + const now = 1_700_000_000_000; + const policy: WorkflowBudgetPolicy = { ...DEFAULT_WORKFLOW_BUDGET_POLICY, maxPhysicalSends: 1 }; + const seeded = admitWorkflowTurn("root-s", "worker", policy, undefined, now); + seeded?.lease.release(); + chargeWorkflowSends("root-s", 1, now); + for (let i = 0; i < WORKFLOW_EVENT_CAPACITY * 2; i += 1) { + admitWorkflowTurn("root-s", "worker", policy, undefined, now + 1 + i); + } + expect(listWorkflowBudgetEvents(1_000).length).toBe(WORKFLOW_EVENT_CAPACITY); + }); + + test("clearing one root moves its ceilings and nothing else", () => { + const now = 1_700_000_000_000; + const policy: WorkflowBudgetPolicy = { ...DEFAULT_WORKFLOW_BUDGET_POLICY, maxPhysicalSends: 2 }; + const held = admitWorkflowTurn("root-t", "worker", policy, "child-1", now); + expect(held?.admitted).toBe(true); + chargeWorkflowSends("root-t", 2, now); + expect(workflowSendCeilingReached("root-t", policy, now)).toBe(true); + + const before = clearWorkflowBudgetForRoot("root-t", policy, now); + expect(before?.sends).toBe(2); + expect(before?.children).toBe(1); + + const after = workflowBudgetSnapshot("root-t", policy, now); + expect(after?.sends).toBe(0); + expect(after?.children).toBe(0); + // The turn holding a slot is still holding it: zeroing `active` would let its release drive + // the count negative and hand out concurrency that is already taken. + expect(after?.active).toBe(1); + // And the lifetime total survives, so clearing a ceiling cannot launder the record of what + // the root actually did. + expect(after?.lifetimeSends).toBe(2); + expect(workflowSendCeilingReached("root-t", policy, now)).toBe(false); + held?.lease.release(); + + const [latest] = listWorkflowBudgetEvents(1); + expect(latest?.kind).toBe("cleared"); + expect(latest?.rootId).toBe("root-t"); + expect(latest?.sends).toBe(2); + }); + + test("clearing a count ceiling does not forgive spend", () => { + // The dangerous version of this feature. A count ceiling is a rate guard an operator may + // reasonably wave off; a token ceiling is money, and one button must not do both. + const ledger = createSpendReservationLedger({ journal: memoryJournal(), policy: spendPolicy(100), now: () => 1_000 }); + const spend = (sendId: string) => ({ sendId, inputTokens: 60, outputCeilingTokens: 40 }); + expect(admitWorkflowTurn("root-u", "interactive", DEFAULT_WORKFLOW_BUDGET_POLICY, + undefined, 1_000, spend("s1"), ledger)?.admitted).toBe(true); + + clearWorkflowBudgetForRoot("root-u", DEFAULT_WORKFLOW_BUDGET_POLICY, 1_000); + + const denied = admitWorkflowTurn("root-u", "interactive", DEFAULT_WORKFLOW_BUDGET_POLICY, + undefined, 1_000, spend("s2"), ledger); + expect(denied?.admitted).toBe(false); + if (denied && !denied.admitted) expect(denied.reason).toBe("workflow-spend-exhausted"); + }); + + test("clearing an untracked root reports that rather than inventing one", () => { + expect(clearWorkflowBudgetForRoot("never-seen")).toBeUndefined(); + expect(workflowBudgetSnapshot("never-seen")).toBeUndefined(); + expect(listWorkflowBudgetEvents(1)).toEqual([]); + }); + + test("tracked roots are listed most recently active first and bounded", () => { + const now = 1_700_000_000_000; + for (const [index, root] of ["root-v", "root-w", "root-x"].entries()) { + admitWorkflowTurn(root, "worker", DEFAULT_WORKFLOW_BUDGET_POLICY, undefined, now + index) + ?.lease.release(); + } + const listed = listTrackedWorkflowRoots(2, DEFAULT_WORKFLOW_BUDGET_POLICY, now + 10); + expect(listed.length).toBe(2); + expect(listed[0]?.rootId).toBe("root-x"); + expect(listed[1]?.rootId).toBe("root-w"); + }); +}); + diff --git a/tests/server/management-workflow-budget-routes.test.ts b/tests/server/management-workflow-budget-routes.test.ts new file mode 100644 index 0000000000..0fb8c98e91 --- /dev/null +++ b/tests/server/management-workflow-budget-routes.test.ts @@ -0,0 +1,108 @@ +import { beforeEach, describe, expect, test } from "bun:test"; +import { handleManagementAPI } from "../../src/server/management-api"; +import { ManagementRequest as Request } from "../helpers/management-auth"; +import type { OcxConfig } from "../../src/types"; +import { + admitWorkflowTurn, + chargeWorkflowSends, + resetWorkflowBudgetsForTest, +} from "../../src/lib/workflow-budget"; + +const config = { providers: [] } as unknown as OcxConfig; + +beforeEach(() => { + resetWorkflowBudgetsForTest(); +}); + +async function call(path: string, init?: RequestInit): Promise { + const url = new URL("http://localhost" + path); + const response = await handleManagementAPI(new Request(url, init), url, config); + if (!response) throw new Error("management API did not handle " + (init?.method ?? "GET") + " " + path); + return response; +} + +function seedChargedRoot(rootId: string, sends = 3): void { + const decision = admitWorkflowTurn(rootId, "interactive"); + if (decision?.admitted !== true) throw new Error("failed to admit " + rootId); + chargeWorkflowSends(rootId, sends); +} + +describe("GET /api/workflow-budget", () => { + test("?root= returns the snapshot for a root that was admitted and charged", async () => { + seedChargedRoot("root-a", 3); + + const response = await call("/api/workflow-budget?root=root-a"); + expect(response.status).toBe(200); + const body = await response.json() as { + root: { rootId: string; sends: number } | null; + events: unknown[]; + }; + expect(body.root).not.toBeNull(); + expect(body.root?.rootId).toBe("root-a"); + expect(body.root?.sends).toBe(3); + expect(Array.isArray(body.events)).toBe(true); + }); + + test("an unknown root returns root: null rather than 404", async () => { + const response = await call("/api/workflow-budget?root=never-seen"); + expect(response.status).toBe(200); + const body = await response.json() as { root: unknown; events: unknown[] }; + expect(body.root).toBeNull(); + expect(body.events).toEqual([]); + }); +}); + +describe("POST /api/workflow-budget/clear", () => { + test("a tracked root returns cleared: true and the before-snapshot, and a following GET shows sends at 0", async () => { + seedChargedRoot("root-a", 4); + + const cleared = await call("/api/workflow-budget/clear", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ root: "root-a" }), + }); + expect(cleared.status).toBe(200); + const payload = await cleared.json() as { + cleared: boolean; + root: string; + before: { sends: number }; + }; + expect(payload.cleared).toBe(true); + expect(payload.root).toBe("root-a"); + expect(payload.before.sends).toBe(4); + + const after = await call("/api/workflow-budget?root=root-a"); + expect(after.status).toBe(200); + const body = await after.json() as { root: { sends: number } | null }; + expect(body.root?.sends).toBe(0); + }); + + test("an untracked root is 404 unknown_root", async () => { + const response = await call("/api/workflow-budget/clear", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ root: "never-seen" }), + }); + expect(response.status).toBe(404); + const body = await response.json() as { error: { code: string } }; + expect(body.error.code).toBe("unknown_root"); + }); + + test("a missing or blank root is 400 invalid_root", async () => { + const missing = await call("/api/workflow-budget/clear", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({}), + }); + expect(missing.status).toBe(400); + expect((await missing.json() as { error: { code: string } }).error.code).toBe("invalid_root"); + + const blank = await call("/api/workflow-budget/clear", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ root: " " }), + }); + expect(blank.status).toBe(400); + expect((await blank.json() as { error: { code: string } }).error.code).toBe("invalid_root"); + }); +}); From 42a49a11fbefb4210a195b514f9c33d49afde65d Mon Sep 17 00:00:00 2001 From: JUN Date: Tue, 15 Sep 2026 05:41:30 +0900 Subject: [PATCH 2/5] fix(server): put the workflow refusal on the request log and expose its header (#4546) CI caught one test and an independent plan review caught two real gaps. The test failure was mine and it was the fixture-assumption mistake again: the tracked-roots ordering test released each lease, and release() stamps lastSeenMs from the wall clock because it feeds eviction ordering rather than a ceiling. Three injected timestamps collapsed into three near-identical real ones. The leases now stay open, which is what the test was actually about. The review's blocking finding was that skipping the request-log row was a choice, not a constraint. addFinalRequestLog takes whatever model and provider the context carries, and the /v1/responses caller already seeds unknown/unknown before the turn runs -- the same placeholder the native passthrough path writes. So the refusal now writes a real row through that context: terminalSource synthetic, a local reason, and an error code naming the ceiling, which wins over the generic 429 classification in the logs column. Only /v1/responses threads it, because the workflow root is the Codex x-codex-parent-thread-id header and no other inbound wire carries it. Second finding: the marker header was invisible to the dashboard. The data plane never sets Access-Control-Expose-Headers, so cross-origin JavaScript could not read it and the marker was useful to curl and to nothing else. The refusal now exposes it. The wire body is deliberately still rate_limit_error / rate_limit_exceeded. The review asked for a distinct error code through classifyError; that changes how every client classifies a 429 from this proxy, and the surface an operator actually reads is the log row, which now carries the name. Also recorded the two routes in structure/gui-and-management-api.md. Local suite, typecheck, install and build: NOT RUN. Hosted CI at the exact head. --- src/server/index.ts | 11 ++++--- src/server/workflow-refusal.ts | 46 ++++++++++++++++++++++---- structure/gui-and-management-api.md | 1 + tests/lib/workflow-budget.test.ts | 51 ++++++++++++++++++++++++++--- 4 files changed, 94 insertions(+), 15 deletions(-) diff --git a/src/server/index.ts b/src/server/index.ts index 042a9744bb..ad5f63782a 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -135,7 +135,7 @@ import { } from "./request-log"; import { sessionLaneIdFromRequest } from "./request-log-conversation"; import { admitWorkflowTurn, type WorkflowLane } from "../lib/workflow-budget"; -import { workflowRefusalResponse } from "./workflow-refusal"; +import { workflowRefusalResponse, type WorkflowRefusalLog } from "./workflow-refusal"; export { addFinalRequestLog, filterRequestLogs, @@ -1291,6 +1291,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server Promise, + refusalLog?: WorkflowRefusalLog, ): Promise { const lease = tryAdmitTurn(sessionLaneIdFromRequest(req.headers)); if (!lease) return serverBusyResponse(req, "active turns", policy); @@ -1308,9 +1309,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server { if (workflow?.admitted) workflow.lease.release(); }; let response: Response; @@ -2585,7 +2584,9 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server { const lines: string[] = []; @@ -423,6 +427,41 @@ describe("a refusal an operator can read, name and clear (#4546)", () => { workflowRefusalResponse("workflow-sends-exhausted", logCtx); expect(logCtx.terminalSource).toBe("synthetic"); expect(logCtx.localTerminalReason).toBe("workflow_sends_exhausted"); + // A locally assigned code wins over the 429 classification, so this is what the logs + // column shows instead of a generic rate limit. + expect(logCtx.errorCode).toBe("workflow_sends_exhausted"); + }); + + test("a refusal before the body is parsed still leaves a row in the logs", () => { + // The defect this closes: the HTTP admission check returns before the turn runs, so a + // refused request left no trace at all on /api/logs. The model and provider stay + // "unknown" because they genuinely never resolved -- the same placeholder the native + // passthrough path already writes -- and the row says who refused and why. + clearRequestLogsForTests(); + const before = getRequestLogEntries().length; + const logCtx = { model: "unknown", provider: "unknown" } as RequestLogContext; + const refusal = workflowRefusalResponse("workflow-children-exhausted", undefined, { + requestId: "req-refusal-1", + start: Date.now() - 5, + logCtx, + }); + expect(refusal.status).toBe(429); + + const written = getRequestLogEntries(); + expect(written.length).toBe(before + 1); + const row = written.find(entry => entry.id === "req-refusal-1"); + expect(row?.terminalSource).toBe("synthetic"); + expect(row?.localTerminalReason).toBe("workflow_children_exhausted"); + expect(row?.errorCode).toBe("workflow_children_exhausted"); + clearRequestLogsForTests(); + }); + + test("the ceiling name is readable by a browser dashboard, not only by curl", () => { + // A header the data plane never exposes is invisible to cross-origin JavaScript, which + // would have made this marker useful to curl and to nothing else. + const refusal = workflowRefusalResponse("workflow-sends-exhausted"); + expect(refusal.headers.get("Access-Control-Expose-Headers")) + .toContain(WORKFLOW_LOCAL_REFUSAL_HEADER); }); test("every refusal lands on the record with the counts that caused it", () => { @@ -507,9 +546,14 @@ describe("a refusal an operator can read, name and clear (#4546)", () => { test("tracked roots are listed most recently active first and bounded", () => { const now = 1_700_000_000_000; + // The leases are deliberately left open. `release()` stamps `lastSeenMs` from the wall + // clock -- it feeds eviction ordering, not a ceiling -- which would collapse the injected + // ordering this test is about into three near-identical real timestamps. for (const [index, root] of ["root-v", "root-w", "root-x"].entries()) { - admitWorkflowTurn(root, "worker", DEFAULT_WORKFLOW_BUDGET_POLICY, undefined, now + index) - ?.lease.release(); + const admitted = admitWorkflowTurn( + root, "worker", DEFAULT_WORKFLOW_BUDGET_POLICY, undefined, now + index, + ); + expect(admitted?.admitted).toBe(true); } const listed = listTrackedWorkflowRoots(2, DEFAULT_WORKFLOW_BUDGET_POLICY, now + 10); expect(listed.length).toBe(2); @@ -517,4 +561,3 @@ describe("a refusal an operator can read, name and clear (#4546)", () => { expect(listed[1]?.rootId).toBe("root-w"); }); }); - From 23c27a79ce00f233b869cf6920d6f270b6e5239a Mon Sep 17 00:00:00 2001 From: JUN Date: Tue, 15 Sep 2026 05:47:33 +0900 Subject: [PATCH 3/5] fix(lib): look the refusal row up by requestId, which is the field it has (#4546) The row was written correctly -- the count assertion passed -- but the lookup read entry.id, and RequestLogEntry names that field requestId. find() returned undefined and the terminalSource assertion read a property of nothing. Local suite, typecheck, install and build: NOT RUN. Hosted CI at the exact head. --- tests/lib/workflow-budget.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/lib/workflow-budget.test.ts b/tests/lib/workflow-budget.test.ts index 1b41f32273..d767d6c0eb 100644 --- a/tests/lib/workflow-budget.test.ts +++ b/tests/lib/workflow-budget.test.ts @@ -449,7 +449,7 @@ describe("a refusal an operator can read, name and clear (#4546)", () => { const written = getRequestLogEntries(); expect(written.length).toBe(before + 1); - const row = written.find(entry => entry.id === "req-refusal-1"); + const row = written.find(entry => entry.requestId === "req-refusal-1"); expect(row?.terminalSource).toBe("synthetic"); expect(row?.localTerminalReason).toBe("workflow_children_exhausted"); expect(row?.errorCode).toBe("workflow_children_exhausted"); From ccb318c8ac16f59a9c985f1c925f6d398afbcd08 Mon Sep 17 00:00:00 2001 From: JUN Date: Tue, 15 Sep 2026 06:02:01 +0900 Subject: [PATCH 4/5] fix(server): reach every surface's log, and actually let a browser read the header (#4546) Second review round found that both of the first round's fixes stopped short. The claim that only /v1/responses carries the Codex parent-thread header was wrong. /v1/responses/compact carries it too -- its existing tests send it -- and runAdmittedHttpTurn reads it for every caller, so a compact refusal still left nothing on /api/logs. Every inbound surface that opens a request-log row now threads it: compact, images, context history, alpha search, messages, chat completions, audio transcriptions and live. /v1/messages/count_tokens is the one that does not, because it opens no row at all. Exposing the marker header was pointless while the admission refusal returned a raw Response. Access-Control-Expose-Headers without Access-Control-Allow-Origin is unreadable to cross-origin JavaScript, so the header remained useful to curl and to nothing else. The refusal is wrapped in withCors now. The pre-dispatch ceiling check in the responses path is a second refusal taken after admission already succeeded, so nothing inside the budget module saw it: it was the one refusal an operator could hit that left no event behind. It records one now, through a seam that only a caller which decided the refusal itself uses, so admitWorkflowTurn's own denials are not double-counted. The review also pointed out that a unit test on the helper proves the helper and not the wiring, and the wiring is exactly where this went wrong twice. A source guard now asserts that every runAdmittedHttpTurn call site but one threads a refusal row, and that the refusal is CORS-wrapped. Local suite, typecheck, install and build: NOT RUN. Hosted CI at the exact head. --- src/lib/workflow-budget.ts | 24 ++++++++++++++++++++++++ src/server/index.ts | 22 +++++++++++----------- src/server/responses/core.ts | 2 +- src/server/workflow-refusal.ts | 5 +++++ tests/lib/workflow-budget.test.ts | 17 +++++++++++++++++ 5 files changed, 58 insertions(+), 12 deletions(-) diff --git a/src/lib/workflow-budget.ts b/src/lib/workflow-budget.ts index 125890d0fa..0a5cb18cfc 100644 --- a/src/lib/workflow-budget.ts +++ b/src/lib/workflow-budget.ts @@ -278,6 +278,30 @@ export function listWorkflowBudgetEvents(limit: number = WORKFLOW_EVENT_CAPACITY return budgetEvents.slice(-wanted).reverse(); } +/** + * Record a refusal decided outside `admitWorkflowTurn`. + * + * The pre-dispatch ceiling check in the responses path is a second refusal, taken after + * admission already succeeded, so nothing in this module sees it. Without this it was the one + * refusal an operator could hit that left no event behind. + */ +export function recordWorkflowRefusalEvent( + rootId: string | undefined, + reason: WorkflowDenial, + now: number = Date.now(), +): void { + if (!rootId) return; + const state = roots.get(rootId); + recordBudgetEvent({ + at: now, + kind: "refused", + rootId, + reason, + sends: state ? windowedSends(state, now) : 0, + children: state ? windowedChildren(state, now) : 0, + }); +} + export type WorkflowLane = "interactive" | "worker"; export interface WorkflowAdmission { diff --git a/src/server/index.ts b/src/server/index.ts index ad5f63782a..d26814b0ed 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -1309,7 +1309,9 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server { if (workflow?.admitted) workflow.lease.release(); }; let response: Response; @@ -2422,7 +2424,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server { .toContain(WORKFLOW_LOCAL_REFUSAL_HEADER); }); + test("every inbound surface that opens a log row threads its refusal into one", async () => { + // A unit test on the helper proves the helper. It does not prove the wiring, and the + // wiring is where this went wrong twice: the refusal originally reached no surface's log + // at all, and the fix first reached only one of nine. Exposing the header was likewise + // pointless until the refusal was CORS-wrapped, because without an allow-origin a browser + // cannot read an exposed header either. + const source = await Bun.file(repoPath("src/server/index.ts")).text(); + const callSites = source.match(/return runAdmittedHttpTurn\(/g) ?? []; + const threaded = source.match(/, \{ requestId, start, logCtx \}\);/g) ?? []; + expect(callSites.length).toBeGreaterThan(1); + // Exactly one surface has no log context to thread: /v1/messages/count_tokens opens no + // request-log row at all. Every other one must, or a refusal there leaves no trace. + expect(callSites.length - threaded.length).toBe(1); + expect(source).toContain("withCors(workflowRefusalResponse("); + }); + test("every refusal lands on the record with the counts that caused it", () => { const now = 1_700_000_000_000; const policy: WorkflowBudgetPolicy = { ...DEFAULT_WORKFLOW_BUDGET_POLICY, maxPhysicalSends: 2 }; From 00d0d36ba70e374ee5769bc16c8d7f56a84181ea Mon Sep 17 00:00:00 2001 From: JUN Date: Tue, 15 Sep 2026 06:08:26 +0900 Subject: [PATCH 5/5] test(server): stop pinning a call's terminator in the loopback CORS oracles (#4546) Two source-oracle tests asserted the literal "req,\n policy,\n ));" inside the Anthropic and chat branches. The invariant they exist for is that withCors finishes with the receiving listener's policy rather than the public config. The closing "));" was the outer runAdmittedHttpTurn call's terminator, which has nothing to do with that, and both went red the moment that call gained a fourth argument. The assertions now stop at the closing paren of withCors, so they still fail on config and no longer fail on an unrelated argument. Local suite, typecheck, install and build: NOT RUN. Hosted CI at the exact head. --- tests/server/loopback-listener-admission.test.ts | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/tests/server/loopback-listener-admission.test.ts b/tests/server/loopback-listener-admission.test.ts index 140cebbbfb..8065b80bbe 100644 --- a/tests/server/loopback-listener-admission.test.ts +++ b/tests/server/loopback-listener-admission.test.ts @@ -78,8 +78,12 @@ describe("loopback listener policy view", () => { source.slice(countTokensStart, messagesStart), source.slice(messagesStart, chatStart), ]) { - expect(branch).toContain("req,\n policy,\n ));"); - expect(branch).not.toContain("req,\n config,\n ));"); + // The tail stops at the closing paren of withCors on purpose. Pinning the call's own + // terminator pinned something this test does not care about: when runAdmittedHttpTurn + // gained a fourth argument (#4546) both of these went red while the invariant they + // exist for -- policy, never config -- was untouched. + expect(branch).toContain("req,\n policy,\n )"); + expect(branch).not.toContain("req,\n config,\n )"); } }); }); @@ -120,8 +124,8 @@ describe("local client inference wires on the loopback listener (#4236)", () => expect(chatStart).toBeGreaterThan(-1); const branch = source.slice(chatStart, nextRoute); expect(branch).toContain("handleChatCompletions(req, config, logCtx"); - expect(branch).toContain("req,\n policy,\n ));"); - expect(branch).not.toContain("req,\n config,\n ));"); + expect(branch).toContain("req,\n policy,\n )"); + expect(branch).not.toContain("req,\n config,\n )"); }); });