review: enforce the run's spend ceiling in code, in real dollars

.changeset/review-spend-ceiling.md

@@ -0,0 +1,61 @@
+---
+"review": minor
+---
+
+review: enforce the run's spend ceiling in code, in real dollars
+
+Dispatch now carries a budget ledger. Work may start while the run's spend is
+under a dispatch budget of `CEILING_USD - LANDING_RESERVE_USD`; crossing it
+refuses further dispatches, aborts the agents already in flight, and discloses
+every shed as a note line rather than as a silently missing dimension.
+
+Why in code rather than at the proxy. The only ceiling today is gh-aw's
+api-proxy credit cap (`maxAiCredits: 2500` in the compiled lock), which is
+denominated in list-price credits rather than the dollars Khan pays, cannot shed
+gracefully (it fails the run), and disappears with the proxy when the migration
+completes. This supersedes the intent of #314, which tried to fix the
+denomination at the proxy and was blocked on a gh-aw version; the proxy cap
+stays as a coarse list-price backstop while it exists.
+
+The numbers are derived, not chosen, and a test pins them so a change to live
+spend behaviour is a review conversation: **$12.50** ceiling, from the lock's
+2500-credit allowance minus the detection pass's measured ~$0.27 which this
+ledger does not govern, against a measured ~$0.50 for a full case's sub-agents;
+**$1.50** landing reserve, from the validator dispatch's measured ~$0.30 plus
+the reconciler. The reserve is the part that is easy to omit and expensive to
+miss: a ceiling with nothing held back turns an over-budget run into a wasted
+one, because the run can no longer validate or post what it already paid to
+find. Each consumer compiles its own lock, so a consumer with a different
+allowance wants a different ceiling.
+
+Enforcement is per turn where it can be. The runner asks the ledger at each turn
+boundary whether the agent may continue, so a single heavy reviewer cannot
+outspend a whole wave before anyone notices, and stopping happens at a turn
+boundary rather than by throwing: an agent stopped for budget has usually
+already found something worth keeping. The ledger has exactly one writer (the
+dispatcher, once per completed attempt, so retries and refusal fallbacks each
+pay), and the per-turn check is a read-only probe against it, which is what
+keeps the accounting from double counting. Concurrency is approximated the way
+the investigation cap approximates it: an in-flight agent cannot see its
+siblings' unsettled spend, so a wave can overshoot by at most the number of
+agents running at once, and `overshootUsd` measures that rather than assuming it
+away.
+
+`REVIEW_SPEND_ENFORCEMENT=proxy-only` is the rollback, and it measures without
+enforcing: the record still reports `crossed`, still lists the sheds, and the
+notice is loud, so a rolled-back run is not a blind run. The record names which
+enforcement was in force, per the standing rule that where enforcement changes
+over time the artifact says which one produced it.
+
+The record itself (`DispatchResult.spend`: ceiling, reserve, spend, overshoot,
+sheds, enforcement) is the run/cost/outcome telemetry shape, staged in both
+`dispatch-result.json` and the `out/` copy the artifact upload carries. It is
+deliberately a schema rather than an ad-hoc log line, because the same shape is
+what a second consumer would need.
+
+One property this does NOT restore, recorded in the module rather than implied:
+awf's cap is un-weakenable because it lives in a container image no Khan repo
+can edit, and this ceiling lives in a repo whose PRs this reviewer reviews. The
+provider-side workspace limit that would have replaced that property is not
+available, so CODEOWNERS on the ledger (naming two owners, since one stalls when
+away and none enforces nothing) plus required review is the whole mitigation.

CODEOWNERS

@@ -0,0 +1,17 @@
+# Ownership for the files where a quiet edit changes what this repo SPENDS or
+# what it is allowed to do. Everything else in this repo is reviewed by whoever
+# is around; these are the paths where "whoever is around" is not enough.
+#
+# The spend ceiling especially. It is the only limit on a review run's cost once
+# the migration removes gh-aw's api-proxy credit cap, and unlike that cap it
+# lives in a repo whose pull requests this reviewer reviews. A provider-side
+# workspace limit would have restored the property of living somewhere no PR can
+# reach; it is not available, so this file is the mitigation. Two owners rather
+# than one, because a single-owner entry stalls the moment that owner is away
+# and an ownerless entry enforces nothing at all.
+#
+# Pin bumps and policy edits under these paths are behaviour changes: reviewed
+# as such, never batch-merged.
+
+/workflows/review/lib/spend-ledger.ts   @jwbron @jeresig
+/CODEOWNERS                             @jwbron @jeresig

workflows/review/lib/dispatch-calls.ts

@@ -79,6 +79,14 @@ export type AgentResult = {
     refused?: boolean;
     /** The output came through the structured-final tool, pre-validated. */
     structured?: boolean;
+    /**
+     * The turn loop ended because the run's spend ceiling was reached, not
+     * because the agent finished. Load-bearing for disclosure: an agent stopped
+     * for budget has usually found SOMETHING, so its output is still used, and
+     * the run has to say that this dimension was cut short rather than
+     * completed. Silence here would be indistinguishable from "nothing found".
+     */
+    stoppedForBudget?: boolean;
 };
 
 /** The model seam; the Pi-backed production runner lives in the CLI entry. */
@@ -117,6 +125,12 @@ export type PerAgentReport = {
     fellBackTo?: string;
     /** The result arrived via the structured-final tool (pre-validated). */
     structuredFinal?: boolean;
+    /**
+     * This agent's turn loop ended on the run's spend ceiling. Kept per agent
+     * rather than only run-wide, because "which dimension got cut short" is the
+     * question a reader of the artifact actually has.
+     */
+    stoppedForBudget?: boolean;
     failed?: string;
 };
 
@@ -140,6 +154,27 @@ export type AgentDispatcherOptions = {
     validatorFor: (
         name: string,
     ) => (payload: Record<string, unknown>) => string | null;
+    /**
+     * Asked before each dispatch whether the run can still afford one. The
+     * spend ledger answers. Refusing here rather than aborting mid-flight is
+     * the cheaper of the two disclosures: nothing is spent, and the dimension
+     * is shed with a note instead of half-run.
+     */
+    mayDispatch?: (name: string) => boolean;
+    /**
+     * Called once per COMPLETED attempt with what it cost. The ledger's only
+     * writer; a retry and a refusal fallback are separate attempts and each
+     * one really was paid for, so each is recorded.
+     */
+    recordSpend?: (name: string, usd: number) => void;
+    /**
+     * Disclose an attempt the spend ceiling cut. `budget-stop` is a dispatch
+     * that completed early because the per-turn probe said stop (its partial
+     * findings are kept, and the cut is disclosed); `run-failed` is a thrown
+     * attempt, which the caller attributes to the budget only when the
+     * ledger's abort signal is what killed it.
+     */
+    recordAborted?: (name: string, why: "budget-stop" | "run-failed") => void;
 };
 
 export type AgentDispatcher = {
@@ -172,6 +207,9 @@ export const createAgentDispatcher = (
         maxTurns,
         timeoutMs,
         validatorFor,
+        mayDispatch,
+        recordSpend,
+        recordAborted,
     } = options;
 
     /**
@@ -192,6 +230,26 @@ export const createAgentDispatcher = (
         malformedNote?: string,
         modelOverride?: string,
     ): Promise<string | null> => {
+        if (mayDispatch !== undefined && !mayDispatch(name)) {
+            // Refused before spending. Staged as an out-file like every other
+            // outcome so the dispatch gate reads one shape, and reported as a
+            // failure cause so the run's own artifact names the reason.
+            writeOut(
+                name,
+                JSON.stringify({
+                    error: "spend ceiling reached before dispatch",
+                }),
+            );
+            report({
+                name,
+                model: agents.get(name)?.model ?? "",
+                usd: 0,
+                turns: 0,
+                wallMs: 0,
+                failed: "budget",
+            });
+            return null;
+        }
         const definition = agents.get(name);
         if (definition === undefined) {
             writeOut(name, JSON.stringify({error: "agent definition missing"}));
@@ -254,6 +312,7 @@ export const createAgentDispatcher = (
                         : {stopReason: result.stopReason}),
                     failed: "refused",
                 });
+                recordSpend?.(name, result.usd);
                 return dispatchAgent(name, malformedNote, fallback);
             }
             writeOut(name, result.output);
@@ -277,7 +336,14 @@ export const createAgentDispatcher = (
                 ...(malformedNote === undefined ? {} : {retried: true}),
                 ...(modelOverride === undefined ? {} : {fellBackTo: model}),
                 ...(result.structured === true ? {structuredFinal: true} : {}),
+                ...(result.stoppedForBudget === true
+                    ? {stoppedForBudget: true}
+                    : {}),
             });
+            recordSpend?.(name, result.usd);
+            if (result.stoppedForBudget === true) {
+                recordAborted?.(name, "budget-stop");
+            }
             return result.output;
         } catch (error) {
             writeOut(
@@ -298,6 +364,7 @@ export const createAgentDispatcher = (
                     : {fellBackTo: modelOverride}),
                 failed: "run-failed",
             });
+            recordAborted?.(name, "run-failed");
             return null;
         }
     };

workflows/review/lib/dispatch-runner-pi.test.ts

@@ -701,6 +701,78 @@ describe("createPiRunner", () => {
         expect(stopped).toBe(true);
     });
 
+    it("stops at the turn boundary when onTurnCost says abort, and flags it", async () => {
+        // The per-turn budget probe: cost lands per turn_end, the probe sees
+        // the agent's cumulative spend, and an abort answer ends the loop at
+        // the turn boundary (shouldStopAfterTurn), flagged stoppedForBudget.
+        const probes: number[] = [];
+        let stopped = false;
+        loop = ({config, emit}) => {
+            const shouldStop = config["shouldStopAfterTurn"] as () => boolean;
+            emit(turnEnd('{"findings": []}', 5));
+            stopped = shouldStop();
+            if (!stopped) {
+                emit(turnEnd("second turn", 5));
+            }
+            return Promise.resolve([]);
+        };
+        const runner = await createPiRunner({
+            onTurnCost: (spent) => {
+                probes.push(spent);
+                return spent >= 5 ? "abort" : "continue";
+            },
+        });
+        const result = await runner(request());
+        expect(stopped).toBe(true);
+        expect(probes).toEqual([5]);
+        expect(result.stoppedForBudget).toBe(true);
+        expect(result.turns).toBe(1);
+        expect(result.usd).toBe(5);
+    });
+
+    it("never flags stoppedForBudget while the probe says continue", async () => {
+        loop = async ({emit}) => {
+            emit(turnEnd('{"findings": []}', 0.1));
+            return [];
+        };
+        const runner = await createPiRunner({onTurnCost: () => "continue"});
+        const result = await runner(request());
+        expect(result.stoppedForBudget).toBeUndefined();
+    });
+
+    it("aborts an in-flight request when the run-wide signal fires", async () => {
+        // The case onTurnCost cannot cover: the ledger's signal stops the
+        // siblings running beside the agent that crossed.
+        const controller = new AbortController();
+        loop = ({signal}) =>
+            new Promise((_resolve, reject) => {
+                signal?.addEventListener("abort", () => reject(signal.reason), {
+                    once: true,
+                });
+            });
+        const runner = await createPiRunner({
+            abortSignal: controller.signal,
+        });
+        const pending = runner(request());
+        controller.abort(new Error("review spend ceiling reached"));
+        await expect(pending).rejects.toThrow(/spend ceiling/);
+    });
+
+    it("hands a pre-aborted run signal to the loop already aborted", async () => {
+        const controller = new AbortController();
+        controller.abort(new Error("review spend ceiling reached"));
+        let sawAborted: boolean | undefined;
+        loop = async ({signal}) => {
+            sawAborted = signal?.aborted;
+            return [];
+        };
+        const runner = await createPiRunner({
+            abortSignal: () => controller.signal,
+        });
+        await runner(request());
+        expect(sawAborted).toBe(true);
+    });
+
     it("reports the turn cap as max_turns, not as a clean finish", async () => {
         // Out of turns and finished-with-prose otherwise return the same
         // shape, and `dispatch.ts` would spend its one re-dispatch correcting

workflows/review/lib/dispatch-runner-pi.ts

@@ -670,6 +670,30 @@ export type PiRunnerOptions = {
      * time to tell which one sheds findings.
      */
     systemPrompt?: string;
+    /**
+     * Asked at each turn boundary whether to keep going, given what THIS agent
+     * has spent so far. The spend ledger answers (lib/spend-ledger.ts), as a
+     * read-only probe: the ledger's accounting is written once per completed
+     * dispatch, and this only decides whether to stop early.
+     *
+     * Per turn rather than per dispatch because that is the only granularity
+     * where the answer is still useful: cost is known when a turn ends, and one
+     * heavy reviewer can outspend a whole wave of light ones, so a check that
+     * runs between dispatches learns about the overshoot after paying it.
+     */
+    onTurnCost?: (agentSpentUsd: number) => "continue" | "abort";
+    /**
+     * Run-wide abort. The spend ledger fires this when a COMPLETED dispatch
+     * pushes the run past its budget, which is the case `onTurnCost` cannot
+     * cover: that probe only stops the agent doing the spending, while this
+     * stops the siblings already in flight beside it. Without it, "crossing
+     * aborts in-flight agents" would be true only of the agent that crossed.
+     *
+     * A thunk is accepted because the ledger's signal is phase-dependent (the
+     * landing phase swaps in a fresh signal so reserve-funded dispatches are
+     * not born aborted); it is resolved once per request.
+     */
+    abortSignal?: AbortSignal | (() => AbortSignal | undefined);
 };
 
 export const createPiRunner = async (
@@ -719,6 +743,21 @@ export const createPiRunner = async (
     return async (request: AgentRequest): Promise<AgentResult> => {
         const started = Date.now();
         const abort = new AbortController();
+        // Link the run-wide abort in both directions in time: already-aborted
+        // means never start, and aborting later means stop now.
+        const runAbort =
+            typeof options.abortSignal === "function"
+                ? options.abortSignal()
+                : options.abortSignal;
+        if (runAbort?.aborted === true) {
+            abort.abort(runAbort.reason);
+        } else {
+            runAbort?.addEventListener(
+                "abort",
+                () => abort.abort(runAbort.reason),
+                {once: true},
+            );
+        }
         let timedOut = false;
         const timer = setTimeout(() => {
             timedOut = true;
@@ -742,6 +781,13 @@ export const createPiRunner = async (
         let usd = 0;
         let turns = 0;
         let toolCalls = 0;
+        /**
+         * Set when the spend ledger says stop. Read by shouldStopAfterTurn, so
+         * the loop ends at a TURN boundary rather than by throwing: an agent
+         * stopped for budget has usually already found something, and killing
+         * it mid-turn would throw that away along with the turn's cost.
+         */
+        let budgetExhausted = false;
         let stopReason: string | undefined;
         let errorMessage: string | undefined;
         let rawStopReason: string | undefined;
@@ -780,7 +826,9 @@ export const createPiRunner = async (
                      * "end the turn now".
                      */
                     shouldStopAfterTurn: () =>
-                        captured !== undefined || turns >= request.maxTurns,
+                        captured !== undefined ||
+                        turns >= request.maxTurns ||
+                        budgetExhausted,
                 },
                 (event: Record<string, unknown>) => {
                     if (event["type"] === "tool_execution_end") {
@@ -835,6 +883,12 @@ export const createPiRunner = async (
                         | {cost?: {total?: number}}
                         | undefined;
                     usd += Number(usage?.cost?.total ?? 0);
+                    // Cumulative for this agent, because the probe is a
+                    // question about the run's total and the caller adds this
+                    // agent's in-flight spend to what it has already settled.
+                    if (options.onTurnCost?.(usd) === "abort") {
+                        budgetExhausted = true;
+                    }
                     const content = (message?.["content"] ?? []) as TextBlock[];
                     const text = content
                         .filter((block) => block.type === "text")
@@ -874,6 +928,7 @@ export const createPiRunner = async (
                         stopReason === "refusal" || rawStopReason === "refusal",
                     wallMs: Date.now() - started,
                     structured: true,
+                    ...(budgetExhausted ? {stoppedForBudget: true} : {}),
                 };
             }
             // Out of turns and finished-with-prose return the same shape: a
@@ -903,6 +958,7 @@ export const createPiRunner = async (
                 refused:
                     stopReason === "refusal" || rawStopReason === "refusal",
                 wallMs: Date.now() - started,
+                ...(budgetExhausted ? {stoppedForBudget: true} : {}),
             };
         } catch (error) {
             // A payload the tool already accepted is complete and validated:
@@ -926,6 +982,7 @@ export const createPiRunner = async (
                         stopReason === "refusal" || rawStopReason === "refusal",
                     wallMs: Date.now() - started,
                     structured: true,
+                    ...(budgetExhausted ? {stoppedForBudget: true} : {}),
                 };
             }
             if (timedOut) {