From 3afb3892471dc61df24a93daced267c0b6d725f8 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 10 Sep 2026 01:05:00 +0800 Subject: [PATCH] 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 --- scripts/check_test_wiring.py | 63 ++++++++++---- .../run_scanner_heal_scoped_ack_evidence.py | 87 ++++++++++++------- 2 files changed, 105 insertions(+), 45 deletions(-) diff --git a/scripts/check_test_wiring.py b/scripts/check_test_wiring.py index 853cf9a98..62e4a1334 100755 --- a/scripts/check_test_wiring.py +++ b/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", diff --git a/scripts/run_scanner_heal_scoped_ack_evidence.py b/scripts/run_scanner_heal_scoped_ack_evidence.py index b2b6ec899..e0afe9d8d 100755 --- a/scripts/run_scanner_heal_scoped_ack_evidence.py +++ b/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()