feat(builder): give a recipe a spending cap and show it on the job page

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 = "13"
+SCHEMA = "14"
 
 # 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
@@ -283,6 +283,9 @@ class Job:
     # defaults filled in (launch_of): what "New run like this" rebuilds a
     # 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.
+    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
     # turns into one view per agent. [] for a job of one agent.
@@ -830,6 +833,17 @@ def launch_of(cfg: dict) -> dict:
     }
 
 
+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")):
+        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"))}
+
+
 def parse_job(job_dir: str, reuse=None) -> Job:
     """One job directory. ``reuse(trial_dir)``: a Trial already parsed from
     files that have not changed since (a workspace's cache, for a running
@@ -906,6 +920,7 @@ def parse_job(job_dir: str, reuse=None) -> Job:
     ctl = std.get("control") if isinstance(std.get("control"), dict) else {}
     job.control_source = str(ctl.get("source") or "control") if ctl.get("name") == job.agent_label.lower() else ""
     job.launch_id = str(std.get("launch") or "")
+    job.budget = _budget(res.get("budget"))
     job.standard = bool(std)
     # a runner's job was never a `harbor run`: there is no command to rebuild
     job.harbor_command = "" if job.runner else redact(harbor_command(

app/nullius/standard/merge.py

@@ -26,7 +26,7 @@
 import json
 import os
 import shutil
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 from nullius.standard import job
 
@@ -154,6 +154,21 @@ def check(job_dirs: List[str], force: bool = False) -> dict:
     return {"shards": shards, "by_unit": by_unit, "notes": notes, "missing": missing, "units_total": total, "differ": differ}
 
 
+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):
+        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),
+            "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),
+            "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 []]}
+
+
 def merge(job_dirs: List[str], out: str, force: bool = False) -> dict:
     """Merge the shard jobs ``job_dirs`` into the new job ``out``.
     {out, trials, merged_from, notes, missing}."""
@@ -183,14 +198,16 @@ def merge(job_dirs: List[str], out: str, force: bool = False) -> dict:
     errored = sum(1 for r in results if r.get("exception_info"))
     rewards = [(r.get("verifier_result") or {}).get("rewards") or {} for r in results]
     scores = [w["score"] for w in rewards if "score" in w]
+    budget = _budget([s["result"].get("budget") for s in shards])
     with open(os.path.join(out, "result.json"), "w", encoding="utf-8") as f:
         # ISO stamps in UTC sort as their instants
         json.dump({"id": name, "started_at": min(starts) if starts else None, "finished_at": max(ends) if ends else None,
                    "n_total_trials": planned,
                    "stats": {"n_completed_trials": len(results) - errored, "n_errored_trials": errored,
                              "n_cancelled_trials": max(0, planned - len(results)), "n_running_trials": 0, "n_pending_trials": 0},
                    "standard": {"passed": sum(1 for w in rewards if w.get("reward") == 1.0),
-                                "scored": len(scores), "mean_score": job.mean_score(results)}}, f, indent=1)
+                                "scored": len(scores), "mean_score": job.mean_score(results)},
+                   **({"budget": budget} if budget else {})}, f, indent=1)
     cfg = json.loads(json.dumps(shards[0]["config"]))
     cfg["job_name"] = name
     std = cfg.setdefault("standard", {})

app/nullius/standard/recipe.py

@@ -31,13 +31,14 @@
 """
 from __future__ import annotations
 
+import math
 import re
 import shlex
 from typing import Callable, Dict, List, Optional
 
 from nullius import hosts, providers
 from nullius.providers import PROVIDER_LABEL, PROVIDERS, STEP
-from nullius.standard import hostreq, definitions, match, ports, prove
+from nullius.standard import hostreq, definitions, ledger, match, ports, prove
 
 HARNESS = "standard"
 # The builder's runner card (command-builder-core.js, runners()): not a definition file, so builtin like Harbor's.
@@ -253,7 +254,8 @@ def _int(v, low: int, high: Optional[int] = None) -> Optional[int]:
 
 
 def compile(recipe: dict, defs: Optional[Dict[str, List[dict]]] = None, *,
-            docker_unusable: Optional[Callable[[], Optional[str]]] = None) -> dict:
+            docker_unusable: Optional[Callable[[], Optional[str]]] = None,
+            unpriced: Optional[Callable[[List[str]], List[str]]] = None) -> dict:
     """{errors, notes, review, jobs, confirm, script, parts, model}
     for one recipe; errors name the step
     that edits what is wrong. And ``shards``: one per machine of *Where*, each
@@ -265,7 +267,15 @@ def compile(recipe: dict, defs: Optional[Dict[str, List[dict]]] = None, *,
     machine, why docker cannot run here (``container.unusable``) - a reason
     is the Where step's error. Not given, it is not asked: compiling stays a
     function of the definitions alone, and the routes that preview or start
-    a run pass it."""
+    a run pass it.
+
+    ``max_usd``: the run's spending cap (``standard.ledger``), one amount for
+    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.
+    ``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."""
     d = defs or _all()
     errors, notes = [], []
     add = lambda step, message: errors.append({"step": STEP[step], "message": message})
@@ -413,6 +423,13 @@ def compile(recipe: dict, defs: Optional[Dict[str, List[dict]]] = None, *,
     # the leaderboard's own models, as the run's provider spells them; the official ones stay the run's default
     sim_model = providers.served_as(lb["simulator"], provider_) if lb.get("simulator") else sim_model
     judge_model = providers.served_as(lb["judge"], provider_) if lb.get("judge") else judge_model
+    # the run's spending cap: an amount in USD, or none
+    cap = recipe.get("max_usd")
+    if cap in ("", None):
+        cap = 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
     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.")
@@ -426,6 +443,25 @@ def compile(recipe: dict, defs: Optional[Dict[str, List[dict]]] = None, *,
         why = docker_unusable()
         if why:
             add("where", f"The docker backend cannot run on this machine: {why}. Choose local processes, or a host with docker.")
+    # a job that calls no model (a control of a benchmark with no judge) spends nothing and takes no share
+    spends = lambda p: not (p["control"] and not (b.get("score") or {}).get("judge") and not needs["embeddings"])
+    paying = len(dss) * sum(1 for p in picks if spends(p)) * len(targets or [None])
+    share = None
+    if cap is not None and paying:
+        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 errors:
         return out
     nouns = (b.get("units") or {}).get("nouns", "units")
@@ -458,6 +494,7 @@ def command_for(d: dict, p: dict, index: int) -> str:
                  "--dataset", d["id"], *(["--limit", str(limit)] if isinstance(limit, int) else []), "-j", str(conc),
                  *[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 []),
                  *(["--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}")],
@@ -537,6 +574,8 @@ def row(p: dict) -> list:
                       + (f" · {repeat} attempts each (pass^1 to pass^{repeat})" if repeat > 1 else "")
                       + (" · a setup counts " + next(p["o"]["units"]["nouns"] for p in picks if p.get("setup"))
                          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["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:

app/nullius/standard/run.py

@@ -527,7 +527,8 @@ def start(self) -> str:
         work = [(i, u, r) for r in range(1, self.repeat + 1) for i, u in mine]
         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)
+        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
         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 = []

app/static/builder/command-builder-core.js

@@ -135,6 +135,8 @@
     if ((Number.isInteger(raw.concurrency) && raw.concurrency >= 1) || raw.concurrency === '') r.concurrency = raw.concurrency;
     // attempts per unit and the judge's and user simulator's models: a reopened recipe runs as it was saved
     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;
     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';

app/static/builder/command-builder.js

@@ -623,7 +623,9 @@
       return '<div class="nb-sectionhead"><h2>Scope</h2></div><div class="nb-two">'
         + field('Number of ' + u.nouns, 'limit', rec.limit, {type: 'number', min: 1, placeholder: all, hint: 'The first ones' + per + ', in its order.'})
         + field('Concurrent ' + u.nouns, 'concurrency', rec.concurrency, {type: 'number', min: 1, max: 32, placeholder: '1'}) + '</div>'
-        + field('Attempts per ' + u.noun, 'repeat', rec.repeat || '', {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'}));
     }
     return '';
   }

app/static/runs/board.css

@@ -212,6 +212,8 @@
 .rp .popm { position: absolute; z-index: 6; left: 0; top: calc(100% + 4px); display: grid; grid-template-columns: auto auto; gap: 4px 14px; padding: 9px 12px;
   border-radius: 8px; background: var(--surface-2); border: 1px solid var(--border); box-shadow: var(--shadow); font-size: 12px; color: var(--text-faint); white-space: nowrap; }
 .rp .popm b { font: 500 12px var(--mono); color: var(--text); text-align: right; }
+/* the units a spending cap left out: ids, as many as there are */
+.rp .popm.cap-units { grid-template-columns: repeat(4, auto); max-height: 220px; overflow-y: auto; font-family: var(--mono); color: var(--text-dim); }
 
 /* linked preview: a hovered row lights its board marks and dims the rest; board counts never change */
 .rp .board.pv .hl:not(.lr) { opacity: .28; }

app/static/runs/live-board.js

@@ -87,7 +87,13 @@
     }
     // 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(`<span><b>${count(Math.round(s.tokens_per_sec))}</b> tokens / s, last 30 s</span>`);
-    if (T.cost_usd != null) bits.push(`<span><b>$${Number(T.cost_usd).toFixed(2)}</b> spent</span>`);
+    // 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
+    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 <b>$${capped(s.budget.max_usd)}</b> 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(`<span><b>$${Number(T.cost_usd).toFixed(cap && T.cost_usd < 0.01 ? 4 : 2)}</b> spent${cap}</span>`);
+    else if (cap) bits.push(`<span>spending cap <b>$${capped(s.budget.max_usd)}</b></span>`);
     if (s.wall_sec) bits.push(`<span>wall <b>${clock(s.wall_sec)}</b> per trial</span>`);
     return bits.join('');
   }

app/templates/runs/job.html

@@ -56,6 +56,14 @@
 </div>
 {% endif %}
 
+{% if job.budget and job.budget.stopped %}
+{# The run's own spending cap ended it (a Nullius standard run's --max-usd): nothing failed, so the notice is neutral. #}
+<div class="alert" id="cap-stopped">
+  <strong>Stopped at its spending cap</strong> 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.
+</div>
+{% endif %}
+
 {% if live_board and live_board.error %}
 <div class="alert warn"><strong>The live feed could not be read</strong> — <span class="mono">{{ live_board.error }}</span>. The sections below come from the trial results and are unaffected.</div>
 {% endif %}
@@ -305,6 +313,9 @@
     <div class="sp"><details class="pop"><summary aria-label="Token split: input, cached, output"><span class="tw"></span>in · cached · out</summary>
       <div class="popm" role="group" aria-label="Token split"><span>input</span><b>{{ '{:,}'.format(job.total_input_tokens) }}</b><span>cached (part of input)</span><b>{{ '{:,}'.format(job.total_cache_tokens) }}</b><span>output</span><b>{{ '{:,}'.format(job.total_output_tokens) }}</b></div></details></div>
     <div class="kv" data-tip="{{ 'A control never calls a model, so this is a real zero' if job.is_control else 'Only recorded by agents that report it' }}"><span class="k">Cost</span><span class="v">{{ job.total_cost_usd|usd(job.kind) }}</span></div>
+    {% if job.budget %}<div class="kv" id="run-cap" data-tip="The most this run could spend: its relay refuses a call that would pass it{% if job.budget.unsettled_usd %}. {{ job.budget.unsettled_usd|usd }} stays booked for calls whose cost was never reported{% endif %}"><span class="k">Spending cap</span><span class="v">{{ job.budget.max_usd|cap }}</span></div>
+    {% if job.budget.not_run %}<div class="sp"><details class="pop"><summary aria-label="The units not run"><span class="tw"></span>{{ job.budget.not_run|length }} not run</summary>
+      <div class="popm cap-units" role="group" aria-label="The units not run">{% for u in job.budget.not_run %}<span>{{ u }}</span>{% endfor %}</div></details></div>{% endif %}{% endif %}
     {% if job.mean_score is not none %}<div class="kv" data-tip="{% if job.n_scored_parts %}The mean score over every one of the {{ job.n_scored_parts }} scored, whichever trial it is under{% else %}The mean of each item's graded score (an F1, partial credit){% endif %}{% if not job.scored_only %}; the rates count an item resolved at its benchmark's pass mark{% endif %}"><span class="k">Mean score</span><span class="v">{{ '%.2f'|format(job.mean_score) }}</span></div>{% endif %}
     {% for g in job.score_by_group %}<div class="kv by-group" data-tip="The mean score of the {{ g.n }} scored with {{ job.parts_by }} {{ g.group }}"><span class="k">{{ job.parts_by|capitalize }} {{ g.group }}</span><span class="v">{{ '%.2f'|format(g.mean) }} <small class="faint">{{ g.n }}</small></span></div>{% endfor %}
     {% if job.duration_min %}<div class="kv" data-tip="Start to finish, the whole run"><span class="k">Duration</span><span class="v">{{ job.duration_min|runtime }}</span></div>{% endif %}