Record per-arm bench cost and apply the cost check

changelog.d/bench-cost-capture.md

@@ -0,0 +1,3 @@
+### Added
+- `scripts/rival_eval_cost.py run ...` (a wrapper around the unchanged, hash-pinned `scripts/rival_eval.py`) records real per-arm usage (`prompt_tokens`, `completion_tokens`, `spend_usd`) from the PCSS response or `claude -p` JSON into `results.json` `cost_per_arm`, with a per-output sidecar so resumed runs keep it; arms without reported usage stay unmeasured.
+- `scripts/bench_verdict.py --cost results.json` prints per-arm cost per item and the Perun/rival ratio and exits 0 when Perun cost <= rival cost, 1 when Perun costs more, 2 on a usage error, 3 when unmeasured (a distinct code from a loss).

docs/perun-vs-rivals.md

@@ -1,5 +1,7 @@
 # Perun vs other skill bundles
 
+Harness: run rival rows with `scripts/rival_eval_cost.py run ...`, a wrapper around the hash-pinned `scripts/rival_eval.py` that also records per-arm cost into `results.json`; `scripts/bench_verdict.py --cost results.json` reads it (exit 0 Perun cost <= rival, 1 Perun costs more, 2 usage error, 3 unmeasured).
+
 **Bottom line (2026-10-17 to 2026-10-19 held-out runs, free PCSS models, judge protocol v2).** Five skills have a confirmed preference-only win over their rival (the judge preferred Perun with the 95% lower bound above 0.5; cost and user outcome were not measured, so none is a win on cost or outcome) (product-output-safety's is narrow: lower bound 0.518 on its preregistered replication, one earlier fresh set below 0.5); idea-critic won once (round 9, narrowly) and the preregistered replication on a second fresh set did not reproduce it, so it is not counted; four more are not wins (growth-analytics among them: its round-4 edit shipped on a paired gain but its win-rate lower bound is 0.483, point estimate 0.550; a round-5 candidate got no verdict because no judge was inside the slot-1 gate, and was not shipped) and none is unproven. A win needs the 95% CI lower bound above 0.5; every row below is a fresh held-out set of n=40 or more, with ties counted as 0.5. Rows with n=10 are not mixed in: they are the older history further down and are not comparable to this table.
 
 | Skill | Rival | Bench dir (`docs/bench/`) | Judge | n | Win rate [95% CI] | Verdict |

scripts/bench_verdict.py

@@ -25,7 +25,8 @@
 exits 1 if a Verdict*/Result* cell claims a Win without `cost $A vs $B` (A <= B) plus
 `outcome <name> X vs Y (higher|lower) is better` (X not worse than Y), or a
 `bench: docs/bench/<dir>` whose verdict_spec.json recomputes to WIN, or labels a preference-only win without "unmeasured".
-Reads files only.
+`bench_verdict.py --cost results.json` prints per-arm cost per item and the Perun/rival ratio (see cost_report). Exit codes:
+0 Perun cost <= rival cost, 1 Perun costs more, 2 usage error (no file argument), 3 unmeasured (no usage, partial, mixed units). Reads files only.
 """
 import functools
 import json
@@ -177,7 +178,39 @@ def check_doc(text, root=REPO):
     return bad
 
 
+def cost_report(res):
+    """Per-arm cost per item and the Perun/rival ratio from a rival_eval results.json `cost_per_arm`, plus the cost check
+    (Perun <= rival). Unit is dollars when an arm has a positive spend_usd, else generator tokens (prompt + completion); arms
+    whose units differ are unmeasured. Arms with no usage, or with only some items measured, are unmeasured: ok is None, never guessed."""
+    arms = res.get("cost_per_arm") or {}
+    per = {}
+    for a in ("PERUN", "RIVAL"):
+        c = arms.get(a) or {}
+        if not c.get("n_measured") or c["n_measured"] != c.get("n_items"):
+            return {"unit": None, "ok": None, "reason": f"{a} cost unmeasured"}
+        per[a] = c
+    units = {a: "usd" if (c.get("spend_usd") or 0) > 0 else "tokens" for a, c in per.items()}
+    if len(set(units.values())) > 1:
+        return {"unit": None, "ok": None, "reason": f"arm units differ: {units}"}
+    unit = units["PERUN"]
+    each = {a: (c["spend_usd"] if unit == "usd" else c["prompt_tokens"] + c["completion_tokens"]) / c["n_measured"] for a, c in per.items()}
+    if not each["RIVAL"]:  # no ratio; Perun > 0 costs more, both zero means nothing was measured
+        return {"unit": unit, "per_item": each, "ratio": None, "ok": False if each["PERUN"] else None, "reason": "RIVAL cost is zero"}
+    return {"unit": unit, "per_item": each, "ratio": each["PERUN"] / each["RIVAL"], "ok": each["PERUN"] <= each["RIVAL"]}
+
+
 def main(argv):
+    if argv[:1] == ["--cost"]:
+        if len(argv) != 2:
+            print("usage: bench_verdict.py --cost results.json", file=sys.stderr)
+            return 2
+        try:
+            rep = cost_report(json.loads(Path(argv[1]).read_text()))
+        except (OSError, ValueError, AttributeError, KeyError, TypeError) as e:  # unreadable results: unmeasured, never "costs more"
+            print(f"cost unmeasured: {type(e).__name__}: {e}", file=sys.stderr)
+            return 3
+        print(json.dumps(rep, indent=2))
+        return 3 if rep["ok"] is None else 0 if rep["ok"] else 1
     if argv[:1] == ["--check-doc"]:
         bad = check_doc(Path(argv[1]).read_text())
         for row in bad:

scripts/ci-tests.d/bench-cost-capture.sh

@@ -0,0 +1,2 @@
+# Registers the per-arm cost capture tests with CI (see the "Per-PR tests" step in ci.yml). Offline; no model is called.
+python3 scripts/test_bench_cost_capture.py

scripts/rival-eval-premerge.sh

@@ -1,13 +1,13 @@
 #!/usr/bin/env bash
-# Local pre-merge hook point for the rival harness (scripts/rival_eval.py): scores only the skills this branch touches
+# Local pre-merge hook point for the rival harness (scripts/rival_eval_cost.py, a wrapper that adds cost capture to the hash-pinned scripts/rival_eval.py): scores only the skills this branch touches
 # against their rivals on the frozen held-out sets and prints win rate, 95% CI and cost per skill. Runs on the developer's
 # machine with a local PCSS_KEY (environment or repo-root .env); there is no CI secret, so CI never runs it.
 # Advisory by default (exit 0, a missing key is a skip); RIVAL_EVAL_STRICT=1 exits 1 when any touched skill FAILs.
 # Wire it as the pre-push extra tier:  DCR_PREPUSH_CMD="bash scripts/rival-eval-premerge.sh" git push
 # or run it by hand before landing. RIVAL_EVAL_BASE (default origin/main) is the diff base.
 set -u
 ROOT="$(cd "$(dirname "$0")/.." && pwd)"
-out="$(python3 "$ROOT/scripts/rival_eval.py" run --changed "${RIVAL_EVAL_BASE:-origin/main}" --out "${TMPDIR:-/tmp}/rival-eval" 2>&1)"
+out="$(python3 "$ROOT/scripts/rival_eval_cost.py" run --changed "${RIVAL_EVAL_BASE:-origin/main}" --out "${TMPDIR:-/tmp}/rival-eval" 2>&1)"
 rc=$?
 printf '%s\n' "$out"
 if printf '%s' "$out" | grep -q 'PCSS_KEY missing'; then echo "rival-eval skipped: no PCSS_KEY"; exit 0; fi

scripts/rival_eval_cost.py

@@ -0,0 +1,141 @@
+#!/usr/bin/env python3
+"""Per-arm cost capture for rival_eval.py, without editing it (its sha256 is frozen by test_evil_twin_bench_r2.py).
+
+usage: rival_eval_cost.py run ...   (same arguments as rival_eval.py)
+
+`install()` wraps rival_eval's `generate`, `save` and `run_skill` and shims its `urllib`/`subprocess` so that the real usage
+reported by the PCSS response (`usage.prompt_tokens`, `completion_tokens`, `cost`) or the `claude -p` result line
+(`usage`, `total_cost_usd`) is written next to each generated output as `<id>-<hash>.usage.json` (so a resumed run keeps the
+numbers of cached outputs), and `run_skill` adds `cost_per_arm` to results.json. Nothing is estimated: an arm whose items
+reported no usage stays unmeasured. `bench_verdict.py --cost results.json` reads the result. Side effects: results.json is
+rewritten after run_skill; no network or process is added beyond what rival_eval already does.
+"""
+import functools
+import io
+import json
+import subprocess
+import sys
+import threading
+import urllib.request
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import rival_eval as r  # noqa: E402
+
+_tl = threading.local()  # usage summed over the current generate() call's attempts on this thread (generation runs in a thread pool)
+_TOKENS = ("input_tokens", "cache_creation_input_tokens", "cache_read_input_tokens")
+
+
+def chat_usage(body):
+    """PCSS response body (dict) -> usage dict, or {} when the API reports no token counts. spend_usd is None (unmeasured) without usage.cost."""
+    u = body.get("usage") or {}
+    if "prompt_tokens" in u and "completion_tokens" in u:
+        return dict(prompt_tokens=u["prompt_tokens"], completion_tokens=u["completion_tokens"], spend_usd=u.get("cost"))
+    return {}
+
+
+def claude_usage(stdout):
+    """`claude -p` stream-json stdout -> usage dict from its result line, or {} when absent."""
+    for line in stdout.splitlines():
+        try:
+            j = json.loads(line)
+        except ValueError:
+            continue
+        u = j.get("usage") or {}
+        if j.get("type") == "result" and "total_cost_usd" in j and "input_tokens" in u and "output_tokens" in u:
+            return dict(prompt_tokens=sum(u.get(k) or 0 for k in _TOKENS), completion_tokens=u["output_tokens"], spend_usd=float(j["total_cost_usd"] or 0))
+    return {}
+
+
+def arm_cost(arm_usage, n_items):
+    """Sum per-item usage dicts (items without a reported usage are left out, never estimated) -> per-arm cost record."""
+    paid = bool(arm_usage) and all(u.get("spend_usd") is not None for u in arm_usage)  # one missing price: dollars unmeasured
+    return dict(n_measured=len(arm_usage), n_items=n_items,
+                **{k: round(sum(u[k] for u in arm_usage), 6) for k in ("prompt_tokens", "completion_tokens")},
+                spend_usd=round(sum(u["spend_usd"] for u in arm_usage), 6) if paid else None)
+
+
+class _Shim:
+    """A module look-alike: `over` attributes replace the real ones, everything else is the real module's."""
+    def __init__(self, mod, **over):
+        self._mod, self._over = mod, over
+
+    def __getattr__(self, k):
+        return self._over[k] if k in self._over else getattr(self._mod, k)
+
+
+def _add(u):
+    """Add one attempt's usage to this thread's running total. Retries are billed too, so an attempt without usage makes the item unmeasured."""
+    if not u:
+        _tl.miss = True
+        return
+    t = getattr(_tl, "usage", None) or {}
+    _tl.usage = dict(prompt_tokens=t.get("prompt_tokens", 0) + u["prompt_tokens"],
+                     completion_tokens=t.get("completion_tokens", 0) + u["completion_tokens"],
+                     spend_usd=None if (t and t.get("spend_usd") is None) or u["spend_usd"] is None else (t.get("spend_usd") or 0) + u["spend_usd"])
+
+
+def _urlopen(*a, **k):
+    u = {}
+    try:
+        body = urllib.request.urlopen(*a, **k).read()
+        try:
+            u = chat_usage(json.loads(body))
+        except (ValueError, AttributeError):
+            pass
+        return io.BytesIO(body)
+    finally:
+        _add(u)
+
+
+def _run(*a, **k):
+    u = {}
+    try:
+        p = subprocess.run(*a, **k)
+        if isinstance(p.stdout, str):
+            u = claude_usage(p.stdout)
+        return p
+    finally:
+        _add(u)
+
+
+def install():
+    """Wrap rival_eval in place (idempotent). Call before rival_eval.main()."""
+    if getattr(r, "_cost_installed", False):
+        return
+    r._cost_installed = True
+    r.urllib = _Shim(urllib, request=_Shim(urllib.request, urlopen=_urlopen))
+    r.subprocess = _Shim(subprocess, run=_run)
+    gen, save, run_skill = r.generate, r.save, r.run_skill
+
+    @functools.wraps(gen)
+    def generate(*a, **k):
+        _tl.usage, _tl.miss = {}, False
+        txt = gen(*a, **k)
+        _tl.pending = {} if _tl.miss else dict(_tl.usage)  # the output about to be saved on this thread owns this usage
+        return txt
+
+    @functools.wraps(save)
+    def save_(p, text):
+        save(p, text)
+        u, _tl.pending = getattr(_tl, "pending", None), None
+        if u:
+            p.with_suffix(".usage.json").write_text(json.dumps(u))
+
+    @functools.wraps(run_skill)
+    def run_skill_(skill, split, backend, out, jobs, cap):
+        res = run_skill(skill, split, backend, out, jobs, cap)
+        scs = r.load_scenarios(skill, split)
+        d = max((out / skill).glob(f"{split}-*/results.json"), key=lambda f: f.stat().st_mtime).parent  # the dir run_skill just wrote
+        uses = {a: [json.loads(f.read_text()) for sc in scs
+                    if (f := d / "outputs" / a / f"{sc['id']}-{r.sha(sc['prompt'])[:8]}.usage.json").exists()] for a in r.ARMS}
+        res["cost_per_arm"] = {a: arm_cost(uses[a], len(scs)) for a in r.ARMS}
+        (d / "results.json").write_text(json.dumps(res, indent=1) + "\n")
+        return res
+
+    r.generate, r.save, r.run_skill = generate, save_, run_skill_
+
+
+if __name__ == "__main__":
+    install()
+    sys.exit(r.main())