diff --git a/.claude/rules/running.md b/.claude/rules/running.md index 4880f77..6bb6d71 100644 --- a/.claude/rules/running.md +++ b/.claude/rules/running.md @@ -16,8 +16,9 @@ through the execution backend (`docs/builder.md`). are included; configuration and credential stores are excluded. Several machines run one shard each. - **Keys**: a run gets only its provider's key and endpoint, as environment - (over stdin to a host), never written; logs are redacted. The standard's - agents and upstream scripts get only the relay's word, never a key. + (over stdin to a host, and the key to a docker relay as a file); logs are + redacted. The standard's agents and upstream scripts get only the relay's + word, never a key. - What a recipe needs of each machine is asked before Run and again as it starts (`nullius/prerequisites.py`). `runs//run.json` lets a restarted Nullius pick a run up again. diff --git a/AGENTS.md b/AGENTS.md index 3aa76e5..199ab26 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -61,7 +61,7 @@ and its resources. The source layout stays under `app/`. `feed`: the live feed any Harbor writes, from its own trial hooks), `importers/` (`atif` ATIF, `otel` OpenTelemetry, and the `cli`), `judge/` (`questions`, `state`, `client`, `store`, `batch`, `reconcile`, `lanes`), `standard/` (the Nullius standard, `docs/standard.md`: `ports`, `names` (capability vocabulary and legacy aliases), `definitions` (with an agent's or capability's versions), `match`, - `jsonl`, `acp`, `cancellation` (interruptible protocol waits), `relay`, `install`, `run`, `container` (the docker backend), `forward` (a licence server's one door from it, or a benchmark's MCP server's), `mcp_machine` and `mcp_bridge` (a benchmark's MCP server run on the machine, in its driver's environment, and the agent's bridge to it), `hostreq` (what a benchmark's options need of the machine its driver runs on: the check and the install) and `egress` (an own image's check's proxy to public addresses), `job`, `conform`, `fidelity` (ACP against the agent's own CLI), `prove` (a benchmark's controls and proofs), `shard` and `merge` (one run over many machines), `recipe` (the builder's standard recipe), `howto` (the How to add steps) and the `cli`; its + `jsonl`, `acp`, `cancellation` (interruptible protocol waits), `relay`, `ledger` (a run's spending cap: each call booked before it is sent), `install`, `run`, `container` (the docker backend), `forward` (a licence server's one door from it, or a benchmark's MCP server's), `mcp_machine` and `mcp_bridge` (a benchmark's MCP server run on the machine, in its driver's environment, and the agent's bridge to it), `hostreq` (what a benchmark's options need of the machine its driver runs on: the check and the install) and `egress` (an own image's check's proxy to public addresses), `job`, `conform`, `fidelity` (ACP against the agent's own CLI), `prove` (a benchmark's controls and proofs), `shard` and `merge` (one run over many machines), `recipe` (the builder's standard recipe), `howto` (the How to add steps) and the `cli`; its `payload` snapshots the running backend and effective catalogue for a remote host; conformance kits under `standard/kits/`, each a benchmark: a `driver` and, for the tools kit, an MCP server `mcp_kit`) - the definitions it reads are in the catalogue: diff --git a/app/nullius/catalogue/capabilities/mem0/capability.py b/app/nullius/catalogue/capabilities/mem0/capability.py index afd1e5d..7e4c44c 100644 --- a/app/nullius/catalogue/capabilities/mem0/capability.py +++ b/app/nullius/catalogue/capabilities/mem0/capability.py @@ -12,7 +12,8 @@ ``add {session}`` - the session's turns as Mem0 messages (a role other than user or assistant is a user turn naming its speaker), - added for one user, the session's id and date as metadata + added for one user, the session's id and date as metadata, + the date also as ``timestamp`` ``recall {question}`` - ``{memory}``: what Mem0's search returns, best first, each memory headed by its session's date @@ -111,8 +112,10 @@ def add(self, s): self.dates[sid] = str(s.get("date") or "") self.warnings.take() t0 = time.time() - got = self.mem.add(messages, user_id=USER, metadata={"session": sid, "date": self.dates[sid]}, - infer=setting("MEM0_INFER", "on") != "off") + meta = {"session": sid, "date": self.dates[sid]} + if self.dates[sid]: + meta["timestamp"] = self.dates[sid] + got = self.mem.add(messages, user_id=USER, metadata=meta, infer=setting("MEM0_INFER", "on") != "off") items = got.get("results") if isinstance(got, dict) else got return {"trace": {"messages": len(messages), "seconds": round(time.time() - t0, 3), "written": [{k: it.get(k) for k in ("id", "event", "memory", "previous_memory") if it.get(k) is not None} diff --git a/app/nullius/execution.py b/app/nullius/execution.py index 8fad11e..812f1f6 100644 --- a/app/nullius/execution.py +++ b/app/nullius/execution.py @@ -78,9 +78,9 @@ run without that endpoint fails before it starts. They are read on this machine and handed to the run as environment (API_TOKEN, OPENAI_API_KEY, LLM_API_KEY; API_URL, OPENAI_BASE_URL, -LLM_BASE_URL) - over stdin to the launcher, never in argv, never written -anywhere; a stdin cut short starts nothing and fails the run at that part -(``_launch``), and one that holds a line break is never sent +LLM_BASE_URL) - over stdin to the launcher, never in argv (the key to a +docker relay as a file); a stdin cut short starts nothing and fails the +run at that part (``_launch``), and one that holds a line break is never sent (``_line_break``) - and they are the only credentials it has: its commands get this process's environment less every secret in Settings, whichever provider it belongs to (``withheld``); a host gets nothing of it but stdin. diff --git a/app/nullius/runs/classify.py b/app/nullius/runs/classify.py index 2b3e817..2b37bf2 100644 --- a/app/nullius/runs/classify.py +++ b/app/nullius/runs/classify.py @@ -57,6 +57,7 @@ "modelnotfounderror", # the model name is wrong; the agent never ran "networkconnectionerror", # DNS, refused connection, TLS — transport, not skill "unknownapierror", # Harbor's own "some provider error"; an ApiError without the prefix + "spendingcaperror", # the run's own spending cap refused the call (the standard's relay): no attempt was had } # What Harbor's agents raise when the agent itself fails — its budget spent, @@ -131,7 +132,8 @@ ), API_ERROR: ( "The model API failed some other way — overloaded, unreachable, out of " - "quota, or the key was rejected. The agent never got a fair attempt, so " + "quota, or the key was rejected — or the run's own spending cap refused " + "the call. The agent never got a fair attempt, so " "this is infrastructure too and is excluded from the adjusted rate." ), SETUP_ERROR: ( diff --git a/app/nullius/standard/cli.py b/app/nullius/standard/cli.py index efc0fb0..837044b 100644 --- a/app/nullius/standard/cli.py +++ b/app/nullius/standard/cli.py @@ -6,7 +6,7 @@ PYTHONPATH=app python3 -m nullius.standard.cli new benchmark|agent|capability KEY [--family F] [--port P] PYTHONPATH=app python3 -m nullius.standard.cli new version KEY VERSION PYTHONPATH=app python3 -m nullius.standard.cli conform AGENT|CAPABILITY --model M [--provider P] [--port PORT ...] [--repeat K] [--option NAME=VALUE ...] [--unsandboxed | --backend docker] - PYTHONPATH=app python3 -m nullius.standard.cli run BENCHMARK[@VERSION] AGENT|CAPABILITY --model M [--provider P] [--limit N] [-j N] [--option NAME=VALUE ...] [--bench-option NAME=VALUE ...] [--price MODEL=INPUT,OUTPUT[,CACHED] ...] [--unsandboxed | --backend docker] + PYTHONPATH=app python3 -m nullius.standard.cli run BENCHMARK[@VERSION] AGENT|CAPABILITY --model M [--provider P] [--limit N] [-j N] [--option NAME=VALUE ...] [--bench-option NAME=VALUE ...] [--price MODEL=INPUT,OUTPUT[,CACHED] ...] [--max-usd USD] [--unsandboxed | --backend docker] PYTHONPATH=app python3 -m nullius.standard.cli run BENCHMARK[@VERSION] nop|oracle [--dataset D] [--limit N] PYTHONPATH=app python3 -m nullius.standard.cli prove BENCHMARK[@VERSION] [--dataset D] [--limit N] PYTHONPATH=app python3 -m nullius.standard.cli run ... --shard I/N [--launch ID] @@ -469,7 +469,7 @@ def cmd_run(a): concurrency=a.concurrency, jobs_root=a.jobs, name=a.name, unsandboxed=a.unsandboxed, judge_model=a.judge_model, backend=a.backend, simulator_model=a.simulator_model, options=options(a.option), shard=_shard(a.shard), launch=a.launch, licences=held, bench_options=options(a.bench_option), repeat=a.repeat, remove_images=a.remove_images, - prices=prices(a.price), **({"env": run_env(a.provider)} if a.from_run else {})) + prices=prices(a.price), max_usd=a.max_usd, **({"env": run_env(a.provider)} if a.from_run else {})) except run.Refused as e: print("refused:", e) return 2 @@ -624,6 +624,8 @@ def main(argv=None): help="one of the benchmark's own options (repeatable: a search, a mode); the rest keep their defaults") p.add_argument("--price", action="append", metavar="MODEL=INPUT,OUTPUT[,CACHED]", help="this run's price for a model in USD per million tokens, over your price list (repeatable; nullius prices)") + p.add_argument("--max-usd", type=float, metavar="USD", + help="the run's spending cap: no call goes out once the spend plus the calls in flight would pass it") v = sub.add_parser("prove", help="prove a benchmark: its controls, its units, no leak; record what passed") v.add_argument("benchmark") v.add_argument("--dataset", help="one dataset (default: each)") diff --git a/app/nullius/standard/container.py b/app/nullius/standard/container.py index 794f239..88b8e47 100644 --- a/app/nullius/standard/container.py +++ b/app/nullius/standard/container.py @@ -32,12 +32,13 @@ bridge *and* on that network as ``relay``; the agent's container on the internal network alone. Its model endpoint is ``http://relay:8080//v1`` and nothing else answers it. The real key reaches the relay's container -only, as ``-e NULLIUS_RELAY_UPSTREAM_KEY`` with the value in the docker -program's own environment - never on an argument list, which ``ps`` and -``docker inspect`` show. The relay publishes its port on 127.0.0.1 so the -driver, which stays on this machine (it owns the answers), reaches the -judge's route and a capability benchmark's reader. Its record folder is mounted -read-write, so the host reads each call as the local relay's are read. +only, as an owner-only file under :func:`key_root`, mounted read-only and +removed from this machine once the relay listens: never on an argument list +(``ps``) or in the container's environment (``docker inspect``). The relay +publishes its port on 127.0.0.1 so the driver, which stays on this machine +(it owns the answers), reaches the judge's route and a capability +benchmark's reader. Its record folder is mounted read-write, so the host +reads each call as the local relay's are read. **A licence server is one more door.** A unit whose environment needs a licence of kind ``server`` (``host:port``) gets a forwarder beside its relay @@ -102,6 +103,7 @@ import urllib.request from typing import Dict, Iterable, List, Optional, Tuple +from nullius import paths from nullius.standard import definitions, install DOCKER = "docker" @@ -114,6 +116,8 @@ EGRESS_IN = "/opt/nullius/egress.py" # the check's proxy's one file, in its container EGRESS_PORT = 3128 EGRESS_ALIAS = "egress" +RELAY_KEY_IN = "/opt/nullius/relay-key" # the key's one file, in the relay's container +KEY_STALE = 600 # older than any start can take: a killed run left it IMAGE_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" NODE_VERSION = "22.23.3" @@ -465,26 +469,35 @@ def install_argv(image: str, where: str, cmd: List[str]) -> List[str]: "-e", f"HOME={where}/.npm-home", "-e", f"npm_config_cache={where}/.npm-cache", image] + list(cmd) -def relay_argv(image: str, *, name: str, job: str, app: str, record: str, base: str, model: Optional[str], - routes: Dict[str, str], prices: Optional[dict] = None, side_models=(), - key_header: str = "authorization", hosts: Optional[Dict[str, str]] = None) -> List[str]: +def relay_argv(image: str, *, name: str, job: str, app: str, record: str, key_file: str, base: str, + model: Optional[str], routes: Dict[str, str], prices: Optional[dict] = None, side_models=(), + key_header: str = "authorization", hosts: Optional[Dict[str, str]] = None, + ledger: Optional[str] = None) -> List[str]: """``docker run`` for a unit's relay: on the default bridge (its way to the - provider), its port published on 127.0.0.1 only. The key is named, never - given: ``-e NULLIUS_RELAY_UPSTREAM_KEY``. ``hosts``: names it resolves - (``--add-host``), the machine's own for an endpoint on its loopback.""" + provider), its port published on 127.0.0.1 only. The key is never given: + ``key_file``, its file, is mounted read-only at ``RELAY_KEY_IN`` with + ``--mount``, which fails when the engine cannot see it (``-v`` would mount + an empty folder). ``hosts``: names it resolves + (``--add-host``), the machine's own for an endpoint on its loopback. + ``ledger``: the file of the run's spending cap, its folder mounted so + this relay books its calls where every other relay of the job does.""" a = [DOCKER, "run", "-d", "--name", name, "--label", f"{LABEL}={job}", "--user", _user(), "--cap-drop", "ALL", "--security-opt", "no-new-privileges", "--init", *[w for n, ip in sorted((hosts or {}).items()) for w in ("--add-host", f"{n}:{ip}")], - "-p", f"127.0.0.1::{RELAY_PORT}", "-v", f"{app}:{app}:ro", "-v", f"{record}:{record}:rw", "-w", app, - "-e", "PYTHONPATH=" + app, "-e", "PYTHONDONTWRITEBYTECODE=1", "-e", "NULLIUS_RELAY_UPSTREAM_KEY", + "-p", f"127.0.0.1::{RELAY_PORT}", "-v", f"{app}:{app}:ro", "-v", f"{record}:{record}:rw", + *(["-v", f"{os.path.dirname(ledger)}:{os.path.dirname(ledger)}:rw"] if ledger else []), + "--mount", f"type=bind,source={key_file},target={RELAY_KEY_IN},readonly", "-w", app, + "-e", "PYTHONPATH=" + app, "-e", "PYTHONDONTWRITEBYTECODE=1", image, "python3", "-m", "nullius.standard.relay", "serve", "--listen", f"0.0.0.0:{RELAY_PORT}", - "--base", base, "--record", record, "--public", f"http://{RELAY_ALIAS}:{RELAY_PORT}"] + "--base", base, "--record", record, "--key-file", RELAY_KEY_IN, "--public", f"http://{RELAY_ALIAS}:{RELAY_PORT}"] if model: a += ["--model", model] for tag, m in sorted(routes.items()): a += ["--route", f"{tag}={m}"] if prices: a += ["--prices", json.dumps(prices, separators=(",", ":"))] + if ledger: + a += ["--ledger", ledger] for m in side_models: a += ["--side-model", m] if key_header != "authorization": @@ -646,8 +659,10 @@ class Unit: def __init__(self, job: str, tag: str, record_dir: str, *, image: str, app: str, base: str, key: str, model: Optional[str] = None, routes: Optional[Dict[str, str]] = None, prices: Optional[dict] = None, side_models=(), key_header: str = "authorization", servers: Iterable[str] = (), - doors: Iterable[Tuple[str, int, str]] = (), hosts: Optional[Dict[str, str]] = None): + doors: Iterable[Tuple[str, int, str]] = (), hosts: Optional[Dict[str, str]] = None, + ledger: Optional[str] = None): self.job, self.tag, self.dir, self.image, self.app = job, tag, record_dir, image, app + self.ledger = ledger # the file of the run's spending cap, when it has one self.servers = list(servers) # licence servers, host:port each: a forwarder apiece self.doors = list(doors) # servers on this machine: (alias, port, socket), a door apiece self.side_models, self.key_header = tuple(side_models), key_header @@ -657,6 +672,7 @@ def __init__(self, job: str, tag: str, record_dir: str, *, image: str, app: str, self.host: Optional[str] = None self.started: List[str] = [] # containers to remove self.network = False + self.key_dir: Optional[str] = None # the key file's folder, until the relay listens os.makedirs(record_dir, exist_ok=True) def start(self) -> "Unit": @@ -664,12 +680,14 @@ def start(self) -> "Unit": _docker(["network", "create", "--internal", "--label", f"{LABEL}={self.job}", self.names["network"]]) self.network = True argv = relay_argv(self.image, name=self.names["relay"], job=self.job, app=self.app, record=self.dir, - base=self.base, model=self.model, routes=self.routes, prices=self.prices, - side_models=self.side_models, key_header=self.key_header, hosts=self.hosts) + key_file=self._key_file(), base=self.base, model=self.model, routes=self.routes, + prices=self.prices, side_models=self.side_models, key_header=self.key_header, + hosts=self.hosts, ledger=self.ledger) self.started.append(self.names["relay"]) - _docker(argv[1:], env=client_env({"NULLIUS_RELAY_UPSTREAM_KEY": self.key})) + _docker(argv[1:]) _docker(["network", "connect", "--alias", RELAY_ALIAS, self.names["network"], self.names["relay"]]) self._wait(self.names["relay"], "the relay's container") + self._forget_key() port = _docker(["port", self.names["relay"], f"{RELAY_PORT}/tcp"]).stdout.strip().splitlines()[0] self.host = "http://" + port.replace("0.0.0.0", "127.0.0.1") self.reach = {} @@ -682,16 +700,35 @@ def start(self) -> "Unit": raise return self + def _key_file(self) -> str: + """Write the key to an owner-only file in a new folder under + :func:`key_root`, and return its path. Not the record folder: it is kept.""" + root = key_root() + os.makedirs(root, exist_ok=True) + os.chmod(root, 0o700) + forget_stale_keys() + self.key_dir = tempfile.mkdtemp(prefix=key_prefix(self.job), dir=root) + path = os.path.join(self.key_dir, "key") + with os.fdopen(os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w") as f: + f.write(self.key) + return path + + def _forget_key(self): + if self.key_dir: + shutil.rmtree(self.key_dir, ignore_errors=True) + self.key_dir = None + def _forward(self, i: int, address: str) -> str: """A licence server's forwarder, started and on the unit's network - under the server's name: the address the agent reaches it by.""" + under the server's name: the address the agent reaches it by. It joins + that network once it listens, after it resolves the server's name.""" alias, host, port = server_alias(address, i) name = f"{self.names['agent']}-lic{i}" self.started.append(name) _docker(forward_argv(self.image, name=name, job=self.job, port=port)[1:], env=client_env({"NULLIUS_FORWARD_TO": f"{host}:{port}"})) - _docker(["network", "connect", "--alias", alias, self.names["network"], name]) self._wait(name, "the licence server's forwarder") + _docker(["network", "connect", "--alias", alias, self.names["network"], name]) return f"{alias}:{port}" def _door(self, i: int, alias: str, port: int, sock: str) -> None: @@ -745,8 +782,9 @@ def wrap(self, cmd: List[str], env: dict, workdir: str, mounted: List[Tuple[str, return argv, client_env({k: v for k, v in inside.items() if k != "PATH"}) def stop(self): - """Remove the unit's containers and network; the relay's log is kept - beside its record first.""" + """Remove the unit's containers and network, and the key's file if the + relay has not read it; the relay's log is kept beside its record first.""" + self._forget_key() if self.names["relay"] in self.started: try: r = _docker(["logs", self.names["relay"]], check=False, timeout=30) @@ -1368,8 +1406,34 @@ def stop(self): self.own_net = False +def key_root() -> str: + """The relays' key folders, in the user's cache: Colima and Lima share the + home with their engine, not the temporary folder.""" + return os.path.join(paths.cache_dir(), "relay-keys") + + +def forget_stale_keys(): + """Remove every job's key folders older than ``KEY_STALE``.""" + old = time.time() - KEY_STALE + for d in glob.glob(os.path.join(glob.escape(key_root()), "*")): + try: + if os.lstat(d).st_mtime < old: + shutil.rmtree(d, ignore_errors=True) + except OSError: + pass + + +def key_prefix(job: str) -> str: + """A job's key folder prefix: a hash, so no job's prefix starts another's.""" + return f"nullius-relay-key-{hashlib.sha256(job.encode()).hexdigest()[:12]}-" + + def sweep(job: str): - """Remove every container and network a job's units left, by its label.""" + """Remove every container and network a job's units left, by its label, + its relays' key folders, and stale key folders.""" + for d in glob.glob(os.path.join(glob.escape(key_root()), glob.escape(key_prefix(job)) + "*")): + shutil.rmtree(d, ignore_errors=True) + forget_stale_keys() try: ids = _docker(["ps", "-aq", "--filter", f"label={LABEL}={job}"], check=False).stdout.split() if ids: diff --git a/app/nullius/standard/forward.py b/app/nullius/standard/forward.py index 0d82f8c..1aac8d4 100644 --- a/app/nullius/standard/forward.py +++ b/app/nullius/standard/forward.py @@ -7,7 +7,10 @@ beside the relay (``container.Unit``), on the default bridge and on the unit's network under the server's own name, listening on the server's port and passing every connection to that one address - and to nothing else, -whatever the agent sends. It reads nothing of what it passes. +whatever the agent sends. It reads nothing of what it passes. It resolves the +server's name once, before it listens: on the unit's network, that name is +the forwarder's own. A name that does not resolve ends it with an +``unresolved`` event. A benchmark's MCP server run on the machine (``mcp_machine``) is reached the same way: a forwarder on the unit's internal network only, the server's @@ -44,14 +47,26 @@ def _pump(src: socket.socket, dst: socket.socket) -> None: pass -def _serve(conn: socket.socket, host: str, port) -> None: - try: - if port is None: # a Unix socket: host is its path - up = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) +def _connect(targets) -> socket.socket: + """The first of ``targets`` (family, address) that accepts, or the last error.""" + error = OSError("no address to connect to") + for family, addr in targets: + up = None + try: + up = socket.socket(family, socket.SOCK_STREAM) # in the try: a family this kernel lacks is skipped up.settimeout(30) - up.connect(host) - else: - up = socket.create_connection((host, port), timeout=30) + up.connect(addr) + return up + except OSError as e: + error = e + if up is not None: + up.close() + raise error + + +def _serve(conn: socket.socket, targets) -> None: + try: + up = _connect(targets) up.settimeout(None) except OSError as e: print(json.dumps({"event": "unreachable", "error": str(e)[:200]}), flush=True) @@ -78,6 +93,14 @@ def main(argv=None) -> int: ap.error("--to, or NULLIUS_FORWARD_TO: where to forward") lhost, lport = _split(a.listen) host, port = _split(a.to) + if port is None: + targets = [(socket.AF_UNIX, host)] + else: + try: + targets = [(f, addr) for f, _, _, _, addr in socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)] + except OSError as e: + print(json.dumps({"event": "unresolved", "host": host, "error": str(e)[:200]}), flush=True) + return 1 srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srv.bind((lhost, lport)) @@ -85,7 +108,7 @@ def main(argv=None) -> int: print(json.dumps({"event": "listening", "port": lport}), flush=True) while True: conn, _ = srv.accept() - threading.Thread(target=_serve, args=(conn, host, port), daemon=True).start() + threading.Thread(target=_serve, args=(conn, targets), daemon=True).start() if __name__ == "__main__": diff --git a/app/nullius/standard/job.py b/app/nullius/standard/job.py index 95aaa0a..58cc1c3 100644 --- a/app/nullius/standard/job.py +++ b/app/nullius/standard/job.py @@ -486,7 +486,7 @@ def write_config(job_dir: str, *, job: str, bench: dict, other: dict, model: str #: an error kind naming the machine or the benchmark's side rather than the agent: the live board's infra_error _INFRA = ("Setup", "Verifier", "Api", "RateLimit", "Environment", "Docker", "Build", "Connection", "Network", "Authentication", - "ModelNotFound") + "ModelNotFound", "SpendingCap") def feed_outcome(kind: str, score: Optional[dict], exception: Optional[tuple]) -> dict: @@ -532,11 +532,14 @@ def pass_hat(results: List[dict]) -> Optional[dict]: "at": {str(k): round(sum(1 - comb(n - sum(v[:n]), k) / comb(n, k) for v in whole.values()) / len(whole), 4) for k in range(1, n + 1)}} -def write_result(job_dir: str, job: str, t0: float, t1: float, results: List[dict], planned: int): +def write_result(job_dir: str, job: str, t0: float, t1: float, results: List[dict], planned: int, budget: Optional[dict] = None): + """``budget``: a run with a spending cap (``run.Run.budget``): the cap, + what was spent, whether it stopped the run, the units it cut off and + those never begun.""" errored = sum(1 for r in results if r.get("exception_info")) hat = pass_hat(results) with open(os.path.join(job_dir, "result.json"), "w") as f: json.dump({"id": job, "started_at": iso(t0), "finished_at": iso(t1), "n_total_trials": planned, "stats": {"n_completed_trials": len(results) - errored, "n_errored_trials": errored, "n_cancelled_trials": planned - len(results), "n_running_trials": 0, "n_pending_trials": 0}, - **({"pass_hat_k": hat} if hat else {})}, f, indent=1) + **({"pass_hat_k": hat} if hat else {}), **({"budget": budget} if budget else {})}, f, indent=1) diff --git a/app/nullius/standard/ledger.py b/app/nullius/standard/ledger.py new file mode 100644 index 0000000..72f07a6 --- /dev/null +++ b/app/nullius/standard/ledger.py @@ -0,0 +1,183 @@ +"""A run's spending cap: one small file every relay of the job books its calls in. + +A capped run (``run --max-usd``) gives each of its relays - the units' own, +in this process or in their containers, and the benchmark's - the same +ledger. Before a call goes out, its relay books the most the call is likely +to cost (:func:`bound`). The call is sent only while what was spent, what the +calls in flight have booked and this booking stay within the cap. The first +call that does not fit is refused and closes the ledger: every later call is +refused too, so a run stops at one point instead of letting its small calls +through. + +A reply turns the booking into what the call cost (``settle``). A call the +provider answered with an error gives its booking back (``release``). One +whose cost was never known - a reply with no token count, a reply cut off - +keeps its whole booking, as ``unsettled``: it may have been billed. + +A booking is an estimate, not a bill: the prompt's characters over three as +input tokens, and as output the limit the call names, else the model's own +(its price row's ``max_output``), else ``OUTPUT``. A call that costs more +than was booked for it takes the spend past the cap by the difference, and +the next call is refused. + +The file is rewritten whole under a lock held for one read and one write +(``flock`` on a file beside it, as an install is locked): a relay that dies +holding it lets it go. +""" +from __future__ import annotations + +import contextlib +import fcntl +import json +import math +import os +import uuid +from typing import Optional, Tuple + +OUTPUT = 4096 # output tokens booked when neither the call nor the model's price row names a limit +ERROR = "SpendingCapError" # what a unit the cap ended carries: infrastructure, never a miss (runs.classify) +_LIMITS = ("max_completion_tokens", "max_output_tokens", "max_tokens") + + +class Unavailable(Exception): + """The ledger could not be read or written: a capped call is then not sent.""" + + +def bound(req: dict, raw: bytes, row: dict, path: str = "") -> float: + """The most 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: the limit the call names, else + the row's, else ``OUTPUT``; an embeddings call has none.""" + tokens = math.ceil(max(0, len(raw) - _blobs(req)) / 3) + rate = max(row["input"], row.get("cache_write") or 0) + out = 0 + if not path.split("?")[0].endswith("/embeddings"): + named = [req.get(k) for k in _LIMITS if isinstance(req.get(k), int) and not isinstance(req.get(k), bool) and req.get(k) > 0] + out = named[0] if named else row.get("max_output") or OUTPUT + return (tokens * rate + out * row["output"]) / 1e6 + + +def _blobs(v) -> int: + """Characters of a request that are an encoded file, not a prompt: a + ``data:`` address, a block's base64 ``data``.""" + if isinstance(v, str): + return len(v) if v.startswith("data:") else 0 + if isinstance(v, list): + return sum(_blobs(x) for x in v) + if isinstance(v, dict): + if v.get("type") == "base64" and isinstance(v.get("data"), str): + return len(v["data"]) + return sum(_blobs(x) for x in v.values()) + return 0 + + +class Ledger: + def __init__(self, path: str): + self.path = path + + @classmethod + def create(cls, path: str, max_usd: float) -> "Ledger": + os.makedirs(os.path.dirname(path), exist_ok=True) + led = cls(path) + led._write({"max_usd": float(max_usd), "spent_usd": 0.0, "unsettled_usd": 0.0, "held": {}, "closed": False}) + return led + + @contextlib.contextmanager + def _locked(self): + try: + with open(self.path + ".lock", "a") as f: + fcntl.flock(f, fcntl.LOCK_EX) + try: + yield self._read() + finally: + fcntl.flock(f, fcntl.LOCK_UN) + except (OSError, ValueError, KeyError, TypeError) as e: + raise Unavailable(f"the spending cap's ledger could not be used: {type(e).__name__}: {e}") + + def _read(self) -> dict: + with open(self.path, encoding="utf-8") as f: + doc = json.load(f) + float(doc["max_usd"]), float(doc["spent_usd"]), float(doc["unsettled_usd"]), sum(doc["held"].values()) + return doc + + def _write(self, doc: dict) -> None: + """Written whole beside the file, then renamed: a reader that takes no lock never finds half of it.""" + part = f"{self.path}.{os.getpid()}.{uuid.uuid4().hex[:8]}.part" + with open(part, "w", encoding="utf-8") as f: + json.dump(doc, f) + os.replace(part, self.path) + + def reserve(self, amount: float) -> Tuple[Optional[str], dict]: + """(the booking's id, the state before it) when ``amount`` fits; + (None, the state) when it does not, or the ledger is closed already - + and it is closed from then on.""" + with self._locked() as doc: + seen = _state(doc) + if doc["closed"] or amount > seen["left_usd"]: + if not doc["closed"]: + doc["closed"] = True + self._write(doc) + return None, seen + booking = uuid.uuid4().hex[:12] + doc["held"][booking] = amount + self._write(doc) + return booking, seen + + def settle(self, booking: str, cost: Optional[float]) -> None: + """The call's cost in place of its booking; None, its cost was never + known: the whole booking is kept, as unsettled.""" + with self._locked() as doc: + amount = doc["held"].pop(booking, None) + if amount is None: + return + if cost is None: + doc["unsettled_usd"] += amount + else: + doc["spent_usd"] += max(0.0, float(cost)) + self._write(doc) + + def release(self, booking: str) -> None: + """The booking given back: the provider answered with an error, or was never reached.""" + with self._locked() as doc: + if doc["held"].pop(booking, None) is not None: + self._write(doc) + + def state(self) -> dict: + """``max_usd``, ``spent_usd``, ``unsettled_usd``, ``held_usd`` (the + calls in flight), ``left_usd`` and ``closed``, read without the lock.""" + try: + return _state(self._read()) + except (OSError, ValueError, KeyError, TypeError) as e: + raise Unavailable(f"the spending cap's ledger could not be read: {type(e).__name__}: {e}") + + def closed(self) -> bool: + """Whether a call was refused: nothing more of the run is begun. A + ledger that cannot be read is closed - no call it guards is sent.""" + try: + return bool(self.state()["closed"]) + except Unavailable: + return True + + +def _state(doc: dict) -> dict: + held = sum(doc["held"].values()) + return {"max_usd": doc["max_usd"], "spent_usd": doc["spent_usd"], "unsettled_usd": doc["unsettled_usd"], "held_usd": held, + "left_usd": doc["max_usd"] - doc["spent_usd"] - doc["unsettled_usd"] - held, "closed": bool(doc["closed"])} + + +def usd(v: float) -> str: + """An amount as a person reads it: cents, and up to two digits more when it has them.""" + whole, _, cents = f"{v:.4f}".rstrip("0").partition(".") + return f"{whole}.{cents.ljust(2, '0')}" + + +def refusal(state: dict, amount: float) -> str: + """Why a call was not sent, as its caller and the record read it.""" + if state["closed"]: + return (f"an earlier call did not fit the run's spending cap of {usd(state['max_usd'])} USD " + f"({usd(state['spent_usd'])} spent): no further call of this run is sent") + flight = state["held_usd"] + state["unsettled_usd"] + return (f"the run's spending cap of {usd(state['max_usd'])} USD: {usd(state['spent_usd'])} spent" + + (f", {usd(flight)} booked by calls in flight or never settled" if flight else "") + + f", and this call may cost up to {usd(amount)}; it was not sent, and no further call of this run is") diff --git a/app/nullius/standard/relay.py b/app/nullius/standard/relay.py index f90dca3..bd12da3 100644 --- a/app/nullius/standard/relay.py +++ b/app/nullius/standard/relay.py @@ -23,6 +23,11 @@ the provider's headers come back the same way (its request id, its rate limits). It never answers for the provider: a 429 reaches the caller and the record as it was sent, which is what makes rate limits measurable. + +One answer is the relay's own: a run with a spending cap (``ledger``) books +each call before it goes out, and a call that does not fit is not sent. Its +caller gets a 402 that names the relay and the cap, and the record says +``refused``, so it never reads as the provider's error. """ from __future__ import annotations @@ -38,6 +43,7 @@ from typing import Dict, List, Optional from nullius.redact import redact +from nullius.standard import ledger as ledgers PLACEHOLDER_KEY = "nullius-relay" _TAG = re.compile(r"^[A-Za-z0-9_-]{1,64}$") @@ -330,7 +336,8 @@ def close(self): class Relay: def __init__(self, base: str, key: str, record_dir: str, prices: Optional[Dict[str, dict]] = None, - model: Optional[str] = None, side_models=(), key_header: str = "authorization", fields: bool = False): + model: Optional[str] = None, side_models=(), key_header: str = "authorization", fields: bool = False, + ledger: Optional[str] = None): """``base``: the provider's endpoint (``https://api.openai.com/v1``). ``model``: when given, every call is sent as this model, whatever the caller named - the run's model is the recipe's, not the agent's. A @@ -345,8 +352,12 @@ def __init__(self, base: str, key: str, record_dir: str, prices: Optional[Dict[s run's, recorded with ``model_asked``. ``key_header``: how the provider takes its key - a bearer ``authorization``, or Anthropic's ``x-api-key``. ``fields``: each record also holds the request's fields as sent - (:func:`request_fields`) - ``fidelity``'s, never a normal run's.""" + (:func:`request_fields`) - ``fidelity``'s, never a normal run's. + ``ledger``: the file of the run's spending cap (``ledger``), shared + by every relay of the job: a call is booked there before it is sent, + and not sent when it does not fit.""" self.fields = fields + self.ledger = ledgers.Ledger(ledger) if ledger else None self.base, self.key, self.dir = base.rstrip("/"), key, record_dir self.prices, self.model = prices or {}, model self.side_models = {a: (b or a) for a, _, b in (str(m).partition("=") for m in side_models or ())} @@ -434,8 +445,15 @@ def _forward(self, h: BaseHTTPRequestHandler): if req.get("stream") and rest.split("?")[0].endswith("/chat/completions"): req.setdefault("stream_options", {})["include_usage"] = True raw = json.dumps(req).encode() + booking = booked = None + if self.ledger is not None and h.command == "POST": + booking, booked, why = self._book(req if isinstance(req, dict) else {}, raw, rest) + if why: + self._refuse(h, tag, rest, req, asked, why) + return t0 = time.time() conn, r, back, body = None, None, [], b"" + cut = False # the reply was cut off: the call may have been billed try: conn, r = self.pool.send(h.command, rest, raw if h.command == "POST" else None, self._headers(h.headers)) status, kind, back = r.status, r.getheader("Content-Type") or "application/json", returned(r.headers) @@ -448,7 +466,7 @@ def _forward(self, h: BaseHTTPRequestHandler): try: body, whole = r.read(), True except Exception as e: # cut off mid-reply - status, kind, back = 502, "application/json", [] + status, kind, back, cut = 502, "application/json", [], True body = json.dumps({"error": {"message": "the endpoint's reply was cut off: " + self._said(e)}}).encode() def record(body): @@ -459,6 +477,7 @@ def record(body): rec["side"] = True if self.fields and isinstance(req, dict) and req: rec["fields"] = request_fields(req) + cost = None if status == 200: rec["reply"] = read_reply(rest, body, streamed) cost, how = self._cost(rec["model"], rec["reply"].get("usage")) @@ -466,6 +485,15 @@ def record(body): rec["cost_usd"], rec["cost_from"] = cost, how else: rec["error"] = _said(status, body) + if booking is not None: + rec["booked_usd"] = booked + try: + if status == 200 or cut: # what it cost; unknown (no token count, cut off), the booking is kept whole + self.ledger.settle(booking, cost) + else: # an error of the provider's, or never reached: not billed + self.ledger.release(booking) + except ledgers.Unavailable: + pass # the booking stays held, against the cap with self.lock: with open(os.path.join(self.dir, tag + ".jsonl"), "a") as f: f.write(json.dumps(rec) + "\n") @@ -515,6 +543,41 @@ def record(body): else: conn.close() + def _book(self, req: dict, raw: bytes, rest: str) -> tuple: + """(booking, its amount, None) for a call the cap lets out; (None, + its amount, why) for one it does not. A model with no price is not + sent either: a capped run sends no call it cannot price.""" + row = self.prices.get(req.get("model")) + if not row: + return None, None, f"no price for the model {req.get('model')!r}: a run with a spending cap sends no call it cannot price" + try: + amount = ledgers.bound(req, raw, row, rest) + booking, seen = self.ledger.reserve(amount) + except ledgers.Unavailable as e: + return None, None, str(e) + return booking, amount, None if booking else ledgers.refusal(seen, amount) + + def _refuse(self, h: BaseHTTPRequestHandler, tag: str, rest: str, req, asked, why: str) -> None: + """A call the run's spending cap did not let out: answered here and + recorded as the relay's own (``refused``), nothing sent.""" + t = time.time() + rec = {"t0": t, "t1": t, "path": rest, "model": req.get("model") if isinstance(req, dict) else None, "model_asked": asked, + "status": 402, "refused": "cap", "error": why, "request": _request_tail(req) if isinstance(req, dict) else {}, + "headers": sorted(k.lower() for k in forwarded(h.headers))} + with self.lock: + with open(os.path.join(self.dir, tag + ".jsonl"), "a") as f: + f.write(json.dumps(rec) + "\n") + body = json.dumps({"type": "error", "error": {"type": "spending_cap", "code": "spending_cap", + "message": "nullius relay: " + why}}).encode() + h.send_response(402) + for n, v in (("Content-Type", "application/json"), ("X-Nullius-Refused", "cap"), ("Content-Length", str(len(body)))): + h.send_header(n, v) + h.end_headers() + try: + h.wfile.write(body) + except OSError: + pass + # The address ``url`` hands out: the listening one unless the caller sets # another - a sidecar listens on 0.0.0.0 and is reached as ``relay``. public: Optional[str] = None @@ -553,7 +616,12 @@ def failure(calls: List[dict]) -> Optional[tuple]: """When a unit's model calls ended failing, the exception a trial carries for it - named so ``nullius.runs.classify`` sorts it as the infrastructure it was - and the endpoint's words. None when the last call - succeeded (a failure the agent recovered from is not the unit's).""" + succeeded (a failure the agent recovered from is not the unit's). A call + the run's spending cap refused is the unit's wherever it falls: once one + is refused, every later one is.""" + cut = next((c for c in calls if c.get("refused") == "cap"), None) + if cut: + return ledgers.ERROR, cut.get("error") or "" if not calls or calls[-1].get("status") == 200: return None last = calls[-1] @@ -570,19 +638,16 @@ def failure(calls: List[dict]) -> Optional[tuple]: # -- the relay as a process of its own: a container backend's sidecar ---------- -KEY_VAR = "NULLIUS_RELAY_UPSTREAM_KEY" - - def serve(argv=None) -> int: """``python3 -m nullius.standard.relay serve --listen HOST:PORT --base URL - --record DIR [--model M] [--route TAG=MODEL ...] [--side-model M ...] [--key-header NAME] - [--public URL] [--prices JSON]``. + --record DIR --key-file PATH [--model M] [--route TAG=MODEL ...] [--side-model M ...] [--key-header NAME] + [--public URL] [--prices JSON] [--ledger FILE]``. ``--base`` is the provider's endpoint, ``--route`` a route sent as another model (a benchmark's judge), ``--public`` the address ``url`` hands out. - The real key is read from ``NULLIUS_RELAY_UPSTREAM_KEY`` and nowhere else: - an argument is visible to anyone who lists processes or inspects a - container. One JSON line, ``{"listening": "http://host:port"}``, says it + The real key is read from ``--key-file`` and nowhere else: an argument or + an environment variable is visible to anyone who lists processes or + inspects a container. One JSON line, ``{"listening": "http://host:port"}``, says it is ready; SIGTERM (or SIGINT) stops it.""" import argparse import signal @@ -590,20 +655,27 @@ def serve(argv=None) -> int: ap.add_argument("--listen", default="127.0.0.1:0", help="HOST:PORT") ap.add_argument("--base", required=True, help="the provider's endpoint") ap.add_argument("--record", required=True, help="the folder each route's calls are recorded in") + ap.add_argument("--key-file", required=True, help="the file that holds the provider's key") ap.add_argument("--model", help="every call sent as this model") ap.add_argument("--route", action="append", default=[], help="TAG=MODEL") ap.add_argument("--side-model", action="append", default=[], help="a model the agent may call as asked") ap.add_argument("--key-header", default="authorization", choices=("authorization", "x-api-key")) ap.add_argument("--public", help="the address url() hands out, when not the listening one") ap.add_argument("--prices", help="a JSON object: model -> USD per million tokens, for a cost estimate") + ap.add_argument("--ledger", help="the file of the run's spending cap: a call is booked there before it is sent") a = ap.parse_args(argv) - key = os.environ.get(KEY_VAR) + try: + with open(a.key_file) as f: + key = f.read().strip() + except OSError as e: + print(json.dumps({"error": f"cannot read --key-file: {e}"}), flush=True) + return 2 if not key: - print(json.dumps({"error": f"{KEY_VAR} is not set"}), flush=True) + print(json.dumps({"error": f"--key-file {a.key_file} contains no key"}), flush=True) return 2 host, _, port = a.listen.rpartition(":") r = Relay(a.base, key, a.record, prices=json.loads(a.prices) if a.prices else None, model=a.model, - side_models=a.side_model, key_header=a.key_header) + side_models=a.side_model, key_header=a.key_header, ledger=a.ledger) for route in a.route: tag, _, model = route.partition("=") if not _TAG.match(tag) or not model: @@ -625,5 +697,5 @@ def serve(argv=None) -> int: if __name__ == "__main__": import sys if sys.argv[1:2] != ["serve"]: - sys.exit("python3 -m nullius.standard.relay serve --listen HOST:PORT --base URL --record DIR [--model M]") + sys.exit("python3 -m nullius.standard.relay serve --listen HOST:PORT --base URL --record DIR --key-file PATH [--model M]") sys.exit(serve(sys.argv[2:])) diff --git a/app/nullius/standard/run.py b/app/nullius/standard/run.py index b84d328..e99e51b 100644 --- a/app/nullius/standard/run.py +++ b/app/nullius/standard/run.py @@ -65,7 +65,7 @@ from nullius import paths, prices as price_list, progress, providers from nullius.redact import redact from nullius.harbor import feed as live_feed -from nullius.standard import acp, container, definitions, hostreq, install, job, jsonl, match, mcp_machine, relay, shard as shards +from nullius.standard import acp, container, definitions, hostreq, install, job, jsonl, ledger as ledgers, match, mcp_machine, relay, shard as shards # the totals a unit's end carries onto the live board, from its trial's agent_result FEED_TOTALS = ("n_input_tokens", "n_cache_tokens", "n_output_tokens", "cost_usd") @@ -308,7 +308,7 @@ def __init__(self, bench: dict, other: dict, *, model: str, provider: str, datas backend: str = "local", options: Optional[dict] = None, shard: Optional[tuple] = None, launch: Optional[str] = None, repeat: int = 1, simulator_model: Optional[str] = None, licences: Optional[Licences] = None, bench_options: Optional[dict] = None, which=shutil.which, - remove_images: bool = False, prices: Optional[dict] = None): + remove_images: bool = False, prices: Optional[dict] = None, max_usd: Optional[float] = None): """``proving``: the run is a conformance kit's, the one run that may start before the port it checks is proven - the family must still match. ``unsandboxed``: the caller's explicit yes to running an agent on the @@ -328,6 +328,11 @@ def __init__(self, bench: dict, other: dict, *, model: str, provider: str, datas (``simulator``): this model instead of its official one. ``prices``: this run's own prices, by model, over the user's list (``nullius.prices``): what a call's cost is estimated from. + ``max_usd``: the run's spending cap (``ledger``): every relay of the + job books a call before sending it, and once one does not fit no + further call goes out and nothing more is begun. Refused when a model + the run calls has no price: a call that cannot be priced cannot be + booked. ``licences``: what the run holds of the benchmark's licences (:class:`Licences`); by default its store on this machine. ``bench_options``: the benchmark's own options for this run @@ -402,6 +407,20 @@ def __init__(self, bench: dict, other: dict, *, model: str, provider: str, datas # (nullius.prices), this run's own prices over it. Read once, here: a run never fetches a price. self.run_prices = dict(prices or {}) self.prices = {**price_list.table(), **self.run_prices} + self.max_usd, self.ledger = None, None + if max_usd is not None: + try: + cap = float(max_usd) + except (TypeError, ValueError): + cap = 0.0 + if not 0 < cap < float("inf"): + raise Refused(f"--max-usd {max_usd!r}: the run's spending cap in USD, above zero") + unpriced = [m for m in self.models_called() if m not in self.prices] + if unpriced: + raise Refused("a spending cap needs a price for every model the run calls, and there is none for " + + ", ".join(unpriced) + ": nullius prices update, nullius prices set MODEL INPUT OUTPUT, " + "or --price MODEL=INPUT,OUTPUT") + self.max_usd = cap lacking = hostreq.missing(bench, needs["host"], which) if lacking: raise Refused(f"{', '.join(f'{k}={v}' for k, v in self.bench_options.items())} needs on this machine " @@ -479,6 +498,9 @@ def start(self) -> str: + (["dataset"] if self.ds.get("url") else []) + (["driver"] if install.driver_dir(self.bench) else []) + ([] if control or self.upstream else ["install"]) + ["units"])) + if self.max_usd is not None and self.base: + # one ledger for the job, in a folder of its own: all a unit's relay container is given of the job besides its record + self.ledger = ledgers.Ledger.create(os.path.join(self.job_dir, ".work", "ledger", "ledger.json"), self.max_usd) try: self.ready() except Refused as e: @@ -511,6 +533,7 @@ def start(self) -> str: results = [] u0 = time.time() self._halt, self._refused, self.halted = threading.Event(), [], None + self._cut: List[str] = [] # units the spending cap cut off # a model server on this machine's loopback, which the relays' containers cannot reach as written self.loopback = (container.Loopback(self.base).start() if self.backend == "docker" and self.base and container.on_loopback(self.base) else None) @@ -542,7 +565,8 @@ def work_unit(iur): finally: # a Stop or Ctrl-C: the units not begun are not begun, those running end with their agents pool.shutdown(wait=True, cancel_futures=True) - results = [f.result() for f in futures if not f.cancelled() and f.exception() is None and f.result() is not None] + begun = [not f.cancelled() and f.exception() is None and f.result() is not None for f in futures] + results = [f.result() for f, ran in zip(futures, begun) if ran] for d in self.drivers: d.close() if self.bench_relay is not None: @@ -551,11 +575,20 @@ def work_unit(iur): container.sweep(self.name) # whatever an interrupted unit left if self.loopback is not None: self.loopback.stop() - job.write_result(self.job_dir, self.name, t0, time.time(), results, len(work)) + budget = self.budget([u for n, (_, u, _) in enumerate(work) if n >= len(begun) or not begun[n]]) + job.write_result(self.job_dir, self.name, t0, time.time(), results, len(work), budget=budget) errored = sum(1 for x in results if x.get("exception_info")) self.feed.emit("job.end", n_completed=len(results) - errored, n_errored=errored, n_cancelled=len(work) - len(results)) p.emit("units", "done", seconds=round(time.time() - u0, 1)) self.n_verdicts, self.n_ended = sum(1 for x in results if x.get("verifier_result")), len(results) + if budget and budget["stopped"]: + spent = "unknown" if budget["spent_usd"] is None else ledgers.usd(budget["spent_usd"]) + p.emit("cap", detail=f"a call did not fit the spending cap of {ledgers.usd(budget['max_usd'])} USD", + spent_usd=spent, cut_off=len(budget["cut_off"]), not_run=len(budget["not_run"])) + self.say(f"stopped: a call did not fit the spending cap of {ledgers.usd(budget['max_usd'])} USD " + f"({spent} spent): {len(budget['cut_off'])} cut off, {len(budget['not_run'])} not run") + elif budget: + self.say(f"spent {ledgers.usd(budget['spent_usd'] or 0)} of the {ledgers.usd(budget['max_usd'])} USD cap") if self.halted: self.say(f"stopped: the endpoint refused {REFUSED_IN_A_ROW} units in a row ({self.halted[0]}: " f"{redact(str(self.halted[1]))[:200]}); the {len(work) - len(results)} units not begun were not run") @@ -626,7 +659,7 @@ def ready(self) -> None: # own, recorded in the job's .work/relay-benchmark beside the units', never counted as the agent's. if self.embed_model and self.bench_relay is None: self.bench_relay = relay.Relay(self.base, self.key, os.path.join(self.job_dir, ".work", "relay-benchmark"), - prices=self.prices, model=self.embed_model, + prices=self.prices, model=self.embed_model, ledger=self.ledger_file(), key_header=PROVIDERS[self.provider].get("key_header", "authorization")).start() def runtimes_needed(self) -> List[str]: @@ -657,6 +690,9 @@ def phase(self, name: str, unit: str, phase: str, timeout: Optional[float] = Non def unit(self, i: int, unit: str, r: int = 1) -> Optional[dict]: if self._stopped.is_set() or (getattr(self, "_halt", None) is not None and self._halt.is_set()): return None # the endpoint refused the run: nothing more is begun + if getattr(self, "ledger", None) is not None and self.ledger.closed(): + self._halt.set() # the spending cap refused a call: every later one is refused too + return None tag = f"u{i:04d}" + (f"r{r}" if self.repeat > 1 else "") self.local.attempt = r trial = os.path.join(self.job_dir, self.trial_name(i, unit, r)) @@ -759,6 +795,13 @@ def unit(self, i: int, unit: str, r: int = 1) -> Optional[dict]: if failed and (exc is None or exc[0] not in ("EnvironmentSetupError",)): exc = failed # what the endpoint said outranks how the agent took it score = None + if rl is not None and getattr(self, "ledger", None) is not None: + # a call the spending cap refused, the agent's, the judge's or the simulated user's: the unit was + # cut off by the run's own limit, whatever its agent made of the refusal - never a miss + cut = next((f for f in (relay.failure(rl.calls(t)) for t in (tag, f"{tag}-judge", f"{tag}-simulator")) + if f and f[0] == ledgers.ERROR), None) + if cut: + exc, score = cut, None if self._stopped.is_set() and score is None: # the Stop ended its agent, its relay or its containers under it: what it then failed with is # the Stop's doing, not the agent's nor the environment's (Harbor's name for a stopped trial) @@ -783,8 +826,30 @@ def unit(self, i: int, unit: str, r: int = 1) -> Optional[dict]: **{k: v for k, v in (res.get("agent_result") or {}).items() if k in FEED_TOTALS}, durations={"agent_run": round(produce[1] - produce[0], 3)} if produce[0] is not None else None) self.note_refusal(exc) + if exc and exc[0] == ledgers.ERROR and getattr(self, "_halt", None) is not None: + with self._said: + self._cut.append(unit) + self._halt.set() # the cap is reached: nothing more is begun return res + def budget(self, not_run: List[str]) -> Optional[dict]: + """What ``result.json`` says of the run's spending cap: the cap, + what was spent, what was booked for calls whose cost was never known + (``unsettled_usd``, said only when there is some), whether the cap + stopped the run, the units it cut off and those never begun. None + for a run without a cap.""" + if getattr(self, "ledger", None) is None: + return None + try: + got = self.ledger.state() + except ledgers.Unavailable: + got = {"spent_usd": None, "unsettled_usd": 0.0, "held_usd": 0.0, "closed": True} + unsettled = round(got["unsettled_usd"] + got["held_usd"], 6) + cut = getattr(self, "_cut", []) + return {"max_usd": self.max_usd, "spent_usd": None if got["spent_usd"] is None else round(got["spent_usd"], 6), + **({"unsettled_usd": unsettled} if unsettled else {}), "stopped": bool(got["closed"]), + "cut_off": list(dict.fromkeys(cut)), "not_run": list(dict.fromkeys(not_run)) if got["closed"] else []} + def note_refusal(self, exc: Optional[tuple]) -> None: """A unit's end, counted toward the endpoint refusing the run: a wrong key or an unknown model fails every unit the same way, and running @@ -897,6 +962,11 @@ def prices_record(self) -> Optional[dict]: listed = doc["from"] and any(r["source"] == "list" for r in rows.values()) return {**({"from": doc["from"], "on": doc["on"]} if listed else {}), "models": rows} + def ledger_file(self) -> Optional[str]: + """The file of the run's spending cap, for a relay; None without a cap.""" + led = getattr(self, "ledger", None) + return led.path if led is not None else None + def models_called(self) -> List[str]: """Every model this run's relays may send a call to, as sent: the run's, the judge's, the simulator's, an option's embeddings and the @@ -920,7 +990,7 @@ def unit_relay(self, tag: str, trial: str, setup: dict, env_def: Optional[dict] key_header = PROVIDERS[self.provider].get("key_header", "authorization") if self.backend == "local": rl = relay.Relay(self.base, self.key, record, prices=getattr(self, "prices", None), model=self.model, side_models=side, - key_header=key_header).start() + key_header=key_header, ledger=self.ledger_file()).start() rl.unit_url = rl.url(tag) return rl routes = {tag + "-judge": self.judge_model} if self.judge_model else {} @@ -940,7 +1010,8 @@ def unit_relay(self, tag: str, trial: str, setup: dict, env_def: Optional[dict] rl = container.Unit(self.name, tag, record, image=self.image, app=APP, base=door.base if door else self.base, key=self.key, model=self.model, routes=routes, side_models=side, key_header=key_header, prices=price_list.given(self.models_called(), getattr(self, "prices", None)), # a sidecar's argv: its models' rows - servers=self.agent_servers(env_def), doors=doors, hosts=door.hosts if door else None) + servers=self.agent_servers(env_def), doors=doors, hosts=door.hosts if door else None, + ledger=self.ledger_file()) rl.mcp_listeners = listeners try: try: diff --git a/app/tests/test_classify.py b/app/tests/test_classify.py index 9008a91..7ab35c9 100644 --- a/app/tests/test_classify.py +++ b/app/tests/test_classify.py @@ -92,6 +92,13 @@ def test_both_boards_charge_the_infrastructure_as_the_classifier_does(self): with self.subTest(name, board=board): self.assertIn(outcome(name), ("agent_error", "timeout")) + def test_a_unit_the_runs_spending_cap_cut_off_is_infrastructure_on_both(self): + """The standard's own: its relay refused the call (``standard.ledger``), so the unit had no attempt.""" + from nullius.standard import job, ledger + self.assertEqual(C.from_verdict(ledger.ERROR, None), C.API_ERROR) + self.assertIn(C.API_ERROR, C.INFRA_OUTCOMES) + self.assertEqual(job.feed_outcome("pass", None, (ledger.ERROR, "the run's spending cap"))["outcome"], "infra_error") + class HarborTaxonomy(unittest.TestCase): def test_provider_failures_are_infra(self): diff --git a/app/tests/test_mem0_capability.py b/app/tests/test_mem0_capability.py new file mode 100644 index 0000000..2ed54a7 --- /dev/null +++ b/app/tests/test_mem0_capability.py @@ -0,0 +1,53 @@ +"""The mem0 capability's add: what it hands Mem0, with a stand-in for Mem0.""" +import importlib.util +import os +import sys +import types +import unittest + +HERE = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "nullius", "catalogue", "capabilities", "mem0", "capability.py") + + +class FakeMem0: + def __init__(self): + self.added = [] + + def add(self, messages, **kw): + self.added.append(kw) + return {"results": []} + + def get_all(self, **kw): + return {"results": []} + + +def load(): + out = sys.stdout # the module sends stdout to stderr when it loads + try: + spec = importlib.util.spec_from_file_location("mem0_capability_under_test", HERE) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + finally: + sys.stdout = out + return mod + + +class Add(unittest.TestCase): + def setUp(self): + self.mod = load() + self.m = self.mod.Memory() + self.m.mem = FakeMem0() + + def test_the_session_date_is_also_the_timestamp_mem0_reads(self): + self.m.add({"id": "s1", "date": "2023/05/21 (Sun) 16:22", "turns": [{"role": "user", "text": "hi"}]}) + self.assertEqual(self.m.mem.added[0]["metadata"], + {"session": "s1", "date": "2023/05/21 (Sun) 16:22", "timestamp": "2023/05/21 (Sun) 16:22"}) + + def test_an_undated_session_has_no_timestamp(self): + # an empty timestamp would be an Observation Date of nothing; without one Mem0 keeps its default + self.m.add({"id": "s1", "turns": [{"role": "user", "text": "hi"}]}) + self.assertNotIn("timestamp", self.m.mem.added[0]["metadata"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_standard_container.py b/app/tests/test_standard_container.py index 490d186..ef8ddf3 100644 --- a/app/tests/test_standard_container.py +++ b/app/tests/test_standard_container.py @@ -7,8 +7,10 @@ import json import os import shutil +import stat import subprocess import tempfile +import time import unittest import urllib.error import urllib.request @@ -42,6 +44,7 @@ class ArgvTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, self.tmp, True) + self.enterContext(mock.patch.dict(os.environ, {"NULLIUS_HOME": os.path.join(self.tmp, "profile")})) def test_mounts_each_folder_once_at_its_own_path(self): ws, home, app = (os.path.join(self.tmp, n) for n in ("ws", "home", "app")) @@ -91,25 +94,34 @@ def test_the_agents_container_is_on_the_units_network_with_its_variables_by_name self.assertEqual(a[-2:], ["img", "codex-acp"]) self.assertNotIn("-p", a) # nothing published - def test_the_relays_container_names_the_key_and_never_holds_it(self): - a = container.relay_argv("img", name="nullius-j-u0000-relay", job="j", app="/app", record="/rec", + def test_the_relays_container_is_given_the_keys_file_never_the_key(self): + a = container.relay_argv("img", name="nullius-j-u0000-relay", job="j", app="/app", record="/rec", key_file="/k/key", base="https://api.example/v1", model="m", routes={"u0000-judge": "the-judge"}) - self.assertIn("NULLIUS_RELAY_UPSTREAM_KEY", a) - self.assertFalse(any(w.startswith("NULLIUS_RELAY_UPSTREAM_KEY=") for w in a)) + self.assertEqual(a[a.index("--mount") + 1], "type=bind,source=/k/key,target=/opt/nullius/relay-key,readonly") + self.assertEqual(a[a.index("--key-file") + 1], container.RELAY_KEY_IN) + self.assertFalse(any("KEY" in a[i + 1] for i, w in enumerate(a) if w == "-e")) self.assertIn("127.0.0.1::8080", a) # the driver's way in, this machine only self.assertIn("/app:/app:ro", a) self.assertIn("/rec:/rec:rw", a) self.assertEqual(a[a.index("--route") + 1], "u0000-judge=the-judge") self.assertEqual(a[a.index("--public") + 1], "http://relay:8080") self.assertEqual(a[a.index("--label") + 1], "nullius.job=j") + self.assertNotIn("--ledger", a) + + def test_the_relays_container_is_given_the_runs_ledger_and_its_folder_alone(self): + a = container.relay_argv("img", name="r", job="j", app="/app", record="/rec", key_file="/key", base="b", model="m", routes={}, + ledger="/job/.work/ledger/ledger.json") + self.assertEqual(a[a.index("--ledger") + 1], "/job/.work/ledger/ledger.json") + # where every relay of the job books its calls: that folder, and nothing else of the job but its own record + self.assertEqual([w for w in a if w.endswith(":rw")], ["/rec:/rec:rw", "/job/.work/ledger:/job/.work/ledger:rw"]) def test_an_endpoint_on_this_machines_loopback_is_reached_through_the_bridge(self): self.assertTrue(all(map(container.on_loopback, ["http://localhost:11434/v1", "http://127.0.0.1:8000/v1", "http://[::1]:8000/v1"]))) self.assertFalse(any(map(container.on_loopback, ["https://openrouter.ai/api/v1", "http://172.17.0.1:8000/v1", ""]))) - a = container.relay_argv("img", name="r", job="j", app="/app", record="/rec", base="http://host.docker.internal:4000/v1", + a = container.relay_argv("img", name="r", job="j", app="/app", record="/rec", key_file="/k", base="http://host.docker.internal:4000/v1", model="m", routes={}, hosts={"host.docker.internal": "172.17.0.1"}) self.assertEqual(a[a.index("--add-host") + 1], "host.docker.internal:172.17.0.1") - self.assertNotIn("--add-host", container.relay_argv("img", name="r", job="j", app="/app", record="/rec", base="b", model="m", routes={})) + self.assertNotIn("--add-host", container.relay_argv("img", name="r", job="j", app="/app", record="/rec", key_file="/k", base="b", model="m", routes={})) with mock.patch.object(container, "bridge_gateway", return_value=""): # Docker Desktop: the bridge is elsewhere door = container.Loopback("http://user@localhost:11434/v1?x=1").start() self.assertEqual((door.base, door.hosts), ("http://user@host.docker.internal:11434/v1?x=1", {"host.docker.internal": "host-gateway"})) @@ -144,6 +156,70 @@ def test_wrap_passes_values_in_the_docker_programs_environment_and_leaves_host_p self.assertNotEqual(env["PATH"], "/i/bin:" + container.IMAGE_PATH) # the docker program's own self.assertEqual(u.inside("u0002"), "http://relay:8080/u0002/v1") + def test_a_licence_forwarder_listens_before_it_takes_the_servers_name(self): + u = container.Unit("j", "u0001", os.path.join(self.tmp, "rec"), image="img", app="/app", base="b", key="k") + calls = [] + with mock.patch.object(container, "_docker", side_effect=lambda argv, **kw: calls.append(argv[:2])), \ + mock.patch.object(container, "wait_listening", side_effect=lambda name, *a: calls.append(["wait", name])): + self.assertEqual(u._forward(1, "licence.example:27000"), "licence.example:27000") + name = u.names["agent"] + "-lic1" + self.assertEqual(calls, [["run", "-d"], ["wait", name], ["network", "connect"]]) + + def start(self, fail_on=None): + """A unit started against a stand-in docker: each call's arguments and + environment, and the key's file as it was when the relay was run.""" + u = container.Unit("job", "u0000", os.path.join(self.tmp, "rec"), image="img", app="/app", base="b", key="sk-real") + calls, seen = [], {} + + def docker(args, env=None, **_): + calls.append((args, env or {})) + if args[0] == "run": + src = args[args.index("--mount") + 1].split(",")[1].removeprefix("source=") + with open(src) as f: + seen.update(path=src, key=f.read(), mode=stat.S_IMODE(os.stat(src).st_mode), + dir_mode=stat.S_IMODE(os.stat(os.path.dirname(src)).st_mode)) + if args[:2] == fail_on: + raise container.Unavailable("docker failed") + return subprocess.CompletedProcess(args, 0, "0.0.0.0:32768\n" if args[0] == "port" else "", "") + self.enterContext(mock.patch.object(container, "_docker", docker)) # for the test's stop too + self.enterContext(mock.patch.object(container, "wait_listening")) + if fail_on: + with self.assertRaises(container.Unavailable): + u.start() + else: + u.start() + return u, calls, seen + + def test_the_key_reaches_the_relay_as_an_owner_only_file_removed_once_read(self): + u, calls, seen = self.start() + self.assertEqual((seen["key"], seen["mode"], seen["dir_mode"]), ("sk-real", 0o600, 0o700)) + self.assertEqual(os.path.dirname(os.path.dirname(seen["path"])), container.key_root()) # never the record's folder + self.assertEqual(stat.S_IMODE(os.stat(container.key_root()).st_mode), 0o700) + self.assertFalse(os.path.exists(os.path.dirname(seen["path"]))) # gone once the relay listens + self.assertNotIn("sk-real", json.dumps(calls)) # no argument, no variable + u.stop() + + def test_a_failed_start_removes_the_keys_file(self): + u, calls, seen = self.start(fail_on=["network", "connect"]) # after the relay's run + self.assertEqual(calls[-1][0][:2], ["network", "rm"]) + self.assertFalse(os.path.exists(os.path.dirname(seen["path"]))) + self.assertIsNone(u.key_dir) + + def test_a_sweep_removes_the_jobs_key_folders_and_no_other_jobs(self): + os.makedirs(container.key_root()) + mine, other = (tempfile.mkdtemp(prefix=container.key_prefix(j), dir=container.key_root()) for j in ("job", "job-2")) + with mock.patch.object(container, "_docker", side_effect=container.Unavailable("no docker")): + container.sweep("job") + self.assertEqual((os.path.exists(mine), os.path.exists(other)), (False, True)) + + def test_a_start_removes_stale_key_folders_of_any_job(self): + os.makedirs(container.key_root()) + old, fresh = (tempfile.mkdtemp(prefix=container.key_prefix(j), dir=container.key_root()) for j in ("dead", "live")) + t = time.time() - container.KEY_STALE - 1 + os.utime(old, (t, t)) + self.start()[0].stop() + self.assertEqual((os.path.exists(old), os.path.exists(fresh)), (False, True)) + def test_a_long_job_name_gives_a_short_unique_slug(self): a, b = container.slug("x" * 200), container.slug("x" * 199 + "y") self.assertLessEqual(len(a), 80) @@ -174,6 +250,9 @@ def test_the_relay_is_the_only_way_out_and_nothing_is_left(self): base="http://127.0.0.1:9/v1", key="sk-test-not-real") # an endpoint no one serves try: u.start() + seen = subprocess.run(["docker", "inspect", u.names["relay"]], stdout=subprocess.PIPE, text=True).stdout + self.assertIn(container.RELAY_KEY_IN, seen) + self.assertNotIn("sk-test-not-real", seen) # the key is in no configuration with self.assertRaises(urllib.error.HTTPError) as e: # the host reaches it urllib.request.urlopen(urllib.request.Request(u.url("u0000") + "/chat/completions", data=b"{}"), timeout=30) self.assertEqual(e.exception.code, 502) diff --git a/app/tests/test_standard_forward.py b/app/tests/test_standard_forward.py new file mode 100644 index 0000000..3d45063 --- /dev/null +++ b/app/tests/test_standard_forward.py @@ -0,0 +1,121 @@ +"""The licence server's forwarder (``forward.py``), run in-process: it +resolves its target once, before it listens, tries each address in order, and +ends when the name does not resolve.""" +from __future__ import annotations + +import contextlib +import io +import json +import socket +import threading +import time +import unittest +from unittest import mock + +from nullius.standard import forward + + +def _greeter(greeting: bytes, host: str = "127.0.0.1") -> int: + family, _, _, _, addr = socket.getaddrinfo(host, 0, type=socket.SOCK_STREAM)[0] # where the name goes first + srv = socket.socket(family, socket.SOCK_STREAM) + srv.bind(addr[:2]) + srv.listen(8) + + def serve(): + while True: + conn, _ = srv.accept() + conn.sendall(greeting) + conn.close() + threading.Thread(target=serve, daemon=True).start() + return srv.getsockname()[1] + + +def _free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _forwarding(to: str) -> int: + """A forwarder to ``to`` in a thread: the port it listens on, once it listens.""" + listen = _free_port() + threading.Thread(target=forward.main, args=(["--listen", f"127.0.0.1:{listen}", "--to", to],), daemon=True).start() + deadline = time.time() + 10 + while True: + try: + with socket.socket() as s: # by address: no lookup, which a test may have patched + s.settimeout(1) + s.connect(("127.0.0.1", listen)) + return listen + except OSError: + if time.time() > deadline: + raise + time.sleep(0.05) + + +def _greeting(port: int) -> bytes: + with socket.socket() as c: + c.settimeout(10) + c.connect(("127.0.0.1", port)) + return c.recv(64) + + +class ForwardTest(unittest.TestCase): + def test_a_named_server_is_resolved_once_so_a_later_lookup_cannot_redirect_it(self): + real = _greeter(b"LICENSE-OK", "localhost") + listen = _forwarding(f"localhost:{real}") + decoy = _greeter(b"SELF") # the name now answers as the forwarder's own + later = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", decoy))] + with mock.patch("socket.getaddrinfo", return_value=later): + self.assertEqual(_greeting(listen), b"LICENSE-OK") + + def test_each_connection_takes_the_names_first_address_that_accepts(self): + first, second = _greeter(b"FIRST"), _greeter(b"SECOND") + both = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", first)), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", second))] + with mock.patch("socket.getaddrinfo", return_value=both): + listen = _forwarding("licence.example:27000") + self.assertEqual([_greeting(listen) for _ in range(3)], [b"FIRST"] * 3) + + def test_a_refused_address_is_passed_over_for_the_next(self): + real = _greeter(b"LICENSE-OK") + dead = socket.socket() # bound, never listening: refused, and no one else can take it + self.addCleanup(dead.close) + dead.bind(("127.0.0.1", 0)) + both = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", dead.getsockname()), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", real))] + with mock.patch("socket.getaddrinfo", return_value=both): + listen = _forwarding("licence.example:27000") + self.assertEqual(_greeting(listen), b"LICENSE-OK") + + def test_an_address_family_this_machine_lacks_is_skipped(self): + real, make = _greeter(b"LICENSE-OK"), socket.socket + + def no_ipv6(family=-1, *args, **kwargs): # the greeter's accept makes sockets too + if family == socket.AF_INET6: + raise OSError(97, "Address family not supported by protocol") + return make(family, *args, **kwargs) + with mock.patch("socket.socket", side_effect=no_ipv6): + up = forward._connect([(socket.AF_INET6, ("::1", real)), (socket.AF_INET, ("127.0.0.1", real))]) + with up: + self.assertEqual(up.recv(64), b"LICENSE-OK") + + def test_no_address_at_all_is_an_error_not_a_crash(self): + with self.assertRaises(OSError): + forward._connect([]) + + def test_a_name_that_does_not_resolve_ends_it_with_the_reason(self): + out, ended = io.StringIO(), [] + argv = ["--listen", f"127.0.0.1:{_free_port()}", "--to", "nowhere.invalid:27000"] + with mock.patch("socket.getaddrinfo", side_effect=socket.gaierror("no such name")), \ + contextlib.redirect_stdout(out): + t = threading.Thread(target=lambda: ended.append(forward.main(argv)), daemon=True) + t.start() + t.join(10) # a forwarder that listened anyway never returns + self.assertEqual(ended, [1]) + event = json.loads(next(line for line in out.getvalue().splitlines() if '"unresolved"' in line)) + self.assertEqual((event["event"], event["host"]), ("unresolved", "nowhere.invalid")) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_standard_ledger.py b/app/tests/test_standard_ledger.py new file mode 100644 index 0000000..8ac64c6 --- /dev/null +++ b/app/tests/test_standard_ledger.py @@ -0,0 +1,138 @@ +"""A run's spending cap: a call is booked before it is sent, sent only while +it fits, and the first that does not closes the ledger for every later one.""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import threading +import unittest + +from nullius.standard import ledger + +ROW = {"input": 1.0, "output": 10.0} + + +class LedgerTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.path = os.path.join(self.tmp, "ledger", "ledger.json") + + def test_a_call_is_sent_while_spent_and_in_flight_and_its_booking_fit(self): + led = ledger.Ledger.create(self.path, 1.0) + a, _ = led.reserve(0.4) + b, _ = led.reserve(0.4) + self.assertTrue(a and b and a != b) + self.assertAlmostEqual(led.state()["held_usd"], 0.8) + led.settle(a, 0.05) # the reply: what it cost, in place of what was booked + self.assertAlmostEqual(led.state()["spent_usd"], 0.05) + self.assertAlmostEqual(led.state()["left_usd"], 1.0 - 0.05 - 0.4) + c, _ = led.reserve(0.5) + self.assertTrue(c) + self.assertFalse(led.closed()) + + def test_the_first_call_that_does_not_fit_closes_it_for_every_later_one(self): + led = ledger.Ledger.create(self.path, 1.0) + a, _ = led.reserve(0.7) + refused, seen = led.reserve(0.4) # 0.7 in flight: 0.4 more would pass the cap + self.assertIsNone(refused) + self.assertFalse(seen["closed"]) # the state it was refused in, for its message + self.assertTrue(led.closed()) + led.settle(a, 0.01) + self.assertIsNone(led.reserve(0.0001)[0]) # room again, and still closed: a run stops at one point + said = ledger.refusal(led.state(), 0.0001) + self.assertIn("1.00 USD", said) + self.assertIn("no further call", said) + + def test_a_call_whose_cost_was_never_known_keeps_its_booking_and_an_error_gives_it_back(self): + led = ledger.Ledger.create(self.path, 1.0) + a, _ = led.reserve(0.3) + b, _ = led.reserve(0.3) + led.settle(a, None) # no token count, or cut off: it may have been billed + led.release(b) # the provider's error: not billed + led.release(b) # twice is once + led.settle("unknown", 9.0) # a booking it does not hold changes nothing + got = led.state() + self.assertEqual((got["spent_usd"], got["unsettled_usd"], got["held_usd"]), (0.0, 0.3, 0)) + self.assertAlmostEqual(got["left_usd"], 0.7) + + def test_a_ledger_that_cannot_be_read_lets_no_call_out(self): + led = ledger.Ledger.create(self.path, 1.0) + with open(self.path, "w") as f: + f.write("{not json") + with self.assertRaises(ledger.Unavailable): + led.reserve(0.1) + self.assertTrue(led.closed()) + self.assertTrue(ledger.Ledger(os.path.join(self.tmp, "none.json")).closed()) + + def test_many_calls_at_once_never_book_past_the_cap(self): + led = ledger.Ledger.create(self.path, 1.0) + got = [] + + def ask(): + got.append(led.reserve(0.03)[0]) + + threads = [threading.Thread(target=ask) for _ in range(60)] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertEqual(sum(1 for b in got if b), 33) # 33 x 0.03 fit in 1.00, the 34th does not + self.assertLessEqual(led.state()["held_usd"], 1.0) + + def test_relays_in_other_processes_book_in_the_same_ledger(self): + """What a docker unit's relay does from its container: each process + books and settles; none of it is lost between them.""" + led = ledger.Ledger.create(self.path, 100.0) + app = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + code = ("import sys\nfrom nullius.standard import ledger\nled = ledger.Ledger(sys.argv[1])\n" + "for _ in range(40):\n b, _ = led.reserve(0.5)\n led.settle(b, 0.01)\n") + procs = [subprocess.Popen([sys.executable, "-c", code, self.path], env={"PATH": os.environ.get("PATH", ""), "PYTHONPATH": app}) + for _ in range(4)] + self.assertEqual([p.wait(timeout=60) for p in procs], [0, 0, 0, 0]) + got = led.state() + self.assertAlmostEqual(got["spent_usd"], 4 * 40 * 0.01) + self.assertEqual((got["held_usd"], got["closed"]), (0, False)) + with open(self.path) as f: + self.assertEqual(json.load(f)["held"], {}) + + +class BoundTest(unittest.TestCase): + def bound(self, req, row=ROW, path="/chat/completions"): + return ledger.bound(req, json.dumps(req).encode(), row, path) + + def test_the_limit_the_call_names_is_its_output(self): + req = {"model": "m", "messages": [{"role": "user", "content": "x" * 300}], "max_tokens": 100} + tokens = -(-len(json.dumps(req)) // 3) # its characters over three, rounded up + self.assertAlmostEqual(self.bound(req), (tokens * 1.0 + 100 * 10.0) / 1e6) + for name in ("max_completion_tokens", "max_output_tokens"): + other = {"model": "m", "input": "hi", name: 50} + self.assertAlmostEqual(self.bound(other), (-(-len(json.dumps(other)) // 3) * 1.0 + 50 * 10.0) / 1e6) + + def test_a_call_that_names_none_is_booked_the_models_own_limit_else_the_default(self): + req = {"model": "m", "messages": []} + tokens = -(-len(json.dumps(req)) // 3) + self.assertAlmostEqual(self.bound(req, dict(ROW, max_output=2000)), (tokens + 2000 * 10.0) / 1e6) + self.assertAlmostEqual(self.bound(req), (tokens + ledger.OUTPUT * 10.0) / 1e6) + + def test_input_is_booked_at_the_dearer_of_input_and_cache_write(self): + req = {"model": "m", "messages": [], "max_tokens": 1} + tokens = -(-len(json.dumps(req)) // 3) + self.assertAlmostEqual(self.bound(req, dict(ROW, cache_write=1.25)), (tokens * 1.25 + 10.0) / 1e6) + + def test_an_embeddings_call_has_no_output_and_an_encoded_file_is_not_a_prompt(self): + req = {"model": "e", "input": ["some text"]} + self.assertAlmostEqual(self.bound(req, {"input": 0.02, "output": 5.0}, "/embeddings"), -(-len(json.dumps(req)) // 3) * 0.02 / 1e6) + picture = "A" * 30000 + plain = {"model": "m", "max_tokens": 1, "messages": [{"role": "user", "content": [ + {"type": "image_url", "image_url": {"url": ""}}, {"type": "image", "source": {"type": "base64", "data": ""}}]}]} + with_files = json.loads(json.dumps(plain)) + with_files["messages"][0]["content"][0]["image_url"]["url"] = "data:image/png;base64," + picture + with_files["messages"][0]["content"][1]["source"]["data"] = picture + self.assertAlmostEqual(self.bound(with_files), self.bound(plain)) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_standard_relay.py b/app/tests/test_standard_relay.py index 3d79e5a..88928d9 100644 --- a/app/tests/test_standard_relay.py +++ b/app/tests/test_standard_relay.py @@ -5,6 +5,7 @@ import json import os +import shutil import sys import tempfile import threading @@ -15,7 +16,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from nullius.runs import classify -from nullius.standard import jsonl, relay +from nullius.standard import jsonl, ledger, relay CHAT = {"choices": [{"message": {"role": "assistant", "content": "hello"}}], "usage": {"prompt_tokens": 10, "completion_tokens": 2, "prompt_tokens_details": {"cached_tokens": 4}}} @@ -44,6 +45,8 @@ def do_POST(self): self.reply(429, b'{"error":{"message":"Rate limit reached"}}', "application/json") elif mode == "401": self.reply(401, b'{"error":{"message":"Incorrect API key"}}', "application/json") + elif mode == "no-usage": + self.reply(200, b'{"choices":[{"message":{"role":"assistant","content":"hello"}}]}', "application/json") elif self.path.endswith("/responses"): self.reply(200, RESP_SSE, "text/event-stream") elif body.get("stream"): @@ -177,6 +180,107 @@ def test_an_advertised_address_is_what_url_hands_out(self): r.stop() +class CapTest(unittest.TestCase): + """A relay of a run with a spending cap: a call is booked before it goes + out and is not sent when it does not fit - the one answer that is the + relay's own, and marked so.""" + PRICES = {"m": {"input": 1.0, "cached": 0.1, "output": 10.0}} + + @classmethod + def setUpClass(cls): + cls.up = ThreadingHTTPServer(("127.0.0.1", 0), Upstream) + threading.Thread(target=cls.up.serve_forever, daemon=True).start() + cls.base = f"http://127.0.0.1:{cls.up.server_address[1]}/v1" + + @classmethod + def tearDownClass(cls): + cls.up.shutdown() + + def capped(self, cap): + tmp = tempfile.mkdtemp() + led = ledger.Ledger.create(os.path.join(tmp, "ledger", "ledger.json"), cap) + r = relay.Relay(self.base, "sk-real", os.path.join(tmp, "relay"), prices=self.PRICES, ledger=led.path).start() + self.addCleanup(r.stop) + return r, led + + def post(self, r, tag, body): + req = urllib.request.Request(r.url(tag) + "/chat/completions", data=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}) + try: + with urllib.request.urlopen(req) as got: + return got.status, dict(got.headers.items()), got.read() + except urllib.error.HTTPError as e: + return e.code, dict(e.headers.items()), e.read() + + def test_a_call_that_fits_is_sent_and_its_booking_becomes_what_it_cost(self): + r, led = self.capped(1.0) + body = {"model": "m", "messages": [{"role": "user", "content": "hi"}], "max_tokens": 100} + status, _, _ = self.post(r, "u1", body) + self.assertEqual(status, 200) + rec = r.calls("u1")[-1] + self.assertAlmostEqual(rec["booked_usd"], ledger.bound(body, json.dumps(body).encode(), self.PRICES["m"])) + got = led.state() + self.assertAlmostEqual(got["spent_usd"], rec["cost_usd"]) + self.assertEqual((got["held_usd"], got["unsettled_usd"], got["closed"]), (0, 0.0, False)) + + def test_a_call_that_does_not_fit_is_answered_by_the_relay_and_never_sent(self): + r, led = self.capped(0.0005) # 100 output tokens at 10 USD a million are 0.001 already + sent = len(Upstream.seen) + status, headers, body = self.post(r, "u1", {"model": "m", "messages": [{"role": "user", "content": "hi"}], "max_tokens": 100}) + self.assertEqual((status, headers.get("X-Nullius-Refused")), (402, "cap")) + error = json.loads(body)["error"] + self.assertEqual(error["type"], "spending_cap") + self.assertIn("nullius relay: the run's spending cap of 0.0005 USD", error["message"]) + rec = r.calls("u1")[-1] + self.assertEqual((rec["status"], rec["refused"]), (402, "cap")) + self.assertNotIn("reply", rec) + self.assertTrue(led.closed()) + # a call small enough to fit is refused too: a run stops at one point + status, _, _ = self.post(r, "u1", {"model": "m", "messages": [], "max_tokens": 1}) + self.assertEqual(status, 402) + self.assertEqual(len(Upstream.seen), sent) # neither reached the provider + self.assertEqual(led.state()["spent_usd"], 0.0) + exc, said = relay.failure(r.calls("u1")) + self.assertEqual(exc, ledger.ERROR) + self.assertIn("spending cap", said) + self.assertEqual(classify.from_verdict(exc, None), classify.API_ERROR) # infrastructure, never a miss + self.assertIn(classify.API_ERROR, classify.INFRA_OUTCOMES) + + def test_a_refused_call_is_the_units_whatever_its_agent_did_next(self): + calls = [{"status": 200}, {"status": 402, "refused": "cap", "error": "the run's spending cap"}, {"status": 200}] + self.assertEqual(relay.failure(calls), (ledger.ERROR, "the run's spending cap")) + + def test_an_error_gives_the_booking_back_and_a_reply_without_tokens_keeps_it(self): + r, led = self.capped(1.0) + status, _, _ = self.post(r, "u1", {"model": "m", "mode": "429", "max_tokens": 100}) + self.assertEqual(status, 429) # the provider's own, passed on as ever + got = led.state() + self.assertEqual((got["spent_usd"], got["unsettled_usd"], got["held_usd"]), (0.0, 0.0, 0)) + status, _, _ = self.post(r, "u1", {"model": "m", "mode": "no-usage", "max_tokens": 100}) + self.assertEqual(status, 200) + rec = r.calls("u1")[-1] + self.assertNotIn("cost_usd", rec) + got = led.state() + self.assertEqual((got["spent_usd"], got["held_usd"]), (0.0, 0)) + self.assertAlmostEqual(got["unsettled_usd"], rec["booked_usd"]) # it may have been billed + + def test_a_model_without_a_price_is_not_sent_and_closes_nothing(self): + r, led = self.capped(1.0) + sent = len(Upstream.seen) + status, _, body = self.post(r, "u1", {"model": "another", "messages": []}) + self.assertEqual(status, 402) + self.assertIn("no price for the model 'another'", json.loads(body)["error"]["message"]) + self.assertEqual(len(Upstream.seen), sent) + self.assertFalse(led.closed()) + + def test_a_relay_without_a_cap_books_nothing(self): + r = relay.Relay(self.base, "sk-real", tempfile.mkdtemp(), prices=self.PRICES).start() + self.addCleanup(r.stop) + status, _, _ = self.post(r, "u1", {"model": "another", "messages": []}) + self.assertEqual(status, 200) + self.assertNotIn("booked_usd", r.calls("u1")[-1]) + + class EffortTest(unittest.TestCase): def test_the_record_says_the_reasoning_effort_a_request_asked_for(self): # chat's reasoning_effort, the Responses API's reasoning.effort, and none at all @@ -201,23 +305,26 @@ def tearDownClass(cls): def serve(self, key, *extra): import subprocess + d = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, d, True) + with open(os.path.join(d, "key"), "w") as f: + f.write(key + "\n") env = {"PATH": os.environ.get("PATH", ""), "PYTHONPATH": self.app} - if key: - env[relay.KEY_VAR] = key return subprocess.Popen([sys.executable, "-m", "nullius.standard.relay", "serve", "--listen", "127.0.0.1:0", - "--base", f"http://127.0.0.1:{self.up.server_address[1]}/v1", "--record", tempfile.mkdtemp(), - "--model", "the-recipes", *extra], env=env, stdout=subprocess.PIPE, text=True) + "--base", f"http://127.0.0.1:{self.up.server_address[1]}/v1", "--record", d, + "--key-file", os.path.join(d, "key"), "--model", "the-recipes", *extra], + env=env, stdout=subprocess.PIPE, text=True) - def test_serve_relays_with_the_key_from_its_environment_and_stops_on_sigterm(self): + def test_serve_relays_with_the_key_from_its_file_and_stops_on_sigterm(self): import signal - p = self.serve("sk-from-env", "--route", "u1-judge=the-judge") + p = self.serve("sk-from-file", "--route", "u1-judge=the-judge") try: where = json.loads(p.stdout.readline())["listening"] for tag, model in (("u1", "the-recipes"), ("u1-judge", "the-judge")): req = urllib.request.Request(f"{where}/{tag}/v1/chat/completions", data=b'{"model":"asked"}', headers={"Content-Type": "application/json"}) urllib.request.urlopen(req).read() - self.assertEqual(Upstream.seen[-1][1], "Bearer sk-from-env") + self.assertEqual(Upstream.seen[-1][1], "Bearer sk-from-file") self.assertEqual(Upstream.seen[-1][2]["model"], model) p.send_signal(signal.SIGTERM) self.assertEqual(p.wait(timeout=10), 0) @@ -226,9 +333,36 @@ def test_serve_relays_with_the_key_from_its_environment_and_stops_on_sigterm(sel p.kill() p.stdout.close() - def test_serve_without_the_key_in_its_environment_refuses(self): + def test_serve_books_its_calls_in_the_runs_ledger(self): + """A docker unit's relay: the ledger is a file it is given, shared with every other relay of the job.""" + import signal + led = ledger.Ledger.create(os.path.join(tempfile.mkdtemp(), "ledger", "ledger.json"), 1.0) + p = self.serve("sk-from-env", "--ledger", led.path, "--prices", json.dumps({"the-recipes": {"input": 1.0, "output": 10.0}})) + try: + where = json.loads(p.stdout.readline())["listening"] + req = urllib.request.Request(f"{where}/u1/v1/chat/completions", data=b'{"model":"asked","max_tokens":5}', + headers={"Content-Type": "application/json"}) + urllib.request.urlopen(req).read() + self.assertAlmostEqual(led.state()["spent_usd"], (10 * 1.0 + 2 * 10.0) / 1e6) + p.send_signal(signal.SIGTERM) + self.assertEqual(p.wait(timeout=10), 0) + finally: + if p.poll() is None: + p.kill() + p.stdout.close() + + def test_serve_without_its_key_file_refuses(self): + import subprocess + p = subprocess.Popen([sys.executable, "-m", "nullius.standard.relay", "serve", "--base", "http://127.0.0.1:9/v1", + "--record", tempfile.mkdtemp(), "--key-file", "/nonexistent/key"], + env={"PATH": os.environ.get("PATH", ""), "PYTHONPATH": self.app}, stdout=subprocess.PIPE, text=True) + self.assertIn("cannot read --key-file", p.stdout.readline()) + self.assertEqual(p.wait(timeout=10), 2) + p.stdout.close() + + def test_serve_with_an_empty_key_file_refuses(self): p = self.serve("") - self.assertIn(relay.KEY_VAR, p.stdout.readline()) + self.assertIn("contains no key", p.stdout.readline()) self.assertEqual(p.wait(timeout=10), 2) p.stdout.close() diff --git a/app/tests/test_standard_run.py b/app/tests/test_standard_run.py index c9856e8..390bf1c 100644 --- a/app/tests/test_standard_run.py +++ b/app/tests/test_standard_run.py @@ -13,7 +13,7 @@ from unittest import mock from nullius.runs import classify, parse -from nullius.standard import acp, container, definitions, install, run +from nullius.standard import acp, container, definitions, install, ledger, run ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures", "standard", "catalogue") # A conversation whose user a model plays: it opens, speaks once more, then is done. @@ -292,6 +292,63 @@ def recorded(doc): # the user's own row over the list's: the list did not say this price, so it is not named self.assertEqual(recorded({"models": {"m-1": listed}, "own": {"m-1": own}}), {"models": {"m-1": {**own, "source": "own"}}}) + def calling(self, **kw): + """Eight units, each one model call (11 tokens in, 3 out), one at a + time. Input is free here and the model's own output limit is 100 + tokens at 10 USD a million, so a call is booked 0.001 and costs 0.00003.""" + here = os.path.join(self.tmp, "calling") + os.makedirs(here, exist_ok=True) + with open(os.path.join(here, "driver.py"), "w") as f: + f.write(CALLING_DRIVER) + b = dict(definitions.get("benchmark", "fakebench", ROOT), _here=here, driver={"command": ["python3", "{here}/driver.py"]}) + said = [] + kw = {"prices": {"m-1": {"input": 0.0, "output": 10.0, "max_output": 100}}, **kw} + job_dir, failed = run.run_and_report(b, definitions.get("agent", "fake", ROOT), model="m-1", provider="custom", jobs_root=self.tmp, + concurrency=1, env=self.env, say=said.append, unsandboxed=True, proving=True, **kw) + with open(os.path.join(job_dir, "result.json")) as f: + return job_dir, failed, json.load(f), said + + def test_a_spending_cap_stops_the_run_and_its_result_says_where(self): + # 0.001 booked a call, 0.00003 spent by each: the fourth no longer fits 0.00107 + job_dir, failed, result, said = self.calling(max_usd=0.00107) + self.assertEqual(result["budget"], {"max_usd": 0.00107, "spent_usd": 0.00009, "stopped": True, + "cut_off": ["m3"], "not_run": ["m4", "m5", "m6", "m7"]}) + self.assertEqual((result["n_total_trials"], result["stats"]["n_cancelled_trials"], result["stats"]["n_errored_trials"]), (8, 4, 1)) + self.assertFalse(failed) # three units reached a verdict + got, _ = self.outcomes(job_dir) + self.assertEqual(sorted(got), ["m0", "m1", "m2", "m3"]) # the four never begun have no trial + self.assertEqual(got["m3"].exception_type, ledger.ERROR) + self.assertEqual(got["m3"].outcome, classify.API_ERROR) # infrastructure: never a miss + self.assertTrue(all(got[u].outcome == classify.RESOLVED for u in ("m0", "m1", "m2"))) + calls = self.relay_calls(job_dir) + self.assertEqual([c.get("refused") for c in calls].count("cap"), 1) + sent = [c for c in calls if not c.get("refused")] + self.assertEqual(len(sent), 3) # no call went out past the cap + self.assertTrue(all(abs(c["booked_usd"] - 0.001) < 1e-12 for c in sent)) + self.assertLessEqual(sum(c["cost_usd"] for c in sent), 0.00107) + self.assertTrue(any(s.startswith("stopped: a call did not fit the spending cap of 0.0011 USD (0.0001 spent): 1 cut off, 4 not run") + for s in said), said) + self.assert_no_key(job_dir) + + def test_a_spending_cap_the_run_stays_under_changes_nothing_but_the_result(self): + job_dir, failed, result, said = self.calling(max_usd=1.0) + self.assertEqual(result["budget"], {"max_usd": 1.0, "spent_usd": 0.00024, "stopped": False, "cut_off": [], "not_run": []}) + self.assertEqual(len(self.outcomes(job_dir)[0]), 8) + self.assertIn("spent 0.0002 of the 1.00 USD cap", said) + job_dir, _, result, _ = self.calling(name="uncapped") + self.assertNotIn("budget", result) + self.assertFalse(os.path.exists(os.path.join(job_dir, ".work", "ledger"))) + self.assertFalse(any("booked_usd" in c for c in self.relay_calls(job_dir))) + + def test_a_spending_cap_is_refused_without_a_price_for_each_model_or_an_amount(self): + with self.assertRaises(run.Refused) as e: + self.calling(max_usd=1.0, prices={"another": {"input": 1.0, "output": 1.0}}) + self.assertIn("there is none for m-1", str(e.exception)) + for bad in (0, -1, float("inf"), float("nan"), "much"): + with self.assertRaises(run.Refused, msg=bad): + self.calling(max_usd=bad) + self.assertEqual([n for n in os.listdir(self.tmp) if "fakebench" in n], []) # nothing was begun + def test_a_run_with_no_price_has_no_cost_and_says_nothing_of_prices(self): job_dir = self.go("fakebench", "fake") calls = self.relay_calls(job_dir) diff --git a/docs/builder.md b/docs/builder.md index e4efa5f..c227b43 100644 --- a/docs/builder.md +++ b/docs/builder.md @@ -88,7 +88,7 @@ a declared model list keeps those published choices. | --- | --- | | Order | `nop`, `oracle`, then the agents under evaluation | | Each Run | A new job (recipe name + timestamp); never resumes | -| Key and endpoint | Passed as environment (over stdin to a host), never written; logs redacted. Other secrets withheld | +| Key and endpoint | Passed as environment (over stdin to a host), the key to a docker relay as a file; logs redacted. Other secrets withheld | | A host | Gets a content-addressed snapshot of the running backend and effective catalogue into `~/.nullius/app/`, including source edits and user entries but no configuration stores; a Harbor recipe gets its pin's wheels in `~/.nullius/harbor//wheels`, and `uv` if missing | | Proofs | Local runs save to the user catalogue. A host uses a writable copy in its run's `catalogue/`, isolated from the host's own entries; that copy and its proofs mirror back alongside `jobs/`, without changing the active user catalogue | | Docker | Where a recipe needs it, `docker ps` must answer, or the run stops before it starts | diff --git a/docs/cli/run.md b/docs/cli/run.md index 7b8c06b..1962596 100644 --- a/docs/cli/run.md +++ b/docs/cli/run.md @@ -5,7 +5,7 @@ One benchmark and one agent, capability, control or setup, into a job. ```bash title="Run" nullius standard run [@version] [--model M] [--dataset D] [--limit N] [-j N] \ [--unsandboxed | --backend docker] [--option NAME=VALUE]... [--bench-option NAME=VALUE]... \ - [--price MODEL=INPUT,OUTPUT[,CACHED]]... \ + [--price MODEL=INPUT,OUTPUT[,CACHED]]... [--max-usd USD] \ [--repeat K] [--remove-images] [--judge-model M] [--simulator-model M] [--name NAME] [--jobs DIR] [--provider P] ``` @@ -28,6 +28,7 @@ nullius standard run locomo model --model gpt-4o-mini | `--option NAME=VALUE` | An agent's or capability's option (repeatable) | | `--bench-option NAME=VALUE` | A benchmark's option (repeatable), read as `NULLIUS_OPTION_` | | `--price MODEL=INPUT,OUTPUT[,CACHED]` | This run's price for a model in USD per million tokens (repeatable), over your [price list](../settings/model-prices.md) | +| `--max-usd USD` | The run's spending cap: no call goes out once the spend plus the calls in flight would pass it ([below](#a-spending-cap)) | | `--repeat K` | Each unit K times; `result.json` carries `pass_hat_k` | | `--remove-images` | Remove each image the run pulls or builds when its last unit ends (never with `--repeat`); by default they stay for the next run, listed and removed in Settings, [Sandbox](../settings/sandbox.md) | | `--judge-model` | Instead of the official judge | @@ -46,6 +47,25 @@ nullius standard run locomo model --model gpt-4o-mini | Progress | stderr, one line per stage (`[nullius] stage=install ...`, `stage=units n=10`, `stage=done`, `stage=refused`), also `progress.jsonl` | | The job | Opens in the app; `events.jsonl` feeds the live board | +## A spending cap + +`--max-usd 5` stops a run at 5 USD, for a prepaid or shared key. Every relay of +the job books each call in one ledger before sending it +(`app/nullius/standard/ledger.py`). + +| Rule | Detail | +| --- | --- | +| Prices first | Every model the run calls (its own, the judge, the simulated user, an option's embeddings, an agent's side models) needs a price in your [price list](../settings/model-prices.md) or a `--price`; else the run is refused before it starts | +| A booking | The most a call is likely to cost: its characters over three as input tokens, and as output the limit the call names, else the model's own (its price's output limit), else 4,096 tokens | +| 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`, an API error 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. A cap close to one call's booking stops the run early | +| An estimate | Prices are list prices and a booking is not a bill: a call that costs more than was booked takes the spend past the cap by the difference, and the next call is refused | +| A shard | The cap is that shard's alone: split a budget over the shards yourself | +
- [The standard](../standard.md#a-run) diff --git a/docs/field-guide.md b/docs/field-guide.md index c0fede5..893780d 100644 --- a/docs/field-guide.md +++ b/docs/field-guide.md @@ -231,7 +231,7 @@ Eleven classes. The three marked **yes** under Infrastructure leave the adjusted | `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. | | `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. | +| `api_error` | API error | yes | Any other provider failure: overloaded, unreachable, out of quota, key rejected. Also a standard run's unit cut off by its own spending cap (`SpendingCapError`). | | `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. | | `unknown` | Unknown | no | No verdict, or an exception on neither list. Counted, charged to nobody; the trial page shows the recorded type. | diff --git a/docs/settings/model-prices.md b/docs/settings/model-prices.md index 5dec9ed..f9d6158 100644 --- a/docs/settings/model-prices.md +++ b/docs/settings/model-prices.md @@ -20,6 +20,7 @@ its list is filled once. | Recorded | A job's `config.json` keeps the prices of the models it calls, each with its source (`list`, `own`, or `run` for `--price`), and where and when the list was fetched when a price is the list's | | Docker and hosts | A unit's relay in a container gets its models' prices from the run; a run on a host takes this machine's list with it | | An estimate | List prices, not a bill: [Metrics](../field-guide.md) | +| A spending cap | `--max-usd` books each call at these prices before sending it, so it needs one for every model the run calls: [Run a benchmark](../cli/run.md#a-spending-cap) | | Code | `app/nullius/prices.py` | ## From a shell diff --git a/docs/settings/model-providers.md b/docs/settings/model-providers.md index ee1a136..242cb3c 100644 --- a/docs/settings/model-providers.md +++ b/docs/settings/model-providers.md @@ -37,7 +37,7 @@ your existing `settings.json` ([location and precedence](README.md#save)). | | | | --- | --- | -| A run | Its provider's key and endpoint, as environment, never written | +| A run | Its provider's key and endpoint, as environment (the key to a docker relay as a file) | | A recipe, a log | Nothing; logs are redacted | | A standard run on Anthropic | The relay alone | diff --git a/docs/standard.md b/docs/standard.md index 2a46f71..9f90f84 100644 --- a/docs/standard.md +++ b/docs/standard.md @@ -138,6 +138,7 @@ record); verifier/reward.txt and score.json | Progress | One line per stage on stderr and in `progress.jsonl`: `[nullius] stage=install agent=codex@2.2.2 state=done seconds=41 at=...` (`start`, `done`, `cached`, `failed`; a download's bytes every 5 s) for the base image, an environment, the upstream, a dataset, `driver.install` and the agent; then `stage=units n=10`, `stage=unit i=3 of=10 unit=... outcome=passed`, `stage=done`; `stage=refused` alone for a run refused first. From `stage=units` the job also writes `events.jsonl`, Harbor's live feed | | Outcomes | A last model call that got a 429 is `rate_limited`, a refused key an API error, both outside the agent's rate; a turn past its limit is the agent's timeout. `"kind": "score"` leaves a score and no pass | | The relay | Headers pass through except the credentials, replaced by the relay's key; `agent/calls.jsonl` names them, never their values. A 429 reaches the agent as it came, never retried. Every call goes as the run's model, the one asked recorded as `model_asked`, except `model.side_models`, sent as asked and recorded `side`. Provider `anthropic` speaks Anthropic's Messages (`ANTHROPIC_API_KEY` as `x-api-key`) | +| A spending cap (`--max-usd`) | Every relay of the job books a call in one ledger (`.work/ledger`, `standard/ledger.py`) before sending it; the first that does not fit is refused by the relay itself (HTTP 402, recorded `refused`), and every call after it. A unit cut off is a `SpendingCapError`, an API error outside the agent's rate; units not begun are not run; `result.json` carries `budget`. [Run a benchmark](cli/run.md#a-spending-cap) | ## Safety @@ -150,7 +151,7 @@ The docker backend (`--backend docker`) is the isolation, and needs no such yes. | --- | --- | | One base image | `nullius/base:`: Ubuntu 24.04, git, python3, curl, CA certificates, tmux and Node 22. Environments are layers on it ([Per-task environments](#per-task-environments)), or each task's own image ([Own images](#own-images)) | | The agent's container | Runs as you (`--user`), no capabilities, only on the unit's internal network: no route out. Read-write: the unit's workspace and home, the MCP call log folder. Read-only, at its path here: the agent's install and definition folder, its MCP servers' files. Never `app/` (refused), the benchmark's folder or upstream | -| The relay is a sidecar | On that network as `relay`, and on the default bridge: the agent's one way out, `http://relay:8080//v1`; its port on 127.0.0.1 for the driver. The key is in the relay's container only, passed by name (`-e NULLIUS_RELAY_UPSTREAM_KEY`), never in the agent's environment, files or mounts; `docker inspect` can read it there | +| The relay is a sidecar | On that network as `relay`, and on the default bridge: the agent's one way out, `http://relay:8080//v1`; its port on 127.0.0.1 for the driver. The key reaches the relay's container only, as an owner-only file in the user's cache (`relay-keys/`), mounted read-only and removed from this machine once the relay listens. It stays readable in that container until it stops. `docker inspect` does not show it, and it is never in the agent's environment, files or mounts | | A model server on this machine | An endpoint on its loopback (`http://localhost:11434/v1`) is reached by the relays as `host.docker.internal`, through a forwarder the run keeps on the default bridge's gateway alone; the agents' internal networks do not reach it | | Cleanup | Everything a unit starts is labelled `nullius.job=` and removed when it ends, after an error too; the run sweeps the label | @@ -296,7 +297,7 @@ N machines are N runs of the same command, each with `--shard I/N`, and one merg | Rule | Detail | | --- | --- | -| No central relay | Each unit's relay runs on its own machine with the key it was handed; split a launch over keys of their own if one cannot carry it | +| No central relay | Each unit's relay runs on its own machine with the key it was handed; split a launch over keys of their own if one cannot carry it. A spending cap (`--max-usd`) is each shard's own | | Shards by the unit's id | `sha256(unit id) mod N` (`standard/shard.py`); changing N moves most units. Order: the driver's `units`, the shard, then `--limit` (`--shard 1/2 --limit 6` is the first six units of shard 1) | | A unit keeps its place | `u0042` is the 43rd unit of the whole dataset in every shard | | One launch | Each shard's `config.json` records `standard.shard` (`index`, `of`, `units_total`, `partition`, `planned`, `limit`) and `standard.launch` (`--launch`, else the UTC day, benchmark, agent or capability and model) |