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..0a5cb18cfc 100644 --- a/src/lib/workflow-budget.ts +++ b/src/lib/workflow-budget.ts @@ -162,6 +162,146 @@ 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(); +} + +/** + * 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 { @@ -287,12 +427,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 +456,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 +489,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 +600,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 +611,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 +631,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..d26814b0ed 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, type WorkflowRefusalLog } from "./workflow-refusal"; export { addFinalRequestLog, filterRequestLogs, @@ -1290,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); @@ -1307,11 +1309,9 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server { if (workflow?.admitted) workflow.lease.release(); }; let response: Response; @@ -2424,7 +2424,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server { + 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..7e2562b84e 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, undefined, workflowRootId); } // 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..6b43feb3fe --- /dev/null +++ b/src/server/workflow-refusal.ts @@ -0,0 +1,84 @@ +/** + * 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 { + addFinalRequestLog, + markLocalRequestLogRefusal, + type RequestLogContext, +} from "./request-log"; +import { + WORKFLOW_LOCAL_REFUSAL_HEADER, + workflowDenialSummary, + recordWorkflowRefusalEvent, + type WorkflowDenial, +} from "../lib/workflow-budget"; + +/** + * What a caller needs to hand over for the refusal to become a row on `/api/logs`. + * + * The HTTP admission check refuses before the body is parsed, so its `logCtx` still carries the + * `unknown` model and provider the caller seeded it with. That is the honest record -- this + * request genuinely never resolved either -- and it is the same placeholder the native + * passthrough path already writes. Skipping the row entirely was the worse option: an operator + * reading the logs saw no trace at all of a request the proxy had refused. + */ +export interface WorkflowRefusalLog { + readonly requestId: string; + readonly start: number; + readonly logCtx: RequestLogContext; +} + +/** + * 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. + * + * The row is where an operator actually looks, so it gets the same treatment #4639 established: + * `terminalSource: "synthetic"`, a local reason, and an error code naming the ceiling. Pass + * `logCtx` when the caller is inside a turn that will write its own row, or `refusalLog` when + * the refusal happens before any row exists and this is the only chance to write one. + */ +export function workflowRefusalResponse( + reason: WorkflowDenial, + logCtx?: RequestLogContext, + refusalLog?: WorkflowRefusalLog, + rootId?: string, +): Response { + const summary = workflowDenialSummary(reason); + // Only a caller that decided the refusal ITSELF passes a root id. admitWorkflowTurn already + // records its own denials, so passing one there would double-count them. + if (rootId) recordWorkflowRefusalEvent(rootId, reason); + const recordOn = logCtx ?? refusalLog?.logCtx; + if (recordOn) { + markLocalRequestLogRefusal(recordOn, summary.code); + // A locally assigned code wins in addFinalRequestLog, so this is what names the ceiling in + // the logs column rather than the generic rate-limit classification a 429 would get. + recordOn.errorCode = summary.code; + } + if (refusalLog) { + addFinalRequestLog(refusalLog.requestId, refusalLog.start, refusalLog.logCtx, 429, { + closeReason: "terminal", + }); + } + 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); + // Without this a browser dashboard cannot read the header at all: the data plane never sets + // Access-Control-Expose-Headers, so a cross-origin reader sees only the CORS-safelisted ones. + refusal.headers.set("Access-Control-Expose-Headers", WORKFLOW_LOCAL_REFUSAL_HEADER); + return refusal; +} diff --git a/structure/gui-and-management-api.md b/structure/gui-and-management-api.md index 44faba2956..f7524ca6d9 100644 --- a/structure/gui-and-management-api.md +++ b/structure/gui-and-management-api.md @@ -137,6 +137,7 @@ this document owns is which module holds which area and what invariant that area | Grok and Claude integrations | `src/server/management/agent-settings-routes.ts` — `GET /api/grok`, `PUT /api/grok/selection`, `POST /api/grok/apply`, `GET/PUT /api/claude-desktop`, `POST /api/claude-desktop/apply`, `GET /api/claude-desktop/status`, `GET/PUT /api/claude-code`. Apply writes an external app's profile, so its status probe must read the same resolved path it writes (see [`responses.md`](transports/responses.md)). | | Grok reset coupons | `src/server/management/grok-coupon-routes.ts` — `GET /api/grok/reset-coupons`, `POST /api/grok/reset-coupons/consume`. The dashboard owner is `gui/src/hooks/useGrokResetCoupons.ts` with `gui/src/components/provider-workspace/GrokResetCoupons.tsx`, wired into the xAI OAuth rows of `ProviderAuthPanel`. Redemption truth is the settled ledger `code`, not the HTTP status: a replayed failure returns 200 with `replayed: true`. See [`providers/xai-grok.md`](providers/xai-grok.md). | | Combos | `src/server/management/combo-routes.ts` — `GET/PUT/DELETE /api/combos` own provider combination and failover definitions. | +| Workflow budget | `src/server/management/workflow-budget-routes.ts` — `GET /api/workflow-budget` reads the tracked roots or one root, and `POST /api/workflow-budget/clear` clears exactly one. The clear moves the windowed send ring and the child map and nothing else: `active` belongs to turns still in flight, the spend ledger is a token budget an operator did not ask to forgive, and the lifetime send total survives so a clear cannot launder the record. Both are `deferred-verb` in the route registry — they are owed CLI verbs, and because the ledger is process memory there is no local projection the CLI could read instead. See [`../devlog/_plan/260915_workflow_budget_window/030_wfc_diff_plan.md`](../devlog/_plan/260915_workflow_budget_window/030_wfc_diff_plan.md). | | Codex accounts | `src/codex/auth-api.ts` — `GET/POST/DELETE /api/codex-auth/accounts`, `PUT /api/codex-auth/accounts/alias`, `PUT /api/codex-auth/accounts/pause`, `PUT /api/codex-auth/accounts/pause-exhausted`, `POST /api/codex-auth/accounts/clear-cooldown`, `GET/PUT /api/codex-auth/active`, `PUT /api/codex-auth/auto-switch`, `PUT /api/codex-auth/pool-strategy`, `PUT /api/codex-auth/failover`, `GET /api/codex-auth/quota`, `GET /api/codex-auth/reset-credits` with `POST /api/codex-auth/reset-credits/consume`, and the login flow `POST /api/codex-auth/login`, `POST /api/codex-auth/login/code`, `POST /api/codex-auth/login/cancel`, `GET /api/codex-auth/login-status`. Per-account quota activation uses the existing `GET/PUT /api/settings` surface and `src/codex/quota-auto-refresh.ts`, keeping scheduled spending separate from credential/authentication mutation. Account ids are opaque handles and are serialized so the GUI can address an account; emails are masked and tokens are never serialized. New-account config commits add UI-managed selector bindings in the same config save; deletion deliberately retains existing bindings for fail-closed exact routing and re-add stability. Account mutations request catalog convergence only after config durability and expose only the boolean `catalogRefreshPending` completion projection. | | Sidebar | `src/server/management/sidebar-routes.ts` — `GET/POST /api/github/star` and `GET /api/update/badge`. Sidebar state is cosmetic; a failed fetch degrades silently. | | Logs | `src/server/management/logs-usage-routes.ts` — `GET /api/logs`, `GET /api/claude/inbound-debug`, and `GET /api/debug/injection-logs` join the debug streams described above. | 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..196b125d49 100644 --- a/tests/lib/workflow-budget.test.ts +++ b/tests/lib/workflow-budget.test.ts @@ -7,13 +7,27 @@ 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 { repoPath } from "../helpers/repo-root"; +import { + clearRequestLogsForTests, + getRequestLogEntries, + type RequestLogContext, +} from "../../src/server/request-log"; const memoryJournal = (): SpendJournal & { lines: string[] } => { const lines: string[] = []; @@ -372,3 +386,195 @@ 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"); + // 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.requestId === "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 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 }; + 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; + // 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()) { + 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); + expect(listed[0]?.rootId).toBe("root-x"); + expect(listed[1]?.rootId).toBe("root-w"); + }); +}); 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 )"); }); }); 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"); + }); +});