test(scanner): harden scoped ack evidence proof (#7602)

Require Scanner/Heal G03 scoped ACK evidence fields to carry their own measured provenance and concrete ACK, capability, and mixed-peer observations before release-bundle gate verification.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

scripts/check_test_wiring.py

@@ -440,6 +440,31 @@ SCANNER_HEAL_RELEASE_SCOPED_ACK_CASES = {
         "restarted-peer-reject",
     ),
 }
+SCANNER_HEAL_RELEASE_G03_REQUIRED_TRUE_FIELDS = {
+    "durable_root_publication_proof": (
+        "root_cas_observed",
+        "root_readback_observed",
+        "dirty_fallback_on_missing_proof_observed",
+    ),
+    "scoped_ack_request_identity": (
+        "bucket_incarnation_observed",
+        "exact_generation_observed",
+        "scanner_instance_observed",
+        "participating_peer_set_observed",
+        "whole_cycle_fallback_observed",
+    ),
+    "participating_peer_capability_snapshot": (
+        "capability_probe_observed",
+        "scoped_ack_capability_observed",
+        "probe_only_fallback_observed",
+        "missing_capability_kept_dirty",
+    ),
+    "mixed_peer_ack_fallback_oracle": (
+        "legacy_peer_fallback_observed",
+        "truncated_token_rejected",
+        "restarted_peer_rejected",
+    ),
+}
 SCANNER_HEAL_RELEASE_MIXED_VERSION_CASES = {
     "mixed_version_reader_evidence": (
         "old-writer-new-reader",
@@ -1840,10 +1865,7 @@ def release_bundle_json_artifact_mirrored_fields(gate: str, field: str) -> tuple
             ))
     if gate == "G03":
         fields.append("scoped_ack_cases")
-        if field == "durable_root_publication_proof":
-            fields.extend(("root_cas_observed", "root_readback_observed"))
-        if field == "scoped_ack_request_identity":
-            fields.append("whole_cycle_fallback_observed")
+        fields.extend(SCANNER_HEAL_RELEASE_G03_REQUIRED_TRUE_FIELDS[field])
     if gate == "G04" and field == "root_floor_intent_crash_evidence":
         fields.extend(("durable_intent_cases", "persist_failure_blocks_acceptance"))
     if gate == "G12":
@@ -2192,12 +2214,8 @@ def validate_release_bundle_domain_evidence(gate: str, field: str, evidence: dic
             SCANNER_HEAL_RELEASE_SCOPED_ACK_CASES[field],
             f"{gate}.{field}.scoped_ack_cases",
         )
-        if field == "durable_root_publication_proof":
-            release_bundle_bool_true(evidence.get("root_cas_observed"), f"{gate}.{field}.root_cas_observed")
-            release_bundle_bool_true(evidence.get("root_readback_observed"), f"{gate}.{field}.root_readback_observed")
-        if field == "scoped_ack_request_identity":
-            release_bundle_bool_true(evidence.get("whole_cycle_fallback_observed"),
-                                     f"{gate}.{field}.whole_cycle_fallback_observed")
+        for required in SCANNER_HEAL_RELEASE_G03_REQUIRED_TRUE_FIELDS[field]:
+            release_bundle_bool_true(evidence.get(required), f"{gate}.{field}.{required}")
 
     if gate == "G04" and field == "root_floor_intent_crash_evidence":
         release_bundle_exact_strings(
@@ -3223,6 +3241,10 @@ def write_scanner_heal_release_bundle_fixture(root: Path, directory: Path) -> Pa
             if gate in ("G03", "G09", "R-L"):
                 evidence["versions"] = [baseline_revision, source_revision]
                 evidence["mixed_version_role"] = SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES[(gate, field)]
+            if gate == "G03":
+                evidence["scoped_ack_cases"] = list(SCANNER_HEAL_RELEASE_SCOPED_ACK_CASES[field])
+                for required in SCANNER_HEAL_RELEASE_G03_REQUIRED_TRUE_FIELDS[field]:
+                    evidence[required] = True
             if gate in ("G04", "G07", "R-E", "R-L"):
                 evidence["crash_points"] = ["fixture-before-commit"]
             if gate == "G12":
@@ -3637,11 +3659,8 @@ class SelfTests(unittest.TestCase):
                     evidence["crash_points"] = ["before-commit"]
                 if gate == "G03":
                     evidence["scoped_ack_cases"] = list(SCANNER_HEAL_RELEASE_SCOPED_ACK_CASES[field])
-                    if field == "durable_root_publication_proof":
-                        evidence["root_cas_observed"] = True
-                        evidence["root_readback_observed"] = True
-                    if field == "scoped_ack_request_identity":
-                        evidence["whole_cycle_fallback_observed"] = True
+                    for required in SCANNER_HEAL_RELEASE_G03_REQUIRED_TRUE_FIELDS[field]:
+                        evidence[required] = True
                 if gate == "G04" and field == "root_floor_intent_crash_evidence":
                     evidence["durable_intent_cases"] = list(SCANNER_HEAL_RELEASE_CRASH_BOUNDARY_FIELDS[(gate, field)])
                     evidence["persist_failure_blocks_acceptance"] = True
@@ -4713,6 +4732,20 @@ class SelfTests(unittest.TestCase):
                 lambda item: item.update({"whole_cycle_fallback_observed": False}),
                 "whole_cycle_fallback_observed",
             ),
+            (
+                "scoped-ack-capability-snapshot",
+                "G03",
+                "participating_peer_capability_snapshot",
+                lambda item: item.pop("capability_probe_observed"),
+                "capability_probe_observed",
+            ),
+            (
+                "scoped-ack-mixed-peer-restart",
+                "G03",
+                "mixed_peer_ack_fallback_oracle",
+                lambda item: item.update({"restarted_peer_rejected": False}),
+                "restarted_peer_rejected",
+            ),
             (
                 "durable-intent-cases",
                 "G04",

scripts/run_scanner_heal_scoped_ack_evidence.py

@@ -76,17 +76,13 @@ def field_from_proof(proof: dict[str, Any], field: str) -> dict[str, Any]:
     wiring.require(isinstance(value, dict), f"proof missing {field}")
     for marker in ("fixture", "fixture_only", "dry_run", "synthetic"):
         wiring.require(value.get(marker) is not True, f"{field} is {marker}")
-    if "evidence_type" in value:
-        wiring.require(value["evidence_type"] == "measured", f"{field} must be measured")
-    if "source_revision" in value:
-        wiring.require(value["source_revision"] == proof["source_revision"], f"{field} source revision mismatch")
-    if "run_id" in value:
-        wiring.require(value["run_id"] == proof["run_id"], f"{field} run_id mismatch")
-    if "measurement_window_id" in value:
-        wiring.require(
-            value["measurement_window_id"] == proof["measurement_window_id"],
-            f"{field} measurement window mismatch",
-        )
+    wiring.require(value.get("evidence_type") == "measured", f"{field} must be measured")
+    wiring.require(value.get("source_revision") == proof["source_revision"], f"{field} source revision mismatch")
+    wiring.require(value.get("run_id") == proof["run_id"], f"{field} run_id mismatch")
+    wiring.require(
+        value.get("measurement_window_id") == proof["measurement_window_id"],
+        f"{field} measurement window mismatch",
+    )
     evidence = {
         **value,
         "versions": proof["versions"],
@@ -173,32 +169,33 @@ def build_descriptor(args: argparse.Namespace) -> Path:
 
 def write_self_test_proof(path: Path, source_revision: str) -> None:
     now = datetime.now(timezone.utc).replace(microsecond=0)
+    run_id = f"scoped-ack-{source_revision[:12]}"
+    measurement_window_id = f"scoped-ack-window-{source_revision[:12]}"
+
+    def field_payload(field: str, values: dict[str, Any]) -> dict[str, Any]:
+        return {
+            "evidence_type": "measured",
+            "source_revision": source_revision,
+            "run_id": run_id,
+            "measurement_window_id": measurement_window_id,
+            "scoped_ack_cases": list(wiring.SCANNER_HEAL_RELEASE_SCOPED_ACK_CASES[field]),
+            **{required: True for required in wiring.SCANNER_HEAL_RELEASE_G03_REQUIRED_TRUE_FIELDS[field]},
+            **values,
+        }
+
     proof = {
         "schema": 1,
         "evidence_type": "measured",
         "source_revision": source_revision,
-        "run_id": f"scoped-ack-{source_revision[:12]}",
-        "measurement_window_id": f"scoped-ack-window-{source_revision[:12]}",
+        "run_id": run_id,
+        "measurement_window_id": measurement_window_id,
         "started_at": now.isoformat().replace("+00:00", "Z"),
         "finished_at": now.isoformat().replace("+00:00", "Z"),
         "versions": ["a" * 40, source_revision],
-        "durable_root_publication_proof": {
-            "scoped_ack_cases": list(wiring.SCANNER_HEAL_RELEASE_SCOPED_ACK_CASES["durable_root_publication_proof"]),
-            "root_cas_observed": True,
-            "root_readback_observed": True,
-        },
-        "scoped_ack_request_identity": {
-            "scoped_ack_cases": list(wiring.SCANNER_HEAL_RELEASE_SCOPED_ACK_CASES["scoped_ack_request_identity"]),
-            "whole_cycle_fallback_observed": True,
-        },
-        "participating_peer_capability_snapshot": {
-            "scoped_ack_cases": list(
-                wiring.SCANNER_HEAL_RELEASE_SCOPED_ACK_CASES["participating_peer_capability_snapshot"]
-            ),
-        },
-        "mixed_peer_ack_fallback_oracle": {
-            "scoped_ack_cases": list(wiring.SCANNER_HEAL_RELEASE_SCOPED_ACK_CASES["mixed_peer_ack_fallback_oracle"]),
-        },
+        "durable_root_publication_proof": field_payload("durable_root_publication_proof", {}),
+        "scoped_ack_request_identity": field_payload("scoped_ack_request_identity", {}),
+        "participating_peer_capability_snapshot": field_payload("participating_peer_capability_snapshot", {}),
+        "mixed_peer_ack_fallback_oracle": field_payload("mixed_peer_ack_fallback_oracle", {}),
     }
     wiring.write_json(path, proof)
 
@@ -247,6 +244,36 @@ def run_self_test() -> None:
         else:
             raise ValueError("self-test accepted incomplete root publication proof")
 
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        source_revision = git_head()
+        proof = root / "scoped-ack-proof.json"
+        write_self_test_proof(proof, source_revision)
+        payload = wiring.read_json(proof)
+        payload["participating_peer_capability_snapshot"]["source_revision"] = "b" * 40
+        wiring.write_json(proof, payload)
+        try:
+            build_descriptor(parse_args(["--proof-json", str(proof), "--out-dir", str(root / "out")]))
+        except ValueError as err:
+            wiring.require("source revision mismatch" in str(err), "wrong self-test failure for field source")
+        else:
+            raise ValueError("self-test accepted stale scoped ACK field source")
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        source_revision = git_head()
+        proof = root / "scoped-ack-proof.json"
+        write_self_test_proof(proof, source_revision)
+        payload = wiring.read_json(proof)
+        payload["participating_peer_capability_snapshot"].pop("capability_probe_observed")
+        wiring.write_json(proof, payload)
+        try:
+            build_descriptor(parse_args(["--proof-json", str(proof), "--out-dir", str(root / "out")]))
+        except ValueError as err:
+            wiring.require("capability_probe_observed" in str(err), "wrong self-test failure for capability proof")
+        else:
+            raise ValueError("self-test accepted incomplete capability snapshot proof")
+
     with tempfile.TemporaryDirectory() as tmp:
         root = Path(tmp)
         source_revision = git_head()