feat(standard): stop a run at a spending cap

AGENTS.md

@@ -61,7 +61,7 @@ and its resources. The source layout stays under `app/`.
   `feed`: the live feed any Harbor writes, from its own trial hooks),
   `importers/` (`atif` ATIF, `otel` OpenTelemetry, and the `cli`), `judge/` (`questions`, `state`, `client`, `store`, `batch`, `reconcile`, `lanes`),
   `standard/` (the Nullius standard, `docs/standard.md`: `ports`, `names` (capability vocabulary and legacy aliases), `definitions` (with an agent's or capability's versions), `match`,
-  `jsonl`, `acp`, `cancellation` (interruptible protocol waits), `relay`, `install`, `run`, `container` (the docker backend), `forward` (a licence server's one door from it, or a benchmark's MCP server's), `mcp_machine` and `mcp_bridge` (a benchmark's MCP server run on the machine, in its driver's environment, and the agent's bridge to it), `hostreq` (what a benchmark's options need of the machine its driver runs on: the check and the install) and `egress` (an own image's check's proxy to public addresses), `job`, `conform`, `fidelity` (ACP against the agent's own CLI), `prove` (a benchmark's controls and proofs), `shard` and `merge` (one run over many machines), `recipe` (the builder's standard recipe), `howto` (the How to add steps) and the `cli`; its
+  `jsonl`, `acp`, `cancellation` (interruptible protocol waits), `relay`, `ledger` (a run's spending cap: each call booked before it is sent), `install`, `run`, `container` (the docker backend), `forward` (a licence server's one door from it, or a benchmark's MCP server's), `mcp_machine` and `mcp_bridge` (a benchmark's MCP server run on the machine, in its driver's environment, and the agent's bridge to it), `hostreq` (what a benchmark's options need of the machine its driver runs on: the check and the install) and `egress` (an own image's check's proxy to public addresses), `job`, `conform`, `fidelity` (ACP against the agent's own CLI), `prove` (a benchmark's controls and proofs), `shard` and `merge` (one run over many machines), `recipe` (the builder's standard recipe), `howto` (the How to add steps) and the `cli`; its
   `payload` snapshots the running backend and effective catalogue for a remote host;
   conformance kits under `standard/kits/`, each a benchmark: a `driver` and, for the
   tools kit, an MCP server `mcp_kit`) - the definitions it reads are in the catalogue:

app/nullius/judge/reconcile.py

@@ -65,6 +65,9 @@
     # The judge's option names API errors alongside rate limits and timeouts.
     classify.API_ERROR: ("rate_limit_or_timeout",),
     classify.SETUP_ERROR: ("environment_error",),
+    # The relay's own refusal at the run's spending cap: recorded, never read
+    # from a transcript, so nothing agrees or disagrees with it and no cause suggests it.
+    classify.SPENDING_CAP: (),
     # No claim made: not comparable. The judge's pick becomes a suggestion.
     classify.UNKNOWN: (),
 }

app/nullius/runs/classify.py

@@ -24,11 +24,15 @@
 RATE_LIMITED = "rate_limited"
 API_ERROR = "api_error"
 SETUP_ERROR = "setup_error"
+# The run's own spending cap refused a model call the unit needed (the
+# standard's relay, ``standard.ledger``): the unit was cut off. Nothing
+# failed, and no attempt was had: the money ran out, by the user's own word.
+SPENDING_CAP = "spending_cap"
 AGENT_ERROR = "agent_error"
 UNKNOWN = "unknown"
 
 # Outcomes that reflect harness/infra problems, not the agent's ability.
-INFRA_OUTCOMES = {RATE_LIMITED, API_ERROR, SETUP_ERROR}
+INFRA_OUTCOMES = {RATE_LIMITED, API_ERROR, SETUP_ERROR, SPENDING_CAP}
 
 ORDER = [
     RESOLVED,
@@ -41,6 +45,7 @@
     RATE_LIMITED,
     API_ERROR,
     SETUP_ERROR,
+    SPENDING_CAP,
     UNKNOWN,
 ]
 
@@ -58,6 +63,9 @@
     "networkconnectionerror",     # DNS, refused connection, TLS — transport, not skill
     "unknownapierror",            # Harbor's own "some provider error"; an ApiError without the prefix
 }
+# What a unit the run's spending cap cut off carries (``standard.ledger.ERROR``): recorded by the relay that
+# refused the call, never inferred.
+_CAP_EXCEPTION = "spendingcaperror"
 
 # What Harbor's agents raise when the agent itself fails — its budget spent,
 # its own context filled, its own output truncated, a refusal, its process
@@ -92,6 +100,7 @@
     RATE_LIMITED: "Rate limited",
     API_ERROR: "API error",
     SETUP_ERROR: "Setup error",
+    SPENDING_CAP: "Spending cap",
     UNKNOWN: "Unknown",
 }
 
@@ -138,6 +147,11 @@
         "The environment or image failed to build or start, so the agent never ran. "
         "Infrastructure, not capability — excluded from the adjusted rate."
     ),
+    SPENDING_CAP: (
+        "The run reached its spending cap: its relay refused a model call this "
+        "trial needed, so the trial was cut off. Nothing failed and no attempt "
+        "was had, so it is excluded from the adjusted rate."
+    ),
     UNKNOWN: (
         "No verifier report and no exception, or an exception this reader does not "
         "recognise — a crash in the harness, the verifier or the reader itself. The "
@@ -162,6 +176,8 @@ def from_verdict(exception_type, verdict):
     """
     if exception_type:
         et = exception_type.lower().rpartition(".")[2]  # tolerate dotted paths
+        if et == _CAP_EXCEPTION:
+            return SPENDING_CAP
         # Rate limiting first: it is an Api* class too, but common enough to be
         # worth seeing on its own rather than folded into the generic bucket.
         if "ratelimit" in et or "rate_limit" in et or "429" in et:

app/nullius/runs/live.py

@@ -70,6 +70,8 @@
 #: mean over the trial's parts, how many (``parts``). The snapshot says so only
 #: where there is one: a ``scored`` count, and ``score`` / ``parts`` on that
 #: trial. A feed without one folds to a snapshot with none of those keys.
+#: ``budget`` is a run's spending cap: a unit its relay cut off there (the
+#: standard's ``--max-usd``), outside the rates as ``infra_error`` is.
 OUTCOMES = ("resolved", "unresolved", "scored", "agent_error", "infra_error",
             "cancelled", "timeout", "budget")
 ERROR_OUTCOMES = {"agent_error", "infra_error", "timeout", "budget"}

app/nullius/runs/parse.py

@@ -30,7 +30,7 @@
 # The shape this module produces. A workspace caches parsed Jobs keyed on it,
 # so bump it when a Trial or Job field is added, renamed or means something
 # new — otherwise a cache written by the old parse is read back as current.
-SCHEMA = "12"
+SCHEMA = "13"
 
 # What a rate-limit record in `agent/gateway.log` looks like. The agent's SDK
 # retries a 429 below the harness (an agent's `max_retries` reaching its

app/nullius/standard/cli.py

@@ -6,7 +6,7 @@
     PYTHONPATH=app python3 -m nullius.standard.cli new benchmark|agent|capability KEY [--family F] [--port P]
     PYTHONPATH=app python3 -m nullius.standard.cli new version KEY VERSION
     PYTHONPATH=app python3 -m nullius.standard.cli conform AGENT|CAPABILITY --model M [--provider P] [--port PORT ...] [--repeat K] [--option NAME=VALUE ...] [--unsandboxed | --backend docker]
-    PYTHONPATH=app python3 -m nullius.standard.cli run BENCHMARK[@VERSION] AGENT|CAPABILITY --model M [--provider P] [--limit N] [-j N] [--option NAME=VALUE ...] [--bench-option NAME=VALUE ...] [--price MODEL=INPUT,OUTPUT[,CACHED] ...] [--unsandboxed | --backend docker]
+    PYTHONPATH=app python3 -m nullius.standard.cli run BENCHMARK[@VERSION] AGENT|CAPABILITY --model M [--provider P] [--limit N] [-j N] [--option NAME=VALUE ...] [--bench-option NAME=VALUE ...] [--price MODEL=INPUT,OUTPUT[,CACHED] ...] [--max-usd USD] [--unsandboxed | --backend docker]
     PYTHONPATH=app python3 -m nullius.standard.cli run BENCHMARK[@VERSION] nop|oracle [--dataset D] [--limit N]
     PYTHONPATH=app python3 -m nullius.standard.cli prove BENCHMARK[@VERSION] [--dataset D] [--limit N]
     PYTHONPATH=app python3 -m nullius.standard.cli run ... --shard I/N [--launch ID]
@@ -469,7 +469,7 @@ def cmd_run(a):
                       concurrency=a.concurrency, jobs_root=a.jobs, name=a.name, unsandboxed=a.unsandboxed, judge_model=a.judge_model, backend=a.backend,
                       simulator_model=a.simulator_model, options=options(a.option), shard=_shard(a.shard), launch=a.launch,
                       licences=held, bench_options=options(a.bench_option), repeat=a.repeat, remove_images=a.remove_images,
-                      prices=prices(a.price), **({"env": run_env(a.provider)} if a.from_run else {}))
+                      prices=prices(a.price), max_usd=a.max_usd, **({"env": run_env(a.provider)} if a.from_run else {}))
     except run.Refused as e:
         print("refused:", e)
         return 2
@@ -624,6 +624,8 @@ def main(argv=None):
                            help="one of the benchmark's own options (repeatable: a search, a mode); the rest keep their defaults")
             p.add_argument("--price", action="append", metavar="MODEL=INPUT,OUTPUT[,CACHED]",
                            help="this run's price for a model in USD per million tokens, over your price list (repeatable; nullius prices)")
+            p.add_argument("--max-usd", type=float, metavar="USD",
+                           help="the run's spending cap: no call goes out once the spend plus the calls in flight would pass it")
     v = sub.add_parser("prove", help="prove a benchmark: its controls, its units, no leak; record what passed")
     v.add_argument("benchmark")
     v.add_argument("--dataset", help="one dataset (default: each)")

app/nullius/standard/container.py

@@ -471,17 +471,21 @@ def install_argv(image: str, where: str, cmd: List[str]) -> List[str]:
 
 def relay_argv(image: str, *, name: str, job: str, app: str, record: str, key_file: str, base: str,
                model: Optional[str], routes: Dict[str, str], prices: Optional[dict] = None, side_models=(),
-               key_header: str = "authorization", hosts: Optional[Dict[str, str]] = None) -> List[str]:
+               key_header: str = "authorization", hosts: Optional[Dict[str, str]] = None,
+               ledger: Optional[str] = None) -> List[str]:
     """``docker run`` for a unit's relay: on the default bridge (its way to the
     provider), its port published on 127.0.0.1 only. The key is never given:
     ``key_file``, its file, is mounted read-only at ``RELAY_KEY_IN`` with
     ``--mount``, which fails when the engine cannot see it (``-v`` would mount
     an empty folder). ``hosts``: names it resolves
-    (``--add-host``), the machine's own for an endpoint on its loopback."""
+    (``--add-host``), the machine's own for an endpoint on its loopback.
+    ``ledger``: the file of the run's spending cap, its folder mounted so
+    this relay books its calls where every other relay of the job does."""
     a = [DOCKER, "run", "-d", "--name", name, "--label", f"{LABEL}={job}", "--user", _user(),
          "--cap-drop", "ALL", "--security-opt", "no-new-privileges", "--init",
          *[w for n, ip in sorted((hosts or {}).items()) for w in ("--add-host", f"{n}:{ip}")],
          "-p", f"127.0.0.1::{RELAY_PORT}", "-v", f"{app}:{app}:ro", "-v", f"{record}:{record}:rw",
+         *(["-v", f"{os.path.dirname(ledger)}:{os.path.dirname(ledger)}:rw"] if ledger else []),
          "--mount", f"type=bind,source={key_file},target={RELAY_KEY_IN},readonly", "-w", app,
          "-e", "PYTHONPATH=" + app, "-e", "PYTHONDONTWRITEBYTECODE=1",
          image, "python3", "-m", "nullius.standard.relay", "serve", "--listen", f"0.0.0.0:{RELAY_PORT}",
@@ -492,6 +496,8 @@ def relay_argv(image: str, *, name: str, job: str, app: str, record: str, key_fi
         a += ["--route", f"{tag}={m}"]
     if prices:
         a += ["--prices", json.dumps(prices, separators=(",", ":"))]
+    if ledger:
+        a += ["--ledger", ledger]
     for m in side_models:
         a += ["--side-model", m]
     if key_header != "authorization":
@@ -653,8 +659,10 @@ class Unit:
     def __init__(self, job: str, tag: str, record_dir: str, *, image: str, app: str, base: str, key: str,
                  model: Optional[str] = None, routes: Optional[Dict[str, str]] = None, prices: Optional[dict] = None,
                  side_models=(), key_header: str = "authorization", servers: Iterable[str] = (),
-                 doors: Iterable[Tuple[str, int, str]] = (), hosts: Optional[Dict[str, str]] = None):
+                 doors: Iterable[Tuple[str, int, str]] = (), hosts: Optional[Dict[str, str]] = None,
+                 ledger: Optional[str] = None):
         self.job, self.tag, self.dir, self.image, self.app = job, tag, record_dir, image, app
+        self.ledger = ledger                     # the file of the run's spending cap, when it has one
         self.servers = list(servers)             # licence servers, host:port each: a forwarder apiece
         self.doors = list(doors)                 # servers on this machine: (alias, port, socket), a door apiece
         self.side_models, self.key_header = tuple(side_models), key_header
@@ -674,7 +682,7 @@ def start(self) -> "Unit":
             argv = relay_argv(self.image, name=self.names["relay"], job=self.job, app=self.app, record=self.dir,
                               key_file=self._key_file(), base=self.base, model=self.model, routes=self.routes,
                               prices=self.prices, side_models=self.side_models, key_header=self.key_header,
-                              hosts=self.hosts)
+                              hosts=self.hosts, ledger=self.ledger)
             self.started.append(self.names["relay"])
             _docker(argv[1:])
             _docker(["network", "connect", "--alias", RELAY_ALIAS, self.names["network"], self.names["relay"]])

app/nullius/standard/job.py

@@ -492,11 +492,12 @@ def write_config(job_dir: str, *, job: str, bench: dict, other: dict, model: str
 def feed_outcome(kind: str, score: Optional[dict], exception: Optional[tuple]) -> dict:
     """A unit's end in the live feed's words (``nullius.runs.live``): its
     outcome, with the reward (a pass benchmark) or the score, and the error's
-    kind as evidence."""
+    kind as evidence. A unit the run's spending cap cut off (``ledger.ERROR``)
+    is the feed's ``budget``."""
     score = score or {}
     if exception:
         k = exception[0]
-        return {"outcome": "cancelled" if k == "CancelledError" else "timeout" if "Timeout" in k else "infra_error" if any(w in k for w in _INFRA) else "agent_error",
+        return {"outcome": "cancelled" if k == "CancelledError" else "budget" if k == "SpendingCapError" else "timeout" if "Timeout" in k else "infra_error" if any(w in k for w in _INFRA) else "agent_error",
                 "exc": k, "evidence": (exception[1] or k)[:200]}
     if kind == "pass":
         reward = score.get("reward")
@@ -532,11 +533,14 @@ def pass_hat(results: List[dict]) -> Optional[dict]:
             "at": {str(k): round(sum(1 - comb(n - sum(v[:n]), k) / comb(n, k) for v in whole.values()) / len(whole), 4) for k in range(1, n + 1)}}
 
 
-def write_result(job_dir: str, job: str, t0: float, t1: float, results: List[dict], planned: int):
+def write_result(job_dir: str, job: str, t0: float, t1: float, results: List[dict], planned: int, budget: Optional[dict] = None):
+    """``budget``: a run with a spending cap (``run.Run.budget``): the cap,
+    what was spent, whether it stopped the run, the units it cut off and
+    those never begun."""
     errored = sum(1 for r in results if r.get("exception_info"))
     hat = pass_hat(results)
     with open(os.path.join(job_dir, "result.json"), "w") as f:
         json.dump({"id": job, "started_at": iso(t0), "finished_at": iso(t1), "n_total_trials": planned,
                    "stats": {"n_completed_trials": len(results) - errored, "n_errored_trials": errored,
                              "n_cancelled_trials": planned - len(results), "n_running_trials": 0, "n_pending_trials": 0},
-                   **({"pass_hat_k": hat} if hat else {})}, f, indent=1)
+                   **({"pass_hat_k": hat} if hat else {}), **({"budget": budget} if budget else {})}, f, indent=1)