diff --git a/AGENTS.md b/AGENTS.md index 3aa76e5..199ab26 100644 --- a/AGENTS.md +++ b/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: diff --git a/app/nullius/judge/reconcile.py b/app/nullius/judge/reconcile.py index f77d610..8e24d6a 100644 --- a/app/nullius/judge/reconcile.py +++ b/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: (), } diff --git a/app/nullius/runs/classify.py b/app/nullius/runs/classify.py index 2b3e817..d8816fa 100644 --- a/app/nullius/runs/classify.py +++ b/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: diff --git a/app/nullius/runs/live.py b/app/nullius/runs/live.py index 41e91ae..6f2affc 100644 --- a/app/nullius/runs/live.py +++ b/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"} diff --git a/app/nullius/runs/parse.py b/app/nullius/runs/parse.py index 5abc965..aaadc1c 100644 --- a/app/nullius/runs/parse.py +++ b/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 diff --git a/app/nullius/standard/cli.py b/app/nullius/standard/cli.py index efc0fb0..837044b 100644 --- a/app/nullius/standard/cli.py +++ b/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)") diff --git a/app/nullius/standard/container.py b/app/nullius/standard/container.py index 4210924..88b8e47 100644 --- a/app/nullius/standard/container.py +++ b/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"]]) diff --git a/app/nullius/standard/job.py b/app/nullius/standard/job.py index 95aaa0a..ca7e3ae 100644 --- a/app/nullius/standard/job.py +++ b/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) diff --git a/app/nullius/standard/ledger.py b/app/nullius/standard/ledger.py new file mode 100644 index 0000000..ec87b42 --- /dev/null +++ b/app/nullius/standard/ledger.py @@ -0,0 +1,202 @@ +"""A run's spending cap: one small file every relay of the job books its calls in. + +A capped run (``run --max-usd``) gives each of its relays - the units' own, +in this process or in their containers, and the benchmark's - the same +ledger. Before a call goes out, its relay books what the call is likely to +cost (:func:`bound`). The call is sent only while what was spent, what the +calls in flight have booked and this booking stay within the cap. The first +call that does not fit is refused and closes the ledger: every later call is +refused too, so a run stops at one point instead of letting its small calls +through. + +A reply turns the booking into what the call cost (``settle``). A call the +provider answered with an error gives its booking back (``release``). One +whose cost was never known - a reply with no token count, a reply cut off - +keeps its whole booking, as ``unsettled``: it may have been billed. + +What was spent is each reply's real cost. Only a call in flight is an +estimate, since its cost is known when its reply comes back: the prompt's +characters over three as input tokens, and as output the largest reply +the run has had from that model so far (the ledger keeps it, ``seen``) - +``OUTPUT`` before any - and never more than the limit the call names or +the model's own (its price row's ``max_output``). The model's limit alone +would book a short answer at a hundred times its cost, and a few calls +at a time would fill a cap nothing had been spent of. A call that costs +more than was booked for it takes the spend past the cap by the +difference, and the next call is refused. + +The file is rewritten whole under a lock held for one read and one write +(``flock`` on a file beside it, as an install is locked): a relay that dies +holding it lets it go. +""" +from __future__ import annotations + +import contextlib +import fcntl +import json +import math +import os +import uuid +from typing import Optional, Tuple + +OUTPUT = 4096 # output tokens booked for a call before the run has had a reply from its model +ERROR = "SpendingCapError" # what a unit the cap ended carries: its own outcome outside the rate, never a miss (runs.classify) +_LIMITS = ("max_completion_tokens", "max_output_tokens", "max_tokens") + + +class Unavailable(Exception): + """The ledger could not be read or written: a capped call is then not sent.""" + + +def bound(req: dict, raw: bytes, row: dict, path: str = "", seen: Optional[int] = None) -> float: + """What a call is likely to cost, in USD, at ``row`` (a price row, USD + per million tokens). Input: the request's characters over three - its + JSON whole, less what it carries as base64 - at the dearer of the input + and the cache-write price. Output: ``seen``, the largest reply the run + has had from this model, or ``OUTPUT`` before any; never more than the + limit the call names or the row's own. An embeddings call has none.""" + tokens = math.ceil(max(0, len(raw) - _blobs(req)) / 3) + rate = max(row["input"], row.get("cache_write") or 0) + out = 0 + if not path.split("?")[0].endswith("/embeddings"): + named = [req.get(k) for k in _LIMITS if isinstance(req.get(k), int) and not isinstance(req.get(k), bool) and req.get(k) > 0] + out = min(x for x in (named[0] if named else None, row.get("max_output"), seen or OUTPUT) if x) + return (tokens * rate + out * row["output"]) / 1e6 + + +def _blobs(v) -> int: + """Characters of a request that are an encoded file, not a prompt: a + ``data:`` address, a block's base64 ``data``.""" + if isinstance(v, str): + return len(v) if v.startswith("data:") else 0 + if isinstance(v, list): + return sum(_blobs(x) for x in v) + if isinstance(v, dict): + if v.get("type") == "base64" and isinstance(v.get("data"), str): + return len(v["data"]) + return sum(_blobs(x) for x in v.values()) + return 0 + + +class Ledger: + def __init__(self, path: str): + self.path = path + + @classmethod + def create(cls, path: str, max_usd: float) -> "Ledger": + os.makedirs(os.path.dirname(path), exist_ok=True) + led = cls(path) + led._write({"max_usd": float(max_usd), "spent_usd": 0.0, "unsettled_usd": 0.0, "held": {}, "closed": False, "seen": {}}) + return led + + @contextlib.contextmanager + def _locked(self): + try: + with open(self.path + ".lock", "a") as f: + fcntl.flock(f, fcntl.LOCK_EX) + try: + yield self._read() + finally: + fcntl.flock(f, fcntl.LOCK_UN) + except (OSError, ValueError, KeyError, TypeError) as e: + raise Unavailable(f"the spending cap's ledger could not be used: {type(e).__name__}: {e}") + + def _read(self) -> dict: + with open(self.path, encoding="utf-8") as f: + doc = json.load(f) + float(doc["max_usd"]), float(doc["spent_usd"]), float(doc["unsettled_usd"]), sum(doc["held"].values()) + if not isinstance(doc.get("seen"), dict): + doc["seen"] = {} + return doc + + def _write(self, doc: dict) -> None: + """Written whole beside the file, then renamed: a reader that takes no lock never finds half of it.""" + part = f"{self.path}.{os.getpid()}.{uuid.uuid4().hex[:8]}.part" + with open(part, "w", encoding="utf-8") as f: + json.dump(doc, f) + os.replace(part, self.path) + + def reserve(self, amount: float) -> Tuple[Optional[str], dict]: + """(the booking's id, the state before it) when ``amount`` fits; + (None, the state) when it does not, or the ledger is closed already - + and it is closed from then on.""" + with self._locked() as doc: + seen = _state(doc) + if doc["closed"] or amount > seen["left_usd"]: + if not doc["closed"]: + doc["closed"] = True + self._write(doc) + return None, seen + booking = uuid.uuid4().hex[:12] + doc["held"][booking] = amount + self._write(doc) + return booking, seen + + def settle(self, booking: str, cost: Optional[float], model: Optional[str] = None, output: Optional[int] = None) -> None: + """The call's cost in place of its booking; None, its cost was never + known: the whole booking is kept, as unsettled. ``output``: the + reply's output tokens, kept when they are the most ``model`` has + answered in this run - what its next calls are booked at.""" + with self._locked() as doc: + amount = doc["held"].pop(booking, None) + if amount is None: + return + if model and isinstance(output, int) and not isinstance(output, bool) and output > doc["seen"].get(model, 0): + doc["seen"][model] = output + if cost is None: + doc["unsettled_usd"] += amount + else: + doc["spent_usd"] += max(0.0, float(cost)) + self._write(doc) + + def release(self, booking: str) -> None: + """The booking given back: the provider answered with an error, or was never reached.""" + with self._locked() as doc: + if doc["held"].pop(booking, None) is not None: + self._write(doc) + + def largest(self, model) -> Optional[int]: + """The most output tokens ``model`` has answered a call of this run with; None before any. Read without the lock.""" + try: + return self._read()["seen"].get(model) or None + except (OSError, ValueError, KeyError, TypeError) as e: + raise Unavailable(f"the spending cap's ledger could not be read: {type(e).__name__}: {e}") + + def state(self) -> dict: + """``max_usd``, ``spent_usd``, ``unsettled_usd``, ``held_usd`` (the + calls in flight), ``left_usd`` and ``closed``, read without the lock.""" + try: + return _state(self._read()) + except (OSError, ValueError, KeyError, TypeError) as e: + raise Unavailable(f"the spending cap's ledger could not be read: {type(e).__name__}: {e}") + + def closed(self) -> bool: + """Whether a call was refused: nothing more of the run is begun. A + ledger that cannot be read is closed - no call it guards is sent.""" + try: + return bool(self.state()["closed"]) + except Unavailable: + return True + + +def _state(doc: dict) -> dict: + held = sum(doc["held"].values()) + return {"max_usd": doc["max_usd"], "spent_usd": doc["spent_usd"], "unsettled_usd": doc["unsettled_usd"], "held_usd": held, + "left_usd": doc["max_usd"] - doc["spent_usd"] - doc["unsettled_usd"] - held, "closed": bool(doc["closed"])} + + +def usd(v: float) -> str: + """An amount as a person reads it: cents, and up to two digits more when it has them.""" + whole, _, cents = f"{v:.4f}".rstrip("0").partition(".") + return f"{whole}.{cents.ljust(2, '0')}" + + +def refusal(state: dict, amount: float) -> str: + """Why a call was not sent, as its caller and the record read it.""" + if state["closed"]: + return (f"an earlier call did not fit the run's spending cap of {usd(state['max_usd'])} USD " + f"({usd(state['spent_usd'])} spent): no further call of this run is sent") + flight = state["held_usd"] + state["unsettled_usd"] + return (f"the run's spending cap of {usd(state['max_usd'])} USD: {usd(state['spent_usd'])} spent" + + (f", {usd(flight)} booked by calls in flight or never settled" if flight else "") + + f", and this call may cost up to {usd(amount)}; it was not sent, and no further call of this run is") diff --git a/app/nullius/standard/relay.py b/app/nullius/standard/relay.py index 169760a..9804324 100644 --- a/app/nullius/standard/relay.py +++ b/app/nullius/standard/relay.py @@ -23,6 +23,11 @@ the provider's headers come back the same way (its request id, its rate limits). It never answers for the provider: a 429 reaches the caller and the record as it was sent, which is what makes rate limits measurable. + +One answer is the relay's own: a run with a spending cap (``ledger``) books +each call before it goes out, and a call that does not fit is not sent. Its +caller gets a 402 that names the relay and the cap, and the record says +``refused``, so it never reads as the provider's error. """ from __future__ import annotations @@ -38,6 +43,7 @@ from typing import Dict, List, Optional from nullius.redact import redact +from nullius.standard import ledger as ledgers PLACEHOLDER_KEY = "nullius-relay" _TAG = re.compile(r"^[A-Za-z0-9_-]{1,64}$") @@ -330,7 +336,8 @@ def close(self): class Relay: def __init__(self, base: str, key: str, record_dir: str, prices: Optional[Dict[str, dict]] = None, - model: Optional[str] = None, side_models=(), key_header: str = "authorization", fields: bool = False): + model: Optional[str] = None, side_models=(), key_header: str = "authorization", fields: bool = False, + ledger: Optional[str] = None): """``base``: the provider's endpoint (``https://api.openai.com/v1``). ``model``: when given, every call is sent as this model, whatever the caller named - the run's model is the recipe's, not the agent's. A @@ -345,8 +352,12 @@ def __init__(self, base: str, key: str, record_dir: str, prices: Optional[Dict[s run's, recorded with ``model_asked``. ``key_header``: how the provider takes its key - a bearer ``authorization``, or Anthropic's ``x-api-key``. ``fields``: each record also holds the request's fields as sent - (:func:`request_fields`) - ``fidelity``'s, never a normal run's.""" + (:func:`request_fields`) - ``fidelity``'s, never a normal run's. + ``ledger``: the file of the run's spending cap (``ledger``), shared + by every relay of the job: a call is booked there before it is sent, + and not sent when it does not fit.""" self.fields = fields + self.ledger = ledgers.Ledger(ledger) if ledger else None self.base, self.key, self.dir = base.rstrip("/"), key, record_dir self.prices, self.model = prices or {}, model self.side_models = {a: (b or a) for a, _, b in (str(m).partition("=") for m in side_models or ())} @@ -434,8 +445,15 @@ def _forward(self, h: BaseHTTPRequestHandler): if req.get("stream") and rest.split("?")[0].endswith("/chat/completions"): req.setdefault("stream_options", {})["include_usage"] = True raw = json.dumps(req).encode() + booking = booked = None + if self.ledger is not None and h.command == "POST": + booking, booked, why = self._book(req if isinstance(req, dict) else {}, raw, rest) + if why: + self._refuse(h, tag, rest, req, asked, why) + return t0 = time.time() conn, r, back, body = None, None, [], b"" + cut = False # the reply was cut off: the call may have been billed try: conn, r = self.pool.send(h.command, rest, raw if h.command == "POST" else None, self._headers(h.headers)) status, kind, back = r.status, r.getheader("Content-Type") or "application/json", returned(r.headers) @@ -448,7 +466,7 @@ def _forward(self, h: BaseHTTPRequestHandler): try: body, whole = r.read(), True except Exception as e: # cut off mid-reply - status, kind, back = 502, "application/json", [] + status, kind, back, cut = 502, "application/json", [], True body = json.dumps({"error": {"message": "the endpoint's reply was cut off: " + self._said(e)}}).encode() def record(body): @@ -459,6 +477,7 @@ def record(body): rec["side"] = True if self.fields and isinstance(req, dict) and req: rec["fields"] = request_fields(req) + cost = None if status == 200: rec["reply"] = read_reply(rest, body, streamed) cost, how = self._cost(rec["model"], rec["reply"].get("usage")) @@ -466,6 +485,15 @@ def record(body): rec["cost_usd"], rec["cost_from"] = cost, how else: rec["error"] = _said(status, body) + if booking is not None: + rec["booked_usd"] = booked + try: + if status == 200 or cut: # what it cost; unknown (no token count, cut off), the booking is kept whole + self.ledger.settle(booking, cost, rec["model"], ((rec.get("reply") or {}).get("usage") or {}).get("output")) + else: # an error of the provider's, or never reached: not billed + self.ledger.release(booking) + except ledgers.Unavailable: + pass # the booking stays held, against the cap with self.lock: with open(os.path.join(self.dir, tag + ".jsonl"), "a") as f: f.write(json.dumps(rec) + "\n") @@ -515,6 +543,41 @@ def record(body): else: conn.close() + def _book(self, req: dict, raw: bytes, rest: str) -> tuple: + """(booking, its amount, None) for a call the cap lets out; (None, + its amount, why) for one it does not. A model with no price is not + sent either: a capped run sends no call it cannot price.""" + row = self.prices.get(req.get("model")) + if not row: + return None, None, f"no price for the model {req.get('model')!r}: a run with a spending cap sends no call it cannot price" + try: + amount = ledgers.bound(req, raw, row, rest, self.ledger.largest(req.get("model"))) + booking, seen = self.ledger.reserve(amount) + except ledgers.Unavailable as e: + return None, None, str(e) + return booking, amount, None if booking else ledgers.refusal(seen, amount) + + def _refuse(self, h: BaseHTTPRequestHandler, tag: str, rest: str, req, asked, why: str) -> None: + """A call the run's spending cap did not let out: answered here and + recorded as the relay's own (``refused``), nothing sent.""" + t = time.time() + rec = {"t0": t, "t1": t, "path": rest, "model": req.get("model") if isinstance(req, dict) else None, "model_asked": asked, + "status": 402, "refused": "cap", "error": why, "request": _request_tail(req) if isinstance(req, dict) else {}, + "headers": sorted(k.lower() for k in forwarded(h.headers))} + with self.lock: + with open(os.path.join(self.dir, tag + ".jsonl"), "a") as f: + f.write(json.dumps(rec) + "\n") + body = json.dumps({"type": "error", "error": {"type": "spending_cap", "code": "spending_cap", + "message": "nullius relay: " + why}}).encode() + h.send_response(402) + for n, v in (("Content-Type", "application/json"), ("X-Nullius-Refused", "cap"), ("Content-Length", str(len(body)))): + h.send_header(n, v) + h.end_headers() + try: + h.wfile.write(body) + except OSError: + pass + # The address ``url`` hands out: the listening one unless the caller sets # another - a sidecar listens on 0.0.0.0 and is reached as ``relay``. public: Optional[str] = None @@ -553,7 +616,12 @@ def failure(calls: List[dict]) -> Optional[tuple]: """When a unit's model calls ended failing, the exception a trial carries for it - named so ``nullius.runs.classify`` sorts it as the infrastructure it was - and the endpoint's words. None when the last call - succeeded (a failure the agent recovered from is not the unit's).""" + succeeded (a failure the agent recovered from is not the unit's). A call + the run's spending cap refused is the unit's wherever it falls: once one + is refused, every later one is.""" + cut = next((c for c in calls if c.get("refused") == "cap"), None) + if cut: + return ledgers.ERROR, cut.get("error") or "" if not calls or calls[-1].get("status") == 200: return None last = calls[-1] @@ -573,7 +641,7 @@ def failure(calls: List[dict]) -> Optional[tuple]: def serve(argv=None) -> int: """``python3 -m nullius.standard.relay serve --listen HOST:PORT --base URL --record DIR --key-file PATH [--model M] [--route TAG=MODEL ...] [--side-model M ...] [--key-header NAME] - [--public URL] [--prices JSON]``. + [--public URL] [--prices JSON] [--ledger FILE]``. ``--base`` is the provider's endpoint, ``--route`` a route sent as another model (a benchmark's judge), ``--public`` the address ``url`` hands out. @@ -594,6 +662,7 @@ def serve(argv=None) -> int: ap.add_argument("--key-header", default="authorization", choices=("authorization", "x-api-key")) ap.add_argument("--public", help="the address url() hands out, when not the listening one") ap.add_argument("--prices", help="a JSON object: model -> USD per million tokens, for a cost estimate") + ap.add_argument("--ledger", help="the file of the run's spending cap: a call is booked there before it is sent") a = ap.parse_args(argv) try: with open(a.key_file) as f: @@ -606,7 +675,7 @@ def serve(argv=None) -> int: return 2 host, _, port = a.listen.rpartition(":") r = Relay(a.base, key, a.record, prices=json.loads(a.prices) if a.prices else None, model=a.model, - side_models=a.side_model, key_header=a.key_header) + side_models=a.side_model, key_header=a.key_header, ledger=a.ledger) for route in a.route: tag, _, model = route.partition("=") if not _TAG.match(tag) or not model: diff --git a/app/nullius/standard/run.py b/app/nullius/standard/run.py index b84d328..e99e51b 100644 --- a/app/nullius/standard/run.py +++ b/app/nullius/standard/run.py @@ -65,7 +65,7 @@ from nullius import paths, prices as price_list, progress, providers from nullius.redact import redact from nullius.harbor import feed as live_feed -from nullius.standard import acp, container, definitions, hostreq, install, job, jsonl, match, mcp_machine, relay, shard as shards +from nullius.standard import acp, container, definitions, hostreq, install, job, jsonl, ledger as ledgers, match, mcp_machine, relay, shard as shards # the totals a unit's end carries onto the live board, from its trial's agent_result FEED_TOTALS = ("n_input_tokens", "n_cache_tokens", "n_output_tokens", "cost_usd") @@ -308,7 +308,7 @@ def __init__(self, bench: dict, other: dict, *, model: str, provider: str, datas backend: str = "local", options: Optional[dict] = None, shard: Optional[tuple] = None, launch: Optional[str] = None, repeat: int = 1, simulator_model: Optional[str] = None, licences: Optional[Licences] = None, bench_options: Optional[dict] = None, which=shutil.which, - remove_images: bool = False, prices: Optional[dict] = None): + remove_images: bool = False, prices: Optional[dict] = None, max_usd: Optional[float] = None): """``proving``: the run is a conformance kit's, the one run that may start before the port it checks is proven - the family must still match. ``unsandboxed``: the caller's explicit yes to running an agent on the @@ -328,6 +328,11 @@ def __init__(self, bench: dict, other: dict, *, model: str, provider: str, datas (``simulator``): this model instead of its official one. ``prices``: this run's own prices, by model, over the user's list (``nullius.prices``): what a call's cost is estimated from. + ``max_usd``: the run's spending cap (``ledger``): every relay of the + job books a call before sending it, and once one does not fit no + further call goes out and nothing more is begun. Refused when a model + the run calls has no price: a call that cannot be priced cannot be + booked. ``licences``: what the run holds of the benchmark's licences (:class:`Licences`); by default its store on this machine. ``bench_options``: the benchmark's own options for this run @@ -402,6 +407,20 @@ def __init__(self, bench: dict, other: dict, *, model: str, provider: str, datas # (nullius.prices), this run's own prices over it. Read once, here: a run never fetches a price. self.run_prices = dict(prices or {}) self.prices = {**price_list.table(), **self.run_prices} + self.max_usd, self.ledger = None, None + if max_usd is not None: + try: + cap = float(max_usd) + except (TypeError, ValueError): + cap = 0.0 + if not 0 < cap < float("inf"): + raise Refused(f"--max-usd {max_usd!r}: the run's spending cap in USD, above zero") + unpriced = [m for m in self.models_called() if m not in self.prices] + if unpriced: + raise Refused("a spending cap needs a price for every model the run calls, and there is none for " + + ", ".join(unpriced) + ": nullius prices update, nullius prices set MODEL INPUT OUTPUT, " + "or --price MODEL=INPUT,OUTPUT") + self.max_usd = cap lacking = hostreq.missing(bench, needs["host"], which) if lacking: raise Refused(f"{', '.join(f'{k}={v}' for k, v in self.bench_options.items())} needs on this machine " @@ -479,6 +498,9 @@ def start(self) -> str: + (["dataset"] if self.ds.get("url") else []) + (["driver"] if install.driver_dir(self.bench) else []) + ([] if control or self.upstream else ["install"]) + ["units"])) + if self.max_usd is not None and self.base: + # one ledger for the job, in a folder of its own: all a unit's relay container is given of the job besides its record + self.ledger = ledgers.Ledger.create(os.path.join(self.job_dir, ".work", "ledger", "ledger.json"), self.max_usd) try: self.ready() except Refused as e: @@ -511,6 +533,7 @@ def start(self) -> str: results = [] u0 = time.time() self._halt, self._refused, self.halted = threading.Event(), [], None + self._cut: List[str] = [] # units the spending cap cut off # a model server on this machine's loopback, which the relays' containers cannot reach as written self.loopback = (container.Loopback(self.base).start() if self.backend == "docker" and self.base and container.on_loopback(self.base) else None) @@ -542,7 +565,8 @@ def work_unit(iur): finally: # a Stop or Ctrl-C: the units not begun are not begun, those running end with their agents pool.shutdown(wait=True, cancel_futures=True) - results = [f.result() for f in futures if not f.cancelled() and f.exception() is None and f.result() is not None] + begun = [not f.cancelled() and f.exception() is None and f.result() is not None for f in futures] + results = [f.result() for f, ran in zip(futures, begun) if ran] for d in self.drivers: d.close() if self.bench_relay is not None: @@ -551,11 +575,20 @@ def work_unit(iur): container.sweep(self.name) # whatever an interrupted unit left if self.loopback is not None: self.loopback.stop() - job.write_result(self.job_dir, self.name, t0, time.time(), results, len(work)) + budget = self.budget([u for n, (_, u, _) in enumerate(work) if n >= len(begun) or not begun[n]]) + job.write_result(self.job_dir, self.name, t0, time.time(), results, len(work), budget=budget) errored = sum(1 for x in results if x.get("exception_info")) self.feed.emit("job.end", n_completed=len(results) - errored, n_errored=errored, n_cancelled=len(work) - len(results)) p.emit("units", "done", seconds=round(time.time() - u0, 1)) self.n_verdicts, self.n_ended = sum(1 for x in results if x.get("verifier_result")), len(results) + if budget and budget["stopped"]: + spent = "unknown" if budget["spent_usd"] is None else ledgers.usd(budget["spent_usd"]) + p.emit("cap", detail=f"a call did not fit the spending cap of {ledgers.usd(budget['max_usd'])} USD", + spent_usd=spent, cut_off=len(budget["cut_off"]), not_run=len(budget["not_run"])) + self.say(f"stopped: a call did not fit the spending cap of {ledgers.usd(budget['max_usd'])} USD " + f"({spent} spent): {len(budget['cut_off'])} cut off, {len(budget['not_run'])} not run") + elif budget: + self.say(f"spent {ledgers.usd(budget['spent_usd'] or 0)} of the {ledgers.usd(budget['max_usd'])} USD cap") if self.halted: self.say(f"stopped: the endpoint refused {REFUSED_IN_A_ROW} units in a row ({self.halted[0]}: " f"{redact(str(self.halted[1]))[:200]}); the {len(work) - len(results)} units not begun were not run") @@ -626,7 +659,7 @@ def ready(self) -> None: # own, recorded in the job's .work/relay-benchmark beside the units', never counted as the agent's. if self.embed_model and self.bench_relay is None: self.bench_relay = relay.Relay(self.base, self.key, os.path.join(self.job_dir, ".work", "relay-benchmark"), - prices=self.prices, model=self.embed_model, + prices=self.prices, model=self.embed_model, ledger=self.ledger_file(), key_header=PROVIDERS[self.provider].get("key_header", "authorization")).start() def runtimes_needed(self) -> List[str]: @@ -657,6 +690,9 @@ def phase(self, name: str, unit: str, phase: str, timeout: Optional[float] = Non def unit(self, i: int, unit: str, r: int = 1) -> Optional[dict]: if self._stopped.is_set() or (getattr(self, "_halt", None) is not None and self._halt.is_set()): return None # the endpoint refused the run: nothing more is begun + if getattr(self, "ledger", None) is not None and self.ledger.closed(): + self._halt.set() # the spending cap refused a call: every later one is refused too + return None tag = f"u{i:04d}" + (f"r{r}" if self.repeat > 1 else "") self.local.attempt = r trial = os.path.join(self.job_dir, self.trial_name(i, unit, r)) @@ -759,6 +795,13 @@ def unit(self, i: int, unit: str, r: int = 1) -> Optional[dict]: if failed and (exc is None or exc[0] not in ("EnvironmentSetupError",)): exc = failed # what the endpoint said outranks how the agent took it score = None + if rl is not None and getattr(self, "ledger", None) is not None: + # a call the spending cap refused, the agent's, the judge's or the simulated user's: the unit was + # cut off by the run's own limit, whatever its agent made of the refusal - never a miss + cut = next((f for f in (relay.failure(rl.calls(t)) for t in (tag, f"{tag}-judge", f"{tag}-simulator")) + if f and f[0] == ledgers.ERROR), None) + if cut: + exc, score = cut, None if self._stopped.is_set() and score is None: # the Stop ended its agent, its relay or its containers under it: what it then failed with is # the Stop's doing, not the agent's nor the environment's (Harbor's name for a stopped trial) @@ -783,8 +826,30 @@ def unit(self, i: int, unit: str, r: int = 1) -> Optional[dict]: **{k: v for k, v in (res.get("agent_result") or {}).items() if k in FEED_TOTALS}, durations={"agent_run": round(produce[1] - produce[0], 3)} if produce[0] is not None else None) self.note_refusal(exc) + if exc and exc[0] == ledgers.ERROR and getattr(self, "_halt", None) is not None: + with self._said: + self._cut.append(unit) + self._halt.set() # the cap is reached: nothing more is begun return res + def budget(self, not_run: List[str]) -> Optional[dict]: + """What ``result.json`` says of the run's spending cap: the cap, + what was spent, what was booked for calls whose cost was never known + (``unsettled_usd``, said only when there is some), whether the cap + stopped the run, the units it cut off and those never begun. None + for a run without a cap.""" + if getattr(self, "ledger", None) is None: + return None + try: + got = self.ledger.state() + except ledgers.Unavailable: + got = {"spent_usd": None, "unsettled_usd": 0.0, "held_usd": 0.0, "closed": True} + unsettled = round(got["unsettled_usd"] + got["held_usd"], 6) + cut = getattr(self, "_cut", []) + return {"max_usd": self.max_usd, "spent_usd": None if got["spent_usd"] is None else round(got["spent_usd"], 6), + **({"unsettled_usd": unsettled} if unsettled else {}), "stopped": bool(got["closed"]), + "cut_off": list(dict.fromkeys(cut)), "not_run": list(dict.fromkeys(not_run)) if got["closed"] else []} + def note_refusal(self, exc: Optional[tuple]) -> None: """A unit's end, counted toward the endpoint refusing the run: a wrong key or an unknown model fails every unit the same way, and running @@ -897,6 +962,11 @@ def prices_record(self) -> Optional[dict]: listed = doc["from"] and any(r["source"] == "list" for r in rows.values()) return {**({"from": doc["from"], "on": doc["on"]} if listed else {}), "models": rows} + def ledger_file(self) -> Optional[str]: + """The file of the run's spending cap, for a relay; None without a cap.""" + led = getattr(self, "ledger", None) + return led.path if led is not None else None + def models_called(self) -> List[str]: """Every model this run's relays may send a call to, as sent: the run's, the judge's, the simulator's, an option's embeddings and the @@ -920,7 +990,7 @@ def unit_relay(self, tag: str, trial: str, setup: dict, env_def: Optional[dict] key_header = PROVIDERS[self.provider].get("key_header", "authorization") if self.backend == "local": rl = relay.Relay(self.base, self.key, record, prices=getattr(self, "prices", None), model=self.model, side_models=side, - key_header=key_header).start() + key_header=key_header, ledger=self.ledger_file()).start() rl.unit_url = rl.url(tag) return rl routes = {tag + "-judge": self.judge_model} if self.judge_model else {} @@ -940,7 +1010,8 @@ def unit_relay(self, tag: str, trial: str, setup: dict, env_def: Optional[dict] rl = container.Unit(self.name, tag, record, image=self.image, app=APP, base=door.base if door else self.base, key=self.key, model=self.model, routes=routes, side_models=side, key_header=key_header, prices=price_list.given(self.models_called(), getattr(self, "prices", None)), # a sidecar's argv: its models' rows - servers=self.agent_servers(env_def), doors=doors, hosts=door.hosts if door else None) + servers=self.agent_servers(env_def), doors=doors, hosts=door.hosts if door else None, + ledger=self.ledger_file()) rl.mcp_listeners = listeners try: try: diff --git a/app/static/runs/board.css b/app/static/runs/board.css index 85fd86e..69a70ce 100644 --- a/app/static/runs/board.css +++ b/app/static/runs/board.css @@ -40,6 +40,7 @@ .rp .oc-no_patch { --oc: var(--no_patch); } .rp .oc-agent_error { --oc: var(--agent_error); } .rp .oc-rate_limited { --oc: var(--rate_limited); } .rp .oc-api_error { --oc: var(--api_error); } .rp .oc-setup_error { --oc: var(--setup_error); } .rp .oc-unknown { --oc: var(--unknown-mark); } +.rp .oc-spending_cap { --oc: var(--spending_cap); } .rp .sw { display: inline-block; width: 10px; height: 10px; border-radius: 2px; background: var(--oc); flex: none; } .rp .odot { display: inline-block; width: 8px; height: 8px; border-radius: 50%; background: var(--oc); flex: none; } .rp .adv { display: inline-block; font: 500 10.5px/16px var(--sans); letter-spacing: .06em; text-transform: uppercase; color: var(--text-faint); diff --git a/app/static/runs/launches.css b/app/static/runs/launches.css index ee5f7c2..1d02ee6 100644 --- a/app/static/runs/launches.css +++ b/app/static/runs/launches.css @@ -64,6 +64,7 @@ .la-mix .oc-no_patch { background: var(--no_patch); } .la-mix .oc-agent_error { background: var(--agent_error); } .la-mix .oc-rate_limited { background: var(--rate_limited); } .la-mix .oc-api_error { background: var(--api_error); } .la-mix .oc-setup_error { background: var(--setup_error); } .la-mix .oc-unknown { background: var(--unknown); } +.la-mix .oc-spending_cap { background: var(--spending_cap); } .la-mix span + span { box-shadow: inset 1.5px 0 0 var(--surface); } .la-go { text-align: right; white-space: nowrap; font-size: 12.5px; } .launch-foot { display: flex; flex-wrap: wrap; align-items: center; gap: 8px 16px; padding: 10px 18px; border-top: 1px solid var(--border-soft); font-size: 12.5px; color: var(--text-dim); } diff --git a/app/static/runs/live-board.css b/app/static/runs/live-board.css index db01ab6..4ef11b1 100644 --- a/app/static/runs/live-board.css +++ b/app/static/runs/live-board.css @@ -18,7 +18,7 @@ .lb { --lb-track: color-mix(in srgb, var(--text-faint) 16%, var(--surface)); --lb-hair: var(--border-soft); } .lb .lbo-resolved { --c: var(--resolved); } .lb .lbo-unresolved { --c: var(--failed_verification); } .lb .lbo-scored { --c: var(--scored); } .lb .lbo-agent_error { --c: var(--agent_error); } .lb .lbo-infra_error { --c: var(--rate_limited); } -.lb .lbo-timeout { --c: var(--patch_failed_tests); } .lb .lbo-budget { --c: var(--api_error); } .lb .lbo-cancelled { --c: var(--unknown); } +.lb .lbo-timeout { --c: var(--patch_failed_tests); } .lb .lbo-budget { --c: var(--spending_cap); } .lb .lbo-cancelled { --c: var(--unknown); } .lb .lbp-env_setup { --c: var(--rate_limited); } .lb .lbp-agent_setup { --c: var(--patch_failed_tests); } .lb .lbp-agent_run { --c: var(--accent); } .lb .lbp-verify { --c: var(--ph-verify); } diff --git a/app/static/runs/live-board.js b/app/static/runs/live-board.js index e19c404..b969a07 100644 --- a/app/static/runs/live-board.js +++ b/app/static/runs/live-board.js @@ -28,7 +28,7 @@ const PHASES = ['env_setup', 'agent_setup', 'agent_run', 'verify']; const PHASE_LABEL = {env_setup: 'env setup', agent_setup: 'agent setup', agent_run: 'agent run', verify: 'verify'}; const OUTCOMES = ['resolved', 'unresolved', 'scored', 'agent_error', 'infra_error', 'timeout', 'budget', 'cancelled']; - const OUTCOME_LABEL = {resolved: 'Resolved', unresolved: 'Unresolved', scored: 'Scored', agent_error: 'Agent error', infra_error: 'Infra error', cancelled: 'Cancelled', timeout: 'Timeout', budget: 'Budget'}; + const OUTCOME_LABEL = {resolved: 'Resolved', unresolved: 'Unresolved', scored: 'Scored', agent_error: 'Agent error', infra_error: 'Infra error', cancelled: 'Cancelled', timeout: 'Timeout', budget: 'Spending cap'}; const STATE_LABEL = {preparing: 'Preparing', starting: 'Starting', running: 'Running', finished: 'Finished', unknown: 'Waiting for the feed'}; const FLIGHT_SHOWN = 12; // past this many in flight, the nearest their limit come first and the rest wait behind a button const MANY_TILES = 60; // past this many trials, the tiles are drawn smaller @@ -99,7 +99,7 @@ const C = s.counts || {}; const queued = C.total != null ? Math.max(0, C.total - C.completed - C.running) : null; const rate = s.elapsed_sec > 60 && C.completed ? C.completed / (s.elapsed_sec / 60) : null; - const ran = C.completed - (oc.infra_error || 0); + const ran = C.completed - (oc.infra_error || 0) - (oc.budget || 0); const when = state === 'finished' ? 'on the feed' : 'so far'; // the feed's outcomes, not the classifier's: Results has those // A benchmark that gives a score and no pass or fail has no solve rate: the mean score, over every part scored, stands there. const scored = (s.trials || []).filter(t => t.state === 'done' && t.outcome === 'scored' && t.score != null); diff --git a/app/static/style.css b/app/static/style.css index 9631c7b..e228d89 100644 --- a/app/static/style.css +++ b/app/static/style.css @@ -35,6 +35,7 @@ --rate_limited: #7d8595; --api_error: #4aa8ee; /* 35.8 from its nearest, rate_limited */ --setup_error: #e58ac0; + --spending_cap: #b68a6c; /* bronze, a coin: 34.8 from its nearest, rate_limited */ --unknown: #454c58; /* Tool categories and run phases own their colours rather than borrowing @@ -109,6 +110,7 @@ --rate_limited: #626b7a; --api_error: #125db2; /* 43.5 from its nearest, rate_limited */ --setup_error: #a03089; + --spending_cap: #895741; /* 34.5 from its nearest, patch_failed_tests */ --unknown: #aab1bb; /* Same treatment, inverted: worst pair 16.7. */ @@ -998,6 +1000,7 @@ tr.control-row td { color: var(--text-dim); } .oc-rate_limited::before, .seg.oc-rate_limited, .tag.oc-rate_limited::before { background: var(--rate_limited); } .oc-api_error::before, .seg.oc-api_error, .tag.oc-api_error::before { background: var(--api_error); } .oc-setup_error::before, .seg.oc-setup_error, .tag.oc-setup_error::before { background: var(--setup_error); } +.oc-spending_cap::before, .seg.oc-spending_cap, .tag.oc-spending_cap::before { background: var(--spending_cap); } .seg.oc-unknown { background: var(--unknown); } /* outcome tag inside a table row */ @@ -1019,15 +1022,16 @@ tr.control-row td { color: var(--text-dim); } .tag.oc-rate_limited { --outcome-color: var(--rate_limited); } .tag.oc-api_error { --outcome-color: var(--api_error); } .tag.oc-setup_error { --outcome-color: var(--setup_error); } +.tag.oc-spending_cap { --outcome-color: var(--spending_cap); } .tag.oc-unknown { --outcome-color: var(--unknown); } -.tag:is(.oc-resolved, .oc-scored, .oc-patch_failed_tests, .oc-failed_verification, .oc-patch_not_applied, .oc-no_patch, .oc-agent_error, .oc-rate_limited, .oc-api_error, .oc-setup_error, .oc-unknown) { +.tag:is(.oc-resolved, .oc-scored, .oc-patch_failed_tests, .oc-failed_verification, .oc-patch_not_applied, .oc-no_patch, .oc-agent_error, .oc-rate_limited, .oc-api_error, .oc-setup_error, .oc-spending_cap, .oc-unknown) { color: color-mix(in srgb, var(--outcome-color) 75%, var(--text)); background: color-mix(in srgb, var(--outcome-color) 16%, var(--surface)); border-color: color-mix(in srgb, var(--outcome-color) 42%, var(--border)); font-family: var(--sans); font-size: 12px; font-weight: 500; line-height: 1.45; padding: .22rem .55rem; border-radius: 5px; } -.tag:is(.oc-resolved, .oc-scored, .oc-patch_failed_tests, .oc-failed_verification, .oc-patch_not_applied, .oc-no_patch, .oc-agent_error, .oc-rate_limited, .oc-api_error, .oc-setup_error, .oc-unknown)::before { width: 7px; height: 7px; flex: none; } +.tag:is(.oc-resolved, .oc-scored, .oc-patch_failed_tests, .oc-failed_verification, .oc-patch_not_applied, .oc-no_patch, .oc-agent_error, .oc-rate_limited, .oc-api_error, .oc-setup_error, .oc-spending_cap, .oc-unknown)::before { width: 7px; height: 7px; flex: none; } .tag.oc-unknown { color: var(--text-dim); } .tag.oc-unknown::before { background: var(--text-dim); } diff --git a/app/tests/test_classify.py b/app/tests/test_classify.py index 9008a91..9bd7ca1 100644 --- a/app/tests/test_classify.py +++ b/app/tests/test_classify.py @@ -92,6 +92,20 @@ def test_both_boards_charge_the_infrastructure_as_the_classifier_does(self): with self.subTest(name, board=board): self.assertIn(outcome(name), ("agent_error", "timeout")) + def test_a_unit_the_runs_spending_cap_cut_off_has_its_own_outcome_outside_the_rate_on_both(self): + """The standard's own: its relay refused the call (``standard.ledger``), so the unit had no attempt. + Not an API error: nothing failed. The live feed's word for it is ``budget``.""" + from nullius.runs import live + from nullius.standard import job, ledger + self.assertEqual(C.from_verdict(ledger.ERROR, None), C.SPENDING_CAP) + self.assertEqual(C.from_verdict("nullius.standard." + ledger.ERROR, None), C.SPENDING_CAP) + self.assertIn(C.SPENDING_CAP, C.INFRA_OUTCOMES) + self.assertEqual(C.ORDER.index(C.SPENDING_CAP), C.ORDER.index(C.UNKNOWN) - 1) + self.assertNotIn("—", C.DESCRIPTIONS[C.SPENDING_CAP]) + end = job.feed_outcome("pass", None, (ledger.ERROR, "the run's spending cap")) + self.assertEqual((end["outcome"], end["exc"]), ("budget", ledger.ERROR)) + self.assertIn("budget", live.OUTCOMES) + class HarborTaxonomy(unittest.TestCase): def test_provider_failures_are_infra(self): @@ -235,7 +249,7 @@ def test_every_outcome_is_ordered(self): """ORDER drives sorting; an outcome missing from it raises on compare.""" produced = {C.RESOLVED, C.SCORED, C.PATCH_FAILED_TESTS, C.FAILED_VERIFICATION, C.PATCH_NOT_APPLIED, C.NO_PATCH, C.AGENT_ERROR, - C.RATE_LIMITED, C.API_ERROR, C.SETUP_ERROR, C.UNKNOWN} + C.RATE_LIMITED, C.API_ERROR, C.SETUP_ERROR, C.SPENDING_CAP, C.UNKNOWN} self.assertEqual(produced, set(C.ORDER)) diff --git a/app/tests/test_judge.py b/app/tests/test_judge.py index 65e1f95..b90fad5 100644 --- a/app/tests/test_judge.py +++ b/app/tests/test_judge.py @@ -61,7 +61,7 @@ def test_likely_outcome_is_asked_of_unknown_trials_only_over_the_classifiers_own from nullius.runs import classify q = questions.QUESTIONS["likely_outcome"] self.assertEqual((q["type"], q["state"], q["when"]), ("choice", "failure", "unknown")) - self.assertEqual(sorted(q["criteria"]), sorted([l for l in classify.ORDER if l not in (classify.UNKNOWN, classify.SCORED)] + ["other"])) # the classifier's labels, unknown itself and scored (only a benchmark's score makes it) excluded, an escape + self.assertEqual(sorted(q["criteria"]), sorted([l for l in classify.ORDER if l not in (classify.UNKNOWN, classify.SCORED, classify.SPENDING_CAP)] + ["other"])) # the classifier's labels, unknown itself, scored (only a benchmark's score makes it) and the spending cap (only the relay's refusal makes it) excluded, an escape self.assertIn("likely_outcome", questions.for_trial(False, "unknown")); self.assertNotIn("failure_cause", questions.for_trial(True, "resolved")) for outcome in ("resolved", "patch_failed_tests", "agent_error", None): self.assertNotIn("likely_outcome", questions.for_trial(outcome == "resolved", outcome), outcome) diff --git a/app/tests/test_standard_container.py b/app/tests/test_standard_container.py index c967492..ef8ddf3 100644 --- a/app/tests/test_standard_container.py +++ b/app/tests/test_standard_container.py @@ -106,6 +106,14 @@ def test_the_relays_container_is_given_the_keys_file_never_the_key(self): self.assertEqual(a[a.index("--route") + 1], "u0000-judge=the-judge") self.assertEqual(a[a.index("--public") + 1], "http://relay:8080") self.assertEqual(a[a.index("--label") + 1], "nullius.job=j") + self.assertNotIn("--ledger", a) + + def test_the_relays_container_is_given_the_runs_ledger_and_its_folder_alone(self): + a = container.relay_argv("img", name="r", job="j", app="/app", record="/rec", key_file="/key", base="b", model="m", routes={}, + ledger="/job/.work/ledger/ledger.json") + self.assertEqual(a[a.index("--ledger") + 1], "/job/.work/ledger/ledger.json") + # where every relay of the job books its calls: that folder, and nothing else of the job but its own record + self.assertEqual([w for w in a if w.endswith(":rw")], ["/rec:/rec:rw", "/job/.work/ledger:/job/.work/ledger:rw"]) def test_an_endpoint_on_this_machines_loopback_is_reached_through_the_bridge(self): self.assertTrue(all(map(container.on_loopback, ["http://localhost:11434/v1", "http://127.0.0.1:8000/v1", "http://[::1]:8000/v1"]))) diff --git a/app/tests/test_standard_ledger.py b/app/tests/test_standard_ledger.py new file mode 100644 index 0000000..cf293d0 --- /dev/null +++ b/app/tests/test_standard_ledger.py @@ -0,0 +1,172 @@ +"""A run's spending cap: a call is booked before it is sent, sent only while +it fits, and the first that does not closes the ledger for every later one.""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import threading +import unittest + +from nullius.standard import ledger + +ROW = {"input": 1.0, "output": 10.0} + + +class LedgerTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.path = os.path.join(self.tmp, "ledger", "ledger.json") + + def test_the_ledger_keeps_the_largest_reply_each_model_has_answered_with(self): + led = ledger.Ledger.create(self.path, 1.0) + self.assertIsNone(led.largest("m")) # before any reply + for cost, model, output in ((0.01, "m", 300), (0.01, "m", 120), (0.01, "other", 9000), (None, "m", None), (0.01, "m", True)): + booking, _ = led.reserve(0.02) + led.settle(booking, cost, model, output) + self.assertEqual((led.largest("m"), led.largest("other"), led.largest("never")), (300, 9000, None)) + with open(self.path) as f: # a ledger written before the field is read as one with none + doc = json.load(f) + del doc["seen"] + with open(self.path, "w") as f: + json.dump(doc, f) + self.assertIsNone(led.largest("m")) + booking, _ = led.reserve(0.02) + led.settle(booking, 0.01, "m", 50) + self.assertEqual(led.largest("m"), 50) + + def test_a_call_is_sent_while_spent_and_in_flight_and_its_booking_fit(self): + led = ledger.Ledger.create(self.path, 1.0) + a, _ = led.reserve(0.4) + b, _ = led.reserve(0.4) + self.assertTrue(a and b and a != b) + self.assertAlmostEqual(led.state()["held_usd"], 0.8) + led.settle(a, 0.05) # the reply: what it cost, in place of what was booked + self.assertAlmostEqual(led.state()["spent_usd"], 0.05) + self.assertAlmostEqual(led.state()["left_usd"], 1.0 - 0.05 - 0.4) + c, _ = led.reserve(0.5) + self.assertTrue(c) + self.assertFalse(led.closed()) + + def test_the_first_call_that_does_not_fit_closes_it_for_every_later_one(self): + led = ledger.Ledger.create(self.path, 1.0) + a, _ = led.reserve(0.7) + refused, seen = led.reserve(0.4) # 0.7 in flight: 0.4 more would pass the cap + self.assertIsNone(refused) + self.assertFalse(seen["closed"]) # the state it was refused in, for its message + self.assertTrue(led.closed()) + led.settle(a, 0.01) + self.assertIsNone(led.reserve(0.0001)[0]) # room again, and still closed: a run stops at one point + said = ledger.refusal(led.state(), 0.0001) + self.assertIn("1.00 USD", said) + self.assertIn("no further call", said) + + def test_a_call_whose_cost_was_never_known_keeps_its_booking_and_an_error_gives_it_back(self): + led = ledger.Ledger.create(self.path, 1.0) + a, _ = led.reserve(0.3) + b, _ = led.reserve(0.3) + led.settle(a, None) # no token count, or cut off: it may have been billed + led.release(b) # the provider's error: not billed + led.release(b) # twice is once + led.settle("unknown", 9.0) # a booking it does not hold changes nothing + got = led.state() + self.assertEqual((got["spent_usd"], got["unsettled_usd"], got["held_usd"]), (0.0, 0.3, 0)) + self.assertAlmostEqual(got["left_usd"], 0.7) + + def test_a_ledger_that_cannot_be_read_lets_no_call_out(self): + led = ledger.Ledger.create(self.path, 1.0) + with open(self.path, "w") as f: + f.write("{not json") + with self.assertRaises(ledger.Unavailable): + led.reserve(0.1) + self.assertTrue(led.closed()) + self.assertTrue(ledger.Ledger(os.path.join(self.tmp, "none.json")).closed()) + + def test_many_calls_at_once_never_book_past_the_cap(self): + led = ledger.Ledger.create(self.path, 1.0) + got = [] + + def ask(): + got.append(led.reserve(0.03)[0]) + + threads = [threading.Thread(target=ask) for _ in range(60)] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertEqual(sum(1 for b in got if b), 33) # 33 x 0.03 fit in 1.00, the 34th does not + self.assertLessEqual(led.state()["held_usd"], 1.0) + + def test_relays_in_other_processes_book_in_the_same_ledger(self): + """What a docker unit's relay does from its container: each process + books and settles; none of it is lost between them.""" + led = ledger.Ledger.create(self.path, 100.0) + app = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + code = ("import sys\nfrom nullius.standard import ledger\nled = ledger.Ledger(sys.argv[1])\n" + "for _ in range(40):\n b, _ = led.reserve(0.5)\n led.settle(b, 0.01)\n") + procs = [subprocess.Popen([sys.executable, "-c", code, self.path], env={"PATH": os.environ.get("PATH", ""), "PYTHONPATH": app}) + for _ in range(4)] + self.assertEqual([p.wait(timeout=60) for p in procs], [0, 0, 0, 0]) + got = led.state() + self.assertAlmostEqual(got["spent_usd"], 4 * 40 * 0.01) + self.assertEqual((got["held_usd"], got["closed"]), (0, False)) + with open(self.path) as f: + self.assertEqual(json.load(f)["held"], {}) + + +class BoundTest(unittest.TestCase): + def bound(self, req, row=ROW, path="/chat/completions"): + return ledger.bound(req, json.dumps(req).encode(), row, path) + + def test_the_limit_the_call_names_is_its_output(self): + req = {"model": "m", "messages": [{"role": "user", "content": "x" * 300}], "max_tokens": 100} + tokens = -(-len(json.dumps(req)) // 3) # its characters over three, rounded up + self.assertAlmostEqual(self.bound(req), (tokens * 1.0 + 100 * 10.0) / 1e6) + for name in ("max_completion_tokens", "max_output_tokens"): + other = {"model": "m", "input": "hi", name: 50} + self.assertAlmostEqual(self.bound(other), (-(-len(json.dumps(other)) // 3) * 1.0 + 50 * 10.0) / 1e6) + + def test_a_call_that_names_none_is_booked_the_models_own_limit_else_the_default(self): + req = {"model": "m", "messages": []} + tokens = -(-len(json.dumps(req)) // 3) + self.assertAlmostEqual(self.bound(req, dict(ROW, max_output=2000)), (tokens + 2000 * 10.0) / 1e6) + self.assertAlmostEqual(self.bound(req), (tokens + ledger.OUTPUT * 10.0) / 1e6) + + def test_a_call_in_flight_is_booked_the_largest_reply_the_run_has_seen_never_past_a_limit(self): + """Its cost is known only when its reply comes back: until then it counts as the largest reply the run + has had from that model, not as the model's limit, which a short answer is a hundredth of.""" + req = {"model": "m", "messages": []} + tokens = -(-len(json.dumps(req)) // 3) + out = lambda row=ROW, seen=None, **named: round((ledger.bound({**req, **named}, json.dumps({**req, **named}).encode(), row, "/chat/completions", seen) * 1e6 + - -(-len(json.dumps({**req, **named})) // 3)) / 10.0) + self.assertEqual(out(seen=300), 300) # what its replies have been + self.assertEqual(out(), ledger.OUTPUT) # before any reply + self.assertEqual(out(dict(ROW, max_output=131072)), ledger.OUTPUT) # not the model's whole limit + self.assertEqual(out(dict(ROW, max_output=131072), seen=900), 900) + self.assertEqual(out(seen=300, max_tokens=100), 100) # never more than the call asks for + self.assertEqual(out(dict(ROW, max_output=2000), seen=5000), 2000) # nor than the model can answer + self.assertEqual(out(max_tokens=64000), ledger.OUTPUT) # a large limit named is no reply seen + self.assertEqual(out(seen=7000, max_tokens=64000), 7000) + self.assertAlmostEqual(ledger.bound(req, json.dumps(req).encode(), ROW, "/chat/completions", 300), (tokens + 300 * 10.0) / 1e6) + + def test_input_is_booked_at_the_dearer_of_input_and_cache_write(self): + req = {"model": "m", "messages": [], "max_tokens": 1} + tokens = -(-len(json.dumps(req)) // 3) + self.assertAlmostEqual(self.bound(req, dict(ROW, cache_write=1.25)), (tokens * 1.25 + 10.0) / 1e6) + + def test_an_embeddings_call_has_no_output_and_an_encoded_file_is_not_a_prompt(self): + req = {"model": "e", "input": ["some text"]} + self.assertAlmostEqual(self.bound(req, {"input": 0.02, "output": 5.0}, "/embeddings"), -(-len(json.dumps(req)) // 3) * 0.02 / 1e6) + picture = "A" * 30000 + plain = {"model": "m", "max_tokens": 1, "messages": [{"role": "user", "content": [ + {"type": "image_url", "image_url": {"url": ""}}, {"type": "image", "source": {"type": "base64", "data": ""}}]}]} + with_files = json.loads(json.dumps(plain)) + with_files["messages"][0]["content"][0]["image_url"]["url"] = "data:image/png;base64," + picture + with_files["messages"][0]["content"][1]["source"]["data"] = picture + self.assertAlmostEqual(self.bound(with_files), self.bound(plain)) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_standard_relay.py b/app/tests/test_standard_relay.py index 4fbea5e..a6f5e7e 100644 --- a/app/tests/test_standard_relay.py +++ b/app/tests/test_standard_relay.py @@ -16,7 +16,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from nullius.runs import classify -from nullius.standard import jsonl, relay +from nullius.standard import jsonl, ledger, relay CHAT = {"choices": [{"message": {"role": "assistant", "content": "hello"}}], "usage": {"prompt_tokens": 10, "completion_tokens": 2, "prompt_tokens_details": {"cached_tokens": 4}}} @@ -45,6 +45,8 @@ def do_POST(self): self.reply(429, b'{"error":{"message":"Rate limit reached"}}', "application/json") elif mode == "401": self.reply(401, b'{"error":{"message":"Incorrect API key"}}', "application/json") + elif mode == "no-usage": + self.reply(200, b'{"choices":[{"message":{"role":"assistant","content":"hello"}}]}', "application/json") elif self.path.endswith("/responses"): self.reply(200, RESP_SSE, "text/event-stream") elif body.get("stream"): @@ -178,6 +180,123 @@ def test_an_advertised_address_is_what_url_hands_out(self): r.stop() +class CapTest(unittest.TestCase): + """A relay of a run with a spending cap: a call is booked before it goes + out and is not sent when it does not fit - the one answer that is the + relay's own, and marked so.""" + PRICES = {"m": {"input": 1.0, "cached": 0.1, "output": 10.0}} + + @classmethod + def setUpClass(cls): + cls.up = ThreadingHTTPServer(("127.0.0.1", 0), Upstream) + threading.Thread(target=cls.up.serve_forever, daemon=True).start() + cls.base = f"http://127.0.0.1:{cls.up.server_address[1]}/v1" + + @classmethod + def tearDownClass(cls): + cls.up.shutdown() + + def capped(self, cap): + tmp = tempfile.mkdtemp() + led = ledger.Ledger.create(os.path.join(tmp, "ledger", "ledger.json"), cap) + r = relay.Relay(self.base, "sk-real", os.path.join(tmp, "relay"), prices=self.PRICES, ledger=led.path).start() + self.addCleanup(r.stop) + return r, led + + def post(self, r, tag, body): + req = urllib.request.Request(r.url(tag) + "/chat/completions", data=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}) + try: + with urllib.request.urlopen(req) as got: + return got.status, dict(got.headers.items()), got.read() + except urllib.error.HTTPError as e: + return e.code, dict(e.headers.items()), e.read() + + def test_a_call_that_fits_is_sent_and_its_booking_becomes_what_it_cost(self): + r, led = self.capped(1.0) + body = {"model": "m", "messages": [{"role": "user", "content": "hi"}], "max_tokens": 100} + status, _, _ = self.post(r, "u1", body) + self.assertEqual(status, 200) + rec = r.calls("u1")[-1] + self.assertAlmostEqual(rec["booked_usd"], ledger.bound(body, json.dumps(body).encode(), self.PRICES["m"])) + got = led.state() + self.assertAlmostEqual(got["spent_usd"], rec["cost_usd"]) + self.assertEqual((got["held_usd"], got["unsettled_usd"], got["closed"]), (0, 0.0, False)) + + def test_a_call_is_booked_at_what_the_runs_replies_have_been_once_one_came_back(self): + """Fifteen calls at the model's limit would fill a cap nothing was spent of: after the first reply + (2 output tokens here) a call that names no limit is booked those, at the same price.""" + r, led = self.capped(1.0) + body = {"model": "m", "messages": [{"role": "user", "content": "hi"}]} + raw = json.dumps(body).encode() + self.assertEqual(self.post(r, "u1", body)[0], 200) + first = r.calls("u1")[-1] + self.assertAlmostEqual(first["booked_usd"], ledger.bound(body, raw, self.PRICES["m"])) # no reply yet: the start + self.assertEqual(led.largest("m"), first["reply"]["usage"]["output"]) + self.assertEqual(self.post(r, "u2", body)[0], 200) + second = r.calls("u2")[-1] + self.assertAlmostEqual(second["booked_usd"], ledger.bound(body, raw, self.PRICES["m"], seen=first["reply"]["usage"]["output"])) + self.assertLess(second["booked_usd"], first["booked_usd"]) + self.assertAlmostEqual(led.state()["spent_usd"], first["cost_usd"] + second["cost_usd"]) # what was spent is what the calls cost + + def test_a_call_that_does_not_fit_is_answered_by_the_relay_and_never_sent(self): + r, led = self.capped(0.0005) # 100 output tokens at 10 USD a million are 0.001 already + sent = len(Upstream.seen) + status, headers, body = self.post(r, "u1", {"model": "m", "messages": [{"role": "user", "content": "hi"}], "max_tokens": 100}) + self.assertEqual((status, headers.get("X-Nullius-Refused")), (402, "cap")) + error = json.loads(body)["error"] + self.assertEqual(error["type"], "spending_cap") + self.assertIn("nullius relay: the run's spending cap of 0.0005 USD", error["message"]) + rec = r.calls("u1")[-1] + self.assertEqual((rec["status"], rec["refused"]), (402, "cap")) + self.assertNotIn("reply", rec) + self.assertTrue(led.closed()) + # a call small enough to fit is refused too: a run stops at one point + status, _, _ = self.post(r, "u1", {"model": "m", "messages": [], "max_tokens": 1}) + self.assertEqual(status, 402) + self.assertEqual(len(Upstream.seen), sent) # neither reached the provider + self.assertEqual(led.state()["spent_usd"], 0.0) + exc, said = relay.failure(r.calls("u1")) + self.assertEqual(exc, ledger.ERROR) + self.assertIn("spending cap", said) + self.assertEqual(classify.from_verdict(exc, None), classify.SPENDING_CAP) # its own outcome, never a miss + self.assertIn(classify.SPENDING_CAP, classify.INFRA_OUTCOMES) + + def test_a_refused_call_is_the_units_whatever_its_agent_did_next(self): + calls = [{"status": 200}, {"status": 402, "refused": "cap", "error": "the run's spending cap"}, {"status": 200}] + self.assertEqual(relay.failure(calls), (ledger.ERROR, "the run's spending cap")) + + def test_an_error_gives_the_booking_back_and_a_reply_without_tokens_keeps_it(self): + r, led = self.capped(1.0) + status, _, _ = self.post(r, "u1", {"model": "m", "mode": "429", "max_tokens": 100}) + self.assertEqual(status, 429) # the provider's own, passed on as ever + got = led.state() + self.assertEqual((got["spent_usd"], got["unsettled_usd"], got["held_usd"]), (0.0, 0.0, 0)) + status, _, _ = self.post(r, "u1", {"model": "m", "mode": "no-usage", "max_tokens": 100}) + self.assertEqual(status, 200) + rec = r.calls("u1")[-1] + self.assertNotIn("cost_usd", rec) + got = led.state() + self.assertEqual((got["spent_usd"], got["held_usd"]), (0.0, 0)) + self.assertAlmostEqual(got["unsettled_usd"], rec["booked_usd"]) # it may have been billed + + def test_a_model_without_a_price_is_not_sent_and_closes_nothing(self): + r, led = self.capped(1.0) + sent = len(Upstream.seen) + status, _, body = self.post(r, "u1", {"model": "another", "messages": []}) + self.assertEqual(status, 402) + self.assertIn("no price for the model 'another'", json.loads(body)["error"]["message"]) + self.assertEqual(len(Upstream.seen), sent) + self.assertFalse(led.closed()) + + def test_a_relay_without_a_cap_books_nothing(self): + r = relay.Relay(self.base, "sk-real", tempfile.mkdtemp(), prices=self.PRICES).start() + self.addCleanup(r.stop) + status, _, _ = self.post(r, "u1", {"model": "another", "messages": []}) + self.assertEqual(status, 200) + self.assertNotIn("booked_usd", r.calls("u1")[-1]) + + class EffortTest(unittest.TestCase): def test_the_record_says_the_reasoning_effort_a_request_asked_for(self): # chat's reasoning_effort, the Responses API's reasoning.effort, and none at all @@ -230,6 +349,24 @@ def test_serve_relays_with_the_key_from_its_file_and_stops_on_sigterm(self): p.kill() p.stdout.close() + def test_serve_books_its_calls_in_the_runs_ledger(self): + """A docker unit's relay: the ledger is a file it is given, shared with every other relay of the job.""" + import signal + led = ledger.Ledger.create(os.path.join(tempfile.mkdtemp(), "ledger", "ledger.json"), 1.0) + p = self.serve("sk-from-env", "--ledger", led.path, "--prices", json.dumps({"the-recipes": {"input": 1.0, "output": 10.0}})) + try: + where = json.loads(p.stdout.readline())["listening"] + req = urllib.request.Request(f"{where}/u1/v1/chat/completions", data=b'{"model":"asked","max_tokens":5}', + headers={"Content-Type": "application/json"}) + urllib.request.urlopen(req).read() + self.assertAlmostEqual(led.state()["spent_usd"], (10 * 1.0 + 2 * 10.0) / 1e6) + p.send_signal(signal.SIGTERM) + self.assertEqual(p.wait(timeout=10), 0) + finally: + if p.poll() is None: + p.kill() + p.stdout.close() + def test_serve_without_its_key_file_refuses(self): import subprocess p = subprocess.Popen([sys.executable, "-m", "nullius.standard.relay", "serve", "--base", "http://127.0.0.1:9/v1", diff --git a/app/tests/test_standard_run.py b/app/tests/test_standard_run.py index c9856e8..52d418d 100644 --- a/app/tests/test_standard_run.py +++ b/app/tests/test_standard_run.py @@ -13,7 +13,7 @@ from unittest import mock from nullius.runs import classify, parse -from nullius.standard import acp, container, definitions, install, run +from nullius.standard import acp, container, definitions, install, ledger, run ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures", "standard", "catalogue") # A conversation whose user a model plays: it opens, speaks once more, then is done. @@ -292,6 +292,64 @@ def recorded(doc): # the user's own row over the list's: the list did not say this price, so it is not named self.assertEqual(recorded({"models": {"m-1": listed}, "own": {"m-1": own}}), {"models": {"m-1": {**own, "source": "own"}}}) + def calling(self, **kw): + """Eight units, each one model call (11 tokens in, 3 out), one at a + time. Input is free here and the model's own output limit is 10 + tokens at 10 USD a million: the first call is booked 0.0001, and once + its reply is seen each call is booked what a reply is, 0.00003, its cost.""" + here = os.path.join(self.tmp, "calling") + os.makedirs(here, exist_ok=True) + with open(os.path.join(here, "driver.py"), "w") as f: + f.write(CALLING_DRIVER) + b = dict(definitions.get("benchmark", "fakebench", ROOT), _here=here, driver={"command": ["python3", "{here}/driver.py"]}) + said = [] + kw = {"prices": {"m-1": {"input": 0.0, "output": 10.0, "max_output": 10}}, **kw} + job_dir, failed = run.run_and_report(b, definitions.get("agent", "fake", ROOT), model="m-1", provider="custom", jobs_root=self.tmp, + concurrency=1, env=self.env, say=said.append, unsandboxed=True, proving=True, **kw) + with open(os.path.join(job_dir, "result.json")) as f: + return job_dir, failed, json.load(f), said + + def test_a_spending_cap_stops_the_run_and_its_result_says_where(self): + # 0.00003 spent by each call and as much booked for the next: with 0.00009 spent the fourth no longer fits 0.00011 + job_dir, failed, result, said = self.calling(max_usd=0.00011) + self.assertEqual(result["budget"], {"max_usd": 0.00011, "spent_usd": 0.00009, "stopped": True, + "cut_off": ["m3"], "not_run": ["m4", "m5", "m6", "m7"]}) + self.assertEqual((result["n_total_trials"], result["stats"]["n_cancelled_trials"], result["stats"]["n_errored_trials"]), (8, 4, 1)) + self.assertFalse(failed) # three units reached a verdict + got, _ = self.outcomes(job_dir) + self.assertEqual(sorted(got), ["m0", "m1", "m2", "m3"]) # the four never begun have no trial + self.assertEqual(got["m3"].exception_type, ledger.ERROR) + self.assertEqual(got["m3"].outcome, classify.SPENDING_CAP) # its own outcome, outside the rate: never a miss + self.assertTrue(all(got[u].outcome == classify.RESOLVED for u in ("m0", "m1", "m2"))) + calls = self.relay_calls(job_dir) + self.assertEqual([c.get("refused") for c in calls].count("cap"), 1) + sent = [c for c in calls if not c.get("refused")] + self.assertEqual(len(sent), 3) # no call went out past the cap + self.assertEqual(sorted(round(c["booked_usd"], 8) for c in sent), [0.00003, 0.00003, 0.0001]) # the start once, then what a reply has been + self.assertLessEqual(sum(c["cost_usd"] for c in sent), 0.00011) + self.assertTrue(any(s.startswith("stopped: a call did not fit the spending cap of 0.0001 USD (0.0001 spent): 1 cut off, 4 not run") + for s in said), said) + self.assert_no_key(job_dir) + + def test_a_spending_cap_the_run_stays_under_changes_nothing_but_the_result(self): + job_dir, failed, result, said = self.calling(max_usd=1.0) + self.assertEqual(result["budget"], {"max_usd": 1.0, "spent_usd": 0.00024, "stopped": False, "cut_off": [], "not_run": []}) + self.assertEqual(len(self.outcomes(job_dir)[0]), 8) + self.assertIn("spent 0.0002 of the 1.00 USD cap", said) + job_dir, _, result, _ = self.calling(name="uncapped") + self.assertNotIn("budget", result) + self.assertFalse(os.path.exists(os.path.join(job_dir, ".work", "ledger"))) + self.assertFalse(any("booked_usd" in c for c in self.relay_calls(job_dir))) + + def test_a_spending_cap_is_refused_without_a_price_for_each_model_or_an_amount(self): + with self.assertRaises(run.Refused) as e: + self.calling(max_usd=1.0, prices={"another": {"input": 1.0, "output": 1.0}}) + self.assertIn("there is none for m-1", str(e.exception)) + for bad in (0, -1, float("inf"), float("nan"), "much"): + with self.assertRaises(run.Refused, msg=bad): + self.calling(max_usd=bad) + self.assertEqual([n for n in os.listdir(self.tmp) if "fakebench" in n], []) # nothing was begun + def test_a_run_with_no_price_has_no_cost_and_says_nothing_of_prices(self): job_dir = self.go("fakebench", "fake") calls = self.relay_calls(job_dir) diff --git a/docs/cli/run.md b/docs/cli/run.md index 7b8c06b..aa42329 100644 --- a/docs/cli/run.md +++ b/docs/cli/run.md @@ -5,7 +5,7 @@ One benchmark and one agent, capability, control or setup, into a job. ```bash title="Run" nullius standard run [@version] [--model M] [--dataset D] [--limit N] [-j N] \ [--unsandboxed | --backend docker] [--option NAME=VALUE]... [--bench-option NAME=VALUE]... \ - [--price MODEL=INPUT,OUTPUT[,CACHED]]... \ + [--price MODEL=INPUT,OUTPUT[,CACHED]]... [--max-usd USD] \ [--repeat K] [--remove-images] [--judge-model M] [--simulator-model M] [--name NAME] [--jobs DIR] [--provider P] ``` @@ -28,6 +28,7 @@ nullius standard run locomo model --model gpt-4o-mini | `--option NAME=VALUE` | An agent's or capability's option (repeatable) | | `--bench-option NAME=VALUE` | A benchmark's option (repeatable), read as `NULLIUS_OPTION_` | | `--price MODEL=INPUT,OUTPUT[,CACHED]` | This run's price for a model in USD per million tokens (repeatable), over your [price list](../settings/model-prices.md) | +| `--max-usd USD` | The run's spending cap: no call goes out once the spend plus the calls in flight would pass it ([below](#a-spending-cap)) | | `--repeat K` | Each unit K times; `result.json` carries `pass_hat_k` | | `--remove-images` | Remove each image the run pulls or builds when its last unit ends (never with `--repeat`); by default they stay for the next run, listed and removed in Settings, [Sandbox](../settings/sandbox.md) | | `--judge-model` | Instead of the official judge | @@ -46,6 +47,25 @@ nullius standard run locomo model --model gpt-4o-mini | Progress | stderr, one line per stage (`[nullius] stage=install ...`, `stage=units n=10`, `stage=done`, `stage=refused`), also `progress.jsonl` | | The job | Opens in the app; `events.jsonl` feeds the live board | +## A spending cap + +`--max-usd 5` stops a run at 5 USD, for a prepaid or shared key. Every relay of +the job books each call in one ledger before sending it +(`app/nullius/standard/ledger.py`). + +| Rule | Detail | +| --- | --- | +| Prices first | Every model the run calls (its own, the judge, the simulated user, an option's embeddings, an agent's side models) needs a price in your [price list](../settings/model-prices.md) or a `--price`; else the run is refused before it starts | +| A booking | What a call in flight is counted as, since its cost is known only when its reply comes back: its characters over three as input tokens, and as output the largest reply the run has had from that model so far, 4,096 tokens before any; never more than the limit the call names or the model's own | +| Sent or refused | A call is sent only while what was spent, what the calls in flight have booked and its own booking stay within the cap. The first that does not fit is refused, and so is every call after it: the run stops at one point | +| Settled | A reply turns the booking into what the call cost. A provider's error gives the booking back. A reply with no token count, or cut off, keeps its whole booking as unsettled: it may have been billed | +| A refused call | Answered by the relay with HTTP 402, naming the relay and the cap, and recorded `refused`: never the provider's error | +| The units | One cut off by the cap ends as `SpendingCapError`: the outcome Spending cap, outside the agent's rate, never a miss. Units not begun are not run | +| The result | `result.json` carries `budget`: `max_usd`, `spent_usd`, `unsettled_usd` when there is some, `stopped`, `cut_off` and `not_run`; stderr says `stage=cap` | +| Several at a time | The calls in flight take room: with `-j 4`, four bookings must fit at once. The first calls, before any reply, are booked 4,096 output tokens each | +| An estimate | What was spent is each reply's cost, the provider's own figure when its reply carries one. A booking is not a bill: a reply longer than any before it costs more than was booked and takes the spend past the cap by the difference; the next call is then refused | +| A shard | The cap is that shard's alone: split a budget over the shards yourself | +
- [The standard](../standard.md#a-run) diff --git a/docs/field-guide.md b/docs/field-guide.md index c0fede5..93549c6 100644 --- a/docs/field-guide.md +++ b/docs/field-guide.md @@ -150,7 +150,7 @@ How the agent worked, not whether it won. Computed over layer 3. | Field | What it is | Source | Works on | | --- | --- | --- | --- | -| `outcome` | One of eleven classes: from the exception if there was one, else from the verifier's verdict. Full vocabulary below. | inferred | mixed | +| `outcome` | One of twelve classes: from the exception if there was one, else from the verifier's verdict. Full vocabulary below. | inferred | mixed | | `score · n_parts` | A graded score the benchmark gave, and how many parts it averages. With no reward and no exception, the outcome is *Scored*. | measured | mixed | | `reward` | The verifier's raw reward. Where no report says more, 1.0 or above is a solve. | measured | any | | `f2p_pass / f2p_fail` | Checks that had to start passing: SWE-bench's FAIL_TO_PASS, or every test of a Terminal-Bench task. | measured | mixed | @@ -219,7 +219,7 @@ Computed here over the trajectory's tool calls, when present. They describe cond ## Outcome vocabulary {#outcomes} -Eleven classes. The three marked **yes** under Infrastructure leave the adjusted solve rate's denominator. +Twelve classes. The four marked **yes** under Infrastructure leave the adjusted solve rate's denominator. | Outcome | Label | Infrastructure | Meaning | | --- | --- | --- | --- | @@ -233,6 +233,7 @@ Eleven classes. The three marked **yes** under Infrastructure leave the adjusted | `rate_limited` | Rate limited | yes | An HTTP 429 ended the trial. | | `api_error` | API error | yes | Any other provider failure: overloaded, unreachable, out of quota, key rejected. | | `setup_error` | Setup error | yes | The environment or image never came up, or the unit was not run for want of a secret or docker it needs, so the agent never ran. | +| `spending_cap` | Spending cap | yes | A standard run reached its spending cap (`--max-usd`): its relay refused a model call the unit needed (`SpendingCapError`). Nothing failed; no attempt was had. | | `unknown` | Unknown | no | No verdict, or an exception on neither list. Counted, charged to nobody; the trial page shows the recorded type. | ## Caveats {#caveats} diff --git a/docs/settings/model-prices.md b/docs/settings/model-prices.md index 5dec9ed..f9d6158 100644 --- a/docs/settings/model-prices.md +++ b/docs/settings/model-prices.md @@ -20,6 +20,7 @@ its list is filled once. | Recorded | A job's `config.json` keeps the prices of the models it calls, each with its source (`list`, `own`, or `run` for `--price`), and where and when the list was fetched when a price is the list's | | Docker and hosts | A unit's relay in a container gets its models' prices from the run; a run on a host takes this machine's list with it | | An estimate | List prices, not a bill: [Metrics](../field-guide.md) | +| A spending cap | `--max-usd` books each call at these prices before sending it, so it needs one for every model the run calls: [Run a benchmark](../cli/run.md#a-spending-cap) | | Code | `app/nullius/prices.py` | ## From a shell diff --git a/docs/standard.md b/docs/standard.md index d625ac8..04910e0 100644 --- a/docs/standard.md +++ b/docs/standard.md @@ -138,6 +138,7 @@ record); verifier/reward.txt and score.json | Progress | One line per stage on stderr and in `progress.jsonl`: `[nullius] stage=install agent=codex@2.2.2 state=done seconds=41 at=...` (`start`, `done`, `cached`, `failed`; a download's bytes every 5 s) for the base image, an environment, the upstream, a dataset, `driver.install` and the agent; then `stage=units n=10`, `stage=unit i=3 of=10 unit=... outcome=passed`, `stage=done`; `stage=refused` alone for a run refused first. From `stage=units` the job also writes `events.jsonl`, Harbor's live feed | | Outcomes | A last model call that got a 429 is `rate_limited`, a refused key an API error, both outside the agent's rate; a turn past its limit is the agent's timeout. `"kind": "score"` leaves a score and no pass | | The relay | Headers pass through except the credentials, replaced by the relay's key; `agent/calls.jsonl` names them, never their values. A 429 reaches the agent as it came, never retried. Every call goes as the run's model, the one asked recorded as `model_asked`, except `model.side_models`, sent as asked and recorded `side`. Provider `anthropic` speaks Anthropic's Messages (`ANTHROPIC_API_KEY` as `x-api-key`) | +| A spending cap (`--max-usd`) | Every relay of the job books a call in one ledger (`.work/ledger`, `standard/ledger.py`) before sending it; the first that does not fit is refused by the relay itself (HTTP 402, recorded `refused`), and every call after it. A unit cut off is a `SpendingCapError`, the outcome `spending_cap` outside the agent's rate; units not begun are not run; `result.json` carries `budget`. [Run a benchmark](cli/run.md#a-spending-cap) | ## Safety @@ -296,7 +297,7 @@ N machines are N runs of the same command, each with `--shard I/N`, and one merg | Rule | Detail | | --- | --- | -| No central relay | Each unit's relay runs on its own machine with the key it was handed; split a launch over keys of their own if one cannot carry it | +| No central relay | Each unit's relay runs on its own machine with the key it was handed; split a launch over keys of their own if one cannot carry it. A spending cap (`--max-usd`) is each shard's own | | Shards by the unit's id | `sha256(unit id) mod N` (`standard/shard.py`); changing N moves most units. Order: the driver's `units`, the shard, then `--limit` (`--shard 1/2 --limit 6` is the first six units of shard 1) | | A unit keeps its place | `u0042` is the 43rd unit of the whole dataset in every shard | | One launch | Each shard's `config.json` records `standard.shard` (`index`, `of`, `units_total`, `partition`, `planned`, `limit`) and `standard.launch` (`--launch`, else the UTC day, benchmark, agent or capability and model) |