Reject aborted runs and incomplete classification evidence

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e76f-be98-7ae3-965d-e0b31290a4c4

tests/test_evidence_completeness.py

@@ -0,0 +1,157 @@
+"""A passing prefix or a truncated pack must never become an accepted run."""
+import json
+from copy import deepcopy
+from dataclasses import replace
+
+import pytest
+
+from scenarios.alice_bob_carol import build
+from testdriver import Runner, Verdict
+from testdriver.classification import Classification, classify
+
+
+def run_reference(*mutations, abort_at=None, claims_only=False):
+    world, driver, observer, asset, oracle = build(*mutations)
+    steps = list(asset.scenario.steps)
+    if abort_at is not None:
+        steps[abort_at] = replace(
+            steps[abort_at],
+            action=replace(steps[abort_at].action, permitted_surfaces=frozenset({"browser"})),
+        )
+    case = asset.scenario.use_case
+    if claims_only:
+        # Every claim is judged after grant; the final step has no assertions.
+        case = replace(case, invariants=(), claims=tuple(
+            c for c in case.claims if c.after_step == "s2-grant"
+        ))
+    asset.scenario = replace(asset.scenario, steps=tuple(steps), use_case=case)
+    return Runner(world, driver, observer, oracle).run(asset), asset.scenario
+
+
+@pytest.mark.parametrize("abort_at", [0, 1, 2])
+def test_aborted_run_retains_unreached_claims_and_invariants(abort_at):
+    result, scenario = run_reference(abort_at=abort_at)
+    assert result.verdict is Verdict.INCONCLUSIVE
+    judgments = {(j.assertion_id, j.step_id): j for j in result.judgments}
+    expected = {
+        (a.id, step.id) for step in scenario.steps
+        for a in (*scenario.use_case.invariants, *(
+            c for c in scenario.use_case.claims if c.after_step == step.id
+        ))
+    }
+    assert judgments.keys() == expected
+    for step in scenario.steps[abort_at:]:
+        assert all(j.verdict is Verdict.INCONCLUSIVE
+                   for j in result.judgments if j.step_id == step.id)
+    assert any(o.kind == "surface_violation" for o in result.evidence.observations)
+    # Do not fabricate successful steps or observations to fill the evidence gap.
+    assert len([o for o in result.evidence.observations
+                if o.kind == "state_snapshot"]) == abort_at
+
+
+def test_abort_without_remaining_assertions_still_cannot_pass():
+    result, _ = run_reference(abort_at=2, claims_only=True)
+    assert all(j.verdict is Verdict.PASS for j in result.judgments)
+    assert result.verdict is Verdict.INCONCLUSIVE
+
+
+def test_abort_preserves_a_previously_observed_failure():
+    result, _ = run_reference("M16", abort_at=2)
+    assert result.verdict is Verdict.FAIL
+    assert result.judgment("c-bob-cannot-write").verdict is Verdict.FAIL
+    assert result.judgment("c-bob-revoked").verdict is Verdict.INCONCLUSIVE
+
+
+@pytest.fixture
+def baseline():
+    result, _ = run_reference()
+    return json.loads(result.evidence.to_json())
+
+
+def test_complete_pack_is_still_accepted(baseline):
+    assert classify(baseline, deepcopy(baseline)).classification is Classification.UNCHANGED
+
+
+def test_manifest_comes_from_schedule_even_if_execution_aborts():
+    baseline, _ = run_reference()
+    aborted, _ = run_reference(abort_at=1)
+    assert aborted.evidence.scheduled_steps == baseline.evidence.scheduled_steps
+    assert aborted.evidence.expected_judgments == baseline.evidence.expected_judgments
+
+
+def test_missing_verdict_and_all_snapshots_reproduces_reported_hole(baseline):
+    candidate = deepcopy(baseline)
+    candidate["verdicts"] = candidate["verdicts"][:1]
+    candidate["observations"] = [o for o in candidate["observations"]
+                                 if o["kind"] != "state_snapshot"]
+    outcome = classify(baseline, candidate)
+    assert outcome.classification is Classification.AMBIGUOUS
+    assert not outcome.safe_to_accept
+
+
+@pytest.mark.parametrize("side", ["baseline", "candidate", "both"])
+@pytest.mark.parametrize("damage", [
+    "one-verdict", "all-verdicts", "one-snapshot", "all-snapshots", "empty-snapshot",
+    "s1-snapshot", "missing-realization", "missing-check", "wrong-step",
+    "duplicate-verdict", "duplicate-snapshot", "invalid-verdict", "missing-version",
+    "unfinished", "missing-manifest", "missing-observations", "missing-verdict-field",
+])
+def test_incomplete_evidence_cannot_be_accepted(baseline, side, damage):
+    candidate = deepcopy(baseline)
+
+    def corrupt(pack):
+        if damage == "one-verdict":
+            pack["verdicts"].pop()
+        elif damage == "all-verdicts":
+            pack["verdicts"] = []
+        elif damage in ("one-snapshot", "missing-realization", "missing-check"):
+            kind = {"one-snapshot": "state_snapshot", "missing-realization": "realization",
+                    "missing-check": "realization_check"}[damage]
+            pack["observations"].remove(next(o for o in pack["observations"] if o["kind"] == kind))
+        elif damage == "all-snapshots":
+            pack["observations"] = [o for o in pack["observations"] if o["kind"] != "state_snapshot"]
+        elif damage in ("empty-snapshot", "s1-snapshot", "wrong-step", "duplicate-snapshot"):
+            snapshot = next(o for o in pack["observations"] if o["kind"] == "state_snapshot")
+            if damage == "empty-snapshot":
+                snapshot["data"] = {}
+            elif damage == "s1-snapshot":
+                snapshot["stratum"] = "S1"
+            elif damage == "wrong-step":
+                snapshot["step_id"] = "not-a-scheduled-step"
+            else:
+                pack["observations"].append(deepcopy(snapshot))
+        elif damage == "duplicate-verdict":
+            pack["verdicts"].append(deepcopy(pack["verdicts"][0]))
+        elif damage == "invalid-verdict":
+            pack["verdicts"][0]["verdict"] = "UNKNOWN"
+        elif damage == "missing-version":
+            del pack["sut_version"]
+        elif damage == "unfinished":
+            pack["finished_at"] = None
+        elif damage == "missing-manifest":
+            pack.pop("scheduled_steps", None)
+            pack.pop("expected_judgments", None)
+        elif damage == "missing-observations":
+            del pack["observations"]
+        elif damage == "missing-verdict-field":
+            del pack["verdicts"]
+
+    if side in ("baseline", "both"):
+        corrupt(baseline)
+    if side in ("candidate", "both"):
+        corrupt(candidate)
+    outcome = classify(baseline, candidate)
+    assert outcome.classification is Classification.AMBIGUOUS
+    assert not outcome.safe_to_accept
+    assert outcome.signals.evidence_incomplete
+
+
+def test_truncating_the_manifest_too_cannot_hide_a_missing_baseline_step(baseline):
+    candidate = deepcopy(baseline)
+    missing = candidate.get("scheduled_steps", ["s3-revoke"])[-1]
+    candidate["scheduled_steps"] = ["s1-create", "s2-grant"]
+    candidate["expected_judgments"] = [v for v in candidate.get("expected_judgments", [])
+                                        if v["step_id"] != missing]
+    candidate["verdicts"] = [v for v in candidate["verdicts"] if v["step_id"] != missing]
+    candidate["observations"] = [o for o in candidate["observations"] if o["step_id"] != missing]
+    assert not classify(baseline, candidate).safe_to_accept