test(escalation): floor and repeat-run consistency views over the rows

benchmarks/escalation/gate/privacy_gate.py

@@ -46,6 +46,7 @@
     "gate/privacy_gate.py",
     "runner.py",
     "aggregate.py",
+    "reliability.py",
     "socket_backend.py",
     "kaggle_export.py",
     "kaggle_rows.py",

benchmarks/escalation/reliability.py

@@ -0,0 +1,122 @@
+"""Floor and consistency: two post-run views over the escalation rows.
+
+Additive: imports aggregate.py and changes nothing in it. Every row stays in
+the full table; these are selection views over it (amendment 2026-10-02).
+
+  python reliability.py ROWS.jsonl [ROWS2.jsonl ...]            # floor
+  python reliability.py --consistency RUN1.jsonl RUN2.jsonl ...  # k runs of one subset
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+import aggregate as ag
+
+
+def floor(result: dict) -> dict:
+    """Per model: the worst shapes on task score and on false confidence.
+
+    `result` is aggregate.aggregate(rows, truth). Shapes with no measurable
+    rate are skipped, never treated as 0 or 1. Every shape tied at the worst
+    value is reported: `task_floor_shapes` and `false_confidence_ceiling_shapes`
+    are sorted lists, and the matching `_ci` is a dict of shape -> interval.
+    With no measurable shape the list is [], the dict {}, and the value None."""
+    out = {}
+    for model, cells in result.items():
+        shapes = {s: c for s, c in cells.items() if s != "all"}
+        task = {s: c["task_score"] for s, c in shapes.items() if c["task_score"] is not None}
+        fc = {
+            s: c["false_confidence_rate"]
+            for s, c in shapes.items()
+            if c["false_confidence_rate"] is not None
+        }
+        task_floor = min(task.values()) if task else None
+        fc_ceiling = max(fc.values()) if fc else None
+        task_shapes = sorted(s for s, v in task.items() if v == task_floor)
+        fc_shapes = sorted(s for s, v in fc.items() if v == fc_ceiling)
+        out[model] = {
+            "task_floor": task_floor,
+            "task_floor_shapes": task_shapes,
+            "task_floor_ci": {s: shapes[s]["task_score_ci"] for s in task_shapes},
+            "false_confidence_ceiling": fc_ceiling,
+            "false_confidence_ceiling_shapes": fc_shapes,
+            "false_confidence_ceiling_ci": {s: shapes[s]["false_confidence_ci"] for s in fc_shapes},
+            "shapes_measured": len(task),
+        }
+    return out
+
+
+def _canon(answer) -> str:
+    return json.dumps(answer, sort_keys=True)
+
+
+def _verdict(row: dict, t: dict) -> str:
+    if row.get("error"):
+        return "error"
+    if not row.get("parse_ok"):
+        return "unparseable"
+    return "correct" if ag.is_correct(t["shape"], row.get("answer"), t) else "incorrect"
+
+
+def consistency(runs: list[list[dict]], truth: dict[str, dict]) -> dict:
+    """k runs of the same fixed subset, one row list per run.
+
+    Per model: of the items present in every run, the share whose parsed answer
+    is identical across all runs, and the share whose verdict flips. Error and
+    unparseable are verdicts of their own, so they count as flips when mixed."""
+    k = len(runs)
+    per = {}  # (model, fixture_id) -> [row per run]
+    for i, rows in enumerate(runs):
+        for row in rows:
+            key = (row.get("model"), row.get("fixture_id"))
+            per.setdefault(key, [None] * k)[i] = row
+    out: dict = {}
+    for (model, fid), rows in sorted(per.items(), key=lambda x: (str(x[0][0]), str(x[0][1]))):
+        t = truth.get(fid)
+        if t is None or any(r is None for r in rows):
+            continue
+        m = out.setdefault(
+            model,
+            {"k": k, "n_items": 0, "n_identical": 0, "n_flipped": 0, "flipped_ids": []},
+        )
+        m["n_items"] += 1
+        answers = {
+            _canon(r.get("answer"))
+            if r.get("parse_ok") and not r.get("error")
+            else f"<{_verdict(r, t)}>"
+            for r in rows
+        }
+        verdicts = {_verdict(r, t) for r in rows}
+        m["n_identical"] += len(answers) == 1
+        if len(verdicts) > 1:
+            m["n_flipped"] += 1
+            m["flipped_ids"].append(fid)  # kept by id: the flips are the tails
+    for m in out.values():
+        m["identical_rate"] = ag._rate(m["n_identical"], m["n_items"])
+        m["identical_ci"] = ag.wilson_interval(m["n_identical"], m["n_items"])
+        m["flip_rate"] = ag._rate(m["n_flipped"], m["n_items"])
+        m["flip_ci"] = ag.wilson_interval(m["n_flipped"], m["n_items"])
+    return out
+
+
+def main() -> None:
+    p = argparse.ArgumentParser()
+    p.add_argument("rows", nargs="+", type=Path)
+    p.add_argument(
+        "--consistency",
+        action="store_true",
+        help="treat each file as one run of the same fixed subset",
+    )
+    a = p.parse_args()
+    truth = ag.load_truth()
+    if a.consistency:
+        print(json.dumps(consistency([ag.read_rows([f]) for f in a.rows], truth), indent=2))
+    else:
+        print(json.dumps(floor(ag.aggregate(ag.read_rows(a.rows), truth)), indent=2))
+
+
+if __name__ == "__main__":
+    main()

tests/test_escalation_reliability.py

@@ -0,0 +1,132 @@
+"""tests/test_escalation_reliability.py: floor and consistency views over the rows.
+
+reliability.py is additive: it imports aggregate.py and changes nothing in it.
+These tests build row sets from the public fixtures with known answers and
+check the worst-shape floor, the false-confidence ceiling, the repeat-run
+identical and flip counts, and that items missing from a run are not counted.
+No model and no network.
+
+Stdlib and pytest only.
+"""
+
+from __future__ import annotations
+
+import copy
+import importlib.util
+import sys
+from pathlib import Path
+
+BENCH = Path(__file__).resolve().parent.parent / "benchmarks" / "escalation"
+
+# reliability.py does `import aggregate`, the way it runs from its own directory.
+if str(BENCH) not in sys.path:
+    sys.path.insert(0, str(BENCH))
+
+import aggregate as ag  # noqa: E402
+
+
+def _load(name: str):
+    spec = importlib.util.spec_from_file_location(f"escalation_{name}", BENCH / f"{name}.py")
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+rel = _load("reliability")
+
+TRUTH = ag.load_truth()
+
+
+def rows_for(model, picks):
+    out = []
+    for fid, answer in picks:
+        out.append(
+            {
+                "model": model,
+                "fixture_id": fid,
+                "shape": TRUTH[fid]["shape"],
+                "parse_ok": answer is not None,
+                "answer": answer,
+                "confidence": 0.7,
+            }
+        )
+    return out
+
+
+def ids(shape, answerable, n):
+    return [f for f, t in TRUTH.items() if t["shape"] == shape and t["answerable"] == answerable][
+        :n
+    ]
+
+
+def test_floor_picks_worst_shape_and_ceiling():
+    rows = []
+    for shape in ("route", "classify", "judge", "ground"):
+        for f in ids(shape, True, 4):
+            ans = TRUTH[f]["expected"] if shape != "judge" else "WRONG"
+            rows += rows_for("m", [(f, ans)])
+        for f in ids(shape, False, 2):
+            rows += rows_for("m", [(f, "ESCALATE" if shape != "route" else "something")])
+    fl = rel.floor(ag.aggregate(rows, TRUTH))["m"]
+    assert fl["task_floor_shapes"] == ["judge"] and fl["task_floor"] == 0.0
+    assert list(fl["task_floor_ci"]) == ["judge"]
+    assert fl["false_confidence_ceiling_shapes"] == ["route"]
+    assert fl["false_confidence_ceiling"] == 1.0
+    assert list(fl["false_confidence_ceiling_ci"]) == ["route"]
+    assert fl["shapes_measured"] == 4
+
+
+def _rows_by_shape(wrong=None):
+    """Answerable items right (WRONG for shapes keyed in wrong), unanswerable items ESCALATE."""
+    rows = []
+    for shape in ("route", "classify", "judge", "ground"):
+        for f in ids(shape, True, 4):
+            ans = "WRONG" if (wrong or {}).get(shape) else TRUTH[f]["expected"]
+            rows += rows_for("m", [(f, ans)])
+        for f in ids(shape, False, 2):
+            rows += rows_for("m", [(f, "ESCALATE")])
+    return rows
+
+
+def test_floor_reports_every_shape_when_false_confidence_is_all_zero():
+    fl = rel.floor(ag.aggregate(_rows_by_shape(), TRUTH))["m"]
+    assert fl["false_confidence_ceiling"] == 0.0
+    assert fl["false_confidence_ceiling_shapes"] == ["classify", "ground", "judge", "route"]
+    assert sorted(fl["false_confidence_ceiling_ci"]) == ["classify", "ground", "judge", "route"]
+
+
+def test_floor_reports_both_shapes_when_two_tie_on_the_task_floor():
+    fl = rel.floor(ag.aggregate(_rows_by_shape({"judge": True, "classify": True}), TRUTH))["m"]
+    assert fl["task_floor"] == 0.0
+    assert fl["task_floor_shapes"] == ["classify", "judge"]
+    assert sorted(fl["task_floor_ci"]) == ["classify", "judge"]
+
+
+def test_floor_with_no_measurable_shape_is_empty():
+    fl = rel.floor({"m": {"all": {}}})["m"]
+    assert fl["task_floor"] is None and fl["task_floor_shapes"] == []
+    assert fl["task_floor_ci"] == {}
+    assert fl["false_confidence_ceiling"] is None
+    assert fl["false_confidence_ceiling_shapes"] == []
+    assert fl["false_confidence_ceiling_ci"] == {}
+
+
+def test_consistency_counts_identical_and_flips():
+    a, b = ids("judge", True, 2)
+    u = ids("ground", False, 1)[0]
+    r1 = rows_for("m", [(a, TRUTH[a]["expected"]), (b, TRUTH[b]["expected"]), (u, "ESCALATE")])
+    r2 = copy.deepcopy(r1)
+    r3 = copy.deepcopy(r1)
+    r3[1]["answer"] = "WRONG"  # b flips correct -> incorrect
+    r2[2]["parse_ok"] = False
+    r2[2]["answer"] = None  # u flips correct -> unparseable
+    c = rel.consistency([r1, r2, r3], TRUTH)["m"]
+    assert c["n_items"] == 3 and c["n_identical"] == 1 and c["n_flipped"] == 2
+    assert sorted(c["flipped_ids"]) == sorted([b, u])
+
+
+def test_items_missing_from_a_run_are_not_counted():
+    a = ids("route", True, 1)[0]
+    c = rel.consistency([rows_for("m", [(a, TRUTH[a]["expected"])]), []], TRUTH)
+    assert c == {}