From a55220ef4a12fa1662cb9b613cfe55a1f6ca1bcf Mon Sep 17 00:00:00 2001 From: Remigiusz Antczak Date: Sat, 24 Oct 2026 09:59:20 +0200 Subject: [PATCH] bench: record real per-arm usage and apply the cost check rival_eval_cost.py (a wrapper around the unchanged, hash-pinned rival_eval.py; rival-eval-premerge.sh now runs it) writes cost_per_arm (prompt_tokens, completion_tokens, spend_usd) taken from the PCSS response or claude -p JSON, never estimated, with a per-output sidecar so resumed runs keep it. bench_verdict.py --cost prints per-arm cost per item and the Perun/rival ratio; exit 0 Perun cost <= rival, 1 Perun costs more, 2 usage error, 3 unmeasured. A billed retry adds to the item's usage, and an attempt with unknown usage leaves the item unmeasured. Zero cost on both arms reads unmeasured, and an unreadable results file exits 3. Tests: 21/21 (test_bench_cost_capture) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_014QNFFPWVcPUbeuvwM5fk2Z --- changelog.d/bench-cost-capture.md | 3 + docs/perun-vs-rivals.md | 2 + scripts/bench_verdict.py | 35 ++++- scripts/ci-tests.d/bench-cost-capture.sh | 2 + scripts/rival-eval-premerge.sh | 4 +- scripts/rival_eval_cost.py | 141 +++++++++++++++++ scripts/test_bench_cost_capture.py | 190 +++++++++++++++++++++++ 7 files changed, 374 insertions(+), 3 deletions(-) create mode 100644 changelog.d/bench-cost-capture.md create mode 100644 scripts/ci-tests.d/bench-cost-capture.sh create mode 100644 scripts/rival_eval_cost.py create mode 100644 scripts/test_bench_cost_capture.py diff --git a/changelog.d/bench-cost-capture.md b/changelog.d/bench-cost-capture.md new file mode 100644 index 000000000..fe1d2fc4e --- /dev/null +++ b/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). diff --git a/docs/perun-vs-rivals.md b/docs/perun-vs-rivals.md index ad16125ae..60bfdfa4e 100644 --- a/docs/perun-vs-rivals.md +++ b/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 | diff --git a/scripts/bench_verdict.py b/scripts/bench_verdict.py index 96bff9dd1..12ebd58b9 100644 --- a/scripts/bench_verdict.py +++ b/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 X vs Y (higher|lower) is better` (X not worse than Y), or a `bench: docs/bench/` 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: diff --git a/scripts/ci-tests.d/bench-cost-capture.sh b/scripts/ci-tests.d/bench-cost-capture.sh new file mode 100644 index 000000000..92bee0b9e --- /dev/null +++ b/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 diff --git a/scripts/rival-eval-premerge.sh b/scripts/rival-eval-premerge.sh index fbfba5f6c..9e5de6b6c 100755 --- a/scripts/rival-eval-premerge.sh +++ b/scripts/rival-eval-premerge.sh @@ -1,5 +1,5 @@ #!/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. @@ -7,7 +7,7 @@ # 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 diff --git a/scripts/rival_eval_cost.py b/scripts/rival_eval_cost.py new file mode 100644 index 000000000..fb8b6f947 --- /dev/null +++ b/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 `-.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()) diff --git a/scripts/test_bench_cost_capture.py b/scripts/test_bench_cost_capture.py new file mode 100644 index 000000000..3a831a8ed --- /dev/null +++ b/scripts/test_bench_cost_capture.py @@ -0,0 +1,190 @@ +#!/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_retries_are_summed_or_unmeasured(self): + ok = {"choices": [{"message": {"content": "hi"}}], "usage": {"prompt_tokens": 12, "completion_tokens": 5}} + empty = {"choices": [{"message": {"content": ""}}], "usage": {"prompt_tokens": 10, "completion_tokens": 1}} + def run(*resps): + with tempfile.TemporaryDirectory() as d, mock.patch.object(rc.urllib.request, "urlopen", side_effect=list(resps)), \ + mock.patch.object(r, "load_key", return_value="k"), mock.patch.object(r.time, "sleep"): + 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 + self.assertEqual(run(FakeResp(empty), FakeResp(ok)), U(22, 6, None)) # the billed empty attempt counts + self.assertIsNone(run(OSError("timeout"), FakeResp(ok))) # an attempt with unknown usage: item unmeasured + + 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_both_zero_is_unmeasured(self): + self.assertIsNone(b.cost_report(res([U(0, 0)] * 2, [U(0, 0)] * 2))["ok"]) + + def test_cli_unreadable_results_exit_3(self): + with tempfile.TemporaryDirectory() as d: + f = Path(d) / "results.json" + self.assertEqual(b.main(["--cost", str(f)]), 3) # missing + f.write_text("{not json") + self.assertEqual(b.main(["--cost", str(f)]), 3) + f.write_text("[1]") + self.assertEqual(b.main(["--cost", str(f)]), 3) + + 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()