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,33 @@ 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()}
+    return {"unit": unit, "per_item": each, "ratio": each["PERUN"] / each["RIVAL"] if each["RIVAL"] else None, "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
+        rep = cost_report(json.loads(Path(argv[1]).read_text()))
+        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,124 @@
+#!/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 of the latest call 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 _urlopen(*a, **k):
+    _tl.usage = {}  # a failed or retried call leaves nothing behind
+    body = urllib.request.urlopen(*a, **k).read()
+    try:
+        _tl.usage = chat_usage(json.loads(body))
+    except (ValueError, AttributeError):
+        pass
+    return io.BytesIO(body)
+
+
+def _run(*a, **k):
+    _tl.usage = {}
+    p = subprocess.run(*a, **k)
+    if isinstance(p.stdout, str):
+        _tl.usage = claude_usage(p.stdout)
+    return p
+
+
+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 = {}
+        txt = gen(*a, **k)
+        _tl.pending = 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())

scripts/test_bench_cost_capture.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""Offline tests for per-arm cost capture (rival_eval usage parsing, arm_cost) and bench_verdict.cost_report. No network."""
+import json
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import bench_verdict as b  # noqa: E402
+import rival_eval as r  # noqa: E402
+import rival_eval_cost as rc  # noqa: E402
+import test_rival_eval as T  # noqa: E402
+
+
+def U(p, c, s=0.0):
+    return {"prompt_tokens": p, "completion_tokens": c, "spend_usd": s}
+
+
+class FakeResp:
+    def __init__(self, body):
+        self.body = body
+
+    def read(self, *a):
+        return json.dumps(self.body).encode()
+
+
+class Wrapped(unittest.TestCase):
+    """rival_eval itself is untouched; install() wraps it. Restore its attributes after each test."""
+
+    def setUp(self):
+        self.saved = {k: getattr(r, k) for k in ("generate", "save", "run_skill", "urllib", "subprocess")}
+        self.saved["flag"] = getattr(r, "_cost_installed", False)
+        rc.install()
+
+    def tearDown(self):
+        for k, v in self.saved.items():
+            setattr(r, "_cost_installed" if k == "flag" else k, v)
+
+    def sidecar(self, body):
+        """generate() through the real chat() with a faked HTTP response, then save() -> the usage sidecar content."""
+        with tempfile.TemporaryDirectory() as d, mock.patch.object(rc.urllib.request, "urlopen", return_value=FakeResp(body)), \
+                mock.patch.object(r, "load_key", return_value="k"):
+            p = Path(d) / "o" / "x.txt"
+            r.save(p, r.generate("pcss", "s", "u", None))
+            s = p.with_suffix(".usage.json")
+            return json.loads(s.read_text()) if s.exists() else None
+
+    def test_chat_records_reported_usage(self):
+        body = {"choices": [{"message": {"content": "hi"}}], "usage": {"prompt_tokens": 12, "completion_tokens": 5}}
+        self.assertEqual(self.sidecar(body), U(12, 5, None))  # no price reported: unmeasured, never 0.0
+        body["usage"]["cost"] = 0.004
+        self.assertEqual(self.sidecar(body), U(12, 5, 0.004))
+
+    def test_chat_without_usage_writes_no_sidecar(self):
+        self.assertIsNone(self.sidecar({"choices": [{"message": {"content": "hi"}}]}))
+
+    def test_save_without_generate_writes_no_sidecar(self):
+        with tempfile.TemporaryDirectory() as d:
+            p = Path(d) / "x.txt"
+            r.save(p, "t")
+            self.assertFalse(p.with_suffix(".usage.json").exists())
+
+    def test_claude_call_records_usage_and_cost(self):
+        lines = [json.dumps({"type": "rate_limit_event", "rate_limit_info": {"unifiedWindows": {"w": {"utilization": 0.1}}}}),
+                 json.dumps({"type": "result", "result": "ok", "total_cost_usd": 0.02,
+                             "usage": {"input_tokens": 10, "cache_creation_input_tokens": 100, "cache_read_input_tokens": 5, "output_tokens": 7}})]
+        with tempfile.TemporaryDirectory() as d, mock.patch.object(rc.subprocess, "run", return_value=mock.Mock(stdout="\n".join(lines))):
+            p = Path(d) / "x.txt"
+            r.save(p, r.generate("claude", "s", "u", r.State(1.0)))
+            self.assertEqual(json.loads(p.with_suffix(".usage.json").read_text()), U(115, 7, 0.02))
+
+    def test_run_skill_adds_cost_per_arm_and_resume_keeps_it(self):
+        tmp = Path(tempfile.mkdtemp())
+        name = T.fake_fixture(tmp, T.scen("held", 10, 10), T.scen("tune", 5))
+
+        def urlopen(req, timeout=0):
+            m = json.loads(req.data)
+            sysm, user = m["messages"][0]["content"], m["messages"][1]["content"]
+            if T.is_judge(sysm):
+                v = "1" if user.index("PERUN-ANSWER") < user.index("RIVAL-ANSWER") else "2"
+                return FakeResp({"choices": [{"message": {"content": T.judge_reply(user, v)}}]})
+            n = 3 if "PERUN skill text" in sysm else 6
+            return FakeResp({"choices": [{"message": {"content": ("PERUN-ANSWER" if n == 3 else "RIVAL-ANSWER") + user[-20:]}}],
+                             "usage": {"prompt_tokens": n, "completion_tokens": n}})
+        with mock.patch.object(r, "ROOT", tmp), mock.patch.object(r, "FIX", tmp / "scripts/eval-fixtures/rival"), \
+                mock.patch.object(rc.urllib.request, "urlopen", urlopen), mock.patch.object(r, "load_key", return_value="k"):
+            res = r.run_skill(name, "heldout", "pcss", tmp / "out", 2, 0.5)
+            again = r.run_skill(name, "heldout", "pcss", tmp / "out", 2, 0.5)  # all outputs cached: usage comes from the sidecars
+        for x in (res, again):
+            self.assertEqual(x["cost_per_arm"]["PERUN"], dict(n_measured=10, n_items=10, prompt_tokens=30, completion_tokens=30, spend_usd=None))
+            self.assertEqual(x["cost_per_arm"]["RIVAL"]["prompt_tokens"], 60)
+        saved = json.loads(next((tmp / "out" / name).glob("heldout-*/results.json")).read_text())
+        self.assertEqual(saved["cost_per_arm"], res["cost_per_arm"])
+
+
+class Capture(unittest.TestCase):
+    def test_claude_usage_without_result_is_empty(self):
+        self.assertEqual(rc.claude_usage('{"type": "assistant"}\nnot json'), {})
+
+    def test_arm_cost_sums_only_measured(self):
+        c = rc.arm_cost([U(10, 2, 0.5), U(20, 4, 0.25)], 3)
+        self.assertEqual((c["n_measured"], c["n_items"], c["prompt_tokens"], c["completion_tokens"], c["spend_usd"]), (2, 3, 30, 6, 0.75))
+
+
+def res(p, q, n=2):
+    return {"cost_per_arm": {"PERUN": rc.arm_cost(p, n), "RIVAL": rc.arm_cost(q, n)}}
+
+
+class Report(unittest.TestCase):
+    def test_tokens_when_free_endpoint(self):
+        rep = b.cost_report(res([U(10, 10)] * 2, [U(20, 20)] * 2))
+        self.assertEqual((rep["unit"], rep["ratio"], rep["ok"]), ("tokens", 0.5, True))
+
+    def test_usd_when_paid(self):
+        rep = b.cost_report(res([U(1, 1, 0.3)] * 2, [U(1, 1, 0.1)] * 2))
+        self.assertEqual((rep["unit"], rep["ok"]), ("usd", False))
+
+    def test_partial_or_missing_is_unmeasured(self):
+        self.assertIsNone(b.cost_report(res([U(1, 1)], [U(1, 1)] * 2))["ok"])
+        self.assertIsNone(b.cost_report({"spend_usd": 0.0})["ok"])
+
+    def test_ratio_value_and_direction(self):
+        rep = b.cost_report(res([U(10, 10), U(20, 20)], [U(30, 30), U(50, 50)]))  # perun 30/item, rival 80/item
+        self.assertEqual((rep["per_item"], rep["ratio"], rep["ok"]), ({"PERUN": 30.0, "RIVAL": 80.0}, 0.375, True))
+        rep = b.cost_report(res([U(30, 30), U(50, 50)], [U(10, 10), U(20, 20)]))
+        self.assertEqual((rep["ratio"], rep["ok"]), (80 / 30, False))
+
+    def test_failed_item_is_unmeasured(self):  # 3 scenarios, one Perun generation failed: 2 measured of 3 items
+        rep = b.cost_report({"cost_per_arm": {"PERUN": rc.arm_cost([U(1, 1)] * 2, 3), "RIVAL": rc.arm_cost([U(2, 2)] * 3, 3)}})
+        self.assertIsNone(rep["ok"])
+
+    def test_zero_cost_rival_fails_closed(self):
+        rep = b.cost_report(res([U(1, 1)] * 2, [U(0, 0)] * 2))
+        self.assertEqual((rep["ratio"], rep["ok"]), (None, False))
+
+    def test_mixed_units_unmeasured(self):
+        rep = b.cost_report(res([U(1, 1, 0.1)] * 2, [U(9, 9)] * 2))
+        self.assertIsNone(rep["ok"])
+        self.assertIn("units differ", rep["reason"])
+
+    def test_missing_price_makes_usd_none(self):
+        self.assertIsNone(rc.arm_cost([U(1, 1, 0.1), U(1, 1, None)], 2)["spend_usd"])
+
+    def test_old_results_json_cli_exit_3(self):
+        with tempfile.TemporaryDirectory() as d:
+            f = Path(d) / "results.json"
+            f.write_text(json.dumps({"spend_usd": 0.0, "n": 150}))
+            self.assertEqual(b.main(["--cost", str(f)]), 3)  # unmeasured is not a loss
+
+    def test_cli_usage_error_without_file(self):
+        self.assertEqual(b.main(["--cost"]), 2)
+
+    def test_cli_exit_code(self):
+        with tempfile.TemporaryDirectory() as d:
+            f = Path(d) / "results.json"
+            f.write_text(json.dumps(res([U(5, 5)] * 2, [U(9, 9)] * 2)))
+            self.assertEqual(b.main(["--cost", str(f)]), 0)
+            f.write_text(json.dumps(res([U(9, 9)] * 2, [U(5, 5)] * 2)))
+            self.assertEqual(b.main(["--cost", str(f)]), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()