diff --git a/app/nullius/runs/classify.py b/app/nullius/runs/classify.py index d8816fa..798a388 100644 --- a/app/nullius/runs/classify.py +++ b/app/nullius/runs/classify.py @@ -81,6 +81,7 @@ # one can place it here. _AGENT_EXCEPTIONS = { "agenttimeouterror", # the agent spent its budget — that is a result + "unitspendingcaperror", # the unit spent its own spending cap (the standard's relay): the same, in money "agentsafetyrefusalerror", # the agent (or its provider) chose to refuse "contextwindowexceedederror", # the agent filled its own context "contextlengthexceedederror", # same, from Harbor's LLM layer @@ -131,7 +132,7 @@ NO_PATCH: "The agent finished without producing a patch at all.", AGENT_ERROR: ( "The agent itself failed — it timed out, filled its context, truncated its " - "own output, refused, or its process exited non-zero. Counted against the " + "own output, refused, spent its own spending cap, or its process exited non-zero. Counted against the " "agent, because each of those is a result rather than an interruption." ), RATE_LIMITED: ( diff --git a/app/nullius/runs/parse.py b/app/nullius/runs/parse.py index 2633468..63154d9 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 = "14" +SCHEMA = "15" # 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 @@ -284,7 +284,8 @@ class Job: # recipe from when the run kept none. Empty when config.json is absent. launch: dict = field(default_factory=dict) # A run with a spending cap (a Nullius standard job's ``result.json``, ``budget``): the cap and what was - # spent in USD, whether the cap stopped the run, the units it cut off and those never begun. {} without one. + # spent in USD, whether the cap stopped the run, the units it cut off and those never begun; a cap per unit + # (``unit_max_usd``) and the units that reached it (``over_unit``). {} without one. budget: dict = field(default_factory=dict) # A Harbor job run with several agents or models (`-m a -m b`, or a config # listing several): each one's {name, model, effort}, which split_agents @@ -837,11 +838,16 @@ def _budget(b) -> dict: """A job's recorded spending cap, read leniently: {} unless it names a cap; an amount that is no number is None (not reported), never zero.""" num = lambda v: v if isinstance(v, (int, float)) and not isinstance(v, bool) else None - if not isinstance(b, dict) or not num(b.get("max_usd")): + if not isinstance(b, dict) or not (num(b.get("max_usd")) or num(b.get("unit_max_usd"))): return {} names = lambda v: [str(x) for x in v] if isinstance(v, list) else [] - return {"max_usd": b["max_usd"], "spent_usd": num(b.get("spent_usd")), "unsettled_usd": num(b.get("unsettled_usd")), - "stopped": b.get("stopped") is True, "cut_off": names(b.get("cut_off")), "not_run": names(b.get("not_run"))} + return {"max_usd": num(b.get("max_usd")), "spent_usd": num(b.get("spent_usd")), "unsettled_usd": num(b.get("unsettled_usd")), + # only the run's cap stops a run: without one there is no stop to say + "stopped": b.get("stopped") is True and num(b.get("max_usd")) is not None, + "cut_off": names(b.get("cut_off")), "not_run": names(b.get("not_run")), + "unit_max_usd": num(b.get("unit_max_usd")), "over_unit": names(b.get("over_unit")), + # a merged job: its shards, and how many stopped at their share + "shards": num(b.get("shards")) or 0, "shards_stopped": num(b.get("shards_stopped")) or 0} def parse_job(job_dir: str, reuse=None) -> Job: diff --git a/app/nullius/standard/cli.py b/app/nullius/standard/cli.py index 837044b..16421ed 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] ...] [--max-usd USD] [--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] [--max-usd-per-unit 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,8 @@ 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), max_usd=a.max_usd, **({"env": run_env(a.provider)} if a.from_run else {})) + prices=prices(a.price), max_usd=a.max_usd, max_usd_per_unit=a.max_usd_per_unit, + **({"env": run_env(a.provider)} if a.from_run else {})) except run.Refused as e: print("refused:", e) return 2 @@ -626,6 +627,8 @@ def main(argv=None): 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") + p.add_argument("--max-usd-per-unit", type=float, metavar="USD", + help="each unit's own spending cap: a unit that reaches it is cut off there, and the run goes on") 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/ledger.py b/app/nullius/standard/ledger.py index ec87b42..c8e7fa8 100644 --- a/app/nullius/standard/ledger.py +++ b/app/nullius/standard/ledger.py @@ -16,14 +16,26 @@ 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. +characters over three as input tokens, and as output the average reply +the run has had from that model so far (the ledger keeps their count and +sum, ``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; the largest reply so far would let one long answer do the same to +every call after it. 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. + +A cap per unit (``run --max-usd-per-unit``) is kept in the same file: what +each unit spent and has in flight, beside the run's. A call that does not +fit its unit's cap is refused, and so is every later call of that unit; the +run goes on. A unit's calls are its own and its simulated user's - part of +the task - and never its judge's, which scores it (:func:`unit_of`). + +A call that lacks room only because of calls in flight is not refused: its +relay waits for them to settle (``WAIT``), since their bookings are +estimates and mostly come back. Only a call that would not fit with nothing +in flight, or that waited in vain, closes the ledger. 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 @@ -41,6 +53,10 @@ 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) +# what a unit that reached its own cap carries: it had its attempt and spent its allowance, as one that spends its +# time does - an agent error, counted against the agent (runs.classify) +UNIT_ERROR = "UnitSpendingCapError" +WAIT = 120.0 # seconds a call waits for the calls in flight to settle before it is refused _LIMITS = ("max_completion_tokens", "max_output_tokens", "max_tokens") @@ -48,11 +64,28 @@ class Unavailable(Exception): """The ledger could not be read or written: a capped call is then not sent.""" +def models(other: dict, model, judge, simulator, embed, control: bool = False) -> list: + """Every model a run's relays may send a call to, as sent: the run's own, the judge's, the simulated + user's (never a control's: it holds no conversation), an option's embeddings and the side models of + ``other`` (the agent, capability or setup). What a capped run must have a price for; the run and the + builder's Limits step both read it here.""" + side = [(to or name) for name, _, to in (str(m).partition("=") for m in (other.get("model") or {}).get("side_models") or ())] + return [m for m in dict.fromkeys([model, judge, None if control else simulator, embed, *side]) if m] + + +def unit_of(tag: str) -> Optional[str]: + """The unit a relay's route books for, by its tag (``run``): the unit's own calls and its simulated user's + (``-simulator``) are the task; its judge's (``-judge``) score it and count toward no unit.""" + if tag.endswith("-judge"): + return None + return tag[:-len("-simulator")] if tag.endswith("-simulator") else tag + + 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 + and the cache-write price. Output: ``seen``, the average 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) @@ -83,10 +116,13 @@ def __init__(self, path: str): self.path = path @classmethod - def create(cls, path: str, max_usd: float) -> "Ledger": + def create(cls, path: str, max_usd: Optional[float] = None, unit_max_usd: Optional[float] = None) -> "Ledger": + """``max_usd``: the run's cap, or None for a run with a cap per unit alone. ``unit_max_usd``: each unit's.""" 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": {}}) + led._write({"max_usd": None if max_usd is None else float(max_usd), + "unit_max_usd": None if unit_max_usd is None else float(unit_max_usd), + "spent_usd": 0.0, "unsettled_usd": 0.0, "held": {}, "closed": False, "seen": {}, "units": {}, "of": {}}) return led @contextlib.contextmanager @@ -104,9 +140,8 @@ def _locked(self): 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"] = {} + float(doc["max_usd"] or 0), float(doc["spent_usd"]), float(doc["unsettled_usd"]), sum(doc["held"].values()) + doc["unit_max_usd"], dict(doc["seen"]), dict(doc["units"]), dict(doc["of"]) # all there, or it is no ledger return doc def _write(self, doc: dict) -> None: @@ -116,49 +151,85 @@ def _write(self, doc: dict) -> None: json.dump(doc, f) os.replace(part, self.path) - def reserve(self, amount: float) -> Tuple[Optional[str], dict]: + def reserve(self, amount, unit: Optional[str] = None, model: Optional[str] = None, wait: bool = False) -> 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.""" + and it is closed from then on. ``unit``: the unit the call is part + of (:func:`unit_of`): with a cap per unit, a call that does not fit + its unit's is refused (the state says ``by: unit``) and closes that + unit alone - the run's ledger stays open for the others. ``amount`` + may be a function of what ``model``'s replies have been on average + (:func:`bound`): it is then read here, under the same lock, and + the state says the ``amount`` booked. ``wait``: a call that would + fit with nothing in flight is not refused but told to wait (``by: + wait``), and nothing is closed.""" with self._locked() as doc: - seen = _state(doc) + if callable(amount): + amount = amount(_usual(doc, model)) + seen = dict(_state(doc), by="cap", amount=amount) + each = doc["unit_max_usd"] if unit is not None else None + if each is not None: + u = doc["units"].setdefault(unit, {"spent_usd": 0.0, "unsettled_usd": 0.0, "closed": False}) + flight = sum(doc["held"].get(b, 0.0) for b, w in doc["of"].items() if w == unit) + u["unsettled_usd"] + seen["unit"] = {"max_usd": each, "spent_usd": u["spent_usd"], "held_usd": flight, "closed": u["closed"]} + if u["closed"] or amount > each - u["spent_usd"] - flight: + if wait and not u["closed"] and flight > u["unsettled_usd"] and amount <= each - u["spent_usd"] - u["unsettled_usd"]: + return None, dict(seen, by="wait") # only its own calls in flight are in the way + u["closed"] = True + self._write(doc) + return None, dict(seen, by="unit") if doc["closed"] or amount > seen["left_usd"]: + if wait and not doc["closed"] and seen["held_usd"] > 0 and amount <= doc["max_usd"] - doc["spent_usd"] - doc["unsettled_usd"]: + return None, dict(seen, by="wait") # only the calls in flight are in the way if not doc["closed"]: doc["closed"] = True self._write(doc) return None, seen booking = uuid.uuid4().hex[:12] doc["held"][booking] = amount + if each is not None: + doc["of"][booking] = unit 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.""" + reply's output tokens, added to what ``model`` has answered in this + run - its next calls are booked at their average.""" 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 model and isinstance(output, int) and not isinstance(output, bool) and output > 0: + replies, tokens = doc["seen"].get(model) or (0, 0) + doc["seen"][model] = [replies + 1, tokens + output] + unit = doc["units"].get(doc["of"].pop(booking, None)) if cost is None: doc["unsettled_usd"] += amount else: doc["spent_usd"] += max(0.0, float(cost)) + if unit is not None: # the same, on its unit's own count + unit["unsettled_usd" if cost is None else "spent_usd"] += amount if cost is None else 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: + doc["of"].pop(booking, None) 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.""" + def forget(self, unit: str) -> None: + """A unit that ended: its own count goes, so the file stays the size of the units in flight.""" + with self._locked() as doc: + if doc["units"].pop(unit, None) is not None: + self._write(doc) + + def usual(self, model) -> Optional[int]: + """The output tokens ``model`` has answered a call of this run with, on average; None before any. Read without the lock.""" try: - return self._read()["seen"].get(model) or None + return _usual(self._read(), model) except (OSError, ValueError, KeyError, TypeError) as e: raise Unavailable(f"the spending cap's ledger could not be read: {type(e).__name__}: {e}") @@ -179,10 +250,16 @@ def closed(self) -> bool: return True +def _usual(doc: dict, model) -> Optional[int]: + replies, tokens = doc["seen"].get(model) or (0, 0) + return math.ceil(tokens / replies) if replies else None + + def _state(doc: dict) -> dict: held = sum(doc["held"].values()) + left = float("inf") if doc["max_usd"] is None else doc["max_usd"] - doc["spent_usd"] - doc["unsettled_usd"] - held # no cap on the run: a cap per unit alone 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"])} + "left_usd": left, "closed": bool(doc["closed"])} def usd(v: float) -> str: @@ -193,6 +270,14 @@ def usd(v: float) -> str: def refusal(state: dict, amount: float) -> str: """Why a call was not sent, as its caller and the record read it.""" + if state.get("by") == "unit": + u = state["unit"] + if u["closed"]: + return (f"an earlier call of this unit did not fit its spending cap of {usd(u['max_usd'])} USD " + f"({usd(u['spent_usd'])} spent by the unit): no further call of this unit is sent") + return (f"this unit's spending cap of {usd(u['max_usd'])} USD: {usd(u['spent_usd'])} spent by the unit" + + (f", {usd(u['held_usd'])} booked by its calls in flight or never settled" if u["held_usd"] else "") + + f", and this call may cost up to {usd(amount)}; it was not sent, and no further call of this unit is") 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") diff --git a/app/nullius/standard/merge.py b/app/nullius/standard/merge.py index 00848cc..01adec1 100644 --- a/app/nullius/standard/merge.py +++ b/app/nullius/standard/merge.py @@ -156,17 +156,27 @@ def check(job_dirs: List[str], force: bool = False) -> dict: def _budget(budgets: list) -> Optional[dict]: """The shards' spending caps as one (``run.Run.budget``): the caps and what was spent added up, stopped - when any shard was, their units cut off and not run together. None unless every shard had a cap: a cap - over part of a launch is not the launch's.""" - if not budgets or not all(isinstance(x, dict) and isinstance(x.get("max_usd"), (int, float)) for x in budgets): + when any shard was, their units cut off and not run together; a cap per unit, the same on every shard, is + kept with the units that reached it. None unless every shard had a cap: a cap over part of a launch is + not the launch's.""" + number = lambda v: isinstance(v, (int, float)) and not isinstance(v, bool) + if not budgets or not all(isinstance(x, dict) for x in budgets): + return None + whole = all(number(x.get("max_usd")) for x in budgets) + each = {x.get("unit_max_usd") for x in budgets} + each = each.pop() if len(each) == 1 else None + if not whole and not number(each): return None spent = [x.get("spent_usd") for x in budgets] unsettled = round(sum(x.get("unsettled_usd") or 0 for x in budgets), 6) - return {"max_usd": round(sum(x["max_usd"] for x in budgets), 6), + return {"max_usd": round(sum(x["max_usd"] for x in budgets), 6) if whole else None, "spent_usd": None if any(not isinstance(v, (int, float)) for v in spent) else round(sum(spent), 6), - **({"unsettled_usd": unsettled} if unsettled else {}), "stopped": any(x.get("stopped") for x in budgets), + **({"unsettled_usd": unsettled} if unsettled else {}), "stopped": whole and any(x.get("stopped") for x in budgets), + # how many shards there were and how many stopped: each at its own share, not the launch at the sum + **({"shards": len(budgets), "shards_stopped": sum(1 for x in budgets if x.get("stopped"))} if whole else {}), "cut_off": [u for x in budgets for u in x.get("cut_off") or []], - "not_run": [u for x in budgets for u in x.get("not_run") or []]} + "not_run": [u for x in budgets for u in x.get("not_run") or []], + **({"unit_max_usd": each, "over_unit": [u for x in budgets for u in x.get("over_unit") or []]} if number(each) else {})} def merge(job_dirs: List[str], out: str, force: bool = False) -> dict: diff --git a/app/nullius/standard/recipe.py b/app/nullius/standard/recipe.py index dd8167f..78bdd4b 100644 --- a/app/nullius/standard/recipe.py +++ b/app/nullius/standard/recipe.py @@ -273,6 +273,8 @@ def compile(recipe: dict, defs: Optional[Dict[str, List[dict]]] = None, *, the whole run: its jobs that call a model - each dataset with each participant, on each machine - take equal shares (``--max-usd``), so none spends another's and the amount typed is the most the run spends. + ``max_usd_per_unit``: each unit's own cap (``--max-usd-per-unit``), given + whole to every job that calls a model: a unit is a unit wherever it runs. ``unpriced``: asked, for a recipe with a cap, which of the models it names have no price in the user's list (``nullius.prices``): a run with a cap is refused without one, and the Limits step says so first.""" @@ -430,6 +432,16 @@ def compile(recipe: dict, defs: Optional[Dict[str, List[dict]]] = None, *, elif isinstance(cap, bool) or not isinstance(cap, (int, float)) or not 0 < cap < float("inf"): add("limits", "The spending cap is an amount in USD above zero, or empty for none.") cap = None + noun = (b.get("units") or {}).get("noun", "unit") + per_unit = recipe.get("max_usd_per_unit") + if per_unit in ("", None): + per_unit = None + elif isinstance(per_unit, bool) or not isinstance(per_unit, (int, float)) or not 0 < per_unit < float("inf"): + add("limits", f"The spending cap per {noun} is an amount in USD above zero, or empty for none.") + per_unit = None + elif per_unit < 1e-6: # what the command line can say: a millionth of a dollar + add("limits", f"The spending cap per {noun} is too small.") + per_unit = None limit = recipe.get("limit") if limit not in ("", None) and _int(limit, 1) is None: add("limits", "The number of units is a whole number of at least 1, or empty for all of them.") @@ -451,17 +463,20 @@ def compile(recipe: dict, defs: Optional[Dict[str, List[dict]]] = None, *, share = math.floor(cap / paying * 1e6) / 1e6 # never more than the cap in all if not share: add("limits", "The spending cap is too small" + (f" to share over {paying} jobs." if paying > 1 else ".")) - elif unpriced is not None: - # the models a run names as it sends them (run.Run.models_called): its own or a setup's, the judge's, - # the simulated user's, an option's embeddings - judge, sim = (b.get("score") or {}).get("judge"), b.get("simulator") - called = [*[(p["o"].get("sent") or p.get("model")) if p.get("setup") else model for p in picks if not p["control"]], - (judge_model or providers.served_as(judge["model"], provider)) if judge else "", - (sim_model or providers.served_as(sim["model"], provider)) if sim and any(not p["control"] for p in picks) else "", - *needs["embeddings"][:1]] - missing = unpriced([m for m in dict.fromkeys(called) if isinstance(m, str) and m]) - if missing: - add("limits", f"A spending cap needs a price for {', '.join(missing)}: update the price list in Settings, or set a price there.") + if not paying: + per_unit = None # no job calls a model: nothing to cap + if (share or per_unit) and unpriced is not None: + # the models each job names as it sends them, read where its run reads them (ledger.models): each + # dataset with each choice, the dataset's own embeddings + judge, sim = (b.get("score") or {}).get("judge"), b.get("simulator") + called = [m for x in dss for p in picks for m in ledger.models( + p["o"], None if p["control"] else (p["o"].get("sent") or p.get("model")) if p.get("setup") else model, + (judge_model or providers.served_as(judge["model"], provider)) if judge else None, + (sim_model or providers.served_as(sim["model"], provider)) if sim else None, + (definitions.option_needs(b, per_ds[x["id"]])["embeddings"] or [None])[0], control=p["control"])] + missing = unpriced([m for m in dict.fromkeys(called) if isinstance(m, str) and m]) + if missing: + add("limits", f"A spending cap needs a price for {', '.join(missing)}: update the price list in Settings, or set a price there.") if errors: return out nouns = (b.get("units") or {}).get("nouns", "units") @@ -495,6 +510,7 @@ def command_for(d: dict, p: dict, index: int) -> str: *[w for k, v in options_for_pick(d, p).items() for w in ("--bench-option", f"{k}={v}")], *(["--repeat", str(repeat)] if repeat > 1 else []), *(["--max-usd", f"{share:.6f}".rstrip("0").rstrip(".")] if share and spends(p) else []), + *(["--max-usd-per-unit", f"{per_unit:.6f}".rstrip("0").rstrip(".")] if per_unit and spends(p) else []), *(["--simulator-model", sim_model] if sim_model and not p["control"] else []), *(["--judge-model", judge_model] if judge_model else []), "--name", job_of(d, p, index), *[w for k, v in given.items() for w in ("--option", f"{k}={v}")], @@ -576,6 +592,7 @@ def row(p: dict) -> list: if any(p.get("setup") for p in picks) and not all(p.get("setup") for p in picks) else "")], *([[STEP["limits"], "Spending cap", f"${ledger.usd(cap)}" + (f" in all · ${ledger.usd(share)} for each of {paying} jobs" if paying > 1 else "")]] if share else []), + *([[STEP["limits"], f"Spending cap per {noun}", f"${ledger.usd(per_unit)}"]] if per_unit else []), [STEP["where"], "Where", (f"{of} shards, one per machine: " if of > 1 else "") + ", ".join(where_label(t) for t in targets) + f" · {backend} backend"]] if agents_local: diff --git a/app/nullius/standard/relay.py b/app/nullius/standard/relay.py index 9804324..4aabda2 100644 --- a/app/nullius/standard/relay.py +++ b/app/nullius/standard/relay.py @@ -36,6 +36,7 @@ import json import os import re +import socket import threading import time import urllib.parse @@ -337,7 +338,7 @@ 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, - ledger: Optional[str] = None): + ledger: Optional[str] = None, unit_caps: bool = True): """``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 @@ -358,6 +359,8 @@ def __init__(self, base: str, key: str, record_dir: str, prices: Optional[Dict[s and not sent when it does not fit.""" self.fields = fields self.ledger = ledgers.Ledger(ledger) if ledger else None + # False for the benchmark's own relay: its calls (an option's embeddings) are no one unit's, so only the run's cap holds them + self.unit_caps = unit_caps 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 ())} @@ -446,10 +449,16 @@ def _forward(self, h: BaseHTTPRequestHandler): req.setdefault("stream_options", {})["include_usage"] = True raw = json.dumps(req).encode() booking = booked = None + # a call that names its model is booked, whatever its route, but a token count, which costs nothing and + # reports no usage; one that names none (a cancellation, a body that is no request) has no price to book if self.ledger is not None and h.command == "POST": - booking, booked, why = self._book(req if isinstance(req, dict) else {}, raw, rest) + if isinstance(req, dict) and req.get("model") and not _free(rest): + booking, booked, why, by = self._book(req, raw, rest, tag) + else: # nothing to price: sent as it is, but not once the run's cap is reached + why, by = (("the run's spending cap was reached: no further call of this run is sent", "cap") + if self.ledger.closed() else (None, None)) if why: - self._refuse(h, tag, rest, req, asked, why) + self._refuse(h, tag, rest, req, asked, why, by) return t0 = time.time() conn, r, back, body = None, None, [], b"" @@ -459,6 +468,9 @@ def _forward(self, h: BaseHTTPRequestHandler): status, kind, back = r.status, r.getheader("Content-Type") or "application/json", returned(r.headers) except Exception as e: status, kind = 502, "application/json" + # refused or unknown address: never sent. Anything else (a timeout, a dropped line) came after the + # request may have gone out: it may have been billed, so its booking is kept + cut = not isinstance(e, (ConnectionRefusedError, socket.gaierror)) body = json.dumps({"error": {"message": "the relay could not reach the endpoint: " + self._said(e)}}).encode() streamed = r is not None and "text/event-stream" in kind whole = False # the reply read to its end: its connection can be kept @@ -543,34 +555,47 @@ 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.""" + def _book(self, req: dict, raw: bytes, rest: str, tag: str = "") -> tuple: + """(booking, its amount, None, None) for a call the cap lets out; + (None, its amount, why, by) for one it does not - ``by`` the run's + cap (``cap``), the cap of the unit ``tag`` routes for (``unit``), + ``price`` (a model with no price is not sent either, a capped run + sending no call it cannot price) or ``ledger`` (it could not be + read or written) - and those two are no cap reached.""" 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" + 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", "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.""" + # a call only the calls in flight keep out waits for them: their bookings are estimates, and settle + deadline = time.time() + ledgers.WAIT + while True: + booking, seen = self.ledger.reserve(lambda usual: ledgers.bound(req, raw, row, rest, usual), + ledgers.unit_of(tag) if self.unit_caps and tag else None, req.get("model"), + wait=time.time() < deadline) + if booking or seen["by"] != "wait": + break + time.sleep(0.2) + except ledgers.Unavailable as e: # not a cap reached: the unit is not the cap's, and the run goes on if the ledger reads again + return None, None, str(e), "ledger" + amount = seen["amount"] + return (booking, amount, None, None) if booking else (None, amount, ledgers.refusal(seen, amount), seen["by"]) + + def _refuse(self, h: BaseHTTPRequestHandler, tag: str, rest: str, req, asked, why: str, by: str = "cap") -> None: + """A call a capped run did not let out - at the run's cap (``by`` is + ``cap``), at its unit's own (``unit``), or for want of a price + (``price``): 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 {}, + "status": 402, "refused": by, "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", + body = json.dumps({"type": "error", "error": {"type": "spending_cap", "code": {"unit": "unit_spending_cap", "price": "unpriced_model", "ledger": "ledger_unavailable"}.get(by, "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)))): + for n, v in (("Content-Type", "application/json"), ("X-Nullius-Refused", by), ("Content-Length", str(len(body)))): h.send_header(n, v) h.end_headers() try: @@ -612,6 +637,11 @@ def stop(self): self.pool.close() +def _free(rest: str) -> bool: + """A route that names a model and is not billed: a token count.""" + return rest.split("?")[0].rstrip("/").endswith("/count_tokens") + + 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 @@ -622,6 +652,9 @@ def failure(calls: List[dict]) -> Optional[tuple]: cut = next((c for c in calls if c.get("refused") == "cap"), None) if cut: return ledgers.ERROR, cut.get("error") or "" + over = next((c for c in calls if c.get("refused") == "unit"), None) + if over: # its own cap: the unit's doing, and the run goes on + return ledgers.UNIT_ERROR, over.get("error") or "" if not calls or calls[-1].get("status") == 200: return None last = calls[-1] diff --git a/app/nullius/standard/run.py b/app/nullius/standard/run.py index 34340e6..4eb71d6 100644 --- a/app/nullius/standard/run.py +++ b/app/nullius/standard/run.py @@ -308,7 +308,8 @@ 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, max_usd: Optional[float] = None): + remove_images: bool = False, prices: Optional[dict] = None, max_usd: Optional[float] = None, + max_usd_per_unit: 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 +329,10 @@ 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_per_unit``: each unit's own spending cap, kept in the same + ledger: its own calls and its simulated user's count toward it, its + judge's do not. A unit that reaches it ends as ``ledger.UNIT_ERROR``, + an agent error counted against the agent, and the run goes on. ``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 @@ -407,20 +412,26 @@ 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: + self.max_usd, self.unit_max_usd, self.ledger = None, None, None + for flag, given, what in (("--max-usd", max_usd, "the run's spending cap"), ("--max-usd-per-unit", max_usd_per_unit, "a unit's spending cap")): + if given is None: + continue try: - cap = float(max_usd) + cap = float(given) 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") + raise Refused(f"{flag} {given!r}: {what} in USD, above zero") + if flag == "--max-usd": + self.max_usd = cap + else: + self.unit_max_usd = cap + if self.max_usd is not None or self.unit_max_usd is not None: 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 " @@ -498,9 +509,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: + if (self.max_usd is not None or self.unit_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) + self.ledger = ledgers.Ledger.create(os.path.join(self.job_dir, ".work", "ledger", "ledger.json"), self.max_usd, self.unit_max_usd) try: self.ready() except Refused as e: @@ -528,13 +539,15 @@ def start(self) -> str: self.n_units = len(work) # what the progress counts: every run of every unit self.feed = live_feed.Writer(os.path.join(self.job_dir, live_feed.FEED), self.name) self.feed.emit("job.start", n_total=len(work), n_concurrent=self.concurrency, agents=[self.other["key"]], tasks=units, - **({"budget": {"max_usd": self.max_usd}} if self.ledger is not None else {})) # the live board says the cap + **({"budget": {k: v for k, v in (("max_usd", self.max_usd), ("unit_max_usd", self.unit_max_usd)) if v is not None}} + if self.ledger is not None else {})) # the live board says the cap for pos, (i, u, r) in enumerate(work): self.feed.emit("trial.queued", self.trial_name(i, u, r), agent=self.other["key"], position=pos) results = [] u0 = time.time() self._halt, self._refused, self.halted = threading.Event(), [], None self._cut: List[str] = [] # units the spending cap cut off + self._over: List[str] = [] # units that reached their own cap: cut off there, the run going on # 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) @@ -580,16 +593,22 @@ def work_unit(iur): 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)) + # the units' stage says why it ended early: a line of its own would be a stage that never starts + p.emit("units", "done", seconds=round(time.time() - u0, 1), + **({"detail": f"stopped at the spending cap of {ledgers.usd(budget['max_usd'])} USD: " + f"{len(budget['cut_off'])} cut off, {len(budget['not_run'])} not run"} if budget and budget["stopped"] else {})) 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: + elif budget and budget["max_usd"] is not None: self.say(f"spent {ledgers.usd(budget['spent_usd'] or 0)} of the {ledgers.usd(budget['max_usd'])} USD cap") + elif budget: + self.say(f"spent {ledgers.usd(budget['spent_usd'] or 0)} USD") + if budget and budget.get("over_unit"): + self.say(f"{len(budget['over_unit'])} of the {self.bench['units']['nouns']} reached the cap of " + f"{ledgers.usd(budget['unit_max_usd'])} USD each and were cut off there") 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") @@ -660,7 +679,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, ledger=self.ledger_file(), + prices=self.prices, model=self.embed_model, ledger=self.ledger_file(), unit_caps=False, key_header=PROVIDERS[self.provider].get("key_header", "authorization")).start() def runtimes_needed(self) -> List[str]: @@ -797,12 +816,19 @@ def unit(self, i: int, unit: str, r: int = 1) -> Optional[dict]: 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) + # a call a 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 - never a miss - or by its own, whatever its agent made of the refusal + cuts = [f for f in (relay.failure(rl.calls(t)) for t in (tag, f"{tag}-judge", f"{tag}-simulator")) if f] + cut = next((f for f in cuts if f[0] == ledgers.ERROR), None) or next((f for f in cuts if f[0] == ledgers.UNIT_ERROR), None) if cut: exc, score = cut, None + if exc and exc[0] in ("EnvironmentSetupError", "VerifierError") and getattr(self, "ledger", None) is not None \ + and getattr(self, "bench_relay", None) is not None: + # its driver failed, at setup or at scoring, while the cap refused a call of the driver's own (an option's + # embeddings, on the benchmark's relay): the unit was cut off there. An agent's own error stays the agent's + late = [c for c in self.bench_relay.calls("embeddings") if c.get("refused") == "cap" and c.get("t0", 0) >= t0] + if late: + exc, score = (ledgers.ERROR, late[0].get("error") or ""), 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) @@ -831,14 +857,24 @@ def unit(self, i: int, unit: str, r: int = 1) -> Optional[dict]: with self._said: self._cut.append(unit) self._halt.set() # the cap is reached: nothing more is begun + elif exc and exc[0] == ledgers.UNIT_ERROR: + with self._said: + self._over.append(unit) # its own cap: the others go on + if getattr(self, "ledger", None) is not None and self.unit_max_usd is not None: + try: + self.ledger.forget(tag) # the unit ended: its own count goes from the ledger + except ledgers.Unavailable: + pass 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.""" + stopped the run, the units it cut off and those never begun. With a + cap per unit, that cap and the units that reached it + (``unit_max_usd``, ``over_unit``); ``max_usd`` is then None when the + run itself had none. None for a run without a cap.""" if getattr(self, "ledger", None) is None: return None try: @@ -846,10 +882,13 @@ def budget(self, not_run: List[str]) -> Optional[dict]: 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) + stopped = bool(got["closed"]) and self.max_usd is not None # only the run's cap stops a run 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 []} + **({"unsettled_usd": unsettled} if unsettled else {}), "stopped": stopped, + "cut_off": list(dict.fromkeys(cut)), "not_run": list(dict.fromkeys(not_run)) if stopped else [], + **({"unit_max_usd": self.unit_max_usd, "over_unit": list(dict.fromkeys(getattr(self, "_over", [])))} + if self.unit_max_usd is not None else {})} def note_refusal(self, exc: Optional[tuple]) -> None: """A unit's end, counted toward the endpoint refusing the run: a wrong @@ -973,9 +1012,8 @@ def models_called(self) -> List[str]: run's, the judge's, the simulator's, an option's embeddings and the agent's side models. Any other name a caller asks for is replaced by the run's (``relay``), so these are all a run can be priced for.""" - side = [(b or a) for a, _, b in (str(m).partition("=") for m in (self.other.get("model") or {}).get("side_models") or ())] - mine = [getattr(self, n, None) for n in ("model", "judge_model", "simulator_model", "embed_model")] - return [m for m in dict.fromkeys([*mine, *side]) if m] + return ledgers.models(self.other, *(getattr(self, n, None) for n in ("model", "judge_model", "simulator_model", "embed_model")), + control=self.other.get("_kind") == "control") def unit_relay(self, tag: str, trial: str, setup: dict, env_def: Optional[dict] = None, own: bool = False): """The unit's own relay, started: in this process on the local diff --git a/app/static/builder/command-builder-core.js b/app/static/builder/command-builder-core.js index 389dc20..aea571e 100644 --- a/app/static/builder/command-builder-core.js +++ b/app/static/builder/command-builder-core.js @@ -137,6 +137,8 @@ if (Number.isInteger(raw.repeat) && raw.repeat >= 1 && raw.repeat <= 32) r.repeat = raw.repeat; // the run's spending cap in USD, shared by its jobs (recipe.py): an amount above zero, or none if (typeof raw.max_usd === 'number' && Number.isFinite(raw.max_usd) && raw.max_usd > 0) r.max_usd = raw.max_usd; + // and each unit's own: a unit that reaches it is cut off there, the run going on + if (typeof raw.max_usd_per_unit === 'number' && Number.isFinite(raw.max_usd_per_unit) && raw.max_usd_per_unit > 0) r.max_usd_per_unit = raw.max_usd_per_unit; for (const key of ['judge_model', 'simulator_model']) if (typeof raw[key] === 'string' && raw[key].trim()) r[key] = raw[key].trim(); r.unsandboxed = raw.unsandboxed === true; if(raw.modelFormat === 'provider-id')r.modelFormat='provider-id'; diff --git a/app/static/builder/command-builder.js b/app/static/builder/command-builder.js index ddb03c8..ce03312 100644 --- a/app/static/builder/command-builder.js +++ b/app/static/builder/command-builder.js @@ -625,7 +625,8 @@ + field('Concurrent ' + u.nouns, 'concurrency', rec.concurrency, {type: 'number', min: 1, max: 32, placeholder: '1'}) + '' + field('Attempts per ' + u.noun, 'repeat', rec.repeat || '', {type: 'number', min: 1, max: 32, placeholder: '1'}) // the whole run's, shared by its jobs (recipe.py); a run that calls no model spends nothing: no cap to offer - + (C.standardKeyless(rec) ? '' : field('Spending cap (USD)', 'max_usd', rec.max_usd ?? '', {type: 'number', min: 0, placeholder: 'none'})); + + (C.standardKeyless(rec) ? '' : field('Spending cap (USD)', 'max_usd', rec.max_usd ?? '', {type: 'number', min: 0, placeholder: 'none'}) + + field('Spending cap per ' + u.noun + ' (USD)', 'max_usd_per_unit', rec.max_usd_per_unit ?? '', {type: 'number', min: 0, placeholder: 'none'})); } return ''; } diff --git a/app/static/runs/board.css b/app/static/runs/board.css index d41c48f..042b863 100644 --- a/app/static/runs/board.css +++ b/app/static/runs/board.css @@ -223,7 +223,7 @@ @container board (max-width: 1240px) { .rp .b-flags { flex-basis: 100%; border-left: 0; border-top: 1px solid var(--hair-in); padding-bottom: 10px; } .rp .board:not(.no-judge) { grid-template-areas: "rate out judge" "tot tot judge"; grid-template-rows: auto 1fr; } - .rp .board:not(.no-judge) .b-tot { display: grid; grid-template-columns: 6em max-content max-content; align-content: start; column-gap: 12px; padding: 0 18px 12px; } + .rp .board:not(.no-judge) .b-tot { display: grid; grid-template-columns: minmax(6em, max-content) max-content max-content; align-content: start; column-gap: 12px; padding: 0 18px 12px; } .rp .board:not(.no-judge) .b-tot .lab { grid-column: 1 / -1; height: 30px; font-weight: 400; font-size: 12px; color: var(--text-faint); } .rp .board:not(.no-judge) .b-tot .kv { display: contents; } .rp .board:not(.no-judge) .b-tot .kv > * { height: 28px; display: flex; align-items: center; } diff --git a/app/static/runs/live-board.js b/app/static/runs/live-board.js index 384ee65..59c47d1 100644 --- a/app/static/runs/live-board.js +++ b/app/static/runs/live-board.js @@ -87,13 +87,11 @@ } // measured, or left out: a rate the window could not measure, and a cost nobody reported, are absent, never 0 if (s.tokens_per_sec != null) bits.push(`${count(Math.round(s.tokens_per_sec))} tokens / s, last 30 s`); - // a run with a spending cap (job.start's budget) says it beside what was spent, or alone while nothing was - // the cap as it was given: cents, and up to two digits more when it has them (0.395), never rounded to a cent it is not + if (T.cost_usd != null) bits.push(`$${Number(T.cost_usd).toFixed(2)} spent`); + // a run's spending cap (job.start's budget), said on its own: it counts the judge's and the simulated user's calls + // too, which the agents' cost beside it does not. As it was given: cents, and up to two digits more (0.395) const capped = v => { const [whole, part = ''] = Number(v).toFixed(4).replace(/0+$/, '').split('.'); return whole + '.' + part.padEnd(2, '0'); }; - const cap = s.budget && Number(s.budget.max_usd) > 0 ? ` of the $${capped(s.budget.max_usd)} cap` : ''; - // under a cent, four decimals as Results shows it: beside a cap, $0.00 spent would read as nothing spent - if (T.cost_usd != null) bits.push(`$${Number(T.cost_usd).toFixed(cap && T.cost_usd < 0.01 ? 4 : 2)} spent${cap}`); - else if (cap) bits.push(`spending cap $${capped(s.budget.max_usd)}`); + if (s.budget && Number(s.budget.max_usd) > 0) bits.push(`spending cap $${capped(s.budget.max_usd)}`); if (s.wall_sec) bits.push(`wall ${clock(s.wall_sec)} per trial`); return bits.join(''); } diff --git a/app/templates/runs/job.html b/app/templates/runs/job.html index ec90fb2..61e596c 100644 --- a/app/templates/runs/job.html +++ b/app/templates/runs/job.html @@ -56,10 +56,10 @@ {% endif %} -{% if job.budget and job.budget.stopped %} +{% if job.budget and job.budget.stopped and job.budget.max_usd %} {# The run's own spending cap ended it (a Nullius standard run's --max-usd): nothing failed, so the notice is neutral. #}
- Stopped at its spending cap of {{ job.budget.max_usd|cap }}{% if job.budget.spent_usd is not none %}, with {{ job.budget.spent_usd|usd }} spent{% endif %}: + {% if job.budget.shards %}{{ job.budget.shards_stopped }} of {{ job.budget.shards }} shards stopped at their share of the spending cap{% else %}Stopped at its spending cap{% endif %} of {{ job.budget.max_usd|cap }}{% if job.budget.spent_usd is not none %}, with {{ job.budget.spent_usd|usd }} spent{% endif %}: {{ job.budget.cut_off|length }} cut off, {{ job.budget.not_run|length }} not run.
{% endif %} @@ -313,9 +313,12 @@
in · cached · out
input{{ '{:,}'.format(job.total_input_tokens) }}cached (part of input){{ '{:,}'.format(job.total_cache_tokens) }}output{{ '{:,}'.format(job.total_output_tokens) }}
Cost{{ job.total_cost_usd|usd(job.kind) }}
- {% if job.budget %}
Spending cap{{ job.budget.max_usd|cap }}
+ {% if job.budget and job.budget.max_usd %}
Spending cap{{ job.budget.max_usd|cap }}
{% if job.budget.not_run %}
{{ job.budget.not_run|length }} not run
{% for u in job.budget.not_run %}{{ u }}{% endfor %}
{% endif %}{% endif %} + {% if job.budget and job.budget.unit_max_usd %}
Cap per {{ job.runner_items[0] if job.runner_items else 'trial' }}{{ job.budget.unit_max_usd|cap }}
+ {% if job.budget.over_unit %}
{{ job.budget.over_unit|length }} reached it +
{% for u in job.budget.over_unit %}{{ u }}{% endfor %}
{% endif %}{% endif %} {% if job.mean_score is not none %}
Mean score{{ '%.2f'|format(job.mean_score) }}
{% endif %} {% for g in job.score_by_group %}
{{ job.parts_by|capitalize }} {{ g.group }}{{ '%.2f'|format(g.mean) }} {{ g.n }}
{% endfor %} {% if job.duration_min %}
Duration{{ job.duration_min|runtime }}
{% endif %} diff --git a/app/tests/test_classify.py b/app/tests/test_classify.py index 9bd7ca1..689daa3 100644 --- a/app/tests/test_classify.py +++ b/app/tests/test_classify.py @@ -106,6 +106,14 @@ def test_a_unit_the_runs_spending_cap_cut_off_has_its_own_outcome_outside_the_ra self.assertEqual((end["outcome"], end["exc"]), ("budget", ledger.ERROR)) self.assertIn("budget", live.OUTCOMES) + def test_a_unit_that_reached_its_own_spending_cap_counts_against_the_agent_on_both(self): + """It had its attempt and spent its allowance, as one that spends its time does: a result, in the rate.""" + from nullius.standard import job, ledger + self.assertEqual(C.from_verdict(ledger.UNIT_ERROR, None), C.AGENT_ERROR) + self.assertNotIn(C.AGENT_ERROR, C.INFRA_OUTCOMES) + self.assertNotEqual(ledger.UNIT_ERROR, ledger.ERROR) # recorded apart from the run's cap + self.assertEqual(job.feed_outcome("pass", None, (ledger.UNIT_ERROR, "its own cap"))["outcome"], "agent_error") + class HarborTaxonomy(unittest.TestCase): def test_provider_failures_are_infra(self): diff --git a/app/tests/test_standard_ledger.py b/app/tests/test_standard_ledger.py index cf293d0..7624642 100644 --- a/app/tests/test_standard_ledger.py +++ b/app/tests/test_standard_ledger.py @@ -20,22 +20,102 @@ 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): + def test_the_ledger_keeps_what_each_model_has_answered_with_on_average(self): + """The average, not the largest: one long answer does not raise what every later call is booked.""" led = ledger.Ledger.create(self.path, 1.0) - self.assertIsNone(led.largest("m")) # before any reply + self.assertIsNone(led.usual("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 + self.assertEqual((led.usual("m"), led.usual("other"), led.usual("never")), (210, 9000, None)) + for output in (100,) * 8 + (4000,): # eight short answers and a long one + booking, _ = led.reserve(0.02) + led.settle(booking, 0.01, "steady", output) + self.assertEqual(led.usual("steady"), 534) + with open(self.path) as f: # a file that lacks a part is no ledger: no call it guards is sent 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) + self.assertRaises(ledger.Unavailable, led.usual, "m") + self.assertRaises(ledger.Unavailable, led.reserve, 0.02) + self.assertTrue(led.closed()) + + def test_a_booking_is_made_from_what_the_models_replies_have_been_under_the_same_lock(self): + led = ledger.Ledger.create(self.path, 1.0, 0.5) + told = [] + a, state = led.reserve(lambda usual: told.append(usual) or 0.1, "u1", "m") + self.assertEqual((told, state["amount"]), ([None], 0.1)) # no reply yet + led.settle(a, 0.05, "m", 200) + b, state = led.reserve(lambda usual: usual / 1000, "u1", "m") # the average so far: 200 tokens + self.assertEqual(state["amount"], 0.2) + led.settle(b, 0.01, "m", 100) + led.forget("u1") # the unit ended: its own count goes, the run's stays + with open(self.path) as f: + doc = json.load(f) + self.assertEqual((doc["units"], round(doc["spent_usd"], 6)), ({}, 0.06)) + + def test_a_unit_that_reaches_its_own_cap_is_refused_from_then_on_and_the_run_goes_on(self): + led = ledger.Ledger.create(self.path, None, 0.10) # a cap per unit alone: none on the run + a, _ = led.reserve(0.06, "u1") + self.assertTrue(a) + refused, state = led.reserve(0.06, "u1") # 0.12 in flight would pass its 0.10 + self.assertEqual((refused, state["by"], state["unit"]["closed"]), (None, "unit", False)) + self.assertIn("this unit's spending cap of 0.10 USD", ledger.refusal(state, 0.06)) + again, state = led.reserve(0.001, "u1") # a small one after it: refused too + self.assertEqual((again, state["by"]), (None, "unit")) + self.assertIn("an earlier call of this unit did not fit its spending cap of 0.10 USD", ledger.refusal(state, 0.001)) + self.assertFalse(led.closed()) # the run is not stopped + self.assertTrue(led.reserve(0.06, "u2")[0]) # another unit has its own + self.assertTrue(led.reserve(5.0)[0]) # a call that is no unit's (its judge's) has none + led.settle(a, 0.02) + with open(self.path) as f: + doc = json.load(f) + self.assertEqual((doc["units"]["u1"]["spent_usd"], doc["units"]["u1"]["closed"], doc["of"].get(a)), (0.02, True, None)) + self.assertAlmostEqual(led.state()["spent_usd"], 0.02) + + def test_the_runs_cap_and_a_units_are_held_together_and_only_the_runs_stops_the_run(self): + led = ledger.Ledger.create(self.path, 0.20, 0.15) + self.assertEqual((led.reserve(0.16, "u1")[1]["by"], led.closed()), ("unit", False)) + a, _ = led.reserve(0.12, "u2") + refused, state = led.reserve(0.12, "u3") # fits its unit's 0.15, not the run's 0.20 + self.assertEqual((bool(a), refused, state["by"], led.closed()), (True, None, "cap", True)) + self.assertIn("the run's spending cap of 0.20 USD", ledger.refusal(state, 0.12)) + unsettled, _ = None, led.settle(a, None) # never known: kept on the unit's count too + with open(self.path) as f: + self.assertEqual(json.load(f)["units"]["u2"]["unsettled_usd"], 0.12) + + def test_a_units_calls_are_its_own_and_its_simulated_users_never_its_judges(self): + self.assertEqual([ledger.unit_of(t) for t in ("t-3", "t-3-simulator", "t-3-judge", "judge")], ["t-3", "t-3", None, "judge"]) + + def test_a_call_only_the_calls_in_flight_keep_out_is_told_to_wait_and_nothing_closes(self): + """Fifteen first calls booked before any reply must not end a run nothing was spent of: their bookings + are estimates, and settle.""" + led = ledger.Ledger.create(self.path, 1.0, 0.6) + a, _ = led.reserve(0.7) + again, state = led.reserve(0.5, wait=True) # 0.5 fits a cap of 1.0 once the 0.7 in flight settles + self.assertEqual((again, state["by"], led.closed()), (None, "wait", False)) + led.settle(a, 0.1) + self.assertTrue(led.reserve(0.5, wait=True)[0]) # and it does + never, state = led.reserve(0.5, wait=True) # 0.1 spent and 0.5 in flight: 0.5 more fits alone, so it waits + self.assertEqual((never, state["by"]), (None, "wait")) + big, state = led.reserve(0.95, wait=True) # with nothing in flight it would still not fit: refused, for good + self.assertEqual((big, state["by"], led.closed()), (None, "cap", True)) + # a unit's own calls in flight are waited for the same way, and what it could never fit closes it + units = ledger.Ledger.create(self.path, None, 0.6) + b, _ = units.reserve(0.4, "u1") + self.assertEqual(units.reserve(0.4, "u1", wait=True)[1]["by"], "wait") + self.assertEqual(units.reserve(0.7, "u1", wait=True)[1]["by"], "unit") + # without waiting (the wait is over), what does not fit now is refused + self.assertEqual(units.reserve(0.4, "u2")[0] is not None, True) + self.assertEqual(units.reserve(0.4, "u2")[1]["by"], "unit") + + def test_the_models_a_run_may_call_are_listed_once_for_the_run_and_the_builder(self): + agent = {"model": {"side_models": ["text-embedding-3-small", "fast=gpt-5-nano"]}} + self.assertEqual(ledger.models(agent, "m", "judge-1", "user-1", "embed-1"), + ["m", "judge-1", "user-1", "embed-1", "text-embedding-3-small", "gpt-5-nano"]) + self.assertEqual(ledger.models({}, None, "judge-1", "user-1", None, control=True), ["judge-1"]) # a control holds no conversation + self.assertEqual(ledger.models({}, "m", "m", None, None), ["m"]) def test_a_call_is_sent_while_spent_and_in_flight_and_its_booking_fit(self): led = ledger.Ledger.create(self.path, 1.0) @@ -134,9 +214,9 @@ def test_a_call_that_names_none_is_booked_the_models_own_limit_else_the_default( 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.""" + def test_a_call_in_flight_is_booked_what_the_runs_replies_have_been_never_past_a_limit(self): + """Its cost is known only when its reply comes back: until then it counts as the average reply the run + has had from that model (``seen``), 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 diff --git a/app/tests/test_standard_recipe.py b/app/tests/test_standard_recipe.py index d262c92..8e0aa3d 100644 --- a/app/tests/test_standard_recipe.py +++ b/app/tests/test_standard_recipe.py @@ -142,7 +142,7 @@ def twin(self): return dict(definitions.load("agent", ROOT)[0][0], key="twin", label="Twin agent") def caps(self, out): - return [c.split("--max-usd ")[1].split()[0] if "--max-usd" in c else None for sh in out["shards"] for c in sh["command"].split("\n")] + return [c.split("--max-usd ")[1].split()[0] if "--max-usd " in c else None for sh in out["shards"] for c in sh["command"].split("\n")] def test_one_job_takes_the_whole_cap_and_no_cap_adds_nothing(self): out = recipe.compile(a_recipe(max_usd=2), defs()) @@ -174,6 +174,34 @@ def test_a_job_that_calls_no_model_takes_no_share(self): self.assertEqual((alone["errors"], self.caps(alone)), ([], [None])) self.assertNotIn("Spending cap", [row[1] for row in alone["review"]]) + def test_a_cap_per_unit_goes_whole_to_every_job_that_calls_a_model(self): + out = recipe.compile(a_recipe(others=["fake", "twin"], where=["local", "me@bench"], max_usd=1, max_usd_per_unit=0.05), defs(agent=[self.twin()])) + self.assertEqual(out["errors"], []) + self.assertEqual(self.caps(out), ["0.25"] * 4) # the run's is shared + each = [c.split("--max-usd-per-unit ")[1].split()[0] for sh in out["shards"] for c in sh["command"].split("\n")] + self.assertEqual(each, ["0.05"] * 4) # a unit's is not: a unit is a unit wherever it runs + self.assertIn([recipe.STEP["limits"], "Spending cap per task", "$0.05"], out["review"]) + alone = recipe.compile(a_recipe(max_usd_per_unit=0.05), defs(), unpriced=lambda models: list(models)) + self.assertEqual([e["message"] for e in alone["errors"]], # alone it needs prices too + ["A spending cap needs a price for gpt-5-mini: update the price list in Settings, or set a price there."]) + plain = recipe.compile(a_recipe(max_usd_per_unit=0.05), defs()) + self.assertEqual((plain["errors"], self.caps(plain)), ([], [None])) + self.assertIn("-j 2 --max-usd-per-unit 0.05 --name", plain["script"]) + self.assertNotIn("Spending cap", [row[1] for row in plain["review"]]) + tiny = recipe.compile(a_recipe(max_usd_per_unit=1e-7), defs())["errors"] # less than the command line can say + self.assertEqual([e["message"] for e in tiny], ["The spending cap per task is too small."]) + bad = recipe.compile(a_recipe(max_usd_per_unit=0), defs())["errors"] + self.assertEqual([(e["step"], e["message"]) for e in bad], + [(recipe.STEP["limits"], "The spending cap per task is an amount in USD above zero, or empty for none.")]) + + def test_the_price_question_names_every_model_the_run_will_call(self): + """As ``run.Run.models_called`` lists them: an agent's side models too, each as it is sent.""" + base = self.twin() + memo = dict(base, model={**(base.get("model") or {}), "side_models": ["text-embedding-3-small", "fast=gpt-5-nano"]}) + asked = [] + out = recipe.compile(a_recipe(others=["twin"], max_usd=1), defs(agent=[memo]), unpriced=lambda models: asked.append(list(models)) or []) + self.assertEqual((out["errors"], asked), ([], [["gpt-5-mini", "text-embedding-3-small", "gpt-5-nano"]])) + def test_a_cap_that_is_no_amount_names_limits(self): for bad in (0, -1, "2", True, float("inf"), float("nan"), [2]): with self.subTest(bad): diff --git a/app/tests/test_standard_relay.py b/app/tests/test_standard_relay.py index a6f5e7e..63b6362 100644 --- a/app/tests/test_standard_relay.py +++ b/app/tests/test_standard_relay.py @@ -10,6 +10,7 @@ import tempfile import threading import time +from unittest import mock import unittest import urllib.error import urllib.request @@ -232,13 +233,97 @@ def test_a_call_is_booked_at_what_the_runs_replies_have_been_once_one_came_back( 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(led.usual("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_units_own_cap_refuses_its_calls_and_its_simulated_users_not_its_judges_nor_another_units(self): + tmp = tempfile.mkdtemp() + led = ledger.Ledger.create(os.path.join(tmp, "ledger", "ledger.json"), None, 0.0005) + r = relay.Relay(self.base, "sk-real", os.path.join(tmp, "relay"), prices=self.PRICES, ledger=led.path).start() + self.addCleanup(r.stop) + body = {"model": "m", "messages": [{"role": "user", "content": "hi"}], "max_tokens": 100} # booked 0.001 and more: over 0.0005 + sent = len(Upstream.seen) + status, headers, answer = self.post(r, "u1", body) + self.assertEqual((status, headers.get("X-Nullius-Refused"), json.loads(answer)["error"]["code"]), (402, "unit", "unit_spending_cap")) + self.assertEqual(len(Upstream.seen), sent) # never sent + self.assertEqual(r.calls("u1")[-1]["refused"], "unit") + exc, why = relay.failure(r.calls("u1")) + self.assertEqual(exc, ledger.UNIT_ERROR) + self.assertIn("this unit's spending cap of 0.0005 USD", why) + self.assertEqual(classify.from_verdict(exc, None), classify.AGENT_ERROR) # it spent its allowance: counted against the agent + self.assertNotIn(classify.AGENT_ERROR, classify.INFRA_OUTCOMES) + self.assertEqual(self.post(r, "u1-judge", body)[0], 200) # the scorer is no part of the unit's task + self.assertEqual(self.post(r, "u1-simulator", body)[0], 402) # its simulated user is + self.assertEqual(self.post(r, "u2", body)[0], 200) # another unit: booked what a reply has been, within its own cap + self.assertFalse(led.closed()) + # the benchmark's own relay books for no unit: only the run's cap holds its calls + shared = relay.Relay(self.base, "sk-real", os.path.join(tmp, "shared"), prices=self.PRICES, ledger=led.path, unit_caps=False).start() + self.addCleanup(shared.stop) + self.assertEqual(self.post(shared, "u1", body)[0], 200) + + def test_only_a_call_a_model_answers_is_booked_and_a_missing_price_is_no_cap_reached(self): + r, led = self.capped(1.0) + # a token count, and a body that names no model, cost nothing and report no usage: never booked, so never + # kept against the cap as a call whose cost was not known + for path, body in (("/messages/count_tokens", {"model": "m", "messages": [{"role": "user", "content": "x" * 6000}]}), + ("/responses/abc/cancel", {})): + req = urllib.request.Request(r.url("u1") + path, data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}) + try: + urllib.request.urlopen(req).read() + except urllib.error.HTTPError as e: + e.read() + self.assertEqual(len(r.calls("u1")), 2) + self.assertFalse(any("booked_usd" in c or c.get("refused") for c in r.calls("u1"))) + self.assertEqual((led.state()["held_usd"], led.state()["unsettled_usd"], led.state()["spent_usd"]), (0, 0.0, 0.0)) + # a model with no price is not sent, and that is no cap reached: the unit is not the cap's, and the run goes on + status, headers, answer = self.post(r, "u2", {"model": "another", "messages": [{"role": "user", "content": "hi"}]}) + self.assertEqual((status, headers.get("X-Nullius-Refused"), json.loads(answer)["error"]["code"]), (402, "price", "unpriced_model")) + self.assertEqual(r.calls("u2")[-1]["refused"], "price") + self.assertEqual(relay.failure(r.calls("u2"))[0], "ApiError") + self.assertFalse(led.closed()) + + def test_a_call_waits_for_the_calls_in_flight_and_is_refused_only_when_they_do_not_settle(self): + r, led = self.capped(0.0015) + body = {"model": "m", "messages": [{"role": "user", "content": "hi"}], "max_tokens": 100} # booked a little over 0.001 + held, _ = led.reserve(0.001) # another call, in flight + got = {} + waiting = threading.Thread(target=lambda: got.update(status=self.post(r, "u1", body)[0])) + waiting.start() + time.sleep(0.6) + self.assertEqual((got, led.closed()), ({}, False)) # not refused, not sent: waiting + led.settle(held, 0.0001) # the call in flight comes back, far under its booking + waiting.join(10) + self.assertEqual((got.get("status"), led.closed()), (200, False)) + # one that waits in vain is refused, and the run stops there + stuck, _ = led.reserve(led.state()["left_usd"] - 0.00001) # a call in flight that never comes back, leaving next to nothing + with mock.patch.object(ledger, "WAIT", 0.5): + status, headers, _ = self.post(r, "u2", body) + self.assertEqual((status, headers.get("X-Nullius-Refused"), led.closed()), (402, "cap", True)) + # once the cap is reached, a call with nothing to price is not sent either + req = urllib.request.Request(r.url("u3") + "/responses/abc/cancel", data=b"{}", headers={"Content-Type": "application/json"}) + with self.assertRaises(urllib.error.HTTPError) as refused: + urllib.request.urlopen(req) + self.assertEqual((refused.exception.code, refused.exception.headers.get("X-Nullius-Refused")), (402, "cap")) + refused.exception.read() + + def test_a_call_that_names_a_model_is_booked_on_any_route_and_a_ledger_that_fails_is_no_cap_reached(self): + r, led = self.capped(1.0) + req = urllib.request.Request(r.url("u1") + "/images/generations", data=json.dumps({"model": "m", "prompt": "a cat"}).encode(), + headers={"Content-Type": "application/json"}) + try: + urllib.request.urlopen(req).read() + except urllib.error.HTTPError as e: + e.read() + self.assertIn("booked_usd", r.calls("u1")[-1]) # a route the relay does not know is booked all the same + os.unlink(led.path) # the ledger gone: the call is not sent, and that is no cap + status, headers, answer = self.post(r, "u2", {"model": "m", "messages": [{"role": "user", "content": "hi"}]}) + self.assertEqual((status, headers.get("X-Nullius-Refused"), json.loads(answer)["error"]["code"]), (402, "ledger", "ledger_unavailable")) + self.assertEqual(relay.failure(r.calls("u2"))[0], "ApiError") + 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) diff --git a/app/tests/test_standard_run.py b/app/tests/test_standard_run.py index 93b552d..38302cf 100644 --- a/app/tests/test_standard_run.py +++ b/app/tests/test_standard_run.py @@ -62,6 +62,10 @@ def setup(unit, dir): ''' +# the same, one unit asking far more than the others: its prompt is three thousand characters +UNEVEN_DRIVER = CALLING_DRIVER.replace('{"prompt": "model hi"}', '{"prompt": "model " + ("x" * 3000 if unit == "m5" else "hi")}') + + class Echo(BaseHTTPRequestHandler): """A model that says back the last message, and counts its tokens.""" status = 200 @@ -292,7 +296,7 @@ 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): + def calling(self, driver=CALLING_DRIVER, **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 @@ -300,7 +304,7 @@ def calling(self, **kw): 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) + f.write(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} @@ -309,6 +313,28 @@ def calling(self, **kw): with open(os.path.join(job_dir, "result.json")) as f: return job_dir, failed, json.load(f), said + def test_a_cap_per_unit_cuts_off_the_unit_that_reaches_it_and_the_run_goes_on(self): + """Input at 1 USD a million here: seven units book about 0.0001 a call, one a thousand tokens of prompt, + 0.001 - over its cap of 0.0005. It alone is cut off, against its agent; the seven others run.""" + prices = {"m-1": {"input": 1.0, "output": 10.0, "max_output": 10}} + job_dir, failed, result, said = self.calling(UNEVEN_DRIVER, prices=prices, max_usd_per_unit=0.0005) + self.assertEqual(result["budget"], {"max_usd": None, "spent_usd": 0.000287, "stopped": False, "cut_off": [], "not_run": [], + "unit_max_usd": 0.0005, "over_unit": ["m5"]}) + self.assertFalse(failed) + got, j = self.outcomes(job_dir) + self.assertEqual(sorted(got), [f"m{i}" for i in range(8)]) # every unit was begun: the run went on + self.assertEqual((got["m5"].exception_type, got["m5"].outcome), (ledger.UNIT_ERROR, classify.AGENT_ERROR)) + self.assertTrue(all(got[u].outcome == classify.RESOLVED for u in got if u != "m5")) + self.assertEqual(j.budget["over_unit"], ["m5"]) + calls = self.relay_calls(job_dir) + self.assertEqual(([c.get("refused") for c in calls].count("unit"), len([c for c in calls if not c.get("refused")])), (1, 7)) + self.assertTrue(any(s == "1 of the tasks reached the cap of 0.0005 USD each and were cut off there" for s in said), said) + self.assertFalse(any(s.startswith("stopped:") for s in said), said) + with open(os.path.join(job_dir, "events.jsonl")) as f: + board = live.fold([live.parse_line(line) for line in f if line.strip()]) + self.assertEqual(board.budget, {"unit_max_usd": 0.0005}) + self.assertEqual({t.trial.split("__")[0]: t.outcome for t in board.trials if t.outcome == "agent_error"}, {"m5": "agent_error"}) + 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) diff --git a/app/tests/test_standard_shard.py b/app/tests/test_standard_shard.py index 526e4a1..780b4c2 100644 --- a/app/tests/test_standard_shard.py +++ b/app/tests/test_standard_shard.py @@ -125,10 +125,11 @@ def test_the_merged_job_carries_the_shards_spending_caps_as_one(self): 3: {"max_usd": 0.5, "spent_usd": 0.1, "stopped": False, "cut_off": [], "not_run": []}} dirs = [write_shard(self.tmp, f"run-shard-{i}of3", self.split[i], i, 3, budget=caps[i]) for i in (1, 2, 3)] got = merge.merge(dirs, os.path.join(self.tmp, "capped", "run")) - want = {"max_usd": 1.5, "spent_usd": 0.71, "unsettled_usd": 0.03, "stopped": True, "cut_off": ["t-9"], "not_run": ["t-11", "t-12"]} + want = {"max_usd": 1.5, "spent_usd": 0.71, "unsettled_usd": 0.03, "stopped": True, "cut_off": ["t-9"], "not_run": ["t-11", "t-12"], + "shards": 3, "shards_stopped": 1} # one shard stopped at its share: not the launch at 1.5 self.assertEqual(read(os.path.join(got["out"], "result.json"))["budget"], want) - self.assertEqual(parse.parse_job(got["out"]).budget, want) - self.assertGreaterEqual(int(parse.SCHEMA), 14) # a job cached before the field is read again + self.assertEqual(parse.parse_job(got["out"]).budget, dict(want, unit_max_usd=None, over_unit=[])) + self.assertGreaterEqual(int(parse.SCHEMA), 15) # a job cached before the field is read again # a cap over part of a launch is not the launch's: nothing is claimed shutil.rmtree(dirs[2]) dirs[2] = write_shard(self.tmp, "run-shard-3of3", self.split[3], 3, 3) @@ -140,7 +141,23 @@ def test_a_recorded_cap_is_read_leniently_and_an_amount_missing_is_not_zero(self self.assertEqual(parse._budget(None), {}) self.assertEqual(parse._budget({"max_usd": "2", "spent_usd": 1}), {}) # no cap named: no cap self.assertEqual(parse._budget({"max_usd": 2, "stopped": "yes", "cut_off": "t-1"}), - {"max_usd": 2, "spent_usd": None, "unsettled_usd": None, "stopped": False, "cut_off": [], "not_run": []}) + {"max_usd": 2, "spent_usd": None, "unsettled_usd": None, "stopped": False, "cut_off": [], "not_run": [], + "unit_max_usd": None, "over_unit": [], "shards": 0, "shards_stopped": 0}) + self.assertFalse(parse._budget({"unit_max_usd": 0.5, "stopped": True})["stopped"]) # only the run's cap stops a run + self.assertEqual(parse._budget({"unit_max_usd": 0.5, "over_unit": ["t-2"]})["over_unit"], ["t-2"]) # a cap per unit alone is a cap + + def test_a_launch_whose_shards_did_not_all_have_the_runs_cap_claims_no_stop(self): + some = lambda i: {"max_usd": 0.5 if i < 3 else None, "spent_usd": 0.1, "stopped": i == 1, "cut_off": [], "not_run": [], "unit_max_usd": 0.05, "over_unit": []} + dirs = [write_shard(self.tmp, f"run-shard-{i}of3", self.split[i], i, 3, budget=some(i)) for i in (1, 2, 3)] + got = read(os.path.join(merge.merge(dirs, os.path.join(self.tmp, "mixed", "run"))["out"], "result.json"))["budget"] + self.assertEqual((got["max_usd"], got["stopped"], got["unit_max_usd"], "shards" in got), (None, False, 0.05, False)) + + def test_the_merged_job_keeps_a_cap_per_unit_and_the_units_that_reached_it(self): + each = lambda over: {"max_usd": None, "spent_usd": 0.1, "stopped": False, "cut_off": [], "not_run": [], "unit_max_usd": 0.05, "over_unit": over} + dirs = [write_shard(self.tmp, f"run-shard-{i}of3", self.split[i], i, 3, budget=each(over)) for i, over in ((1, ["t-0"]), (2, []), (3, ["t-7"]))] + got = merge.merge(dirs, os.path.join(self.tmp, "units", "run")) + self.assertEqual(read(os.path.join(got["out"], "result.json"))["budget"], + {"max_usd": None, "spent_usd": 0.3, "stopped": False, "cut_off": [], "not_run": [], "unit_max_usd": 0.05, "over_unit": ["t-0", "t-7"]}) def test_shards_of_different_launches_models_or_versions_are_refused_unless_forced(self): a = write_shard(self.tmp, "a", self.split[1], 1, 3) diff --git a/app/ui_tests/test_capped_job.py b/app/ui_tests/test_capped_job.py index ea8b407..a18d26f 100644 --- a/app/ui_tests/test_capped_job.py +++ b/app/ui_tests/test_capped_job.py @@ -22,7 +22,7 @@ AGENT = {"key": "fake", "_kind": "agent", "_ref": "fake"} -def capped_job(root: Path, name: str, ran: int, planned: int, budget=None, cut=()): +def capped_job(root: Path, name: str, ran: int, planned: int, budget=None, cut=(), cut_by=None): """A standard job of ``planned`` units of which the first ``ran`` have a trial; those in ``cut`` ended at the cap.""" d = root / name d.mkdir(parents=True) @@ -33,7 +33,7 @@ def capped_job(root: Path, name: str, ran: int, planned: int, budget=None, cut=( for k, u in enumerate(units[:ran]): trial = d / f"{u}__u{k:04d}" trial.mkdir() - exc = (ledger.ERROR, "the run's spending cap") if u in cut else None + exc = (ledger.ERROR, "the run's spending cap") if u in cut else ((cut_by or {})[u], "its own spending cap") if u in (cut_by or {}) else None results.append(job.write_trial(str(trial), job=name, bench=BENCH, other=AGENT, model="m-1", unit=u, t0=1_760_000_000.0 + k, t1=1_760_000_005.0 + k, produce=(1_760_000_000.0 + k, 1_760_000_004.0 + k), score=None if exc else {"reward": 1.0}, exception=exc, calls=[], trajectory=None)) @@ -50,6 +50,12 @@ def setUpClass(cls): "not_run": ["t-6", "t-7", "t-8", "t-9"]}) capped_job(tmp / "jobs", "within", 4, 4, budget={"max_usd": 0.395, "spent_usd": 0.004, "stopped": False, "cut_off": [], "not_run": []}) capped_job(tmp / "jobs", "uncapped", 4, 4) + capped_job(tmp / "jobs", "merged", 3, 4, budget={"max_usd": 10.0, "spent_usd": 4.0, "stopped": True, "cut_off": [], "not_run": ["t-3"], + "shards": 4, "shards_stopped": 1}) + capped_job(tmp / "jobs", "unit-only", 4, 4, budget={"max_usd": None, "spent_usd": None, "stopped": True, "cut_off": [], "not_run": [], + "unit_max_usd": 0.002, "over_unit": []}) + capped_job(tmp / "jobs", "per-unit", 4, 4, cut_by={"t-2": ledger.UNIT_ERROR}, + budget={"max_usd": None, "spent_usd": 0.004, "stopped": False, "cut_off": [], "not_run": [], "unit_max_usd": 0.002, "over_unit": ["t-2"]}) cls.stack = ExitStack() cls.stack.enter_context(patch.object(web.shared, "WS_HOME", str(tmp / "ws"))) cls.stack.enter_context(patch.dict(web.app.config, TESTING=True)) @@ -97,12 +103,36 @@ def test_a_run_within_its_cap_says_the_cap_and_no_more_and_one_without_says_noth self.assertNotIn(absent, plain) + def test_a_run_with_a_cap_per_unit_says_it_and_the_units_that_reached_it_count_against_the_agent(self): + html = self.page("per-unit") + self.assertEqual(self.text(html, r'
'), "Cap per task $0.002") + self.assertIn("counts against the agent", unescape(html)) + self.assertIn("1 reached it", html) + self.assertEqual(self.text(html, r'
'), "t-2") + for absent in ('id="run-cap"', 'id="cap-stopped"', "oc-spending_cap"): # the run had no cap of its own, and was not stopped + self.assertNotIn(absent, html) + self.assertRegex(html, r'class="[^"]*oc-agent_error[^"]*"[^>]*data-o="agent_error"') + self.assertIn(ledger.UNIT_ERROR, html) # the recorded type beside the label + + + def test_a_merged_job_says_how_many_shards_stopped_and_a_run_without_its_own_cap_says_no_stop(self): + html = self.page("merged") + self.assertEqual(self.text(html, r'
.*?
'), + "1 of 4 shards stopped at their share of the spending cap of $10.00, with $4.00 spent: 0 cut off, 1 not run.") + alone = self.page("unit-only") # recorded stopped with no cap of the run's: nothing to say, no error + self.assertNotIn('id="cap-stopped"', alone) + self.assertEqual(self.text(alone, r'
'), "Cap per task $0.002") + + class BuilderKeepsTheCap(CompilerCase): def test_a_saved_recipe_keeps_an_amount_above_zero_and_nothing_else(self): out = self.js('const back=v=>C.fromJSON(JSON.stringify({...C.createStandard("gsm8k"),max_usd:v})).max_usd;' 'process.stdout.write(JSON.stringify([back(2),back(0.25),back(0),back(-1),back("2"),back(true),back(null),' '"max_usd" in C.createStandard("gsm8k")].map(v=>v===undefined?"none":v)));') self.assertEqual(out, [2, 0.25, "none", "none", "none", "none", "none", False]) + each = self.js('const back=v=>C.fromJSON(JSON.stringify({...C.createStandard("gsm8k"),max_usd_per_unit:v})).max_usd_per_unit;' + 'process.stdout.write(JSON.stringify([back(0.01),back(0),back("1")].map(v=>v===undefined?"none":v)));') + self.assertEqual(each, [0.01, "none", "none"]) def test_a_harbor_recipe_takes_no_cap(self): """Harbor's calls do not pass the relay, so nothing could hold them to one: the field is the standard's alone.""" @@ -110,7 +140,7 @@ def test_a_harbor_recipe_takes_no_cap(self): + 'process.stdout.write(JSON.stringify(["max_usd" in r,/max.usd/i.test(JSON.stringify(C.compile(r)))]));') self.assertEqual(out, [False, False]) page = (APP / "static" / "builder" / "command-builder.js").read_text(encoding="utf-8") - self.assertEqual(page.count("'max_usd'"), 1) # offered once, on the standard's Limits step + self.assertEqual((page.count("'max_usd'"), page.count("'max_usd_per_unit'")), (1, 1)) # offered once each, on the standard's Limits step self.assertIn("C.standardKeyless(rec) ? '' : field('Spending cap (USD)', 'max_usd'", page) diff --git a/app/web/execute.py b/app/web/execute.py index faadcbd..58af977 100644 --- a/app/web/execute.py +++ b/app/web/execute.py @@ -4,6 +4,7 @@ import json import os import re +import time from flask import abort, jsonify, request, url_for from nullius import execution, hosts, prerequisites, prices, providers from nullius.harbor import source as harbor_source @@ -141,11 +142,15 @@ def _start_standard(recipe: dict): **started[0].summary()) +_PRICED = {"at": float("-inf"), "rows": {}} + + def unpriced(models: list) -> list: """Which of ``models`` have no price in the user's list (``nullius.prices``): what a recipe with a spending cap is told at its Limits step, since a run with a cap is refused without a price for each model it calls.""" - rows = prices.table() - return [m for m in models if m not in rows] + if time.monotonic() - _PRICED["at"] > 5: # the builder asks at every keystroke: the list is read at most every five seconds + _PRICED.update(at=time.monotonic(), rows=prices.table()) + return [m for m in models if m not in _PRICED["rows"]] def docker_problem(): diff --git a/docs/builder.md b/docs/builder.md index dd49d73..0491265 100644 --- a/docs/builder.md +++ b/docs/builder.md @@ -27,7 +27,7 @@ Then [read its results](results.md). | Dataset | One or more; each dataset with each agent is a job of its own. Each shows its proofs and the licences its units may need | | Agent (*Memory* on a memory benchmark) | Reference agents first, or **No agent · Original** for a direct-model benchmark. Memory benchmarks offer **Original memory** alongside compatible memory capabilities. Controls follow separately; each choice has its own settings tab | | Model | **Like the paper** keeps what the paper fixed: the model picker shows only documented paper models when the benchmark lists them (the filter applies to the selected agent or capability too), and the user simulator and judge are their official models. A benchmark that lists no paper models but has an official simulator or judge leaves the model to you. **Like the leaderboard**, on a benchmark that declares its leaderboard's rules, leaves the model to you and fixes what the leaderboard does: its user simulator and judge, the options it sets and at least its attempts per unit. **Any of your choice** opens the provider's full list and lets the simulator and judge be changed. Nothing fixed by the paper: the paper choice is disabled | -| Limits | Task limit (empty is every task), a task-name glob, concurrency, retries, attempts per unit, timeouts in minutes. On the Nullius runner, a spending cap in USD for the whole run: its jobs that call a model share it in equal parts ([A spending cap](cli/run.md#a-spending-cap)) | +| Limits | Task limit (empty is every task), a task-name glob, concurrency, retries, attempts per unit, timeouts in minutes. On the Nullius runner, a [spending cap](cli/run.md#a-spending-cap) for the run and one per unit | | Where | This machine, a Settings host, or several machines, one shard each, merged into one job. The backend: *local processes*, as you, with the explicit yes to running unsandboxed; or *Docker*. A benchmark option's machine setup is installed on a click (`sudo -n`), or its commands are given | | Review | One row per decision and the prerequisites. **Run** | diff --git a/docs/cli/run.md b/docs/cli/run.md index 84c8eaa..a7ec0cf 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]]... [--max-usd USD] \ + [--price MODEL=INPUT,OUTPUT[,CACHED]]... [--max-usd USD] [--max-usd-per-unit USD] \ [--repeat K] [--remove-images] [--judge-model M] [--simulator-model M] [--name NAME] [--jobs DIR] [--provider P] ``` @@ -29,6 +29,7 @@ nullius standard run locomo model --model gpt-4o-mini | `--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)) | +| `--max-usd-per-unit USD` | Each unit's own spending cap: a unit that reaches it is cut off there, and the run goes on ([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 | @@ -49,24 +50,20 @@ nullius standard run locomo model --model gpt-4o-mini ## 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`). +`--max-usd 5` stops a run at 5 USD and `--max-usd-per-unit 0.50` one unit at 0.50, for a +prepaid or shared key (`app/nullius/standard/ledger.py`). Neither is set by default. | 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 | On the command line the cap is that shard's alone. `merge` adds the shards' caps and spending into one `budget`, when every shard had a cap | -| From the builder | *Limits* takes one cap for the whole run. Its jobs that call a model (each dataset with each participant, on each machine) take equal shares, rounded down, so the amount typed is the most the run spends. A model without a price is named there, before Run. Not offered for a Harbor recipe, whose calls do not pass the relay | -| The job page | Run totals shows the cap. A run the cap stopped says so, with what was spent, and lists the units not run; the units cut off are its Spending cap trials. The live board says what was spent of the cap | +| Prices first | Every model the run calls needs a price ([price list](../settings/model-prices.md) or `--price`), else the run is refused | +| What is counted | Each reply's cost; a call in flight as the average reply so far from its model, 4,096 output tokens before any | +| At the run's cap | A call waits while only calls in flight are in its way. The one that does not fit is refused by the relay (HTTP 402), and every call after it | +| Its units | One cut off is the outcome Spending cap, outside the rate; those not begun are not run | +| At a unit's cap | Its own calls and its simulated user's count, its judge's do not. The unit is an Agent error; the run goes on | +| Not exact | A reply longer than the average takes the spend past a cap by the difference, for each call in flight | +| The result | `result.json` carries `budget`, and the job page shows it | +| From the builder | *Limits*: one cap shared in equal parts by the run's jobs, and one per unit. Not for a Harbor recipe | +| Shards | A shard's cap is its own on the command line; `merge` adds them up |
diff --git a/docs/cli/shard-and-merge.md b/docs/cli/shard-and-merge.md index 1720df2..ab07d72 100644 --- a/docs/cli/shard-and-merge.md +++ b/docs/cli/shard-and-merge.md @@ -15,7 +15,7 @@ nullius standard merge jobs_standard/*__shard-* -o jobs_standard/humaneval-oracl | A unit keeps its place | `u0042` is the 43rd unit of the dataset in every shard; shard jobs never share a trial name | | One launch | Each shard's `config.json` records `standard.shard` (`index`, `of`, `units_total`, `partition`, `planned`, `limit`) and `standard.launch`: `--launch`, or the day (UTC), benchmark, agent or capability and model. The runs page groups them until merged | | `merge JOB_DIR... -o OUT` | One job (`app/nullius/standard/merge.py`): trials hard-linked, each `result.json` naming the merged job and shard; the job's `result.json` recomputed; `config.json` with `merged_from` for `shard`. Different launches, versions, datasets, agents, models, backends or N refused unless `--force`; two trials of one unit always. Units no shard ran are named | -| A spending cap | `--max-usd` is each shard's own; the builder shares a recipe's cap over its shards in equal parts. `merge` adds the caps and what was spent into the merged job's `budget`, with every unit cut off or not run, when every shard had a cap | +| A spending cap | Each shard's own; the builder shares a recipe's cap over its shards. `merge` adds them into one `budget` when every shard had one | | `prove --shard I/N` | Controls and leak on that shard, units on all; recorded as `shard` | The builder's *Where* does the same: one shard per machine. diff --git a/docs/field-guide.md b/docs/field-guide.md index 698a6ab..fb1eb56 100644 --- a/docs/field-guide.md +++ b/docs/field-guide.md @@ -139,7 +139,7 @@ How the agent worked, not whether it won. Computed over layer 3. | `total_output_tokens` | Completion tokens. | derived | any | | `total_cost_usd` | The trials' recorded costs, summed: an estimate, not a bill. Shows — when no trial recorded one. | derived | any | | `duration_min` | The whole run, start to finish. | derived | any | -| `budget` | A standard run's spending cap, from its `result.json`: the cap and what its ledger counted as spent (`max_usd`, `spent_usd`), whether the cap stopped the run, and the units cut off or never begun. A merged job's is its shards' added up. Empty without a cap. | measured | any | +| `budget` | A standard run's spending cap, from `result.json`: `max_usd` and `spent_usd`, whether it `stopped` the run, the units `cut_off` and `not_run`, and with a cap per unit `unit_max_usd` and `over_unit`. Empty without a cap. | measured | any | | `agent_label` | The agent's name: the class of a `module:Class` path, or the trials' own name when the config gives none. | measured | any | | `agent_version` | The version the trial's `agent_info` reports. | measured | any | | `is_control` | An `oracle` or `nop` run: a check on the tasks, kept out of rankings and totals. Its row reads as a verdict, not a rate. A standard run is one only when its benchmark declares that control. An imported runner's arm without that declaration is not a control. On a benchmark that scores, a control's score of 1 reads as accepted and 0 as rejected. | measured | any | @@ -230,11 +230,11 @@ Twelve classes. The four marked **yes** under Infrastructure leave the adjusted | `failed_verification` | Failed verification | no | The verifier said no, on a benchmark with no patch detail: Terminal-Bench, or one read only by its reward. | | `patch_not_applied` | Patch did not apply | no | Malformed context lines, or edits against the wrong revision. | | `no_patch` | No patch produced | no | Finished without changing anything. | -| `agent_error` | Agent error | no | Timed out, filled its context, truncated its output, refused, or exited non-zero. A result, counted against the agent. | +| `agent_error` | Agent error | no | Timed out, filled its context, truncated its output, refused, exited non-zero, or reached its own spending cap (`UnitSpendingCapError`). A result, counted against the agent. | | `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. | +| `spending_cap` | Spending cap | yes | The run reached its `--max-usd` and its relay refused a call the unit needed (`SpendingCapError`). 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/standard.md b/docs/standard.md index 2aea931..5884905 100644 --- a/docs/standard.md +++ b/docs/standard.md @@ -138,7 +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`, which the job page shows. The builder's *Limits* takes one cap for the whole run, shared in equal parts by its jobs. [Run a benchmark](cli/run.md#a-spending-cap) | +| A spending cap (`--max-usd`, `--max-usd-per-unit`) | Each relay books a call in the job's ledger (`.work/ledger`, `standard/ledger.py`) before sending it, and refuses the one that does not fit the run's cap or its unit's. [Run a benchmark](cli/run.md#a-spending-cap) | ## Safety @@ -297,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. A spending cap (`--max-usd`) is each shard's own: the builder shares a recipe's cap over its shards in equal parts, and `merge` adds them up | +| 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 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) |