From dea882cac2bcb360055364a35887794f18a631bd Mon Sep 17 00:00:00 2001 From: rudi193-cmd Date: Fri, 2 Oct 2026 15:57:07 -0600 Subject: [PATCH 1/3] test(escalation): floor and repeat-run consistency views over the rows Adds reliability.py (worst-shape floor and false-confidence ceiling; k-run identical/flip consistency) over aggregate.py, which is unchanged, plus three tests. The privacy gate allowlist gains reliability.py. Persona: hanuman Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_014195BDJ4JX3rsqMe52bcWx --- benchmarks/escalation/gate/privacy_gate.py | 1 + benchmarks/escalation/reliability.py | 119 +++++++++++++++++++++ tests/test_escalation_reliability.py | 94 ++++++++++++++++ 3 files changed, 214 insertions(+) create mode 100644 benchmarks/escalation/reliability.py create mode 100644 tests/test_escalation_reliability.py diff --git a/benchmarks/escalation/gate/privacy_gate.py b/benchmarks/escalation/gate/privacy_gate.py index 74aa012..281c420 100644 --- a/benchmarks/escalation/gate/privacy_gate.py +++ b/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", diff --git a/benchmarks/escalation/reliability.py b/benchmarks/escalation/reliability.py new file mode 100644 index 0000000..7ea02ed --- /dev/null +++ b/benchmarks/escalation/reliability.py @@ -0,0 +1,119 @@ +"""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 shape 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.""" + out = {} + for model, cells in result.items(): + shapes = {s: c for s, c in cells.items() if s != "all"} + task = [(c["task_score"], s) for s, c in shapes.items() if c["task_score"] is not None] + fc = [ + (c["false_confidence_rate"], s) + for s, c in shapes.items() + if c["false_confidence_rate"] is not None + ] + worst_task = min(task) if task else (None, None) + worst_fc = max(fc) if fc else (None, None) + out[model] = { + "task_floor": worst_task[0], + "task_floor_shape": worst_task[1], + "task_floor_ci": shapes[worst_task[1]]["task_score_ci"] if worst_task[1] else None, + "false_confidence_ceiling": worst_fc[0], + "false_confidence_ceiling_shape": worst_fc[1], + "false_confidence_ceiling_ci": ( + shapes[worst_fc[1]]["false_confidence_ci"] if worst_fc[1] else None + ), + "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() diff --git a/tests/test_escalation_reliability.py b/tests/test_escalation_reliability.py new file mode 100644 index 0000000..8fa392e --- /dev/null +++ b/tests/test_escalation_reliability.py @@ -0,0 +1,94 @@ +"""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_shape"] == "judge" and fl["task_floor"] == 0.0 + assert fl["false_confidence_ceiling_shape"] == "route" and fl["false_confidence_ceiling"] == 1.0 + assert fl["shapes_measured"] == 4 + + +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 == {} From f0c44cd772b89209e091fce4cbff9db99871adec Mon Sep 17 00:00:00 2001 From: rudi193-cmd Date: Fri, 2 Oct 2026 15:59:21 -0600 Subject: [PATCH 2/3] test(escalation): floor reports every tied worst shape floor() picked the worst shape with min/max over (value, shape) tuples, so a tie went to whichever shape name sorted first or last. On the Day-2 rows gemini-3.1-pro and gpt-5.5 have false confidence 0.0 on all four shapes and both were labelled "route", which is only the alphabetically last shape. task_floor_shape / false_confidence_ceiling_shape become sorted lists task_floor_shapes / false_confidence_ceiling_shapes, and the matching _ci keys become dicts of shape -> interval. No measurable shape gives [] and {} with a None value. Persona: hanuman Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_014195BDJ4JX3rsqMe52bcWx --- benchmarks/escalation/reliability.py | 35 ++++++++++++----------- tests/test_escalation_reliability.py | 42 ++++++++++++++++++++++++++-- 2 files changed, 59 insertions(+), 18 deletions(-) diff --git a/benchmarks/escalation/reliability.py b/benchmarks/escalation/reliability.py index 7ea02ed..8e0cc8f 100644 --- a/benchmarks/escalation/reliability.py +++ b/benchmarks/escalation/reliability.py @@ -17,30 +17,33 @@ def floor(result: dict) -> dict: - """Per model: the worst shape on task score and on false confidence. + """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.""" + 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 = [(c["task_score"], s) for s, c in shapes.items() if c["task_score"] is not None] - fc = [ - (c["false_confidence_rate"], s) + 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 - ] - worst_task = min(task) if task else (None, None) - worst_fc = max(fc) if fc else (None, 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": worst_task[0], - "task_floor_shape": worst_task[1], - "task_floor_ci": shapes[worst_task[1]]["task_score_ci"] if worst_task[1] else None, - "false_confidence_ceiling": worst_fc[0], - "false_confidence_ceiling_shape": worst_fc[1], - "false_confidence_ceiling_ci": ( - shapes[worst_fc[1]]["false_confidence_ci"] if worst_fc[1] else None - ), + "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 diff --git a/tests/test_escalation_reliability.py b/tests/test_escalation_reliability.py index 8fa392e..df0eea2 100644 --- a/tests/test_escalation_reliability.py +++ b/tests/test_escalation_reliability.py @@ -69,11 +69,49 @@ def test_floor_picks_worst_shape_and_ceiling(): 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_shape"] == "judge" and fl["task_floor"] == 0.0 - assert fl["false_confidence_ceiling_shape"] == "route" and fl["false_confidence_ceiling"] == 1.0 + 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_shapes=()): + """Answerable items right (WRONG for wrong_shapes), unanswerable items ESCALATE.""" + rows = [] + for shape in ("route", "classify", "judge", "ground"): + for f in ids(shape, True, 4): + ans = "WRONG" if shape in wrong_shapes 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", "classify")), 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] From 2a8670d898908de716228d4c6e89712fa4856bfc Mon Sep 17 00:00:00 2001 From: rudi193-cmd Date: Fri, 2 Oct 2026 16:21:26 -0600 Subject: [PATCH 3/3] test(escalation): row builder asks no membership of its loop variable _rows_by_shape took a tuple and asked `shape in wrong_shapes` of its loop variable, the _filters_by_membership shape tests/test_scans_fire.py flags as an unplanted scan helper. It builds rows and is not a scan. Take a lookup instead and read it with .get(shape); behaviour is unchanged and the rule is untouched. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_014195BDJ4JX3rsqMe52bcWx Persona: hanuman --- tests/test_escalation_reliability.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_escalation_reliability.py b/tests/test_escalation_reliability.py index df0eea2..0ac8288 100644 --- a/tests/test_escalation_reliability.py +++ b/tests/test_escalation_reliability.py @@ -77,12 +77,12 @@ def test_floor_picks_worst_shape_and_ceiling(): assert fl["shapes_measured"] == 4 -def _rows_by_shape(wrong_shapes=()): - """Answerable items right (WRONG for wrong_shapes), unanswerable items ESCALATE.""" +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 shape in wrong_shapes else TRUTH[f]["expected"] + 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")]) @@ -97,7 +97,7 @@ def test_floor_reports_every_shape_when_false_confidence_is_all_zero(): def test_floor_reports_both_shapes_when_two_tie_on_the_task_floor(): - fl = rel.floor(ag.aggregate(_rows_by_shape(("judge", "classify")), TRUTH))["m"] + 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"]