Leave recorded ceiling-exhaustion rows alone when applying an adjudication

policybench/adjudications.py

@@ -37,6 +37,12 @@ class and the reasoning, so the override stays auditable, and this module
     "reasoning",
 )
 FINAL_SOURCES = frozenset({"llm_error", "parse_contract_failure"})
+# Row classes an adjudication leaves alone: a parse failure stays a parse
+# failure, and a ceiling exhaustion is the run data's own record (collect_audit
+# and the dashboard export force it onto its row), not a judge's verdict.
+NON_SUBSTANTIVE_SOURCES = frozenset(
+    {"parse_contract_failure", "budget_exhausted_at_ceiling"}
+)
 # A developer may affirm a judge's (or override a judge's) verdict as prompt
 # ambiguity, or record a reference engine defect, only when the output is also
 # removed from scoring for every model (see policybench.reference_exclusions);
@@ -207,8 +213,8 @@ def apply_adjudications(
     """Return adjudicated copies of the row annotations and case notes.
 
     An adjudication is a case-level decision: every substantive row of the
-    case (anything but a ``parse_contract_failure`` row, which stays a parse
-    failure) takes the adjudicated source and subtype, so re-applying a
+    case (anything but a :data:`NON_SUBSTANTIVE_SOURCES` row, which keeps its
+    class) takes the adjudicated source and subtype, so re-applying a
     revised record updates rows an earlier application already rewrote. The
     case note's ``case_failure_sources`` and
     ``case_failure_subtypes`` take the adjudicated class, and the case
@@ -219,8 +225,8 @@ def apply_adjudications(
     cases = cases.copy()
     report: list[dict] = []
     for entry in adjudications:
-        row_mask = _case_mask(rows, entry) & (
-            rows["failure_source"].astype(str) != "parse_contract_failure"
+        row_mask = _case_mask(rows, entry) & ~rows["failure_source"].astype(str).isin(
+            NON_SUBSTANTIVE_SOURCES
         )
         case_mask = _case_mask(cases, entry)
         if not case_mask.any():
@@ -271,9 +277,10 @@ def verify_adjudications_applied(
 ) -> None:
     """Raise unless the annotations agree with the complete adjudication record.
 
-    For every entry: no row still carries the judge's non-final verdict; every
-    substantive row of the case (anything but a parse failure) carries the
-    adjudicated source and subtype; the case note carries the adjudicated
+    For every entry: no substantive row of the case (anything but a
+    :data:`NON_SUBSTANTIVE_SOURCES` row, which apply_adjudications leaves
+    alone) still carries the judge's non-final verdict, and every one carries
+    the adjudicated source and subtype; the case note carries the adjudicated
     source and subtype and the exact adjudication sentence, reasoning
     included. A revised record therefore fails verification until it is
     re-applied, and a record edited to disagree with the rows fails closed.
@@ -285,8 +292,13 @@ def verify_adjudications_applied(
         if not case_mask.any() or not row_mask.any():
             raise AdjudicationError(f"adjudicated case missing from annotations: {key}")
         case_rows = rows.loc[row_mask]
-        stray = case_rows[
-            case_rows["failure_source"].astype(str) == entry["judge_failure_source"]
+        substantive = case_rows[
+            ~case_rows["failure_source"].astype(str).isin(NON_SUBSTANTIVE_SOURCES)
+        ]
+        # A recorded ceiling exhaustion is run data even when the judge also
+        # chose that class for the case.
+        stray = substantive[
+            substantive["failure_source"].astype(str) == entry["judge_failure_source"]
         ]
         affirmed = entry["judge_failure_source"] == entry["adjudicated_failure_source"]
         if (
@@ -298,9 +310,6 @@ def verify_adjudications_applied(
                 f"{len(stray)} rows of {key} still carry the judge's "
                 f"{entry['judge_failure_source']} verdict"
             )
-        substantive = case_rows[
-            case_rows["failure_source"].astype(str) != "parse_contract_failure"
-        ]
         off_source = substantive[
             substantive["failure_source"].astype(str)
             != entry["adjudicated_failure_source"]

tests/test_adjudications.py

@@ -5,17 +5,24 @@
 from collections import Counter
 from functools import cache
 from pathlib import Path
+from unittest import mock
 
 import pandas as pd
 import pytest
 
+from policybench import adjudications
 from policybench.adjudications import (
     AdjudicationError,
     apply_adjudications,
     load_adjudications,
+    parse_adjudications,
     unresolved_suspect_cases,
     verify_adjudications_applied,
 )
+from policybench.annotation_taxonomy import (
+    FAILURE_SOURCE_VALUES,
+    FAILURE_SUBTYPE_VALUES,
+)
 
 ROOT = Path(__file__).resolve().parents[1]
 RUN_LABEL = "us_full_run_20260612_policyengine_4_16_1_populace"
@@ -430,6 +437,72 @@ def test_verify_requires_agreement_with_the_complete_record():
         verify_adjudications_applied(bad_rows, out_cases, [_entry()])
 
 
+def _ceiling_frames() -> tuple[pd.DataFrame, pd.DataFrame]:
+    """_frames, with m3's missing answer on scenario_001 recorded by the run as
+    ceiling exhaustion rather than a parse failure."""
+    rows, cases = _frames()
+    rows.loc[2, ["failure_source", "failure_subtype"]] = [
+        "budget_exhausted_at_ceiling",
+        "missing_output",
+    ]
+    return rows, cases
+
+
+def test_apply_leaves_a_recorded_ceiling_exhaustion_alone():
+    """Like a parse failure, a recorded ceiling exhaustion keeps its class and
+    subtype through an adjudication of its case; collect_audit and the
+    dashboard export force it onto the row, so the CSVs must keep it too."""
+    rows, cases = _ceiling_frames()
+    out_rows, out_cases, report = apply_adjudications(rows, cases, [_entry()])
+
+    case_rows = out_rows[out_rows["scenario_id"] == "scenario_001"]
+    assert case_rows[["failure_source", "failure_subtype"]].values.tolist() == [
+        ["llm_error", "age_disability"],
+        ["llm_error", "age_disability"],
+        ["budget_exhausted_at_ceiling", "missing_output"],
+    ]
+    assert report[0]["rows_rewritten"] == 2
+    assert out_cases.loc[0, "case_failure_sources"] == "llm_error"
+    verify_adjudications_applied(out_rows, out_cases, [_entry()])
+
+
+def test_verify_does_not_hold_a_ceiling_row_to_the_adjudicated_class():
+    """The verifier's substantive rows are apply's: a recorded ceiling row is
+    neither off the adjudicated class nor a stray judge verdict, even when the
+    judge chose ceiling exhaustion for the case. The rows apply does rewrite
+    are still held to the record."""
+    rows, cases = _ceiling_frames()
+    entry = _entry(judge_failure_source="budget_exhausted_at_ceiling")
+    out_rows, out_cases, _ = apply_adjudications(rows, cases, [entry])
+    verify_adjudications_applied(out_rows, out_cases, [entry])
+
+    with pytest.raises(AdjudicationError, match="still carry"):
+        verify_adjudications_applied(rows, cases, [_entry()])
+    bad_rows = out_rows.copy()
+    bad_rows.loc[0, "failure_source"] = "reference_model_issue_fixed"
+    with pytest.raises(AdjudicationError, match="source other than"):
+        verify_adjudications_applied(bad_rows, out_cases, [entry])
+
+
+def test_skipping_ceiling_rows_moves_no_committed_annotation():
+    """No committed row records ceiling exhaustion, so the committed record
+    gives the committed annotations whether or not apply skips those rows."""
+    entries = load_adjudications(ANNOTATIONS / "us_adjudications.json")
+    rows = pd.read_csv(ANNOTATIONS / "us_audit_row_annotations.csv")
+    cases = pd.read_csv(ANNOTATIONS / "us_case_notes.csv")
+    with mock.patch.object(
+        adjudications, "NON_SUBSTANTIVE_SOURCES", frozenset({"parse_contract_failure"})
+    ):
+        before = apply_adjudications(rows, cases, entries)
+    after = apply_adjudications(rows, cases, entries)
+    for old, new in zip(before, after):
+        if isinstance(old, pd.DataFrame):
+            pd.testing.assert_frame_equal(new, old)
+        else:
+            assert new == old
+    verify_adjudications_applied(*after[:2], entries)
+
+
 def _suspect_frames() -> tuple[pd.DataFrame, pd.DataFrame]:
     rows, cases = _frames()
     rows.loc[rows["scenario_id"] == "scenario_001", "reference_suspect"] = True
@@ -1545,3 +1618,118 @@ def test_release_20261013_restates_only_the_cases_muse_wonder_reopened(tmp_path)
         if history[:-1] == before.get("judge_previous", []) and history:
             replaced += 1
     assert replaced == len(verdicts) == 50
+
+
+# The rows an adjudication skips, spelled out so the oracle below does not
+# share the module's constant. They are drawn more often than their share of
+# the taxonomy, so most examples hold a row an adjudication must skip.
+_SKIPPED = ("budget_exhausted_at_ceiling", "parse_contract_failure")
+_ROW_SOURCES = st.one_of(
+    st.sampled_from(_SKIPPED),
+    st.sampled_from(FAILURE_SOURCE_VALUES),
+)
+_SUBTYPES = st.sampled_from(FAILURE_SUBTYPE_VALUES)
+
+
+@st.composite
+def _adjudicated_annotations(draw):
+    """Row annotations and case notes for one to three cases of one to four
+    rows, each row of any class and subtype, and a valid adjudication of any
+    judge verdict for some of the cases."""
+    rows, cases, entries = [], [], []
+    for n in range(draw(st.integers(1, 3))):
+        scenario = f"scenario_{n:03d}"
+        count = draw(st.integers(1, 4))
+        for m in range(count):
+            rows.append(
+                ("us", scenario, "ssi", f"m{m}", draw(_ROW_SOURCES), draw(_SUBTYPES))
+            )
+        cases.append(
+            (
+                "us",
+                scenario,
+                "ssi",
+                count,
+                draw(_ROW_SOURCES),
+                draw(_SUBTYPES),
+                f"note {n}.",
+            )
+        )
+        if draw(st.booleans()):
+            adjudicated = draw(
+                st.sampled_from(
+                    ["llm_error", "parse_contract_failure", "prompt_ambiguity"]
+                )
+            )
+            entry = _entry(
+                scenario_id=scenario,
+                judge_failure_source=draw(st.sampled_from(FAILURE_SOURCE_VALUES)),
+                judge_failure_subtype=draw(_SUBTYPES),
+                adjudicated_failure_source=adjudicated,
+                adjudicated_failure_subtype=draw(_SUBTYPES),
+            )
+            if adjudicated == "prompt_ambiguity":
+                entry["excluded_from_scoring"] = True
+            entries.append(entry)
+    row_frame = pd.DataFrame(
+        rows,
+        columns=[
+            "country",
+            "scenario_id",
+            "variable",
+            "model",
+            "failure_source",
+            "failure_subtype",
+        ],
+    )
+    row_frame["reference_suspect"] = False
+    row_frame["annotation"] = "diagnosis"
+    case_frame = pd.DataFrame(
+        cases,
+        columns=[
+            "country",
+            "scenario_id",
+            "variable",
+            "wrong_model_count",
+            "case_failure_sources",
+            "case_failure_subtypes",
+            "case_annotation",
+        ],
+    )
+    return row_frame, case_frame, parse_adjudications({"adjudications": entries}, "")
+
+
+@settings(max_examples=300, deadline=None)
+@given(data=_adjudicated_annotations())
+def test_adjudication_skips_non_substantive_rows_properties(data):
+    """For any annotations and valid record:
+    - a parse-failure or ceiling-exhaustion row keeps its class and subtype,
+      and a row of a case no entry decides is untouched;
+    - every other row of a decided case takes the adjudicated class;
+    - the report counts exactly the rows whose class or subtype changed;
+    - the verifier accepts what apply builds, and a second apply changes
+      nothing.
+    """
+    rows, cases, entries = data
+    out_rows, out_cases, report = apply_adjudications(rows, cases, entries)
+
+    decided = {e["scenario_id"]: e for e in entries}
+    classes = ["failure_source", "failure_subtype"]
+    expected = rows.copy()
+    for index, row in rows.iterrows():
+        entry = decided.get(row["scenario_id"])
+        if entry and row["failure_source"] not in _SKIPPED:
+            expected.loc[index, classes] = [
+                entry["adjudicated_failure_source"],
+                entry["adjudicated_failure_subtype"],
+            ]
+    pd.testing.assert_frame_equal(out_rows, expected)
+    changed = (rows[classes] != out_rows[classes]).any(axis=1)
+    assert [item["rows_rewritten"] for item in report] == [
+        int(changed[rows["scenario_id"] == e["scenario_id"]].sum()) for e in entries
+    ]
+
+    verify_adjudications_applied(out_rows, out_cases, entries)
+    again_rows, again_cases, _ = apply_adjudications(out_rows, out_cases, entries)
+    pd.testing.assert_frame_equal(again_rows, out_rows)
+    pd.testing.assert_frame_equal(again_cases, out_cases)