#!/usr/bin/env python3 """The Jev bench — a decision model against the generating seat, on one card. python3 run.py --arm jev-readout --task a python3 run.py --arm jev-generate --task a --task b --task c python3 run.py --score rows/ # tables only, no model call THE QUESTION, pre-registered in README.md before any row was written: does a typed READOUT — one forward pass, the answer read off the scores of the option letters at the first output position, no generation — match the generating seat on decisions the estate already makes, at a fraction of the latency and energy? WHAT THIS FILE IS CAREFUL ABOUT, each one paid for by an estate trap: * **Residency is proved on the NAMED board, never by /api/ps** (D-20260921-001, and D-20260920-133 before it: ollama enumerates the other board over Vulkan and `size_vram == size` was true on the wrong card). Every arm calls `--verify-pin` first: memory growth on the card the receipts name AND no growth anywhere else, refusing in both directions. * **The first output position is not the first token of the reply.** This model's chat template opens the assistant turn with a thinking channel, so the untouched first token is `<|channel>` and every option letter sits ~13 nats below it. The template's own `enable_thinking=false` closes the channel inside the generation prompt (`<|channel>thought\n`), which is what ollama's `think: false` sends and what the OpenJev helper's `chat_template_kwargs={"enable_thinking": False}` sends to vLLM. Both arms set it, the generating control included, so the two differ in the READOUT and in nothing else. * **Raw logprobs are kept in every row.** The published OpenJev calibration (`READOUT_T=0.85`, `READOUT_NOUL_T=1.829074`) was fitted for OpenJev, on its own targeted extraction, and this bench applies it unchanged to a different model through a server that has no `logprob_token_ids`. That is stated rather than hidden — and because the letter scores themselves are in the file, any temperature can be refitted later without spending a GPU second. * **Energy is the board's own watts, integrated.** A 1 Hz `nvidia-smi` sampler pinned to the card's UUID runs for the length of each arm and nothing else shares the card; idle is read for ten seconds before the first call, and the per-decision figure is (integral - idle x seconds) as well as the raw one. ADDING AN ARM (arm 2 is openjev/openjev-FP8 on both cards under vLLM): add an `Arm` to `ARMS`. An arm names its endpoint, its model string, how it is asked to stop thinking, and whether it reads letters or writes them. Nothing else in the file knows which model is behind it. """ from __future__ import annotations import argparse import collections import datetime as dt import hashlib import json import math import os import statistics import subprocess import sys import time import urllib.error import urllib.request from dataclasses import dataclass, field, asdict HERE = os.path.dirname(os.path.abspath(__file__)) KIT = os.path.join(HERE, "kit") # ---------------------------------------------------------------- the protocol #: OpenJev gives each option a letter and reads the scores of exactly those #: letters at the first output position (model card, "How it works"). LETTERS = [chr(65 + i) for i in range(26)] + [chr(97 + i) for i in range(26)] #: The published calibration, verbatim from the model card's quick start. Both #: are applied unchanged; see the module docstring for what that does and does #: not license us to say about the probabilities. READOUT_T = 0.85 NOUL_T = 1.829074 NOUL_BIAS = 0.0 #: The floor the OLD (untargeted) protocol gives a letter that missed the raw #: top-K. `logprob_token_ids` is a vLLM extra; ollama has no equivalent, so this #: bench is on the old path and says so in every report. FLOOR = -30.0 def sha(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest() def utc() -> str: return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def readout_prompt(state: str, instructions: str, options: list[tuple[str, str]]) -> str: """The OpenJev helper's text-lane layout, `helper/shim.py::_readout_once`.""" lines = "\n".join("[%s] %s: %s" % (LETTERS[i], k, d) for i, (k, d) in enumerate(options)) return ("State:\n%s\n\nQuestion: %s\nOptions:\n%s\n\n" "Answer with the letter of the best option only." % (state, instructions, lines)) def softmax(z: list[float]) -> list[float]: m = max(z) e = [math.exp(v - m) for v in z] s = sum(e) return [v / s for v in e] def choice_confidence(p: list[float]) -> float: """The official formula, as the helper carries it.""" if len(p) == 1: return 1.0 u = 1.0 / len(p) return max(0.0, (max(p) - u) / (1.0 - u)) # -------------------------------------------------------------------- the arms #: benchbox's two boards, by UUID, because an index is not an identity: they are #: the same model of card on different links, and every energy and residency #: reading in this file is keyed on these strings rather than on 0/1. #: #: THE LINK WIDTHS HERE ARE MEASURED UNDER LOAD, and they are the reverse of what #: arm 1's README and arm 2's brief both recorded. Idle readings downtrain to #: gen 1 and `pcie.link.width.max` reports the CARD's capability rather than the #: slot's, which is how the claim got written backwards twice; sampling #: `pcie.link.width.current` through the all-reduce microbench settles it #: (`receipts/phb-linkwidth.csv`). `lspci -vv`'s LnkCap would say it in one line #: and is root-only on this box. CARD0 = "GPU-46890836-d8f7-e868-1977-c1ce28db0d7a" # PCI 04:00.0, trains gen3 x4 CARD1 = "GPU-7aa0be10-974f-6430-52fe-09018a0e2e08" # PCI 2B:00.0, trains gen4 x16 #: ARM 3's board, and it is on a DIFFERENT BOX. `largecard` is inferencebox's RTX PRO 6000 #: Blackwell (sm_120, 97,887 MiB, cap 420 W, PCI 01:00.0) — one card big enough #: to hold a 54 GB BF16 checkpoint whole, so arm 3 has no tensor-parallel and no #: all-reduce at all. That is the whole point of it: arm 2 measured a model #: through a link (§R2.6 — 1.82 GB/s, 6.4 ms of wire under every token), and the #: only way to say what the model costs is to take the link away. #: #: inferencebox's OTHER board, the 3090 `the 3090 board` #: (GPU-fde82ed0-be70-eb1f-2d4d-5538c5b82166, PCI C1:00.0), is NOT named by any #: arm and is NOT stopped for arm 3's window: it carries the mistral seat on #: :11435, pinned resident (`OLLAMA_KEEP_ALIVE=-1`). `verify_pin` reads GROWTH, #: so a board that stays resident and still is quiet by construction — and a #: board that GROWS mid-arm means that seat loaded a second model, which is #: reported as a pin note rather than absorbed. LARGECARD = "GPU-25bc3288-319a-790c-a14e-82acdb8d15b8" # inferencebox, PCI 01:00.0, sm_120 @dataclass(frozen=True) class Arm: """One model, one way of being asked, on the named board or boards. `runtime` is the only thing in this file that knows which server is behind an arm: "ollama" speaks `/api/chat` (arm 1), "vllm" speaks `/v1/chat/completions` and `/tokenize` (arm 2). Everything downstream of `call()` — the readout, the scoring, the energy, the pin — is written once and used by both. """ name: str model: str mode: str # "readout" | "generate" runtime: str = "ollama" host: str = "http://127.0.0.1:11436" cards: tuple[str, ...] = (CARD0,) num_ctx: int = 131072 seed: int = 20260921 quant: str = "Q4_K_M" top_logprobs: int = 20 #: The BOX and the board's own CAP, because arm 3 is on neither of arm 1's. #: Both were literals inside `run_arm` until arm 3 needed different ones, and #: a literal that is right on one box is a lie on the next: inferencebox's largecard #: caps at 420 W, not benchbox's 250, and a row that said 250 would have been #: a typed number where the card has a reading (the house law: derive from #: state, never type the number). box: str = "benchbox" power_cap_w: int = 250 note: str = "" @property def card_uuid(self) -> str: """The first named board. Kept because arm 1's receipts say `card_uuid`.""" return self.cards[0] JEV = "hf.co/mradermacher/jevify-gemma4-26b-a4b-GGUF:Q4_K_M" BASE = "gemma4:26b" ARMS: dict[str, Arm] = { "jev-readout": Arm("jev-readout", JEV, "readout", note="the community adapter, letters read at position 0"), "jev-generate": Arm("jev-generate", JEV, "generate", note="the same adapter and the same prompt, written out"), "base-readout": Arm("base-readout", BASE, "readout", quant="Q4_K_M (ollama default tag)", note="THE CONTROL: the adapter's own base, same readout"), "base-generate": Arm("base-generate", BASE, "generate", quant="Q4_K_M (ollama default tag)", note="THE SEAT'S NORMAL WAY: the base, written out"), } #: ARM 2 — the FP8 decision model on BOTH boards under vLLM, pre-registered in #: README §Arm 2 before a row of it existed. What it changes from arm 1, and #: nothing else: the model (built as a decision model rather than adapted into #: one), the runtime (vLLM, whose logprob path is the one OpenJev's published #: numbers were measured on), and the boards (two, tensor-parallel, over PHB #: with card 0 on four lanes and no peer-to-peer between two GeForce cards — a #: link whose price §A2.8 required as a measured number and §R2.6 reports: a #: 1.82 GB/s ceiling and 6.4 ms of wire under every written token). #: #: What it KEEPS from arm 1, so the decisions stay comparable: the frozen kit, #: the prompt builder, the option order, the seed, the calibration constants, #: `FLOOR = -30`, `top_logprobs = 20` — the wider 64 this server can serve is #: deliberately not used, because a floor counted against a wider top-K is not #: arm 1's measurement (§A2.3). #: #: LICENCE FENCE: CC BY-NC 4.0. Bench-only. These weights never become a seat. OPENJEV_FP8 = "openjev-fp8" # --served-model-name, not the repo path ARMS["openjev-fp8-readout"] = Arm( "openjev-fp8-readout", OPENJEV_FP8, "readout", runtime="vllm", host="http://127.0.0.1:8000", cards=(CARD0, CARD1), num_ctx=16384, quant="FP8 e4m3, 128x128 weight blocks, dynamic activations", top_logprobs=20, note="openjev/openjev-FP8 tensor-parallel across both 3090s; CC BY-NC 4.0, " "bench-only, never a seat") #: The generating control on the SAME model and the same runtime — arm 1's #: readout-versus-generation question asked again where the per-request overhead #: is not ollama's 870 ms. It is not under the gate; it is the only way to say #: whether the readout's saving is real on this substrate. ARMS["openjev-fp8-generate"] = Arm( "openjev-fp8-generate", OPENJEV_FP8, "generate", runtime="vllm", host="http://127.0.0.1:8000", cards=(CARD0, CARD1), num_ctx=16384, quant="FP8 e4m3, 128x128 weight blocks, dynamic activations", note="the same FP8 model and prompt, written out instead of read") #: ARM 3 — THE SAME MODEL, ONE CARD, NO LINK: `openjev/openjev` at BF16 and #: `openjev/openjev-FP8` natively on inferencebox's largecard, pre-registered in README #: §Arm 3 before a row of it existed. #: #: What it changes from arm 2, and nothing else: the BOARD (one RTX PRO 6000 #: Blackwell instead of two 3090s over a host bridge) and, between its own two #: arms, the PRECISION (bfloat16 as published against the FP8 file). Everything #: the decisions are made of is arm 2's: the frozen kit, the prompt builder, the #: option order, the seed, the calibration constants, `FLOOR = -30`, #: `top_logprobs = 20`, `max_model_len = 16384` — so the same 53 / 35 / 108 items #: are reached and the coverage table is comparable row for row. #: #: WHY THE PAIR IS THE POINT. sm_120 has NATIVE FP8 arithmetic, which sm86 does #: not: arm 2's FP8 ran weight-only through Marlin (dequantised into a bf16 #: tensor-core path), so it measured FP8's MEMORY and never FP8's MATH. On #: largecard the FP8 arm should select a native block-scaled FP8 kernel, and the #: startup line that says which is a required receipt — **a Marlin line here is #: a finding, not a detail**, and it would mean the two arms differ in memory #: and not in arithmetic. #: #: LICENCE FENCE, UNCHANGED AND RESTATED: `openjev/openjev` and #: `openjev/openjev-FP8` are CC BY-NC 4.0. Arm 3 is a bench. These weights never #: become a seat — not the docent, not rulesage, not amble, not the long table, #: not the beat lab, not the cove. The window this arm runs in stops seats; it #: never feeds one. OPENJEV_BF16_LARGECARD = "openjev-bf16-largecard" # --served-model-name OPENJEV_FP8_LARGECARD = "openjev-fp8-largecard" # --served-model-name ARMS["openjev-bf16-largecard"] = Arm( "openjev-bf16-largecard", OPENJEV_BF16_LARGECARD, "readout", runtime="vllm", host="http://127.0.0.1:8000", cards=(LARGECARD,), num_ctx=16384, box="inferencebox", power_cap_w=420, quant="none - bfloat16 weights as published (openjev/openjev, rev 5ec9e5fd)", top_logprobs=20, note="THE FP16 ARM: 54.73 GB of BF16 weights whole on one card, no " "tensor-parallel and no all-reduce; CC BY-NC 4.0, bench-only, never a seat") ARMS["openjev-fp8-largecard"] = Arm( "openjev-fp8-largecard", OPENJEV_FP8_LARGECARD, "readout", runtime="vllm", host="http://127.0.0.1:8000", cards=(LARGECARD,), num_ctx=16384, box="inferencebox", power_cap_w=420, quant="FP8 e4m3, 128x128 weight blocks, dynamic activations " "(openjev/openjev-FP8, rev 4ec320f2) - NATIVE on sm_120", top_logprobs=20, note="THE FP8 ARM: the same model's FP8 file on the same card, so the pair " "differs in precision and in nothing else; CC BY-NC 4.0, bench-only") class SeatError(RuntimeError): pass def call(arm: Arm, prompt: str, *, predict: int, timeout: float = 900.0) -> dict: """One decision call, to whichever server this arm names. Both runtimes return the SAME shape, so nothing below this function knows which one answered: `logprobs` is a list of positions, each with a `token`, a `logprob` and a `top_logprobs` list — ollama's own `/api/chat` shape, which arm 1 was written against and which arm 2 therefore adopts rather than forking the reader. """ if arm.runtime == "vllm": return call_vllm(arm, prompt, predict=predict, timeout=timeout) return call_ollama(arm, prompt, predict=predict, timeout=timeout) def post(url: str, body: dict, timeout: float) -> dict: """POST JSON, and raise the SERVER'S OWN SENTENCE rather than the status line. `exceeds the available context size` / `maximum context length` arrive as a 400 whose body names both numbers, and a bench that printed only "400 Bad Request" would have sent someone hunting the harness for a fact the server had already stated. """ req = urllib.request.Request( url, method="POST", data=json.dumps(body).encode("utf-8"), headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=timeout) as fh: return json.loads(fh.read().decode("utf-8")) except urllib.error.HTTPError as exc: try: detail = exc.read().decode("utf-8", "replace")[:600] except Exception: # noqa: BLE001 - best effort only detail = "" raise SeatError("HTTP %s: %s" % (exc.code, detail)) from exc except (urllib.error.URLError, OSError, ValueError) as exc: raise SeatError("%s: %s" % (exc.__class__.__name__, exc)) from exc def call_vllm(arm: Arm, prompt: str, *, predict: int, timeout: float = 900.0) -> dict: """One `/v1/chat/completions` call, normalised into the arm-1 shape. `chat_template_kwargs={"enable_thinking": False}` is what the OpenJev helper sends and it is the vLLM twin of arm 1's `think: false`: it closes the template's thinking channel INSIDE the generation prompt, so the first output position is the answer and not a channel marker. Arm 1 paid for that lesson (§R6) and every row here records the position-0 token so a recurrence is visible rather than inferred. """ body = { "model": arm.model, "messages": [{"role": "user", "content": prompt}], "max_tokens": predict, "temperature": 0, "seed": arm.seed, "logprobs": True, "top_logprobs": arm.top_logprobs, "chat_template_kwargs": {"enable_thinking": False}, } started = time.monotonic() out = post(arm.host + "/v1/chat/completions", body, timeout) seconds = time.monotonic() - started choice = (out.get("choices") or [{}])[0] content = (choice.get("message") or {}).get("content") or "" positions = [] for pos in ((choice.get("logprobs") or {}).get("content") or []): positions.append({ "token": pos.get("token"), "logprob": pos.get("logprob"), "top_logprobs": [{"token": t.get("token"), "logprob": t.get("logprob")} for t in (pos.get("top_logprobs") or [])], }) usage = out.get("usage") or {} return {"logprobs": positions, "message": {"content": content}, "prompt_eval_count": usage.get("prompt_tokens"), "eval_count": usage.get("completion_tokens"), "prompt_eval_duration": None, "total_duration": seconds * 1e9, "done_reason": choice.get("finish_reason"), "_seconds": seconds} def count_tokens_vllm(arm: Arm, prompt: str, timeout: float = 120.0) -> int: """What the SERVER makes of this prompt, before a card spends a second on it. vLLM's `/tokenize` applies the same chat template and the same `chat_template_kwargs` the decision call will, so an overlong page is a refusal decided on the server's own count and not on an estimate. """ out = post(arm.host + "/tokenize", {"model": arm.model, "messages": [{"role": "user", "content": prompt}], "add_generation_prompt": True, "chat_template_kwargs": {"enable_thinking": False}}, timeout) return int(out.get("count") if out.get("count") is not None else len(out.get("tokens") or [])) def call_ollama(arm: Arm, prompt: str, *, predict: int, timeout: float = 900.0) -> dict: """One `/api/chat` call. `think: false` closes the template's thought channel. `/api/chat` and not `/v1/chat/completions` for one reason that matters: it takes `options.num_ctx`, and the largest page in the docent's own bank is about 67,000 tokens. The two endpoints return the same logprobs. """ body = { "model": arm.model, "messages": [{"role": "user", "content": prompt}], "stream": False, "think": False, "logprobs": True, "top_logprobs": 20, "options": {"num_predict": predict, "temperature": 0, "seed": arm.seed, "num_ctx": arm.num_ctx}, } started = time.monotonic() out = post(arm.host + "/api/chat", body, timeout) out["_seconds"] = time.monotonic() - started return out def letter_scores(resp: dict, n: int) -> tuple[list[float], dict]: """The scores of exactly the option letters at the FIRST output position. The exact letter token wins over a look-alike: ` U` must not overwrite `U`, which is the helper's own rule and the reason this is not a dict update. """ lps = resp.get("logprobs") or [] if not lps: raise SeatError("no logprobs at position 0") top = lps[0].get("top_logprobs") or [] lp: dict[str, float] = {} for t in top: tok = t["token"] key = tok if tok in LETTERS else tok.strip() if key not in lp or tok in LETTERS: lp[key] = t["logprob"] raw = [lp.get(LETTERS[i]) for i in range(n)] return raw, {"top0_token": lps[0]["token"], "top0_logprob": lps[0]["logprob"], "letters_in_topk": sum(1 for v in raw if v is not None)} def decide(arm: Arm, state: str, instructions: str, options: list[tuple[str, str]], *, noul: bool = False) -> dict: """One typed decision, by whichever mode the arm names.""" prompt = readout_prompt(state, instructions, options) row: dict = {"prompt_sha256": sha(prompt), "prompt_chars": len(prompt), "n_options": len(options), "mode": arm.mode} if arm.mode == "readout": resp = call(arm, prompt, predict=1) raw, detail = letter_scores(resp, len(options)) detail["top_logprobs_asked"] = arm.top_logprobs z = [(FLOOR if v is None else v) / READOUT_T for v in raw] p = softmax(z) row["letter_logprobs"] = raw row["floored"] = [i for i, v in enumerate(raw) if v is None] row.update(detail) else: resp = call(arm, prompt, predict=8) text = (resp.get("message") or {}).get("content") or "" row["generated"] = text pick = None for ch in text: if ch in LETTERS[:len(options)]: pick = LETTERS.index(ch) break row["parsed"] = pick is not None p = [0.0] * len(options) if pick is None: # unparsed: uniform, and counted as unparsed p = [1.0 / len(options)] * len(options) else: p[pick] = 1.0 if noul: # the helper's yes/no calibration, applied to p(yes) exactly as it is py = min(max(p[0], 1e-4), 1 - 1e-4) z2 = math.log(py / (1 - py)) / NOUL_T + NOUL_BIAS row["noul_raw_p_yes"] = p[0] p = [1 / (1 + math.exp(-z2))] p.append(1 - p[0]) best = max(range(len(p)), key=p.__getitem__) row.update( choice=options[best][0], probs={k: p[i] for i, (k, _) in enumerate(options)}, confidence=choice_confidence(p), seconds=resp["_seconds"], prompt_tokens=resp.get("prompt_eval_count"), eval_tokens=resp.get("eval_count"), prompt_eval_ms=(resp.get("prompt_eval_duration") or 0) / 1e6, total_ms=(resp.get("total_duration") or 0) / 1e6, done_reason=resp.get("done_reason"), ) return row # ------------------------------------------------------------------ the tasks def load(name: str): with open(os.path.join(KIT, name + ".json"), encoding="utf-8") as fh: return json.load(fh) #: Verbatim. The three instruction strings are the whole of what each task asks, #: and they are here rather than in a template so a reader can read them. INSTR_A = ("Does the article in the state answer this reader's question? " "Question: %s") INSTR_B = ("Which section of the article in the state contains the answer to " "this reader's question? Question: %s") INSTR_C = ("Six people are at a dinner table. Which of them said the line " "below?\nLine: %s") A_OPTIONS = [("yes", "The article answers the question."), ("no", "The article does not answer the question.")] def refusal(reason: str, detail: dict) -> dict: """A row that spent no GPU second, and is excluded from every rate. A refused item is NOT a wrong answer and must never be averaged with one; it is also not an absence, because the reader needs to know which items the instrument could not reach and why. So it is written into the same file, in the same order, carrying its own reason and the number that produced it — and `summarise` drops it from the scored set rather than the file. """ row = {"refused": reason, "choice": None, "probs": None, "seconds": None} row.update(detail) return row def task_a(arm: Arm, articles: dict, overlong: dict): for item in load("task_a"): art = articles[item["slug"]] if item["slug"] in overlong: row = refusal("overlong", {"prompt_tokens": overlong[item["slug"]]}) else: row = decide(arm, art["state"], INSTR_A % item["question"], A_OPTIONS, noul=True) row.update(id=item["id"], task="a", slug=item["slug"], label=item["label"], kind=item["kind"], state_sha256=art["state_sha256"]) yield row def task_b(arm: Arm, articles: dict, overlong: dict): for item in load("task_b"): art = articles[item["slug"]] if item["slug"] in overlong: row = refusal("overlong", {"prompt_tokens": overlong[item["slug"]]}) else: by_id = {s["id"]: s["heading"] for s in art["sections"]} options = [(sid, by_id.get(sid, sid)) for sid in item["options"]] row = decide(arm, art["state"], INSTR_B % item["question"], options) row.update(id=item["id"], task="b", slug=item["slug"], label=item.get("label"), state_sha256=art["state_sha256"]) yield row def task_c(arm: Arm, _articles: dict, _overlong: dict): roster = {g["id"]: g["name"] for g in load("manifest")["c_roster"]} for item in load("task_c"): options = [(gid, roster[gid]) for gid in item["options"]] # the LINE is the question, not the state: a state that carried the # course would carry the speaker labels of the turns around it. row = decide(arm, "A dinner table with six guests: %s." % ", ".join(roster[g] for g in item["options"]), INSTR_C % item["text"], options) row.update(id=item["id"], task="c", label=item["label"]) yield row TASKS = {"a": task_a, "b": task_b, "c": task_c} #: Announced up front so a peer reading the box knows how long the cards are #: held. Arm 1's own wall clock, rounded up; arm 2 prints these and then the #: report prints what it actually took. EXPECT_MINUTES = {"a": 20, "b": 15, "c": 5} #: ARM 3 runs inside a 45-minute maintenance window on a LIVE box, so its #: announcement has to be honest about how long it holds largecard rather than #: reprinting benchbox's figures. These are ESTIMATES, derived from arm 2's #: per-decision medians with the link removed (README §Arm 3), and the report #: prints what each task actually took beside them. EXPECT_MINUTES_BY_ARM: dict[str, dict[str, int]] = { "openjev-bf16-largecard": {"a": 5, "b": 4, "c": 2}, "openjev-fp8-largecard": {"a": 4, "b": 3, "c": 1}, } def expect_minutes(arm: "Arm", task: str) -> int: return EXPECT_MINUTES_BY_ARM.get(arm.name, EXPECT_MINUTES).get(task, 15) def census_vllm(arm: Arm, articles: dict) -> dict: """Every page's real token count from `/tokenize`, and which ones will not fit. Costs no forward pass at all, so the whole corpus is measured in seconds. A page over the served context is named here, once, and every task arm refuses its items rather than truncating them (README §A2.5). """ rows, worst, over = {}, 0, {} for slug, art in sorted(articles.items()): prompt = readout_prompt(art["state"], "x", A_OPTIONS) n = count_tokens_vllm(arm, prompt) rows[slug] = {"prompt_tokens": n, "est_tokens": art["est_tokens"], "state_bytes": art["state_bytes"], "fits": n < arm.num_ctx - CTX_HEADROOM} if not rows[slug]["fits"]: over[slug] = n worst = max(worst, n) return {"max_prompt_tokens": worst, "num_ctx": arm.num_ctx, "headroom_tokens": CTX_HEADROOM, "fits": worst < arm.num_ctx - CTX_HEADROOM, "overlong": over, "overlong_pages": len(over), "how": "vllm /tokenize, the server's own chat template", "per_slug": rows} #: A decision needs the answer token and a little room for the template's own #: preamble; a page that lands within this of the ceiling is refused rather than #: gambled on, because a 400 forty calls into a run is not a measurement. CTX_HEADROOM = 256 def census(arm: Arm, articles: dict) -> dict: """Every article's real prompt length, before an arm spends a minute on one. The kit's estimate is the estate's four-bytes-to-the-token guess; this asks the server. A page that does not fit is a REFUSAL here rather than a 400 forty calls into a run, and the numbers it prints are ones the report wants anyway. Measured 2026-09-21: the longest page in the docent's own bank is 82,488 tokens, which is why this bench does not run at the unit's 8,192. """ rows, worst = {}, 0 for slug, art in sorted(articles.items()): prompt = readout_prompt(art["state"], "x", A_OPTIONS) resp = call(arm, prompt, predict=1) n = resp.get("prompt_eval_count") or 0 rows[slug] = {"prompt_tokens": n, "est_tokens": art["est_tokens"], "state_bytes": art["state_bytes"]} worst = max(worst, n) return {"max_prompt_tokens": worst, "num_ctx": arm.num_ctx, "fits": worst < arm.num_ctx, "per_slug": rows} # ------------------------------------------------------------ the card's watts def gpu_rows(uuid: str) -> list[dict]: out = subprocess.run( ["nvidia-smi", "--query-gpu=uuid,memory.used,power.draw,clocks.sm,temperature.gpu", "--format=csv,noheader,nounits"], capture_output=True, text=True, check=True).stdout rows = [] for line in out.strip().splitlines(): f = [x.strip() for x in line.split(",")] rows.append({"uuid": f[0], "memory_mib": float(f[1]), "watts": float(f[2]), "sm_mhz": float(f[3]), "temp_c": float(f[4])}) return rows def verify_pin(uuids, before: list[dict], after: list[dict], floor_mib=200.0) -> dict: """Residency on the NAMED board or boards, refusing in both directions. A named board that did not grow is a pin that did not hold; an unnamed board that did is the Vulkan trap (D-20260920-133) wearing a correct-looking env. Arm 2 names TWO boards, and a tensor-parallel load that grew only one of them is exactly as failed a pin as a single-card load on the wrong card — so `named_grew` is an ALL, never an any. """ named = (uuids,) if isinstance(uuids, str) else tuple(uuids) b = {r["uuid"]: r["memory_mib"] for r in before} a = {r["uuid"]: r["memory_mib"] for r in after} growth = {u: a[u] - b.get(u, 0.0) for u in a} verdict = {"card": named[0], "cards": list(named), "growth_mib": growth, "floor_mib": floor_mib, "named_grew": all(growth.get(u, 0.0) >= floor_mib for u in named), "per_card_grew": {u: growth.get(u, 0.0) >= floor_mib for u in named}, "others_quiet": all(v < floor_mib for u, v in growth.items() if u not in named), "memory_after_mib": a} verdict["ok"] = verdict["named_grew"] and verdict["others_quiet"] return verdict class Watts: """A 1 Hz reading of one board, by UUID, for the length of one arm.""" def __init__(self, uuid: str, path: str): self.uuid, self.path, self.proc = uuid, path, None def __enter__(self): self.fh = open(self.path, "w") self.proc = subprocess.Popen( ["nvidia-smi", "-i", self.uuid, "--query-gpu=timestamp,power.draw,memory.used,utilization.gpu", "--format=csv,noheader,nounits", "-lms", "1000"], stdout=self.fh, stderr=subprocess.DEVNULL) return self def __exit__(self, *exc): if self.proc: self.proc.terminate() try: self.proc.wait(timeout=10) except subprocess.TimeoutExpired: self.proc.kill() self.fh.close() return False @staticmethod def integrate(path: str) -> dict: """Trapezoid over the sampler's own timestamps: joules, not an average.""" ts, w = [], [] for line in open(path, encoding="utf-8"): f = [x.strip() for x in line.split(",")] if len(f) < 2: continue try: t = dt.datetime.strptime(f[0], "%Y/%m/%d %H:%M:%S.%f") ts.append(t.timestamp()) w.append(float(f[1])) except ValueError: continue if len(ts) < 2: return {"samples": len(ts), "joules": None, "seconds": None} j = sum((w[i] + w[i + 1]) / 2 * (ts[i + 1] - ts[i]) for i in range(len(ts) - 1)) return {"samples": len(ts), "seconds": ts[-1] - ts[0], "joules": j, "mean_watts": j / (ts[-1] - ts[0]), "peak_watts": max(w), "min_watts": min(w)} class Boards: """A 1 Hz reading of EVERY named board, by UUID, for the length of one arm. One `nvidia-smi` process for all of them, so the two cards share a clock and a sampling cadence; the uuid rides in each line and the integrator splits on it. `Watts` above is the single-board form arm 1 measured with and reads; this is the same trapezoid over more than one card, because arm 2's energy is the SUM of two boards and an average of two averages would not be joules. """ QUERY = "uuid,timestamp,power.draw,memory.used,utilization.gpu" def __init__(self, uuids, path: str): self.uuids = (uuids,) if isinstance(uuids, str) else tuple(uuids) self.path, self.proc = path, None def __enter__(self): self.fh = open(self.path, "w") self.proc = subprocess.Popen( ["nvidia-smi", "-i", ",".join(self.uuids), "--query-gpu=" + self.QUERY, "--format=csv,noheader,nounits", "-lms", "1000"], stdout=self.fh, stderr=subprocess.DEVNULL) return self def __exit__(self, *exc): if self.proc: self.proc.terminate() try: self.proc.wait(timeout=10) except subprocess.TimeoutExpired: self.proc.kill() self.fh.close() return False @staticmethod def integrate(path: str) -> dict: """Joules per board and their sum, each board over its own timestamps.""" per: dict[str, tuple[list, list]] = {} for line in open(path, encoding="utf-8"): f = [x.strip() for x in line.split(",")] if len(f) < 3: continue try: t = dt.datetime.strptime(f[1], "%Y/%m/%d %H:%M:%S.%f").timestamp() w = float(f[2]) except ValueError: continue ts, ws = per.setdefault(f[0], ([], [])) ts.append(t) ws.append(w) cards, total_j, span = {}, 0.0, 0.0 for uuid, (ts, w) in sorted(per.items()): if len(ts) < 2: cards[uuid] = {"samples": len(ts), "joules": None} continue j = sum((w[i] + w[i + 1]) / 2 * (ts[i + 1] - ts[i]) for i in range(len(ts) - 1)) secs = ts[-1] - ts[0] cards[uuid] = {"samples": len(ts), "seconds": secs, "joules": j, "mean_watts": j / secs, "peak_watts": max(w), "min_watts": min(w)} total_j += j span = max(span, secs) out = {"per_card": cards, "cards": sorted(cards), "joules": total_j if cards else None, "seconds": span or None, "samples": sum(c["samples"] for c in cards.values())} if span: out["mean_watts"] = total_j / span # the SUM of the boards out["peak_watts"] = max((c.get("peak_watts") or 0) for c in cards.values()) return out def idle_reading(uuids, path: str, seconds: float = 10.0) -> dict: """The draw of every named board for ten seconds, with the model resident. Read immediately before each task and subtracted from that task's joules, so "energy per decision" is the work and not the cards existing. """ with Boards(uuids, path): time.sleep(seconds) r = Boards.integrate(path) r["what"] = ("the draw of these cards immediately before the arm, with the " "model resident and no work in flight") return r # ----------------------------------------------------------------- the scoring def brier(rows: list[dict]) -> float | None: """Multi-class Brier: the mean squared error of the whole distribution.""" vals = [] for r in rows: if not r.get("label") or not r.get("probs"): continue vals.append(sum((p - (1.0 if k == r["label"] else 0.0)) ** 2 for k, p in r["probs"].items())) return statistics.fmean(vals) if vals else None def reliability(rows: list[dict], bins: int = 10) -> list[dict]: """Confidence against correctness, in ten bins, with the counts shown.""" out = [{"lo": i / bins, "hi": (i + 1) / bins, "n": 0, "sum_p": 0.0, "correct": 0} for i in range(bins)] for r in rows: if not r.get("label") or not r.get("probs"): continue p = max(r["probs"].values()) i = min(bins - 1, int(p * bins)) out[i]["n"] += 1 out[i]["sum_p"] += p out[i]["correct"] += int(r["choice"] == r["label"]) for b in out: b["mean_p"] = (b["sum_p"] / b["n"]) if b["n"] else None b["accuracy"] = (b["correct"] / b["n"]) if b["n"] else None b["gap"] = (b["mean_p"] - b["accuracy"]) if b["n"] else None return out def summarise(rows: list[dict]) -> dict: """Every rate over the rows the instrument actually REACHED. A refused row (an overlong page, README §A2.5) is in the file and out of the arithmetic: it is neither a wrong answer nor an absence, so it is counted on its own line and excluded from accuracy, Brier, reliability, latency and the token totals. Arm 1 had none of these and its numbers are unchanged by this. """ live = [r for r in rows if not r.get("refused")] scored = [r for r in live if r.get("label")] secs = sorted(r["seconds"] for r in live if r.get("seconds") is not None) refused = collections.Counter(r["refused"] for r in rows if r.get("refused")) s = { "n": len(rows), "n_reached": len(live), "n_labelled": len(scored), "refused": sum(refused.values()), "refused_by_reason": dict(refused), "correct": sum(1 for r in scored if r["choice"] == r["label"]), "errors": sum(1 for r in live if r.get("error")), "brier": brier(scored), "reliability": reliability(scored), "median_seconds": statistics.median(secs) if secs else None, "p95_seconds": (secs[min(len(secs) - 1, int(round(0.95 * (len(secs) - 1))))] if secs else None), "prompt_tokens_total": sum(r.get("prompt_tokens") or 0 for r in live), "eval_tokens_total": sum(r.get("eval_tokens") or 0 for r in live), "unparsed": sum(1 for r in live if r.get("mode") == "generate" and r.get("parsed") is False), "floored_rows": sum(1 for r in live if r.get("floored")), "letters_floored_total": sum(len(r.get("floored") or []) for r in live), } s["accuracy"] = (s["correct"] / s["n_labelled"]) if s["n_labelled"] else None return s # --------------------------------------------------------------------- the run def run_arm(arm: Arm, tasks: list[str], out_dir: str) -> list[dict]: os.makedirs(out_dir, exist_ok=True) articles = load("articles") reports = [] pre, overlong = None, {} if any(t in ("a", "b") for t in tasks): cache = os.path.join(out_dir, "census.%s.json" % arm.model.replace("/", "_").replace(":", "-")) if os.path.exists(cache): pre = json.load(open(cache, encoding="utf-8")) print("census: reusing %s" % cache, flush=True) else: pre = (census_vllm if arm.runtime == "vllm" else census)(arm, articles) with open(cache, "w") as fh: json.dump(pre, fh, indent=1, sort_keys=True) overlong = pre.get("overlong") or {} print("census: longest page %d tokens, num_ctx %d, fits=%s, overlong pages %d" % (pre["max_prompt_tokens"], pre["num_ctx"], pre["fits"], len(overlong)), flush=True) if not pre["fits"] and not overlong: # a runtime with no refusal policy (arm 1's) must still STOP rather # than send a page it knows will 400 forty calls in raise SystemExit("run: the corpus does not fit this context - " "%d tokens against num_ctx %d" % (pre["max_prompt_tokens"], arm.num_ctx)) for t in tasks: stamp = utc() base = os.path.join(out_dir, "%s.%s" % (arm.name, t)) before = gpu_rows(arm.card_uuid) idle = idle_reading(arm.cards, base + ".idle.csv") # THE ANNOUNCEMENT, at the moment the arm starts: box, every board it # will use, the arm, the task, the stamp, and how long it expects to # hold them — so a peer reading the box knows what is on it and for how # long, rather than discovering a load (the 09-17 lesson). print("ANNOUNCE box %s · cards %s · arm %s · task %s · start %s · " "expect ~%d min" % (arm.box, ", ".join(c[:16] for c in arm.cards), arm.name, t, stamp, expect_minutes(arm, t)), flush=True) rows, started = [], time.monotonic() with Boards(arm.cards, base + ".watts.csv"): with open(base + ".jsonl", "w", encoding="utf-8") as fh: for row in TASKS[t](arm, articles, overlong): row["arm"], row["model"], row["quant"] = arm.name, arm.model, arm.quant row["card_uuid"], row["num_ctx"] = arm.card_uuid, arm.num_ctx row["power_cap_w"], row["seed"] = arm.power_cap_w, arm.seed row["stamp"] = utc() rows.append(row) fh.write(json.dumps(row, ensure_ascii=False) + "\n") fh.flush() print(" %-28s %-6s -> %-34s %s" % (row["id"], row.get("label") or "-", (row["choice"] or ("REFUSED: " + row["refused"]))[:34], "%6.2fs" % row["seconds"] if row.get("seconds") else " — (%s tokens)" % row.get("prompt_tokens")), flush=True) wall = time.monotonic() - started after = gpu_rows(arm.card_uuid) energy = Boards.integrate(base + ".watts.csv") pin = verify_pin(arm.cards, before, after) # the card was already loaded on every task after the first, so the pin # is ALSO checked against the absolute reading, not only the growth. after_mib = {r["uuid"]: r["memory_mib"] for r in after} pin["memory_after_named_mib"] = after_mib.get(arm.card_uuid) pin["memory_after_per_card_mib"] = {c: after_mib.get(c) for c in arm.cards} pin["resident"] = (all((after_mib.get(c) or 0) > 2000 for c in arm.cards) and pin["others_quiet"]) rep = {"arm": asdict(arm), "task": t, "started": stamp, "ended": utc(), "wall_seconds": wall, "idle": idle, "energy": energy, "pin": pin, "summary": summarise(rows), "kit": load("manifest"), "census": pre, "overlong": overlong, "refusal_policy": ("README §A2.5: an item whose prompt exceeds the " "served context is refused with its measured token " "count and excluded from every rate; it is never " "truncated"), "readout": {"T": READOUT_T, "NOUL_T": NOUL_T, "NOUL_BIAS": NOUL_BIAS, "floor": FLOOR, "protocol": "openjev old (untargeted) path", "instructions": {"a": INSTR_A, "b": INSTR_B, "c": INSTR_C}}} reached = rep["summary"]["n_reached"] if energy.get("joules") is not None and reached: # per DECISION, so a refused row does not dilute the figure net = energy["joules"] - (idle.get("mean_watts") or 0) * energy["seconds"] rep["energy"]["joules_per_decision"] = energy["joules"] / reached rep["energy"]["net_joules_per_decision"] = net / reached rep["energy"]["decisions"] = reached with open(base + ".report.json", "w", encoding="utf-8") as fh: json.dump(rep, fh, indent=1, sort_keys=True, ensure_ascii=False) reports.append(rep) print(json.dumps(rep["summary"], default=str)[:600], flush=True) return reports def unload(arm: Arm) -> None: """Evict this arm's model, so the next one's residency is a GROWTH proof.""" body = json.dumps({"model": arm.model, "keep_alive": 0, "messages": [{"role": "user", "content": "."}], "stream": False, "options": {"num_predict": 1}}).encode() req = urllib.request.Request(arm.host + "/api/chat", data=body, method="POST", headers={"Content-Type": "application/json"}) try: urllib.request.urlopen(req, timeout=300).read() except Exception as exc: # noqa: BLE001 - best effort print("unload: %s" % exc, flush=True) time.sleep(8) def main(argv=None) -> int: ap = argparse.ArgumentParser() ap.add_argument("--arm", action="append", choices=sorted(ARMS)) ap.add_argument("--task", action="append", choices=sorted(TASKS)) ap.add_argument("--out", default=os.path.join(HERE, "rows")) ap.add_argument("--limit", type=int, default=0, help="smoke: first N items") ap.add_argument("--score", help="re-score a rows dir, calling nothing") ap.add_argument("--unload-between", action="store_true", help="evict the model when the model string changes") a = ap.parse_args(argv) if a.score: for f in sorted(os.listdir(a.score)): if f.endswith(".jsonl"): rows = [json.loads(l) for l in open(os.path.join(a.score, f))] print(f, json.dumps(summarise(rows), default=str)) return 0 if a.limit: for name, fn in list(TASKS.items()): TASKS[name] = (lambda fn=fn, n=a.limit: ( lambda arm, arts, over: ( r for i, r in enumerate(fn(arm, arts, over)) if i < n)))() seen_model = None for name in (a.arm or []): arm = ARMS[name] if a.unload_between and seen_model and seen_model != arm.model: unload(ARMS[seen_model_arm]) seen_model, seen_model_arm = arm.model, name run_arm(arm, a.task or ["a", "b", "c"], a.out) return 0 if __name__ == "__main__": raise SystemExit(main())