"""CPU tests for the JevBench converter, scorer and training guard (deg.bench.jevbench).""" from __future__ import annotations import json import math from pathlib import Path import pytest from deg.api import parse_systemone, systemone_response from deg.bench import jevbench as jb from deg.schema import compile_schema NOUL = {"id": "original-policy-01-0", "family": "policy", "group": "original-policy-01", "split": "public", "expected": "no", "labels": ["no", "yes"], "question": {"type": "noul", "instructions": "Is the refund permitted?", "criteria": {"false": "A condition is missing.", "true": "Every condition holds."}}, "provenance": {"license": "MIT"}, "state": "Policy: refunds require a receipt. The customer has no receipt."} NOUL2 = {**NOUL, "id": "original-policy-01-1", "state": "A receipt is required for any refund; this buyer lost theirs. Refund?"} CHOICE = {"id": "easy-intent-00", "family": "intent", "group": None, "split": "public", "expected": "track_order", "labels": ["track_order", "cancel_order"], "question": {"type": "choice", "instructions": "Which intent?", "criteria": {"cancel_order": "Wants to cancel", "track_order": "Wants to know where an order is"}}, "provenance": {}, "state": "Where is my package? I ordered it last week and it " "still has not arrived at my house."} SCORE = {"id": "original-ordinal-01-0", "family": "ordinal", "group": None, "split": "public", "expected": 2, "labels": ["0", "1", "2"], "question": {"type": "score", "instructions": "How urgent?", "criteria": ["not urgent", "somewhat", "very urgent"]}, "provenance": {}, "state": "The server room is on fire right now."} PROB = {"id": "hard-opus-b-probability-02", "family": "probability", "group": "g-p", "split": "public", "expected": "late", "labels": ["late", "on_time", "lost"], "question": {"type": "choice", "instructions": "Likely outcome?", "criteria": {"late": "arrives late", "lost": "never arrives", "on_time": "arrives on time"}}, "provenance": {"gold_probs": {"late": 0.6, "on_time": 0.3, "lost": 0.1}}, "state": {"log": "Of 40 comparable shipments 24 were late, 12 on time and 4 lost."}} @pytest.fixture() def repo(tmp_path: Path) -> Path: pub = tmp_path / "datasets" / "public" pub.mkdir(parents=True) files = {"easy.jsonl": [CHOICE], "original.jsonl": [NOUL, NOUL2, SCORE], "hard.jsonl": [PROB]} for name, rows in files.items(): (pub / name).write_text("".join(json.dumps(r) + "\n" for r in rows)) res = tmp_path / "results" / "v1.2" res.mkdir(parents=True) (res / "jevbench-v1.2-per-task.json").write_text(json.dumps({"systems": {"x": { "public_tasks": {"easy-intent-00": ["c", 0.5], "original-policy-01-0": ["w", 0.5], "original-policy-01-1": ["c", 0.5], "original-ordinal-01-0": ["c", 0.5], "hard-opus-b-probability-02": ["f", 0.5]}}}})) (res / "jevbench-v1.2-results.json").write_text(json.dumps({"systems": [ {"key": "x", "display": "X", "jevbench_score": 70.0, "axes": {}, "tiers": {}, "cost": {"usd_per_1000": 0.01}}]})) return tmp_path def test_load_tasks_tiers_and_validation(repo: Path) -> None: tasks = jb.load_tasks(repo) assert [t["tier"] for t in tasks] == ["easy", "standard", "standard", "standard", "hard"] bad = {**CHOICE, "expected": "refund"} with pytest.raises(ValueError): jb.validate_task(bad) with pytest.raises(ValueError): jb.validate_task({**NOUL, "state": {"expected": "no"}}) def test_payload_is_the_typesafe_adapter_request() -> None: p = jb.systemone_payload(CHOICE, "jev-latest") assert p == {"model": "jev-latest", "state": CHOICE["state"], "questions": {"decision": {"type": "choice", "instructions": "Which intent?", "criteria": CHOICE["question"]["criteria"]}}} # criteria keep the file's order (the option order JevBench sends) assert list(p["questions"]["decision"]["criteria"]) == ["cancel_order", "track_order"] @pytest.mark.parametrize("task,gold", [(NOUL, False), (CHOICE, "track_order"), (SCORE, 2), (PROB, "late")]) def test_record_is_deg_systemone_schema(task: dict, gold: object) -> None: rec = jb.record({**task, "tier": "x"}) plan = compile_schema(rec["schema"]) (fld,) = plan.fields assert fld.pointer == "/decision" and fld.mode.startswith("decide.") assert rec["value"] == {"decision": gold} assert rec["schema"] == parse_systemone(jb.systemone_payload(task)).schema def _fake_fill(task: dict, probs: list[float]) -> tuple: job = parse_systemone(jb.systemone_payload(task)) kind = task["question"]["type"] meta = ({"mode": "decide.noul", "probability": probs[0]} if kind == "noul" else {"mode": f"decide.{kind}", "probabilities": probs, "confidence": max(probs)}) value = (probs[0] > 0.5 if kind == "noul" else (list(task["question"]["criteria"])[probs.index(max(probs))] if kind == "choice" else probs.index(max(probs)))) resp = {"value": {"decision": value}, "meta": {"/decision": meta}, "usage": {"prompt_tokens": 50, "generated_tokens": 0}} return job, systemone_response(job, resp) def test_probs_from_deg_systemone_answers() -> None: _, body = _fake_fill(NOUL, [0.2]) assert jb.probs_from_answer(NOUL, body["answers"]["decision"]) == pytest.approx( {"yes": 0.2, "no": 0.8}) _, body = _fake_fill(CHOICE, [0.1, 0.9]) assert jb.probs_from_answer(CHOICE, body["answers"]["decision"]) == { "cancel_order": 0.1, "track_order": 0.9} _, body = _fake_fill(SCORE, [0.1, 0.2, 0.7]) assert jb.probs_from_answer(SCORE, body["answers"]["decision"]) == { "0": 0.1, "1": 0.2, "2": 0.7} with pytest.raises(ValueError): jb.probs_from_answer(NOUL, {"noul": True}) def test_score_item_rules() -> None: t = {**CHOICE, "tier": "easy"} s = jb.score_item(t, {"track_order": 0.7, "cancel_order": 0.3}) assert s["valid"] and s["strict_valid"] and s["correct"] and s["confidence"] == 0.7 s = jb.score_item(t, {"track_order": 0.705, "cancel_order": 0.3}) # rounding band assert s["valid"] and not s["strict_valid"] and s["correct"] assert sum(s["probs"].values()) == pytest.approx(1.0) s = jb.score_item(t, {"track_order": 0.9, "cancel_order": 0.3}) # outside the band assert not s["valid"] and s["correct"] is False s = jb.score_item(t, {"track_order": 1.0}) # missing a label key assert not s["valid"] s = jb.score_item(t, {"track_order": 0.5, "cancel_order": 0.5}) # tie -> smallest label assert s["predicted"] == "cancel_order" and s["correct"] is False s = jb.score_item({**SCORE, "tier": "standard"}, {"0": 0.4, "1": 0.2, "2": 0.4}) assert s["predicted"] == "0" and s["ordinal_ev"] == pytest.approx(1.0) # argmax, not EV s = jb.score_item({**PROB, "tier": "hard"}, {"late": 0.5, "on_time": 0.4, "lost": 0.1}) assert s["tvd"] == pytest.approx(0.1) assert jb.score_item(t, None, "boom")["correct"] is False def test_published_formulas_reproduce_jev_1_13_row() -> None: """Jev 1.13.0's v1.3.0 row (results/v1.2/jevbench-v1.2-results.json).""" hist = {"easy": {2: 18, 4: 13, 5: 41}, "standard": {2: 32, 4: 40, 5: 12, 6: 12}, "judge": {2: 68, 9: 78}, "hard": {2: 77, 3: 26, 4: 73, 5: 38, 6: 6}} chances = {k: sum(c / o for o, c in v.items()) / sum(v.values()) for k, v in hist.items()} tiers = {"easy": 1.0, "standard": 0.9895833333333334, "judge": 0.9452054794520548, "hard": 0.740909090909091} assert jb.intelligence(tiers, chances) == pytest.approx(85.69176285262259) assert jb.speed(0.6524335257709026, 0.7221905551850795, "api") == pytest.approx( 83.26811926100174) assert jb.cost(0.03991423595505617) == pytest.approx(51.96616538951724) assert jb.calibration(0.06061111111111118, 1 - 77.42800000000001 / 100) == pytest.approx( 82.6528888888889) axes = {"intelligence": 85.69176285262259, "calibration": 82.6528888888889, "speed": 83.26811926100174, "cost": 51.96616538951724} assert jb.jevbench_score(axes) == pytest.approx(74.40448849535433) def test_near_chance_penalty_and_self_hosted_latency() -> None: axes = {"intelligence": 25.0, "calibration": 80.0, "speed": 80.0, "cost": 80.0} base = math.exp(0.25 * sum(math.log(v) for v in axes.values())) assert jb.jevbench_score(axes) == pytest.approx(base * 0.25) assert jb.adjusted_latency(0.5, "gpu") == pytest.approx(1.15) assert jb.adjusted_latency(0.5, "demo") == pytest.approx(1.0) assert jb.harmonic_view({**axes, "intelligence": 100.0}) == pytest.approx( 4 / (1 / 100 + 3 / 80)) def test_summarize_axes(repo: Path) -> None: tasks = jb.load_tasks(repo) preds = { "easy-intent-00": {"probs": {"track_order": 0.9, "cancel_order": 0.1}, "latency_s": 0.1, "input_tokens": 100}, "original-policy-01-0": {"probs": {"yes": 0.2, "no": 0.8}, "latency_s": 0.2, "input_tokens": 100}, "original-policy-01-1": {"probs": {"yes": 0.6, "no": 0.4}, "latency_s": 0.3, "input_tokens": 100}, "original-ordinal-01-0": {"error": "FillError: x", "latency_s": 0.4}, "hard-opus-b-probability-02": {"probs": {"late": 0.6, "on_time": 0.3, "lost": 0.1}, "latency_s": 1.0, "input_tokens": 400}, } s = jb.summarize(tasks, preds, usd_per_m_input=0.02) assert s["tiers"] == {"easy": 1.0, "standard": pytest.approx(1 / 3), "hard": 1.0} ch = s["tier_chance"] assert ch["standard"] == pytest.approx((0.5 + 0.5 + 1 / 3) / 3) want_i = (0.14 * 100 + 0.28 * jb.chance_corrected(1 / 3, ch["standard"]) + 0.30 * 100) / 0.72 assert s["axes"]["intelligence"] == pytest.approx(want_i) # one hard item, conf .6 correct -> ECE .4; TVD 0 assert s["calibration_detail"]["ece_hard"] == pytest.approx(0.4) assert s["axes"]["calibration"] == pytest.approx((100 * (1 - 0.4 / 0.5) + 100) / 2) assert s["latency"]["standard"]["p50_s"] == pytest.approx(0.3) assert s["cost"]["usd_per_1000"] == pytest.approx(175 * 0.02 / 1000) assert s["paraphrase"] == {"pairs": 1, "both_answered_valid": 1, "both_correct": 0, "consistency": 0.0} assert s["valid"] == 4 and s["attempted"] == 5 def test_leaderboard_public_intelligence(repo: Path) -> None: (row,) = jb.leaderboard(repo) assert row["public_tiers"] == {"easy": 1.0, "standard": pytest.approx(2 / 3), "hard": 0.0} assert row["published_score"] == 70.0 and row["public_intelligence"] is not None def test_contamination_guard(repo: Path) -> None: g = jb.ContaminationGuard(repo) assert g.loaded == 5 assert g.match(NOUL["state"]) # exact long_state = PROB["state"] assert g.match("Intro text. " + json.dumps(long_state, sort_keys=True) + " trailing") assert g.match("Header\n" + CHOICE["state"].upper() + "\nfooter") # embedded, normalised assert g.match("Where is my package? It was ordered yesterday.") is None short = {**SCORE, "id": "easy-intent-11", "state": "Set an alarm for 7 am."} (repo / "datasets" / "public" / "easy.jsonl").write_text( json.dumps(CHOICE) + "\n" + json.dumps(short) + "\n") g2 = jb.ContaminationGuard(repo) assert g2.match("set an alarm for 7 AM please") # the item, lightly edited assert g2.match("hey, i need to set an alarm for 7 am tomorrow but also can you find " "the best pizza place near me for lunch?") is None # ordinary language assert g.match("A completely unrelated support ticket about printer toner.") is None empty = jb.ContaminationGuard(repo / "missing") assert empty.loaded == 0 and empty.match(NOUL["state"]) is None def test_guard_path_refuses_jevbench() -> None: with pytest.raises(ValueError): jb.guard_path("/data/deg/external/jevbench/repo/datasets/public/hard.jsonl") jb.guard_path("/data/deg/datagen/full-off/records.jsonl")