test(scanner): add release bundle evidence descriptors (#7595)

* test(scanner): merge release evidence descriptors Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(scanner): add authority and legacy release descriptors Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(scanner): add scoped ack release descriptor Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(scanner): require all release evidence fields Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(scanner): add mrf release descriptor producer Add a standalone W13 MRF descriptor producer for measured G07/G08/P4 raw artifacts, with fail-closed self-tests for missing responsibility cases, disk-full ENOSPC observations, synthetic artifacts, and short P4 cleanup soaks. Also bind G12 quota path evidence to explicit case lists so W16 descriptors cannot pass without reset and settlement coverage. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(scanner): reuse mrf release descriptor producer Route the W13 runner through the shared MRF descriptor producer so measured runs and offline re-packaging use the same fail-closed validation. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-09-09 21:56:03 +00:00 · 2026-09-09 22:55:43 +08:00
parent 9b4b366209
commit 8be1e9b2c1
17 changed files with 1989 additions and 207 deletions
@@ -33,6 +33,9 @@ script-tests: ## Run shell script tests
 	./scripts/test_hotpath_warp_abba.sh
 	./scripts/test_scanner_validation_harness.sh
 	./scripts/test_scanner_heal_checkpoint_crash_evidence.sh
+	./scripts/test_scanner_heal_authority_evidence.sh
+	./scripts/test_scanner_heal_scoped_ack_evidence.sh
+	./scripts/test_scanner_heal_legacy_rollback_evidence.sh
 	./scripts/test_scanner_heal_g14_multiset_evidence.sh
 	./scripts/test_scanner_heal_scheduler_pressure_evidence.sh
 	./scripts/test_scanner_heal_status_outcome_evidence.sh
@@ -171,7 +171,11 @@
      "lane": "authority-coverage",
      "status": "pending",
      "description": "Complete root and quota authority coverage",
-      "requires": ["root authority evidence", "quota authority evidence"]
+      "requires": ["root authority evidence", "quota authority evidence"],
+      "evidence_fields": [
+        "root_authority_evidence",
+        "quota_authority_evidence"
+      ]
    },
    {
      "gate": "G02",
@@ -179,7 +183,11 @@
      "lane": "checkpoint-and-crash",
      "status": "pending",
      "description": "Bounded checkpoint progress and independent version inventory",
-      "requires": ["bounded checkpoint oracle", "independent version inventory"]
+      "requires": ["bounded checkpoint oracle", "independent version inventory"],
+      "evidence_fields": [
+        "bounded_checkpoint_oracle",
+        "independent_version_inventory"
+      ]
    },
    {
      "gate": "G03",
@@ -201,7 +209,11 @@
      "lane": "checkpoint-and-crash",
      "status": "pending",
      "description": "Crash at every cache, root, floor and intent boundary",
-      "requires": ["cache boundary crash evidence", "root/floor/intent crash evidence"]
+      "requires": ["cache boundary crash evidence", "root/floor/intent crash evidence"],
+      "evidence_fields": [
+        "cache_boundary_crash_evidence",
+        "root_floor_intent_crash_evidence"
+      ]
    },
    {
      "gate": "G05",
@@ -272,7 +284,11 @@
      "lane": "scheduler-pressure",
      "status": "pending",
      "description": "Bounded scheduling and pressure recovery",
-      "requires": ["scheduler bound evidence", "pressure recovery evidence"]
+      "requires": ["scheduler bound evidence", "pressure recovery evidence"],
+      "evidence_fields": [
+        "scheduler_bound_evidence",
+        "pressure_recovery_evidence"
+      ]
    },
    {
      "gate": "G11",
@@ -297,7 +313,11 @@
      "lane": "authority-coverage",
      "status": "pending",
      "description": "Both quota paths during reset and settlement",
-      "requires": ["reset quota-path evidence", "settlement quota-path evidence"]
+      "requires": ["reset quota-path evidence", "settlement quota-path evidence"],
+      "evidence_fields": [
+        "reset_quota_path_evidence",
+        "settlement_quota_path_evidence"
+      ]
    },
    {
      "gate": "G13",
@@ -305,7 +325,12 @@
      "lane": "maintenance-producers",
      "status": "pending",
      "description": "Quorum-minus-one, unknown disks, remount, Object Lock, dry-run, grace and commit tail",
-      "requires": ["quorum-minus-one matrix", "unknown-disk/remount matrix", "Object Lock dry-run grace evidence"]
+      "requires": ["quorum-minus-one matrix", "unknown-disk/remount matrix", "Object Lock dry-run grace evidence"],
+      "evidence_fields": [
+        "quorum_minus_one_matrix",
+        "unknown_disk_remount_matrix",
+        "object_lock_dry_run_grace_evidence"
+      ]
    },
    {
      "gate": "G14",
@@ -334,7 +359,12 @@
      "lane": "scheduler-pressure",
      "status": "pending",
      "description": "Measured cold-walk share and foreground latency/throughput",
-      "requires": ["cold-walk share measurement", "foreground latency/throughput measurement"]
+      "requires": ["cold-walk share measurement", "foreground latency/throughput measurement"],
+      "evidence_fields": [
+        "cold_walk_share_measurement",
+        "foreground_latency_throughput_measurement",
+        "profile_evidence"
+      ]
    },
    {
      "gate": "P2",
@@ -354,7 +384,12 @@
      "lane": "scheduler-pressure",
      "status": "pending",
      "description": "Measured two-hour pressure/heal capacity and recovery window",
-      "requires": ["two-hour pressure measurement", "heal capacity measurement", "recovery-window measurement"]
+      "requires": ["two-hour pressure measurement", "heal capacity measurement", "recovery-window measurement"],
+      "evidence_fields": [
+        "two_hour_pressure_measurement",
+        "heal_capacity_measurement",
+        "recovery_window_measurement"
+      ]
    },
    {
      "gate": "P4",
@@ -376,7 +411,12 @@
      "lane": "checkpoint-and-crash",
      "status": "pending",
      "description": "Fixed-budget real process restart through enumeration and classification",
-      "requires": ["fixed-budget restart evidence", "enumeration evidence", "classification evidence"]
+      "requires": ["fixed-budget restart evidence", "enumeration evidence", "classification evidence"],
+      "evidence_fields": [
+        "fixed_budget_restart_evidence",
+        "enumeration_evidence",
+        "classification_evidence"
+      ]
    },
    {
      "gate": "R-D",
@@ -398,7 +438,12 @@
      "lane": "mixed-version-rollback",
      "status": "pending",
      "description": "Legacy source conflicts, migration gaps and crash-safe source retirement",
-      "requires": ["legacy source-conflict evidence", "migration-gap evidence", "crash-safe source retirement evidence"]
+      "requires": ["legacy source-conflict evidence", "migration-gap evidence", "crash-safe source retirement evidence"],
+      "evidence_fields": [
+        "legacy_source_conflict_evidence",
+        "migration_gap_evidence",
+        "crash_safe_source_retirement_evidence"
+      ]
    }
  ]
 }
@@ -279,6 +279,33 @@ another platform, or `--test mixed-version|rollback` while narrowing a failure.
 It performs a free-space preflight before building so a saturated validation
 host fails before producing partial evidence.

+Operator-collected G01 root/quota authority evidence can be packaged with:
+
+```bash
+scripts/run_scanner_heal_authority_evidence.py \
+  --root-authority-json /path/to/root-authority.json \
+  --quota-authority-json /path/to/quota-authority.json \
+  --out-dir /path/to/authority-descriptor
+```
+
+The producer rejects fixture, dry-run, synthetic, stale-revision, incomplete
+root authority, and incomplete quota authority inputs before writing
+`release-bundle-authority.json`. The descriptor validates only G01; it still
+needs the full bundle assembler and the remaining release lanes before a release
+can be approved.
+
+Operator-collected G03 scoped ACK evidence can be packaged with:
+
+```bash
+scripts/python_bin.sh scripts/run_scanner_heal_scoped_ack_evidence.py \
+  --proof-json /path/to/scoped-ack-proof.json \
+  --out-dir /path/to/scoped-ack-descriptor
+```
+
+The producer requires durable root publication, exact request identity,
+participating peer capability, mixed-peer fallback, and mixed-version
+provenance before writing `release-bundle-scoped-ack.json`.
+
 The W16 recovery-intent and quota-authority lanes can emit raw G04/G12 JSON
 artifacts with:

@@ -309,6 +336,26 @@ pre-mounted small filesystem. P4 is release evidence only when it completes the
 default two-hour soak; `--allow-short-soak` is diagnostic and skips P4 bundle
 gate validation.

+Already collected W13 raw MRF artifacts can be re-packaged and checked without
+rerunning the Rust test with:
+
+```bash
+scripts/python_bin.sh scripts/run_scanner_heal_mrf_evidence.py \
+  --run-dir /path/to/scanner-heal-w13-evidence-run
+```
+
+The producer rejects fixture, dry-run, synthetic, stale-revision, incomplete
+MRF responsibility, missing disk-full ENOSPC observations, and short P4 cleanup
+soaks before writing `release-bundle-w13.json`.
+
+Legacy rollback evidence for R-L is assembled from a measured proof JSON:
+
+```bash
+scripts/python_bin.sh scripts/run_scanner_heal_legacy_rollback_evidence.py \
+  --proof-json /path/to/legacy-rollback-proof.json \
+  --out-dir /path/to/legacy-rollback-descriptor
+```
+
 When the real release lanes have produced their dedicated artifacts, validate
 the complete hard-gate bundle with:

@@ -317,6 +364,23 @@ scripts/python_bin.sh scripts/check_test_wiring.py \
  --check-scanner-heal-release-bundle /path/to/release-evidence.json
 ```

+Lane descriptors can be assembled into that bundle with:
+
+```bash
+scripts/python_bin.sh scripts/check_test_wiring.py \
+  --assemble-scanner-heal-release-bundle \
+  /path/to/release-bundle-g14.json \
+  /path/to/release-bundle-w16.json \
+  /path/to/assembled-release-bundle
+```
+
+The final argument is the new output directory. Every preceding argument is a
+measured descriptor, and each descriptor gate must keep its own passing status,
+lane identity, measured evidence type, complete required fields, relative
+artifact paths, and matching SHA256 hashes. The assembler may verify only the
+lanes already present; a partial assembled bundle remains blocked until all
+release gates are supplied.
+
 The bundle checker is intentionally stricter than the case checker. It requires
 schema 2 registry metadata, `evidence: measured`, the current checkout revision,
 all G01-G14/P1-P4/R-E/R-D/R-L gates, per-gate `status: pass`, lane identity,
@@ -55,10 +55,15 @@ their issue closes.
 | `run.ps1` | dev-tool | Windows counterpart of `run.sh` | — |
 | `probe.sh` | dev-tool | Probe-style e2e run | `make probe-e2e` |
 | `run_scanner_validation_harness.sh` | dev-tool | Scanner validation harness | `docs/operations/scanner-benchmark-runbook.md` |
+| `check_test_wiring.py` | dev-tool | Validates test wiring and assembles one or more measured Scanner/Heal release descriptors into a single release evidence bundle | `.config/scanner-heal-required-tests.json`; `check_test_wiring.py --self-test` |
 | `run_scanner_heal_evidence_case.sh` | dev-tool | Runs one Scanner/Heal release-evidence registry case and checks the produced receipt/oracle | `.config/scanner-heal-required-tests.json`; `check_test_wiring.py --check-scanner-heal` |
+| `run_scanner_heal_authority_evidence.py` | dev-tool | Assembles measured Scanner/Heal G01 root/quota authority release descriptors from operator-collected authority artifacts | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_authority_evidence.sh` |
 | `run_scanner_heal_checkpoint_crash_evidence.py` | dev-tool | Assembles measured Scanner/Heal G02/R-E checkpoint and restart release descriptors from scanner restart diagnostic reports | `diagnose_scanner_enumeration_restart.py`; `test_scanner_heal_checkpoint_crash_evidence.sh` |
 | `run_scanner_heal_g14_multiset_evidence.py` | dev-tool | Assembles measured Scanner/Heal G14 same-window EC8+4 multi-set/multi-pool release descriptors from e2e case directories or an operator-collected proof | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_g14_multiset_evidence.sh` |
 | `run_scanner_heal_g09_upgrade_evidence.sh` | dev-tool | Runs the G09 mixed-version and rollback upgrade E2E lanes against a pinned previous release and verifies the raw evidence artifacts | `docs/testing/ci-gates.md`; `.github/workflows/e2e-upgrade.yml`; `test_scanner_heal_g09_upgrade_evidence.sh` |
+| `run_scanner_heal_scoped_ack_evidence.py` | dev-tool | Assembles measured Scanner/Heal G03 scoped ACK publication and mixed-peer fallback release descriptors | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_scoped_ack_evidence.sh` |
+| `run_scanner_heal_legacy_rollback_evidence.py` | dev-tool | Assembles measured Scanner/Heal R-L legacy source-conflict, migration-gap, and source-retirement release descriptors | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_legacy_rollback_evidence.sh` |
+| `run_scanner_heal_mrf_evidence.py` | dev-tool | Assembles measured Scanner/Heal G07/G08/P4 MRF release descriptors from W13 raw artifacts | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_w13_mrf_evidence.sh` |
 | `run_scanner_heal_scheduler_pressure_evidence.py` | dev-tool | Assembles measured Scanner/Heal G10/P1/P3 scheduler-pressure release descriptors from a completed measured ABBA run, recovery-window proof, and profile artifacts | `docs/operations/scanner-benchmark-runbook.md`; `test_scanner_heal_scheduler_pressure_evidence.sh` |
 | `run_scanner_heal_status_outcome_evidence.py` | dev-tool | Assembles measured Scanner/Heal G05/G06/R-D status-and-outcome release descriptors from status, compatibility, and disposition artifacts | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_status_outcome_evidence.sh` |
 | `run_scanner_heal_maintenance_evidence.py` | dev-tool | Assembles measured Scanner/Heal G11/G13 maintenance-producer release descriptors from operator-collected proof JSON | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_maintenance_evidence.sh` |
@@ -66,6 +71,9 @@ their issue closes.
 | `run_scanner_heal_w16_recovery_evidence.sh` | dev-tool | Runs the W16 recovery-intent and quota authority lanes and writes G04/G12 bundle-ready evidence descriptors | `docs/testing/ci-gates.md`; `test_scanner_heal_w16_recovery_evidence.sh` |
 | `test_scanner_validation_harness.sh` | dev-tool | Self-test for the scanner validation harness | — |
 | `test_scanner_heal_checkpoint_crash_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal checkpoint/crash evidence assembler | — |
+| `test_scanner_heal_authority_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal G01 authority evidence assembler | — |
+| `test_scanner_heal_scoped_ack_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal G03 scoped ACK evidence assembler | — |
+| `test_scanner_heal_legacy_rollback_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal R-L legacy rollback evidence assembler | — |
 | `test_scanner_heal_g14_multiset_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal G14 multi-set/multi-pool evidence assembler | — |
 | `test_scanner_heal_scheduler_pressure_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal scheduler-pressure evidence assembler | — |
 | `test_scanner_heal_status_outcome_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal status-and-outcome evidence assembler | — |
@@ -44,6 +44,24 @@ SCANNER_HEAL_RELEASE_REQUIRED_GATES = (
    "P1", "P2", "P3", "P4", "R-E", "R-D", "R-L",
 )
 SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS = {
+    "G01": (
+        "root_authority_evidence",
+        "quota_authority_evidence",
+    ),
+    "G02": (
+        "bounded_checkpoint_oracle",
+        "independent_version_inventory",
+    ),
+    "G03": (
+        "durable_root_publication_proof",
+        "scoped_ack_request_identity",
+        "participating_peer_capability_snapshot",
+        "mixed_peer_ack_fallback_oracle",
+    ),
+    "G04": (
+        "cache_boundary_crash_evidence",
+        "root_floor_intent_crash_evidence",
+    ),
    "G05": (
        "per_object_outcome_oracle",
        "terminal_retention_bounds",
@@ -62,22 +80,29 @@ SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS = {
        "disk_full_matrix",
        "replica_loss_matrix",
    ),
-    "G03": (
-        "durable_root_publication_proof",
-        "scoped_ack_request_identity",
-        "participating_peer_capability_snapshot",
-        "mixed_peer_ack_fallback_oracle",
-    ),
    "G09": (
        "mixed_version_reader_evidence",
        "mixed_version_writer_evidence",
        "rollback_payload_evidence",
    ),
+    "G10": (
+        "scheduler_bound_evidence",
+        "pressure_recovery_evidence",
+    ),
    "G11": (
        "maintenance_producer_matrix",
        "complete_producer_inventory",
        "segment_activation_preflight",
    ),
+    "G12": (
+        "reset_quota_path_evidence",
+        "settlement_quota_path_evidence",
+    ),
+    "G13": (
+        "quorum_minus_one_matrix",
+        "unknown_disk_remount_matrix",
+        "object_lock_dry_run_grace_evidence",
+    ),
    "G14": (
        "same_window_field_evidence",
        "ec8_4_evidence",
@@ -85,15 +110,30 @@ SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS = {
        "multi_pool_evidence",
        "distributed_segment_invalidation_evidence",
    ),
+    "P1": (
+        "cold_walk_share_measurement",
+        "foreground_latency_throughput_measurement",
+        "profile_evidence",
+    ),
+    "P2": (
+        "post_stop_convergence_measurement",
+        "cold_segment_reuse_measurement",
+    ),
+    "P3": (
+        "two_hour_pressure_measurement",
+        "heal_capacity_measurement",
+        "recovery_window_measurement",
+    ),
    "P4": (
        "mrf_scale_measurement",
        "mrf_replay_cost_measurement",
        "retained_responsibility_evidence",
        "mrf_cleanup_gc_soak_evidence",
    ),
-    "P2": (
-        "post_stop_convergence_measurement",
-        "cold_segment_reuse_measurement",
+    "R-E": (
+        "fixed_budget_restart_evidence",
+        "enumeration_evidence",
+        "classification_evidence",
    ),
    "R-D": (
        "manager_disposition_evidence",
@@ -101,29 +141,34 @@ SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS = {
        "ledger_disposition_evidence",
        "grace_handling",
    ),
+    "R-L": (
+        "legacy_source_conflict_evidence",
+        "migration_gap_evidence",
+        "crash_safe_source_retirement_evidence",
+    ),
 }
 SCANNER_HEAL_RELEASE_BUNDLE_REQUIRED_EVIDENCE_FIELDS = {
-    "G01": ("root_authority_evidence", "quota_authority_evidence"),
-    "G02": ("bounded_checkpoint_oracle", "independent_version_inventory"),
+    "G01": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G01"],
+    "G02": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G02"],
    "G03": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G03"],
-    "G04": ("cache_boundary_crash_evidence", "root_floor_intent_crash_evidence"),
+    "G04": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G04"],
    "G05": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G05"],
    "G06": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G06"],
-    "G07": ("mrf_responsibility_oracle", "commit_boundary_crash_matrix"),
-    "G08": ("mrf_capacity_evidence", "disk_full_matrix", "replica_loss_matrix"),
+    "G07": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G07"],
+    "G08": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G08"],
    "G09": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G09"],
-    "G10": ("scheduler_bound_evidence", "pressure_recovery_evidence"),
+    "G10": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G10"],
    "G11": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G11"],
-    "G12": ("reset_quota_path_evidence", "settlement_quota_path_evidence"),
-    "G13": ("quorum_minus_one_matrix", "unknown_disk_remount_matrix", "object_lock_dry_run_grace_evidence"),
+    "G12": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G12"],
+    "G13": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G13"],
    "G14": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G14"],
-    "P1": ("cold_walk_share_measurement", "foreground_latency_throughput_measurement", "profile_evidence"),
+    "P1": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["P1"],
    "P2": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["P2"],
-    "P3": ("two_hour_pressure_measurement", "heal_capacity_measurement", "recovery_window_measurement"),
+    "P3": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["P3"],
    "P4": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["P4"],
-    "R-E": ("fixed_budget_restart_evidence", "enumeration_evidence", "classification_evidence"),
+    "R-E": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["R-E"],
    "R-D": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["R-D"],
-    "R-L": ("legacy_source_conflict_evidence", "migration_gap_evidence", "crash_safe_source_retirement_evidence"),
+    "R-L": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["R-L"],
 }
 SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES = {
    ("G03", "durable_root_publication_proof"): "durable-root-publication",
@@ -137,6 +182,18 @@ SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES = {
    ("R-L", "migration_gap_evidence"): "migration-gap",
    ("R-L", "crash_safe_source_retirement_evidence"): "crash-safe-source-retirement",
 }
+SCANNER_HEAL_RELEASE_G01_ROOT_AUTHORITY_CASES = (
+    "root-cas-success",
+    "root-readback-success",
+    "incomplete-root-rejected",
+    "stale-root-rejected",
+)
+SCANNER_HEAL_RELEASE_G01_QUOTA_AUTHORITY_CASES = (
+    "quota-floor-readback",
+    "quota-over-limit-rejected",
+    "rejected-object-invisible",
+    "quota-fails-closed-without-authority",
+)
 SCANNER_HEAL_RELEASE_MRF_DURABLE_REPLAY_FIELDS = {
    ("G07", "mrf_responsibility_oracle"),
    ("G07", "commit_boundary_crash_matrix"),
@@ -266,6 +323,19 @@ SCANNER_HEAL_RELEASE_CRASH_BOUNDARY_FIELDS = {
        "process-restart-replay",
    ),
 }
+SCANNER_HEAL_RELEASE_G12_QUOTA_PATH_CASES = {
+    "reset_quota_path_evidence": (
+        "storage-owner-reconstruction",
+        "future-reservation-protocol-fail-closed",
+        "reservation-ledger-retained",
+    ),
+    "settlement_quota_path_evidence": (
+        "distributed-hard-quota-admission",
+        "quota-stats-current-usage-observed",
+        "oversized-put-rejected",
+        "rejected-object-not-visible",
+    ),
+}
 SCANNER_HEAL_RELEASE_G05_PER_OBJECT_OUTCOME_CASES = (
    "object-repaired",
    "object-already-healthy",
@@ -318,6 +388,24 @@ SCANNER_HEAL_RELEASE_RD_GRACE_CASES = (
    "grace-expired-prunes-terminal",
    "restart-preserves-grace-clock",
 )
+SCANNER_HEAL_RELEASE_RL_REQUIRED_CASES = {
+    "legacy_source_conflict_evidence": (
+        "data-movement-owned-target-equivalence",
+        "newer-target-conflict-rejected",
+        "legacy-part-checksum-gap-rejected",
+    ),
+    "migration_gap_evidence": (
+        "encrypted-iam-decrypt-before-normalize",
+        "empty-legacy-source-inherits-prior-responsibilities",
+        "missing-corrupt-empty-sources-fail-closed",
+    ),
+    "crash_safe_source_retirement_evidence": (
+        "source-change-capacity-failure-keeps-old-commit",
+        "torn-inactive-payload-keeps-previous-anchor",
+        "commit-boundary-lost-response-idempotent",
+        "successor-retry-validates-orphan",
+    ),
+}
 SCANNER_HEAL_RELEASE_SCOPED_ACK_CASES = {
    "durable_root_publication_proof": (
        "root-cas-success",
@@ -1677,6 +1765,23 @@ def is_json_artifact_format(value: str) -> bool:

 def release_bundle_json_artifact_mirrored_fields(gate: str, field: str) -> tuple[str, ...]:
    fields: list[str] = []
+    if gate == "G01":
+        if field == "root_authority_evidence":
+            fields.extend((
+                "root_authority_cases",
+                "root_cas_observed",
+                "root_readback_observed",
+                "incomplete_root_rejected",
+                "stale_root_rejected",
+            ))
+        if field == "quota_authority_evidence":
+            fields.extend((
+                "quota_authority_cases",
+                "quota_floor_readback_observed",
+                "over_limit_put_rejected",
+                "rejected_object_invisible",
+                "quota_fails_closed_without_authority",
+            ))
    if gate in ("G03", "G09", "R-L"):
        fields.extend(("versions", "mixed_version_role"))
    if gate == "G02":
@@ -1730,6 +1835,8 @@ def release_bundle_json_artifact_mirrored_fields(gate: str, field: str) -> tuple
            fields.append("whole_cycle_fallback_observed")
    if gate == "G04" and field == "root_floor_intent_crash_evidence":
        fields.extend(("durable_intent_cases", "persist_failure_blocks_acceptance"))
+    if gate == "G12":
+        fields.append("quota_path_cases")
    if gate == "G05":
        if field == "per_object_outcome_oracle":
            fields.extend(("per_object_outcome_cases", "outcome_counts", "status_matches_object_oracle"))
@@ -1853,6 +1960,28 @@ def release_bundle_json_artifact_mirrored_fields(gate: str, field: str) -> tuple
                "grace_retention_observed",
                "grace_expiry_pruned_terminal_records",
            ))
+    if gate == "R-L":
+        if field == "legacy_source_conflict_evidence":
+            fields.extend((
+                "legacy_source_conflict_cases",
+                "source_conflicts_rejected",
+                "takeover_identity_bound",
+                "legacy_checksum_gap_rejected",
+            ))
+        if field == "migration_gap_evidence":
+            fields.extend((
+                "migration_gap_cases",
+                "migration_gap_closed",
+                "legacy_sources_fail_closed",
+                "prior_responsibilities_inherited",
+            ))
+        if field == "crash_safe_source_retirement_evidence":
+            fields.extend((
+                "source_retirement_cases",
+                "source_retirement_is_crash_safe",
+                "old_source_retained_until_successor",
+                "recovered_pending_migration",
+            ))
    return tuple(dict.fromkeys(fields))


@@ -1953,6 +2082,33 @@ def release_bundle_number(value: object, name: str, minimum: int | float = 0) ->


 def validate_release_bundle_domain_evidence(gate: str, field: str, evidence: dict[str, object]) -> None:
+    if gate == "G01":
+        if field == "root_authority_evidence":
+            release_bundle_exact_strings(
+                evidence.get("root_authority_cases"),
+                SCANNER_HEAL_RELEASE_G01_ROOT_AUTHORITY_CASES,
+                f"{gate}.{field}.root_authority_cases",
+            )
+            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")
+            release_bundle_bool_true(evidence.get("incomplete_root_rejected"),
+                                     f"{gate}.{field}.incomplete_root_rejected")
+            release_bundle_bool_true(evidence.get("stale_root_rejected"), f"{gate}.{field}.stale_root_rejected")
+        if field == "quota_authority_evidence":
+            release_bundle_exact_strings(
+                evidence.get("quota_authority_cases"),
+                SCANNER_HEAL_RELEASE_G01_QUOTA_AUTHORITY_CASES,
+                f"{gate}.{field}.quota_authority_cases",
+            )
+            release_bundle_bool_true(evidence.get("quota_floor_readback_observed"),
+                                     f"{gate}.{field}.quota_floor_readback_observed")
+            release_bundle_bool_true(evidence.get("over_limit_put_rejected"),
+                                     f"{gate}.{field}.over_limit_put_rejected")
+            release_bundle_bool_true(evidence.get("rejected_object_invisible"),
+                                     f"{gate}.{field}.rejected_object_invisible")
+            release_bundle_bool_true(evidence.get("quota_fails_closed_without_authority"),
+                                     f"{gate}.{field}.quota_fails_closed_without_authority")
+
    if gate == "G02":
        if field == "bounded_checkpoint_oracle":
            release_bundle_bool_true(evidence.get("checkpoint_progress_bounded"),
@@ -2007,6 +2163,13 @@ def validate_release_bundle_domain_evidence(gate: str, field: str, evidence: dic
        release_bundle_bool_true(evidence.get("persist_failure_blocks_acceptance"),
                                 f"{gate}.{field}.persist_failure_blocks_acceptance")

+    if gate == "G12":
+        release_bundle_exact_strings(
+            evidence.get("quota_path_cases"),
+            SCANNER_HEAL_RELEASE_G12_QUOTA_PATH_CASES[field],
+            f"{gate}.{field}.quota_path_cases",
+        )
+
    if gate == "G05":
        if field == "per_object_outcome_oracle":
            release_bundle_exact_strings(
@@ -2314,6 +2477,44 @@ def validate_release_bundle_domain_evidence(gate: str, field: str, evidence: dic
            release_bundle_bool_true(evidence.get("grace_expiry_pruned_terminal_records"),
                                     f"{gate}.{field}.grace_expiry_pruned_terminal_records")

+    if gate == "R-L":
+        if field == "legacy_source_conflict_evidence":
+            release_bundle_exact_strings(
+                evidence.get("legacy_source_conflict_cases"),
+                SCANNER_HEAL_RELEASE_RL_REQUIRED_CASES[field],
+                f"{gate}.{field}.legacy_source_conflict_cases",
+            )
+            release_bundle_bool_true(evidence.get("source_conflicts_rejected"),
+                                     f"{gate}.{field}.source_conflicts_rejected")
+            release_bundle_bool_true(evidence.get("takeover_identity_bound"),
+                                     f"{gate}.{field}.takeover_identity_bound")
+            release_bundle_bool_true(evidence.get("legacy_checksum_gap_rejected"),
+                                     f"{gate}.{field}.legacy_checksum_gap_rejected")
+        if field == "migration_gap_evidence":
+            release_bundle_exact_strings(
+                evidence.get("migration_gap_cases"),
+                SCANNER_HEAL_RELEASE_RL_REQUIRED_CASES[field],
+                f"{gate}.{field}.migration_gap_cases",
+            )
+            release_bundle_bool_true(evidence.get("migration_gap_closed"),
+                                     f"{gate}.{field}.migration_gap_closed")
+            release_bundle_bool_true(evidence.get("legacy_sources_fail_closed"),
+                                     f"{gate}.{field}.legacy_sources_fail_closed")
+            release_bundle_bool_true(evidence.get("prior_responsibilities_inherited"),
+                                     f"{gate}.{field}.prior_responsibilities_inherited")
+        if field == "crash_safe_source_retirement_evidence":
+            release_bundle_exact_strings(
+                evidence.get("source_retirement_cases"),
+                SCANNER_HEAL_RELEASE_RL_REQUIRED_CASES[field],
+                f"{gate}.{field}.source_retirement_cases",
+            )
+            release_bundle_bool_true(evidence.get("source_retirement_is_crash_safe"),
+                                     f"{gate}.{field}.source_retirement_is_crash_safe")
+            release_bundle_bool_true(evidence.get("old_source_retained_until_successor"),
+                                     f"{gate}.{field}.old_source_retained_until_successor")
+            release_bundle_bool_true(evidence.get("recovered_pending_migration"),
+                                     f"{gate}.{field}.recovered_pending_migration")
+

 def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, gate: str, field: str,
                                     evidence: dict[str, object]) -> str:
@@ -2361,7 +2562,7 @@ def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, ga
            evidence_integer(evidence.get("lock_hold_p95_ms"), f"{gate}.{field}.lock_hold_p95_ms", 0, 2**31 - 1)
            evidence_integer(evidence.get("foreground_latency_p95_ms"),
                             f"{gate}.{field}.foreground_latency_p95_ms", 1, 2**31 - 1)
-    if gate in ("G11", "G13"):
+    if gate in ("G01", "G11", "G12", "G13", "R-L"):
        validate_release_bundle_domain_evidence(gate, field, evidence)
    if gate == "P1" and field == "foreground_latency_throughput_measurement":
        evidence_integer(evidence.get("foreground_latency_p95_ms"),
@@ -2719,66 +2920,80 @@ def copy_release_bundle_artifact(descriptor_path: Path, bundle_dir: Path, eviden
    evidence["sha256"] = source_sha


-def assemble_scanner_heal_release_bundle(root: Path, descriptor_path: Path, directory: Path) -> tuple[Path, dict[str, object]]:
-    """Copy measured Scanner/Heal evidence into one validated release bundle."""
+def assemble_scanner_heal_release_bundle_descriptors(
+    root: Path,
+    descriptor_paths: list[Path],
+    directory: Path,
+) -> tuple[Path, dict[str, object]]:
+    """Copy measured Scanner/Heal evidence descriptors into one validated release bundle."""
    require(not directory.exists(), "scanner/heal release bundle directory must be new")
+    require(descriptor_paths, "scanner/heal release bundle assembly requires at least one descriptor")
    registry = read_json(root / ".config/scanner-heal-required-tests.json")
    requirements, release_schema_capable, _ = scanner_heal_release_requirements(registry)
    require(release_schema_capable, "scanner/heal release bundle assembly requires schema 2 registry")
-    descriptor_path = descriptor_path.resolve(strict=True)
-    descriptor = read_json(descriptor_path)
-    reject_release_bundle_markers(descriptor, "scanner/heal release evidence descriptor")
-    require(descriptor.get("schema") == 1, "unsupported scanner/heal release evidence descriptor schema")
-    require(descriptor.get("evidence") == "measured", "scanner/heal release evidence descriptor must be measured")
-    source_revision = descriptor.get("source_revision")
-    require(isinstance(source_revision, str) and re.fullmatch(r"[0-9a-f]{40}", source_revision) is not None,
-            "scanner/heal release evidence descriptor source revision is invalid")
-    raw_gates = descriptor.get("gates")
-    require(isinstance(raw_gates, dict), "scanner/heal release evidence descriptor missing gates")
-    unknown_gates = sorted(set(raw_gates) - set(requirements))
-    require(not unknown_gates, f"scanner/heal release evidence descriptor has unknown gates: {', '.join(unknown_gates)}")

    bundle_dir = directory.resolve()
    artifact_dir = bundle_dir / "artifacts"
    artifact_dir.mkdir(parents=True)
    assembled_gates = {}
-    for gate in sorted(raw_gates):
-        raw_gate = raw_gates[gate]
-        require(isinstance(raw_gate, dict), f"{gate} descriptor gate must be an object")
-        reject_release_bundle_markers(raw_gate, f"{gate} descriptor gate")
-        raw_fields = raw_gate.get("evidence_fields", raw_gate)
-        require(isinstance(raw_fields, dict), f"{gate} descriptor gate missing evidence fields")
-        required_fields = set(SCANNER_HEAL_RELEASE_BUNDLE_REQUIRED_EVIDENCE_FIELDS[gate])
-        unknown_fields = sorted(set(raw_fields) - required_fields)
-        require(not unknown_fields, f"{gate} descriptor has unknown fields: {', '.join(unknown_fields)}")
-        fields = {}
-        for field in sorted(raw_fields):
-            raw_evidence = raw_fields[field]
-            require(isinstance(raw_evidence, dict), f"{gate}.{field} descriptor evidence must be an object")
-            evidence = json.loads(json.dumps(raw_evidence))
-            reject_release_bundle_markers(evidence, f"{gate}.{field} descriptor evidence")
-            copy_release_bundle_artifact(descriptor_path, bundle_dir, evidence, gate, field, f"{gate}-{field}")
-            profile_artifacts = evidence.get("profile_artifacts")
-            if profile_artifacts is not None:
-                require(isinstance(profile_artifacts, dict), f"{gate}.{field} profile artifacts must be an object")
-                for artifact_kind, item in sorted(profile_artifacts.items()):
-                    require(isinstance(item, dict), f"{gate}.{field}.{artifact_kind} descriptor profile artifact must be an object")
-                    reject_release_bundle_markers(item, f"{gate}.{field}.{artifact_kind} descriptor profile artifact")
-                    copy_release_bundle_artifact(
-                        descriptor_path,
-                        bundle_dir,
-                        item,
-                        gate,
-                        f"{field}.{artifact_kind}",
-                        f"{gate}-{field}-{artifact_kind}",
-                    )
-            fields[field] = evidence
-        assembled_gates[gate] = {
-            "status": "pass",
-            "lane": requirements[gate]["lane"],
-            "evidence_type": "measured",
-            "evidence_fields": fields,
-        }
+    source_revision = None
+    for descriptor_index, raw_descriptor_path in enumerate(descriptor_paths):
+        descriptor_path = raw_descriptor_path.resolve(strict=True)
+        descriptor = read_json(descriptor_path)
+        label = f"scanner/heal release evidence descriptor {descriptor_index + 1}"
+        reject_release_bundle_markers(descriptor, label)
+        require(descriptor.get("schema") == 1, f"unsupported {label} schema")
+        require(descriptor.get("evidence") == "measured", f"{label} must be measured")
+        descriptor_revision = descriptor.get("source_revision")
+        require(isinstance(descriptor_revision, str) and re.fullmatch(r"[0-9a-f]{40}", descriptor_revision) is not None,
+                f"{label} source revision is invalid")
+        if source_revision is None:
+            source_revision = descriptor_revision
+        else:
+            require(descriptor_revision == source_revision, "scanner/heal release descriptor source revisions differ")
+        raw_gates = descriptor.get("gates")
+        require(isinstance(raw_gates, dict), f"{label} missing gates")
+        unknown_gates = sorted(set(raw_gates) - set(requirements))
+        require(not unknown_gates, f"{label} has unknown gates: {', '.join(unknown_gates)}")
+
+        for gate in sorted(raw_gates):
+            require(gate not in assembled_gates, f"scanner/heal release descriptor repeats gate: {gate}")
+            raw_gate = raw_gates[gate]
+            require(isinstance(raw_gate, dict), f"{gate} descriptor gate must be an object")
+            reject_release_bundle_markers(raw_gate, f"{gate} descriptor gate")
+            raw_fields = raw_gate.get("evidence_fields", raw_gate)
+            require(isinstance(raw_fields, dict), f"{gate} descriptor gate missing evidence fields")
+            required_fields = set(SCANNER_HEAL_RELEASE_BUNDLE_REQUIRED_EVIDENCE_FIELDS[gate])
+            unknown_fields = sorted(set(raw_fields) - required_fields)
+            require(not unknown_fields, f"{gate} descriptor has unknown fields: {', '.join(unknown_fields)}")
+            fields = {}
+            for field in sorted(raw_fields):
+                raw_evidence = raw_fields[field]
+                require(isinstance(raw_evidence, dict), f"{gate}.{field} descriptor evidence must be an object")
+                evidence = json.loads(json.dumps(raw_evidence))
+                reject_release_bundle_markers(evidence, f"{gate}.{field} descriptor evidence")
+                copy_release_bundle_artifact(descriptor_path, bundle_dir, evidence, gate, field, f"{gate}-{field}")
+                profile_artifacts = evidence.get("profile_artifacts")
+                if profile_artifacts is not None:
+                    require(isinstance(profile_artifacts, dict), f"{gate}.{field} profile artifacts must be an object")
+                    for artifact_kind, item in sorted(profile_artifacts.items()):
+                        require(isinstance(item, dict), f"{gate}.{field}.{artifact_kind} descriptor profile artifact must be an object")
+                        reject_release_bundle_markers(item, f"{gate}.{field}.{artifact_kind} descriptor profile artifact")
+                        copy_release_bundle_artifact(
+                            descriptor_path,
+                            bundle_dir,
+                            item,
+                            gate,
+                            f"{field}.{artifact_kind}",
+                            f"{gate}-{field}-{artifact_kind}",
+                        )
+                fields[field] = evidence
+            assembled_gates[gate] = {
+                "status": "pass",
+                "lane": requirements[gate]["lane"],
+                "evidence_type": "measured",
+                "evidence_fields": fields,
+            }

    bundle = bundle_dir / "release-evidence.json"
    write_json(bundle, {
@@ -2790,6 +3005,11 @@ def assemble_scanner_heal_release_bundle(root: Path, descriptor_path: Path, dire
    return bundle, scanner_heal_release_bundle_status(root, bundle)


+def assemble_scanner_heal_release_bundle(root: Path, descriptor_path: Path, directory: Path) -> tuple[Path, dict[str, object]]:
+    """Copy one measured Scanner/Heal evidence descriptor into a validated release bundle."""
+    return assemble_scanner_heal_release_bundle_descriptors(root, [descriptor_path], directory)
+
+
 def write_scanner_heal_release_bundle_fixture(root: Path, directory: Path) -> Path:
    """Write a non-approvable release bundle shape fixture for dry-run validation."""
    bundle_dir = directory.resolve()
@@ -2911,6 +3131,8 @@ def write_scanner_heal_release_bundle_fixture(root: Path, directory: Path) -> Pa
                evidence["mixed_version_role"] = SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES[(gate, field)]
            if gate in ("G04", "G07", "R-E", "R-L"):
                evidence["crash_points"] = ["fixture-before-commit"]
+            if gate == "G12":
+                evidence["quota_path_cases"] = list(SCANNER_HEAL_RELEASE_G12_QUOTA_PATH_CASES[field])
            if (gate, field) in SCANNER_HEAL_RELEASE_MRF_DURABLE_REPLAY_FIELDS:
                evidence["replayed_records"] = 1
                evidence["responsibility_anchor_retained"] = True
@@ -3298,6 +3520,22 @@ class SelfTests(unittest.TestCase):
                        "versions_retained": 16,
                        "bytes_retained": 16,
                    })
+                if gate == "G01" and field == "root_authority_evidence":
+                    evidence.update({
+                        "root_authority_cases": list(SCANNER_HEAL_RELEASE_G01_ROOT_AUTHORITY_CASES),
+                        "root_cas_observed": True,
+                        "root_readback_observed": True,
+                        "incomplete_root_rejected": True,
+                        "stale_root_rejected": True,
+                    })
+                if gate == "G01" and field == "quota_authority_evidence":
+                    evidence.update({
+                        "quota_authority_cases": list(SCANNER_HEAL_RELEASE_G01_QUOTA_AUTHORITY_CASES),
+                        "quota_floor_readback_observed": True,
+                        "over_limit_put_rejected": True,
+                        "rejected_object_invisible": True,
+                        "quota_fails_closed_without_authority": True,
+                    })
                if gate in ("G03", "G09", "R-L"):
                    evidence["versions"] = ["a" * 40, source_revision]
                    evidence["mixed_version_role"] = SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES[(gate, field)]
@@ -3313,6 +3551,8 @@ class SelfTests(unittest.TestCase):
                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
+                if gate == "G12":
+                    evidence["quota_path_cases"] = list(SCANNER_HEAL_RELEASE_G12_QUOTA_PATH_CASES[field])
                if gate == "G05" and field == "per_object_outcome_oracle":
                    evidence["per_object_outcome_cases"] = list(SCANNER_HEAL_RELEASE_G05_PER_OBJECT_OUTCOME_CASES)
                    evidence["outcome_counts"] = {"repaired": 4, "healthy": 3, "skipped": 2, "failed": 1}
@@ -3486,6 +3726,21 @@ class SelfTests(unittest.TestCase):
                    evidence["grace_window_seconds"] = 300
                    evidence["grace_retention_observed"] = True
                    evidence["grace_expiry_pruned_terminal_records"] = True
+                if gate == "R-L" and field == "legacy_source_conflict_evidence":
+                    evidence["legacy_source_conflict_cases"] = list(SCANNER_HEAL_RELEASE_RL_REQUIRED_CASES[field])
+                    evidence["source_conflicts_rejected"] = True
+                    evidence["takeover_identity_bound"] = True
+                    evidence["legacy_checksum_gap_rejected"] = True
+                if gate == "R-L" and field == "migration_gap_evidence":
+                    evidence["migration_gap_cases"] = list(SCANNER_HEAL_RELEASE_RL_REQUIRED_CASES[field])
+                    evidence["migration_gap_closed"] = True
+                    evidence["legacy_sources_fail_closed"] = True
+                    evidence["prior_responsibilities_inherited"] = True
+                if gate == "R-L" and field == "crash_safe_source_retirement_evidence":
+                    evidence["source_retirement_cases"] = list(SCANNER_HEAL_RELEASE_RL_REQUIRED_CASES[field])
+                    evidence["source_retirement_is_crash_safe"] = True
+                    evidence["old_source_retained_until_successor"] = True
+                    evidence["recovered_pending_migration"] = True
                if field == "profile_evidence":
                    evidence["resolved_samples"] = 1
                    evidence["allocation_bytes"] = 1024
@@ -3591,6 +3846,71 @@ class SelfTests(unittest.TestCase):
            self.assertEqual(evidence["sha256"], digest(bundle.parent / evidence["artifact"]))
            self.assertTrue((bundle.parent / evidence["artifact"]).is_file())

+    def test_scanner_heal_release_bundle_assembler_merges_measured_descriptors(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, descriptor = self.scanner_heal_release_bundle_fixture(Path(tmp))
+            source = read_json(descriptor)
+            descriptor_paths = []
+            for gate in ("G01", "G09"):
+                partial = descriptor.parent / f"release-bundle-{gate}.json"
+                write_json(partial, {
+                    "schema": 1,
+                    "evidence": "measured",
+                    "source_revision": source["source_revision"],
+                    "gates": {gate: source["gates"][gate]},
+                })
+                descriptor_paths.append(partial)
+
+            with mock.patch("subprocess.check_output", return_value="b" * 40):
+                bundle, status = assemble_scanner_heal_release_bundle_descriptors(
+                    root,
+                    descriptor_paths,
+                    Path(tmp) / "assembled",
+                )
+            self.assertEqual(status["decision"], "blocked")
+            self.assertFalse(status["release_approved"])
+            self.assertIn("G01", status["verified_gates"])
+            self.assertIn("G09", status["verified_gates"])
+            self.assertIn("G02", status["pending_gates"])
+            assembled = read_json(bundle)
+            self.assertEqual(sorted(assembled["gates"]), ["G01", "G09"])
+            g09 = assembled["gates"]["G09"]["evidence_fields"]["mixed_version_reader_evidence"]
+            self.assertTrue((bundle.parent / g09["artifact"]).is_file())
+
+    def test_scanner_heal_release_bundle_assembler_rejects_duplicate_descriptor_gate(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, descriptor = self.scanner_heal_release_bundle_fixture(Path(tmp))
+            with mock.patch("subprocess.check_output", return_value="b" * 40):
+                with self.assertRaisesRegex(ValueError, "repeats gate: G01"):
+                    assemble_scanner_heal_release_bundle_descriptors(
+                        root,
+                        [descriptor, descriptor],
+                        Path(tmp) / "assembled",
+                    )
+
+    def test_scanner_heal_release_bundle_assembler_rejects_mixed_descriptor_revisions(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, descriptor = self.scanner_heal_release_bundle_fixture(Path(tmp))
+            source = read_json(descriptor)
+            g01 = descriptor.parent / "release-bundle-g01.json"
+            g09 = descriptor.parent / "release-bundle-g09.json"
+            write_json(g01, {
+                "schema": 1,
+                "evidence": "measured",
+                "source_revision": source["source_revision"],
+                "gates": {"G01": source["gates"]["G01"]},
+            })
+            write_json(g09, {
+                "schema": 1,
+                "evidence": "measured",
+                "source_revision": "c" * 40,
+                "gates": {"G09": source["gates"]["G09"]},
+            })
+
+            with mock.patch("subprocess.check_output", return_value="b" * 40):
+                with self.assertRaisesRegex(ValueError, "source revisions differ"):
+                    assemble_scanner_heal_release_bundle_descriptors(root, [g01, g09], Path(tmp) / "assembled")
+
    def test_scanner_heal_release_bundle_gate_accepts_single_measured_gate(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root, bundle = self.scanner_heal_release_bundle_fixture(Path(tmp))
@@ -3977,6 +4297,20 @@ class SelfTests(unittest.TestCase):
            ("scheduler-duplicates", "G10", "scheduler_bound_evidence", lambda item: item.pop("duplicate_task_count"), "duplicate_task_count"),
            ("p1-throughput", "P1", "foreground_latency_throughput_measurement", lambda item: item.pop("throughput_ops_per_second"), "throughput_ops_per_second"),
            ("p3-fixed-load", "P3", "two_hour_pressure_measurement", lambda item: item.update({"fixed_offered_load": False}), "fixed offered load"),
+            (
+                "g01-root-cases",
+                "G01",
+                "root_authority_evidence",
+                lambda item: item["root_authority_cases"].remove("stale-root-rejected"),
+                "root_authority_cases missing cases",
+            ),
+            (
+                "g01-quota-fail-closed",
+                "G01",
+                "quota_authority_evidence",
+                lambda item: item.update({"quota_fails_closed_without_authority": False}),
+                "quota_fails_closed_without_authority",
+            ),
            (
                "rd-manager",
                "R-D",
@@ -4098,6 +4432,12 @@ class SelfTests(unittest.TestCase):
                ("G05", "per_object_outcome_oracle"),
                "per_object_outcome_cases missing cases",
            ),
+            (
+                "g01-authority-mirror",
+                lambda payload: payload["root_authority_cases"].remove("incomplete-root-rejected"),
+                ("G01", "root_authority_evidence"),
+                "root_authority_cases missing cases",
+            ),
            (
                "g06-truncation-mirror",
                lambda payload: payload["truncation_cases"].remove("trailing-data-node-status-reject"),
@@ -4128,6 +4468,12 @@ class SelfTests(unittest.TestCase):
                ("R-D", "grace_handling"),
                "grace_cases missing cases",
            ),
+            (
+                "rl-migration-gap-mirror",
+                lambda payload: payload.update({"migration_gap_closed": False}),
+                ("R-L", "migration_gap_evidence"),
+                "migration_gap_closed",
+            ),
        ):
            with self.subTest(fault=fault), tempfile.TemporaryDirectory() as tmp:
                root, bundle = self.scanner_heal_release_bundle_fixture(Path(tmp))
@@ -4208,6 +4554,20 @@ class SelfTests(unittest.TestCase):
                lambda item: item.pop("persist_failure_blocks_acceptance"),
                "persist_failure_blocks_acceptance",
            ),
+            (
+                "g12-reset-quota-cases",
+                "G12",
+                "reset_quota_path_evidence",
+                lambda item: item["quota_path_cases"].remove("future-reservation-protocol-fail-closed"),
+                "quota_path_cases missing cases",
+            ),
+            (
+                "g12-settlement-quota-cases",
+                "G12",
+                "settlement_quota_path_evidence",
+                lambda item: item["quota_path_cases"].remove("rejected-object-not-visible"),
+                "quota_path_cases missing cases",
+            ),
            (
                "mixed-version-cases",
                "G09",
@@ -4222,6 +4582,27 @@ class SelfTests(unittest.TestCase):
                lambda item: item.update({"rollback_payload_replayed": False}),
                "rollback_payload_replayed",
            ),
+            (
+                "legacy-source-conflict-cases",
+                "R-L",
+                "legacy_source_conflict_evidence",
+                lambda item: item["legacy_source_conflict_cases"].remove("newer-target-conflict-rejected"),
+                "legacy_source_conflict_cases missing cases",
+            ),
+            (
+                "migration-gap-closed",
+                "R-L",
+                "migration_gap_evidence",
+                lambda item: item.update({"migration_gap_closed": False}),
+                "migration_gap_closed",
+            ),
+            (
+                "source-retirement-crash-safe",
+                "R-L",
+                "crash_safe_source_retirement_evidence",
+                lambda item: item.update({"source_retirement_is_crash_safe": False}),
+                "source_retirement_is_crash_safe",
+            ),
            (
                "scheduler-bounds",
                "G10",
@@ -4365,25 +4746,14 @@ class SelfTests(unittest.TestCase):
            self.assertIn("ec8-4-multiset", status["pending_lanes"])
            self.assertIn("scheduler-pressure", status["pending_lanes"])
            requirements, _, _ = scanner_heal_release_requirements(read_json(root / ".config/scanner-heal-required-tests.json"))
-            self.assertIn("durable_root_publication_proof", requirements["G03"]["evidence_fields"])
-            self.assertIn("per_object_outcome_oracle", requirements["G05"]["evidence_fields"])
-            self.assertIn("truncation_behavior", requirements["G06"]["evidence_fields"])
-            self.assertIn("disk_full_matrix", requirements["G08"]["evidence_fields"])
-            self.assertIn("mixed_version_writer_evidence", requirements["G09"]["evidence_fields"])
-            self.assertIn("segment_activation_preflight", requirements["G11"]["evidence_fields"])
-            self.assertIn("distributed_segment_invalidation_evidence", requirements["G14"]["evidence_fields"])
-            self.assertIn("cold_segment_reuse_measurement", requirements["P2"]["evidence_fields"])
-            self.assertEqual(
-                tuple(requirements["P4"]["evidence_fields"]),
-                SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["P4"],
-            )
-            self.assertIn("grace_handling", requirements["R-D"]["evidence_fields"])
+            for gate, expected_fields in SCANNER_HEAL_RELEASE_BUNDLE_REQUIRED_EVIDENCE_FIELDS.items():
+                self.assertEqual(tuple(requirements[gate]["evidence_fields"]), expected_fields, gate)

    def test_scanner_heal_required_evidence_fields_cannot_be_removed(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root, run_dir = self.scanner_heal_fixture(Path(tmp))
            registry = read_json(root / ".config/scanner-heal-required-tests.json")
-            for gate in ("G03", "G05", "G06", "G07", "G08", "G09", "G11", "G14", "P2", "P4", "R-D"):
+            for gate, expected_fields in SCANNER_HEAL_RELEASE_BUNDLE_REQUIRED_EVIDENCE_FIELDS.items():
                for requirement in registry["release_requirements"]:
                    if requirement["gate"] == gate:
                        requirement["evidence_fields"] = []
@@ -4397,7 +4767,7 @@ class SelfTests(unittest.TestCase):
                registry = read_json(root / ".config/scanner-heal-required-tests.json")
                for requirement in registry["release_requirements"]:
                    if requirement["gate"] == gate:
-                        requirement["evidence_fields"] = list(SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS[gate])
+                        requirement["evidence_fields"] = list(expected_fields)
                        break

    def test_scanner_heal_ec84_case_rejects_wrong_erasure_geometry(self) -> None:
@@ -5368,9 +5738,13 @@ def main() -> int:
                    return 2
                print(json.dumps(status, sort_keys=True, separators=(",", ":")))
                return 0 if status["decision"] == "verified" else 1
-            if len(sys.argv) == 4 and sys.argv[1] == "--assemble-scanner-heal-release-bundle":
+            if len(sys.argv) >= 4 and sys.argv[1] == "--assemble-scanner-heal-release-bundle":
                try:
-                    bundle, status = assemble_scanner_heal_release_bundle(ROOT, Path(sys.argv[2]), Path(sys.argv[3]))
+                    bundle, status = assemble_scanner_heal_release_bundle_descriptors(
+                        ROOT,
+                        [Path(item) for item in sys.argv[2:-1]],
+                        Path(sys.argv[-1]),
+                    )
                except (OSError, KeyError, TypeError, ValueError, ET.ParseError) as error:
                    print(json.dumps({"schema": 1, "decision": "invalid", "release_approved": False,
                                      "error": str(error)}, sort_keys=True, separators=(",", ":")))
@@ -5382,7 +5756,7 @@ def main() -> int:
                bundle = write_scanner_heal_release_bundle_fixture(ROOT, Path(sys.argv[2]))
                print(bundle)
                return 0
-            raise ValueError("expected --begin-scanner-heal DIR BINARY TEST_BINARY, --finish-scanner-heal DIR EXIT, --check-scanner-heal DIR CASE|release, --check-scanner-heal-release DIR, --check-scanner-heal-release-bundle FILE, --check-scanner-heal-release-bundle-gate FILE GATE, --assemble-scanner-heal-release-bundle DESCRIPTOR DIR, or --write-scanner-heal-release-bundle-fixture DIR")
+            raise ValueError("expected --begin-scanner-heal DIR BINARY TEST_BINARY, --finish-scanner-heal DIR EXIT, --check-scanner-heal DIR CASE|release, --check-scanner-heal-release DIR, --check-scanner-heal-release-bundle FILE, --check-scanner-heal-release-bundle-gate FILE GATE, --assemble-scanner-heal-release-bundle DESCRIPTOR [DESCRIPTOR ...] DIR, or --write-scanner-heal-release-bundle-fixture DIR")
        except (OSError, KeyError, TypeError, ValueError, subprocess.SubprocessError) as error:
            print(f"ERROR: {error}", file=sys.stderr)
            return 1
@@ -5411,7 +5785,8 @@ def main() -> int:
            "usage: check_test_wiring.py [--self-test | --check-core LISTING | --check-profile PROFILE LISTING | "
            "--update-profile PROFILE LISTING PLATFORM | --check-scanner-heal-release-bundle FILE | "
            "--check-scanner-heal-release-bundle-gate FILE GATE | "
-            "--assemble-scanner-heal-release-bundle DESCRIPTOR DIR | --write-scanner-heal-release-bundle-fixture DIR]",
+            "--assemble-scanner-heal-release-bundle DESCRIPTOR [DESCRIPTOR ...] DIR | "
+            "--write-scanner-heal-release-bundle-fixture DIR]",
            file=sys.stderr,
        )
        return 2
@@ -29,7 +29,7 @@ MIN_MINOR=11
 UV_PYTHON_SPEC="${RUSTFS_UV_PYTHON:-3.12}"

 version_ok() {
-    "$1" -c "import sys; raise SystemExit(0 if sys.version_info >= (${MIN_MAJOR}, ${MIN_MINOR}) else 1)" \
+    "$@" -c "import sys; raise SystemExit(0 if sys.version_info >= (${MIN_MAJOR}, ${MIN_MINOR}) else 1)" \
        >/dev/null 2>&1
 }

@@ -55,20 +55,25 @@ if [ "${1:-}" = "--print-interpreter" ]; then
 fi

 if [ -n "${RUSTFS_PYTHON:-}" ]; then
-    if ! command -v "${RUSTFS_PYTHON}" >/dev/null 2>&1; then
+    read -r -a rustfs_python_cmd <<< "${RUSTFS_PYTHON}"
+    if [ "${#rustfs_python_cmd[@]}" -eq 0 ]; then
        echo >&2 "❌ RUSTFS_PYTHON='${RUSTFS_PYTHON}' is not an executable command."
        exit 1
    fi
-    if ! version_ok "${RUSTFS_PYTHON}"; then
+    if ! command -v "${rustfs_python_cmd[0]}" >/dev/null 2>&1; then
+        echo >&2 "❌ RUSTFS_PYTHON='${RUSTFS_PYTHON}' is not an executable command."
+        exit 1
+    fi
+    if ! version_ok "${rustfs_python_cmd[@]}"; then
        echo >&2 "❌ RUSTFS_PYTHON='${RUSTFS_PYTHON}' is older than Python ${MIN_MAJOR}.${MIN_MINOR}."
        echo >&2 "   The repository's checkers import tomllib (Python ${MIN_MAJOR}.${MIN_MINOR}+)."
        exit 1
    fi
    if [ "${print_only}" = "1" ]; then
-        command -v "${RUSTFS_PYTHON}"
+        printf '%s\n' "${RUSTFS_PYTHON}"
        exit 0
    fi
-    exec "${RUSTFS_PYTHON}" "$@"
+    exec "${rustfs_python_cmd[@]}" "$@"
 fi

 for candidate in python3.14 python3.13 python3.12 python3.11 python3 python; do
@@ -0,0 +1,286 @@
+#!/usr/bin/env python3
+"""Assemble measured Scanner/Heal G01 authority release evidence.
+
+The producer consumes operator-collected measured JSON. It packages root and
+quota authority measurements into the common release-bundle descriptor shape
+and lets check_test_wiring.py validate G01 without approving the full release.
+"""
+
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timedelta, timezone
+import json
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+import check_test_wiring as wiring
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def git_head() -> str:
+    return subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
+
+
+def timestamp(value: Any, name: str) -> str:
+    wiring.require(isinstance(value, str) and value.strip(), f"missing {name}")
+    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    wiring.require(parsed.tzinfo is not None, f"{name} must include timezone")
+    return parsed.isoformat().replace("+00:00", "Z")
+
+
+def measured_json(path: Path, source_revision: str, label: str) -> dict[str, Any]:
+    payload = wiring.read_json(path.resolve())
+    wiring.require(isinstance(payload, dict), f"{label} must be a JSON object")
+    for marker in ("fixture", "fixture_only", "dry_run", "synthetic"):
+        wiring.require(payload.get(marker) is not True, f"{label} is {marker}")
+    wiring.require(payload.get("schema") == 1, f"{label} schema must be 1")
+    wiring.require(payload.get("evidence_type") == "measured", f"{label} must be measured")
+    wiring.require(payload.get("source_revision") == source_revision, f"{label} source revision mismatch")
+    return payload
+
+
+def common_evidence(args: argparse.Namespace, source_revision: str) -> dict[str, Any]:
+    duration = wiring.evidence_integer(args.duration_seconds, "duration_seconds", 1, 86400)
+    started_at = timestamp(args.started_at, "started_at") if args.started_at else datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+    if args.finished_at:
+        finished_at = timestamp(args.finished_at, "finished_at")
+    else:
+        started = datetime.fromisoformat(started_at.replace("Z", "+00:00"))
+        finished_at = (started + timedelta(seconds=duration)).isoformat().replace("+00:00", "Z")
+    wiring.require(
+        datetime.fromisoformat(finished_at.replace("Z", "+00:00")) >= datetime.fromisoformat(started_at.replace("Z", "+00:00")),
+        "authority measurement timestamps are inverted",
+    )
+    run_id = args.run_id or f"authority-{source_revision[:12]}"
+    window_id = args.measurement_window_id or f"authority-window-{source_revision[:12]}"
+    wiring.evidence_string(run_id, "run_id", r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}")
+    wiring.evidence_string(window_id, "measurement_window_id", r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}")
+    wiring.require(run_id != window_id, "authority proof must separate run/window identities")
+    return {
+        "evidence_type": "measured",
+        "source_revision": source_revision,
+        "run_id": run_id,
+        "measurement_window_id": window_id,
+        "started_at": started_at,
+        "finished_at": finished_at,
+        "command": [
+            "scripts/run_scanner_heal_authority_evidence.py",
+            "--root-authority-json",
+            "<root-authority-json>",
+            "--quota-authority-json",
+            "<quota-authority-json>",
+        ],
+    }
+
+
+def validate_root_authority(payload: dict[str, Any]) -> None:
+    wiring.release_bundle_exact_strings(
+        payload.get("root_authority_cases"),
+        wiring.SCANNER_HEAL_RELEASE_G01_ROOT_AUTHORITY_CASES,
+        "root_authority_cases",
+    )
+    wiring.release_bundle_bool_true(payload.get("root_cas_observed"), "root_cas_observed")
+    wiring.release_bundle_bool_true(payload.get("root_readback_observed"), "root_readback_observed")
+    wiring.release_bundle_bool_true(payload.get("incomplete_root_rejected"), "incomplete_root_rejected")
+    wiring.release_bundle_bool_true(payload.get("stale_root_rejected"), "stale_root_rejected")
+
+
+def validate_quota_authority(payload: dict[str, Any]) -> None:
+    wiring.release_bundle_exact_strings(
+        payload.get("quota_authority_cases"),
+        wiring.SCANNER_HEAL_RELEASE_G01_QUOTA_AUTHORITY_CASES,
+        "quota_authority_cases",
+    )
+    wiring.release_bundle_bool_true(payload.get("quota_floor_readback_observed"), "quota_floor_readback_observed")
+    wiring.release_bundle_bool_true(payload.get("over_limit_put_rejected"), "over_limit_put_rejected")
+    wiring.release_bundle_bool_true(payload.get("rejected_object_invisible"), "rejected_object_invisible")
+    wiring.release_bundle_bool_true(payload.get("quota_fails_closed_without_authority"), "quota_fails_closed_without_authority")
+
+
+def write_field(out_dir: Path, field: str, common: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]:
+    evidence = {
+        **payload,
+        **common,
+        "evidence_type": "measured",
+        "summary": payload.get("summary") or f"Measured Scanner/Heal G01.{field} evidence.",
+    }
+    artifact = out_dir / "artifacts" / f"G01-{field}.json"
+    artifact.parent.mkdir(parents=True, exist_ok=True)
+    artifact_payload = {
+        "schema": 1,
+        "evidence_type": "measured",
+        "source_revision": evidence["source_revision"],
+        "run_id": evidence["run_id"],
+        "measurement_window_id": evidence["measurement_window_id"],
+        "gate": "G01",
+        "field": field,
+    }
+    for key, value in evidence.items():
+        if key not in {"artifact", "sha256", "artifact_format", "summary", "started_at", "finished_at", "command"}:
+            artifact_payload[key] = value
+    wiring.write_json(artifact, artifact_payload)
+    evidence["artifact"] = artifact.relative_to(out_dir).as_posix()
+    evidence["sha256"] = wiring.digest(artifact)
+    evidence["artifact_format"] = "json"
+    return evidence
+
+
+def build_descriptor(args: argparse.Namespace) -> Path:
+    out_dir = args.out_dir.resolve()
+    wiring.require(not out_dir.exists(), "output directory must be new")
+    source_revision = args.source_revision or git_head()
+    root_authority = measured_json(args.root_authority_json, source_revision, "root authority artifact")
+    quota_authority = measured_json(args.quota_authority_json, source_revision, "quota authority artifact")
+    validate_root_authority(root_authority)
+    validate_quota_authority(quota_authority)
+    out_dir.mkdir(parents=True)
+    common = common_evidence(args, source_revision)
+    fields = {
+        "root_authority_evidence": write_field(out_dir, "root_authority_evidence", common, root_authority),
+        "quota_authority_evidence": write_field(out_dir, "quota_authority_evidence", common, quota_authority),
+    }
+    descriptor = out_dir / "release-bundle-authority.json"
+    wiring.write_json(descriptor, {
+        "schema": 1,
+        "evidence": "measured",
+        "source_revision": source_revision,
+        "gates": {
+            "G01": {
+                "status": "pass",
+                "lane": "authority-coverage",
+                "evidence_type": "measured",
+                "evidence_fields": fields,
+            },
+        },
+    })
+    subprocess.check_call([
+        sys.executable,
+        str(ROOT / "scripts/check_test_wiring.py"),
+        "--check-scanner-heal-release-bundle-gate",
+        str(descriptor),
+        "G01",
+    ], cwd=ROOT)
+    return descriptor
+
+
+def write_self_test_inputs(root: Path, source_revision: str) -> tuple[Path, Path]:
+    root_json = root / "root-authority.json"
+    quota_json = root / "quota-authority.json"
+    wiring.write_json(root_json, {
+        "schema": 1,
+        "evidence_type": "measured",
+        "source_revision": source_revision,
+        "root_authority_cases": list(wiring.SCANNER_HEAL_RELEASE_G01_ROOT_AUTHORITY_CASES),
+        "root_cas_observed": True,
+        "root_readback_observed": True,
+        "incomplete_root_rejected": True,
+        "stale_root_rejected": True,
+    })
+    wiring.write_json(quota_json, {
+        "schema": 1,
+        "evidence_type": "measured",
+        "source_revision": source_revision,
+        "quota_authority_cases": list(wiring.SCANNER_HEAL_RELEASE_G01_QUOTA_AUTHORITY_CASES),
+        "quota_floor_readback_observed": True,
+        "over_limit_put_rejected": True,
+        "rejected_object_invisible": True,
+        "quota_fails_closed_without_authority": True,
+    })
+    return root_json, quota_json
+
+
+def run_self_test() -> None:
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        source_revision = git_head()
+        root_json, quota_json = write_self_test_inputs(root, source_revision)
+        descriptor = build_descriptor(parse_args([
+            "--root-authority-json", str(root_json),
+            "--quota-authority-json", str(quota_json),
+            "--out-dir", str(root / "out"),
+            "--duration-seconds", "60",
+        ]))
+        wiring.require(descriptor.is_file(), "self-test descriptor missing")
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        source_revision = git_head()
+        root_json, quota_json = write_self_test_inputs(root, source_revision)
+        payload = wiring.read_json(root_json)
+        payload["synthetic"] = True
+        wiring.write_json(root_json, payload)
+        try:
+            build_descriptor(parse_args([
+                "--root-authority-json", str(root_json),
+                "--quota-authority-json", str(quota_json),
+                "--out-dir", str(root / "out"),
+            ]))
+        except ValueError as err:
+            wiring.require("synthetic" in str(err), "wrong self-test failure for synthetic root authority")
+        else:
+            raise ValueError("self-test accepted synthetic root authority")
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        source_revision = git_head()
+        root_json, quota_json = write_self_test_inputs(root, source_revision)
+        payload = wiring.read_json(quota_json)
+        payload["quota_authority_cases"] = payload["quota_authority_cases"][:-1]
+        wiring.write_json(quota_json, payload)
+        try:
+            build_descriptor(parse_args([
+                "--root-authority-json", str(root_json),
+                "--quota-authority-json", str(quota_json),
+                "--out-dir", str(root / "out"),
+            ]))
+        except ValueError as err:
+            wiring.require("missing cases" in str(err), "wrong self-test failure for incomplete quota authority")
+        else:
+            raise ValueError("self-test accepted incomplete quota authority")
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--root-authority-json", type=Path)
+    parser.add_argument("--quota-authority-json", type=Path)
+    parser.add_argument("--out-dir", type=Path)
+    parser.add_argument("--source-revision")
+    parser.add_argument("--run-id")
+    parser.add_argument("--measurement-window-id")
+    parser.add_argument("--started-at")
+    parser.add_argument("--finished-at")
+    parser.add_argument("--duration-seconds", type=int, default=60)
+    parser.add_argument("--self-test", action="store_true")
+    args = parser.parse_args(argv)
+    if not args.self_test:
+        if args.root_authority_json is None:
+            parser.error("--root-authority-json is required unless --self-test is used")
+        if args.quota_authority_json is None:
+            parser.error("--quota-authority-json is required unless --self-test is used")
+        if args.out_dir is None:
+            parser.error("--out-dir is required unless --self-test is used")
+    return args
+
+
+def main() -> int:
+    try:
+        args = parse_args()
+        if args.self_test:
+            run_self_test()
+            return 0
+        descriptor = build_descriptor(args)
+        print(f"Authority release descriptor verified: {descriptor}")
+        return 0
+    except (ValueError, OSError, json.JSONDecodeError, subprocess.SubprocessError) as err:
+        print(f"ERROR: {err}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -212,7 +212,7 @@ def proof_from_case_directories(raw_values: list[str], out_dir: Path, source_rev
        "peer_count": oracle["peer_count"],
        "same_window_remote_proof": oracle["same_window_remote_proof"],
        "all_peers_bound_to_generation_window": oracle["all_peers_bound_to_generation_window"],
-        "case_evidence": copy_case_artifacts(out_dir, records, window_id, source_revision),
+        "case_evidence": copy_case_artifacts(out_dir, [selected], window_id, source_revision),
    }


@@ -408,6 +408,13 @@ def run_self_test() -> None:
            "--out-dir", str(root / "out"),
        ]))
        require(descriptor.is_file(), "self-test case-dir descriptor missing")
+        data = read_json(descriptor)
+        evidence = data["gates"]["G14"]["evidence_fields"]["same_window_field_evidence"]
+        artifact = descriptor.parent / evidence["artifact"]
+        case_evidence = read_json(artifact)["case_evidence"]
+        require(len(case_evidence) == 1, "self-test case-dir descriptor copied non-covering case evidence")
+        require(case_evidence[0]["case"] == "background-target-crash-ec8-4-multi-pool",
+                "self-test case-dir descriptor selected the wrong covering case")

    with tempfile.TemporaryDirectory() as tmp:
        root = Path(tmp)
@@ -0,0 +1,334 @@
+#!/usr/bin/env python3
+"""Assemble measured Scanner/Heal R-L legacy rollback release evidence.
+
+This producer consumes one operator-collected measured JSON proof. It packages
+legacy source-conflict, migration-gap, and crash-safe source-retirement evidence
+into the common release-bundle descriptor shape.
+"""
+
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timezone
+import json
+import re
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+import check_test_wiring as wiring
+
+ROOT = Path(__file__).resolve().parents[1]
+RL_FIELDS = (
+    "legacy_source_conflict_evidence",
+    "migration_gap_evidence",
+    "crash_safe_source_retirement_evidence",
+)
+
+
+def git_head() -> str:
+    return subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
+
+
+def timestamp(value: Any, name: str) -> str:
+    wiring.require(isinstance(value, str) and value.strip(), f"missing {name}")
+    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    wiring.require(parsed.tzinfo is not None, f"{name} must include timezone")
+    return parsed.isoformat().replace("+00:00", "Z")
+
+
+def measured_proof(path: Path, source_revision: str) -> dict[str, Any]:
+    proof = wiring.read_json(path)
+    wiring.require(proof.get("schema") == 1, "proof schema must be 1")
+    wiring.require(proof.get("evidence_type") == "measured", "proof must be measured")
+    for marker in ("fixture", "fixture_only", "dry_run", "synthetic"):
+        wiring.require(proof.get(marker) is not True, f"proof is {marker}")
+    wiring.require(proof.get("source_revision") == source_revision, "proof source revision mismatch")
+    started_at = timestamp(proof.get("started_at"), "proof.started_at")
+    finished_at = timestamp(proof.get("finished_at"), "proof.finished_at")
+    wiring.require(
+        datetime.fromisoformat(finished_at.replace("Z", "+00:00"))
+        >= datetime.fromisoformat(started_at.replace("Z", "+00:00")),
+        "proof timestamps are inverted",
+    )
+    wiring.evidence_string(proof.get("run_id"), "proof.run_id", r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}")
+    wiring.evidence_string(
+        proof.get("measurement_window_id"),
+        "proof.measurement_window_id",
+        r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}",
+    )
+    wiring.require(proof["measurement_window_id"] != proof["run_id"], "proof must separate run/window identities")
+    versions = proof.get("versions")
+    wiring.require(
+        isinstance(versions, list)
+        and len(set(versions)) >= 2
+        and all(isinstance(version, str) and re.fullmatch(r"[0-9a-f]{40}", version) for version in versions),
+        "proof requires two source revisions",
+    )
+    wiring.require(source_revision in versions, "proof versions omit tested source revision")
+    return proof
+
+
+def field_from_proof(proof: dict[str, Any], field: str) -> dict[str, Any]:
+    value = proof.get(field)
+    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",
+        )
+    evidence = {
+        **value,
+        "versions": proof["versions"],
+        "mixed_version_role": wiring.SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES[("R-L", field)],
+    }
+    if "crash_points" not in evidence:
+        evidence["crash_points"] = proof.get("crash_points")
+    wiring.validate_release_bundle_domain_evidence("R-L", field, evidence)
+    return evidence
+
+
+def write_field(out_dir: Path, common: dict[str, Any], field: str, field_evidence: dict[str, Any]) -> dict[str, Any]:
+    evidence = {
+        **field_evidence,
+        **common,
+        "evidence_type": "measured",
+        "summary": field_evidence.get("summary") or f"Measured Scanner/Heal R-L {field} evidence.",
+    }
+    artifact = out_dir / "artifacts" / f"R-L-{field}.json"
+    artifact.parent.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "schema": 1,
+        "evidence_type": "measured",
+        "source_revision": evidence["source_revision"],
+        "run_id": evidence["run_id"],
+        "measurement_window_id": evidence["measurement_window_id"],
+        "gate": "R-L",
+        "field": field,
+    }
+    for key, value in evidence.items():
+        if key not in {"artifact", "sha256", "artifact_format", "summary", "started_at", "finished_at", "command"}:
+            payload[key] = value
+    wiring.write_json(artifact, payload)
+    evidence["artifact"] = artifact.relative_to(out_dir).as_posix()
+    evidence["sha256"] = wiring.digest(artifact)
+    evidence["artifact_format"] = "json"
+    return evidence
+
+
+def build_descriptor(args: argparse.Namespace) -> Path:
+    out_dir = args.out_dir.resolve()
+    wiring.require(not out_dir.exists(), "output directory must be new")
+    source_revision = args.source_revision or git_head()
+    proof = measured_proof(args.proof_json.resolve(), source_revision)
+    out_dir.mkdir(parents=True)
+    common = {
+        "evidence_type": "measured",
+        "source_revision": source_revision,
+        "run_id": proof["run_id"],
+        "measurement_window_id": proof["measurement_window_id"],
+        "started_at": timestamp(proof["started_at"], "proof.started_at"),
+        "finished_at": timestamp(proof["finished_at"], "proof.finished_at"),
+        "command": [
+            "scripts/run_scanner_heal_legacy_rollback_evidence.py",
+            "--proof-json",
+            "<proof-json>",
+        ],
+    }
+    descriptor = out_dir / "release-bundle-legacy-rollback.json"
+    gates = {
+        "R-L": {
+            "status": "pass",
+            "lane": "mixed-version-rollback",
+            "evidence_type": "measured",
+            "evidence_fields": {
+                field: write_field(out_dir, common, field, field_from_proof(proof, field))
+                for field in RL_FIELDS
+            },
+        },
+    }
+    wiring.write_json(descriptor, {
+        "schema": 1,
+        "evidence": "measured",
+        "source_revision": source_revision,
+        "gates": gates,
+    })
+    subprocess.check_call([
+        sys.executable,
+        str(ROOT / "scripts/check_test_wiring.py"),
+        "--check-scanner-heal-release-bundle-gate",
+        str(descriptor),
+        "R-L",
+    ], cwd=ROOT)
+    return descriptor
+
+
+def write_self_test_proof(path: Path, source_revision: str) -> None:
+    now = datetime.now(timezone.utc).replace(microsecond=0)
+    proof = {
+        "schema": 1,
+        "evidence_type": "measured",
+        "source_revision": source_revision,
+        "run_id": f"legacy-rollback-{source_revision[:12]}",
+        "measurement_window_id": f"legacy-rollback-window-{source_revision[:12]}",
+        "started_at": now.isoformat().replace("+00:00", "Z"),
+        "finished_at": now.isoformat().replace("+00:00", "Z"),
+        "versions": ["a" * 40, source_revision],
+        "crash_points": ["before-successor-manifest", "after-successor-before-retire"],
+        "legacy_source_conflict_evidence": {
+            "legacy_source_conflict_cases": list(
+                wiring.SCANNER_HEAL_RELEASE_RL_REQUIRED_CASES["legacy_source_conflict_evidence"]
+            ),
+            "source_conflicts_rejected": True,
+            "takeover_identity_bound": True,
+            "legacy_checksum_gap_rejected": True,
+        },
+        "migration_gap_evidence": {
+            "migration_gap_cases": list(wiring.SCANNER_HEAL_RELEASE_RL_REQUIRED_CASES["migration_gap_evidence"]),
+            "migration_gap_closed": True,
+            "legacy_sources_fail_closed": True,
+            "prior_responsibilities_inherited": True,
+        },
+        "crash_safe_source_retirement_evidence": {
+            "source_retirement_cases": list(
+                wiring.SCANNER_HEAL_RELEASE_RL_REQUIRED_CASES["crash_safe_source_retirement_evidence"]
+            ),
+            "source_retirement_is_crash_safe": True,
+            "old_source_retained_until_successor": True,
+            "recovered_pending_migration": True,
+        },
+    }
+    wiring.write_json(path, proof)
+
+
+def run_self_test() -> None:
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        source_revision = git_head()
+        proof = root / "legacy-rollback-proof.json"
+        write_self_test_proof(proof, source_revision)
+        descriptor = build_descriptor(parse_args([
+            "--proof-json", str(proof),
+            "--out-dir", str(root / "out"),
+        ]))
+        wiring.require(descriptor.is_file(), "self-test descriptor missing")
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        source_revision = git_head()
+        proof = root / "legacy-rollback-proof.json"
+        write_self_test_proof(proof, source_revision)
+        payload = wiring.read_json(proof)
+        payload["migration_gap_evidence"]["migration_gap_closed"] = False
+        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("migration_gap_closed" in str(err), "wrong self-test failure for R-L proof")
+        else:
+            raise ValueError("self-test accepted incomplete R-L proof")
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        source_revision = git_head()
+        proof = root / "legacy-rollback-proof.json"
+        write_self_test_proof(proof, source_revision)
+        payload = wiring.read_json(proof)
+        payload["evidence_type"] = "synthetic"
+        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("measured" in str(err), "wrong self-test failure for synthetic proof")
+        else:
+            raise ValueError("self-test accepted synthetic proof")
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        source_revision = git_head()
+        proof = root / "legacy-rollback-proof.json"
+        write_self_test_proof(proof, source_revision)
+        payload = wiring.read_json(proof)
+        payload["versions"] = ["not-a-source-revision".ljust(40, "x"), source_revision]
+        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 revisions" in str(err), "wrong self-test failure for invalid version")
+        else:
+            raise ValueError("self-test accepted invalid source revision")
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        source_revision = git_head()
+        proof = root / "legacy-rollback-proof.json"
+        write_self_test_proof(proof, source_revision)
+        payload = wiring.read_json(proof)
+        payload["finished_at"] = "2026-09-08T00:00:00Z"
+        payload["started_at"] = "2026-09-09T00:00:00Z"
+        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("timestamps are inverted" in str(err), "wrong self-test failure for inverted timestamps")
+        else:
+            raise ValueError("self-test accepted inverted timestamps")
+
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        source_revision = git_head()
+        proof = root / "legacy-rollback-proof.json"
+        write_self_test_proof(proof, source_revision)
+        payload = wiring.read_json(proof)
+        payload["legacy_source_conflict_evidence"]["run_id"] = "legacy-rollback-different-run"
+        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("run_id mismatch" in str(err), "wrong self-test failure for nested run_id")
+        else:
+            raise ValueError("self-test accepted nested run_id mismatch")
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--proof-json", type=Path)
+    parser.add_argument("--out-dir", type=Path)
+    parser.add_argument("--source-revision")
+    parser.add_argument("--self-test", action="store_true")
+    args = parser.parse_args(argv)
+    if not args.self_test:
+        if args.proof_json is None:
+            parser.error("--proof-json is required unless --self-test is used")
+        if args.out_dir is None:
+            parser.error("--out-dir is required unless --self-test is used")
+    return args
+
+
+def main() -> int:
+    try:
+        args = parse_args()
+        if args.self_test:
+            run_self_test()
+            return 0
+        descriptor = build_descriptor(args)
+        print(f"Legacy rollback release descriptor verified: {descriptor}")
+        return 0
+    except (ValueError, OSError, json.JSONDecodeError, subprocess.SubprocessError) as err:
+        print(f"ERROR: {err}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,425 @@
+#!/usr/bin/env python3
+"""Assemble measured Scanner/Heal G07/G08/P4 MRF release evidence.
+
+The producer consumes raw W13 MRF JSON artifacts emitted by the measured Rust
+test and packages them into the common release-bundle descriptor shape. It is
+usable both by the W13 shell runner and by operators who need to re-check an
+already collected MRF evidence directory.
+"""
+
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timedelta, timezone
+import json
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+import check_test_wiring as wiring
+
+ROOT = Path(__file__).resolve().parents[1]
+FIELD_ARTIFACTS = {
+    "G07": {
+        "mrf_responsibility_oracle": "g07-mrf-responsibility/G07-mrf_responsibility_oracle.json",
+        "commit_boundary_crash_matrix": "g07-mrf-responsibility/G07-commit_boundary_crash_matrix.json",
+    },
+    "G08": {
+        "mrf_capacity_evidence": "g08-mrf-capacity/G08-mrf_capacity_evidence.json",
+        "disk_full_matrix": "g08-mrf-capacity/G08-disk_full_matrix.json",
+        "replica_loss_matrix": "g08-mrf-capacity/G08-replica_loss_matrix.json",
+    },
+    "P4": {
+        "mrf_scale_measurement": "p4-mrf-soak/P4-mrf_scale_measurement.json",
+        "mrf_replay_cost_measurement": "p4-mrf-soak/P4-mrf_replay_cost_measurement.json",
+        "retained_responsibility_evidence": "p4-mrf-soak/P4-retained_responsibility_evidence.json",
+        "mrf_cleanup_gc_soak_evidence": "p4-mrf-soak/P4-mrf_cleanup_gc_soak_evidence.json",
+    },
+}
+SELECTION_GATES = {
+    "all": ("G07", "G08", "P4"),
+    "g07": ("G07",),
+    "g08": ("G08",),
+    "p4": ("P4",),
+}
+
+
+def git_head() -> str:
+    return subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
+
+
+def source_revision_for(run_dir: Path, override: str | None) -> str:
+    if override:
+        return override
+    stamp = run_dir / "source-revision.txt"
+    if stamp.is_file():
+        value = stamp.read_text().strip()
+        wiring.require(value, "source-revision.txt is empty")
+        return value
+    return git_head()
+
+
+def timestamp(value: Any, name: str) -> str:
+    wiring.require(isinstance(value, str) and value.strip(), f"missing {name}")
+    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    wiring.require(parsed.tzinfo is not None, f"{name} must include timezone")
+    return parsed.isoformat().replace("+00:00", "Z")
+
+
+def selected_artifacts(selection: str) -> dict[str, dict[str, Path]]:
+    return {
+        gate: {field: Path(relative) for field, relative in FIELD_ARTIFACTS[gate].items()}
+        for gate in SELECTION_GATES[selection]
+    }
+
+
+def descriptor_path_for(run_dir: Path, output: Path | None) -> Path:
+    if output is not None:
+        return output.resolve()
+    return run_dir / "release-bundle-w13.json"
+
+
+def relative_to_descriptor(path: Path, descriptor: Path) -> str:
+    return path.resolve(strict=True).relative_to(descriptor.parent.resolve()).as_posix()
+
+
+def measured_payload(path: Path, source_revision: str, gate: str, field: str) -> dict[str, Any]:
+    payload = wiring.read_json(path.resolve())
+    wiring.require(isinstance(payload, dict), f"{gate}.{field} artifact must be a JSON object")
+    for marker in ("fixture", "fixture_only", "dry_run", "synthetic"):
+        wiring.require(payload.get(marker) is not True, f"{gate}.{field} artifact is {marker}")
+    wiring.require(payload.get("schema") == 1, f"{gate}.{field} artifact schema must be 1")
+    wiring.require(payload.get("evidence_type") == "measured", f"{gate}.{field} artifact must be measured")
+    wiring.require(payload.get("source_revision") == source_revision, f"{gate}.{field} source revision mismatch")
+    wiring.require(payload.get("gate") == gate, f"{gate}.{field} artifact gate mismatch")
+    wiring.require(payload.get("field") == field, f"{gate}.{field} artifact field mismatch")
+    for key in ("run_id", "measurement_window_id"):
+        value = payload.get(key)
+        wiring.require(isinstance(value, str) and value.strip(), f"{gate}.{field} missing {key}")
+    expected_kind = wiring.SCANNER_HEAL_RELEASE_MRF_ARTIFACT_KINDS[(gate, field)]
+    wiring.require(payload.get("artifact_kind") == expected_kind, f"{gate}.{field} artifact kind mismatch")
+    timestamp(payload.get("started_at"), f"{gate}.{field}.started_at")
+    timestamp(payload.get("finished_at"), f"{gate}.{field}.finished_at")
+    command = payload.get("command")
+    wiring.require(
+        isinstance(command, list) and command and all(isinstance(item, str) and item.strip() for item in command),
+        f"{gate}.{field} artifact missing command provenance",
+    )
+    summary = payload.get("summary")
+    wiring.require(isinstance(summary, str) and summary.strip(), f"{gate}.{field} artifact missing summary")
+    if gate == "G08" and field == "disk_full_matrix":
+        for observed in (
+            "journal_write_enospc_observed",
+            "committed_checkpoint_enospc_observed",
+            "cleanup_delete_on_full_filesystem_observed",
+        ):
+            wiring.release_bundle_bool_true(payload.get(observed), f"{gate}.{field}.{observed}")
+        wiring.evidence_integer(payload.get("enospc_filler_bytes"), f"{gate}.{field}.enospc_filler_bytes", 1, 2**63 - 1)
+    return payload
+
+
+def validate_mrf_field(gate: str, field: str, evidence: dict[str, Any]) -> None:
+    if gate == "G07":
+        case_field = {
+            "mrf_responsibility_oracle": "mrf_responsibility_cases",
+            "commit_boundary_crash_matrix": "commit_crash_cases",
+        }[field]
+        wiring.release_bundle_exact_strings(
+            evidence.get(case_field),
+            wiring.SCANNER_HEAL_RELEASE_G07_REQUIRED_CASES[field],
+            f"{gate}.{field}.{case_field}",
+        )
+        wiring.evidence_integer(evidence.get("replayed_records"), f"{gate}.{field}.replayed_records", 1, 2**63 - 1)
+        wiring.release_bundle_bool_true(
+            evidence.get("responsibility_anchor_retained"),
+            f"{gate}.{field}.responsibility_anchor_retained",
+        )
+        wiring.release_bundle_bool_true(
+            evidence.get("successor_snapshot_published"),
+            f"{gate}.{field}.successor_snapshot_published",
+        )
+    if gate == "G08":
+        case_field = {
+            "mrf_capacity_evidence": "capacity_cases",
+            "disk_full_matrix": "disk_full_cases",
+            "replica_loss_matrix": "replica_loss_cases",
+        }[field]
+        wiring.release_bundle_exact_strings(
+            evidence.get(case_field),
+            wiring.SCANNER_HEAL_RELEASE_G08_REQUIRED_CASES[field],
+            f"{gate}.{field}.{case_field}",
+        )
+    if gate == "P4":
+        duration = wiring.evidence_integer(evidence.get("duration_seconds"), f"{gate}.{field}.duration_seconds", 1, 86400)
+        wiring.require(duration >= 900, f"{gate}.{field} requires at least 900 seconds")
+        if field != "mrf_scale_measurement":
+            wiring.evidence_integer(evidence.get("replayed_records"), f"{gate}.{field}.replayed_records", 1, 2**63 - 1)
+            wiring.release_bundle_bool_true(
+                evidence.get("responsibility_anchor_retained"),
+                f"{gate}.{field}.responsibility_anchor_retained",
+            )
+            wiring.release_bundle_bool_true(
+                evidence.get("successor_snapshot_published"),
+                f"{gate}.{field}.successor_snapshot_published",
+            )
+        if field == "retained_responsibility_evidence":
+            wiring.release_bundle_exact_strings(
+                evidence.get("retained_responsibility_cases"),
+                wiring.SCANNER_HEAL_RELEASE_P4_RETAINED_RESPONSIBILITY_CASES,
+                f"{gate}.{field}.retained_responsibility_cases",
+            )
+            retention_window = wiring.evidence_integer(
+                evidence.get("retention_window_seconds"),
+                f"{gate}.{field}.retention_window_seconds",
+                7200,
+                86400,
+            )
+            wiring.require(duration >= retention_window, f"{gate}.{field} duration must cover retention window")
+            wiring.release_bundle_bool_true(evidence.get("idle_cleanup_observed"), f"{gate}.{field}.idle_cleanup_observed")
+            wiring.release_bundle_bool_true(
+                evidence.get("verified_proof_discharge_observed"),
+                f"{gate}.{field}.verified_proof_discharge_observed",
+            )
+        if field == "mrf_cleanup_gc_soak_evidence":
+            wiring.require(duration >= 7200, f"{gate}.{field} requires at least two hours")
+            wiring.release_bundle_exact_strings(
+                evidence.get("cleanup_gc_cases"),
+                wiring.SCANNER_HEAL_RELEASE_MRF_CLEANUP_GC_SOAK_CASES,
+                f"{gate}.{field}.cleanup_gc_cases",
+            )
+            wiring.release_bundle_bool_true(
+                evidence.get("verified_idle_gc_observed"),
+                f"{gate}.{field}.verified_idle_gc_observed",
+            )
+            wiring.require(evidence.get("pending_responsibilities_after_gc") == 0,
+                           f"{gate}.{field} requires zero pending responsibilities after GC")
+            wiring.require(evidence.get("stale_journals_after_gc") == 0,
+                           f"{gate}.{field} requires zero stale journals after GC")
+
+
+def field_evidence(run_dir: Path, descriptor: Path, source_revision: str, gate: str, field: str, relative: Path) -> dict[str, Any]:
+    artifact = run_dir / relative
+    payload = measured_payload(artifact, source_revision, gate, field)
+    evidence = {
+        "artifact": relative_to_descriptor(artifact, descriptor),
+        "sha256": wiring.digest(artifact),
+        "evidence_type": "measured",
+        "source_revision": source_revision,
+        "run_id": payload["run_id"],
+        "measurement_window_id": payload["measurement_window_id"],
+        "started_at": payload["started_at"],
+        "finished_at": payload["finished_at"],
+        "command": payload["command"],
+        "artifact_format": "json",
+        "summary": payload["summary"],
+    }
+    for mirror in wiring.release_bundle_json_artifact_mirrored_fields(gate, field):
+        wiring.require(mirror in payload, f"{gate}.{field} artifact missing {mirror}")
+        evidence[mirror] = payload[mirror]
+    if gate == "P4":
+        evidence["duration_seconds"] = payload["duration_seconds"]
+    validate_mrf_field(gate, field, evidence)
+    return evidence
+
+
+def build_descriptor(args: argparse.Namespace) -> Path:
+    run_dir = args.run_dir.resolve()
+    wiring.require(run_dir.is_dir(), "run directory is missing")
+    descriptor = descriptor_path_for(run_dir, args.out_file)
+    wiring.require(descriptor.parent == run_dir or descriptor.parent.is_relative_to(run_dir),
+                   "descriptor must be written under the run directory")
+    source_revision = source_revision_for(run_dir, args.source_revision)
+    registry = wiring.read_json(ROOT / ".config/scanner-heal-required-tests.json")
+    requirements = {item["gate"]: item for item in registry["release_requirements"]}
+    gates: dict[str, Any] = {}
+    for gate, artifacts in selected_artifacts(args.test).items():
+        gates[gate] = {
+            "status": "pass",
+            "lane": requirements[gate]["lane"],
+            "evidence_type": "measured",
+            "evidence_fields": {
+                field: field_evidence(run_dir, descriptor, source_revision, gate, field, relative)
+                for field, relative in artifacts.items()
+            },
+        }
+    wiring.write_json(descriptor, {
+        "schema": 1,
+        "evidence": "measured",
+        "source_revision": source_revision,
+        "gates": gates,
+    })
+    for gate in gates:
+        subprocess.check_call([
+            sys.executable,
+            str(ROOT / "scripts/check_test_wiring.py"),
+            "--check-scanner-heal-release-bundle-gate",
+            str(descriptor),
+            gate,
+        ], cwd=ROOT)
+    return descriptor
+
+
+def base_payload(source_revision: str, gate: str, field: str, duration_seconds: int = 7200) -> dict[str, Any]:
+    started = datetime.now(timezone.utc).replace(microsecond=0)
+    finished = started + timedelta(seconds=duration_seconds)
+    return {
+        "schema": 1,
+        "evidence_type": "measured",
+        "source_revision": source_revision,
+        "run_id": f"w13-mrf-{gate.lower()}-{source_revision[:12]}",
+        "measurement_window_id": f"w13-mrf-window-{gate.lower()}-{source_revision[:12]}",
+        "started_at": started.isoformat().replace("+00:00", "Z"),
+        "finished_at": finished.isoformat().replace("+00:00", "Z"),
+        "command": ["scripts/run_scanner_heal_w13_mrf_evidence.sh", "--test", gate.lower()],
+        "summary": f"Measured Scanner/Heal {gate}.{field} MRF evidence.",
+        "gate": gate,
+        "field": field,
+        "artifact_kind": wiring.SCANNER_HEAL_RELEASE_MRF_ARTIFACT_KINDS[(gate, field)],
+    }
+
+
+def write_self_test_artifacts(run_dir: Path, source_revision: str) -> None:
+    for gate, artifacts in selected_artifacts("all").items():
+        for field, relative in artifacts.items():
+            payload = base_payload(source_revision, gate, field)
+            if gate == "G07":
+                payload["crash_points"] = ["before-commit", "after-commit"]
+                payload["replayed_records"] = 2
+                payload["responsibility_anchor_retained"] = True
+                payload["successor_snapshot_published"] = True
+                if field == "mrf_responsibility_oracle":
+                    payload["mrf_responsibility_cases"] = list(wiring.SCANNER_HEAL_RELEASE_G07_REQUIRED_CASES[field])
+                else:
+                    payload["commit_crash_cases"] = list(wiring.SCANNER_HEAL_RELEASE_G07_REQUIRED_CASES[field])
+            elif gate == "G08":
+                case_field = {
+                    "mrf_capacity_evidence": "capacity_cases",
+                    "disk_full_matrix": "disk_full_cases",
+                    "replica_loss_matrix": "replica_loss_cases",
+                }[field]
+                payload[case_field] = list(wiring.SCANNER_HEAL_RELEASE_G08_REQUIRED_CASES[field])
+                if field == "disk_full_matrix":
+                    payload.update({
+                        "journal_write_enospc_observed": True,
+                        "committed_checkpoint_enospc_observed": True,
+                        "cleanup_delete_on_full_filesystem_observed": True,
+                        "enospc_filler_bytes": 1024,
+                    })
+            elif gate == "P4":
+                payload["duration_seconds"] = 7200
+                if field != "mrf_scale_measurement":
+                    payload["replayed_records"] = 2
+                    payload["responsibility_anchor_retained"] = True
+                    payload["successor_snapshot_published"] = True
+                if field == "retained_responsibility_evidence":
+                    payload.update({
+                        "retained_responsibility_cases": list(wiring.SCANNER_HEAL_RELEASE_P4_RETAINED_RESPONSIBILITY_CASES),
+                        "retention_window_seconds": 7200,
+                        "idle_cleanup_observed": True,
+                        "verified_proof_discharge_observed": True,
+                    })
+                if field == "mrf_cleanup_gc_soak_evidence":
+                    payload.update({
+                        "cleanup_gc_cases": list(wiring.SCANNER_HEAL_RELEASE_MRF_CLEANUP_GC_SOAK_CASES),
+                        "verified_idle_gc_observed": True,
+                        "pending_responsibilities_after_gc": 0,
+                        "stale_journals_after_gc": 0,
+                    })
+            path = run_dir / relative
+            path.parent.mkdir(parents=True, exist_ok=True)
+            wiring.write_json(path, payload)
+
+
+def expect_failure(args: list[str], needle: str) -> None:
+    try:
+        build_descriptor(parse_args(args))
+    except (ValueError, subprocess.CalledProcessError) as err:
+        wiring.require(needle in str(err), f"wrong self-test failure: {err}")
+    else:
+        raise ValueError("self-test accepted invalid MRF evidence")
+
+
+def run_self_test() -> None:
+    import tempfile
+
+    source_revision = git_head()
+    with tempfile.TemporaryDirectory() as tmp:
+        run_dir = Path(tmp) / "run"
+        run_dir.mkdir()
+        (run_dir / "source-revision.txt").write_text(source_revision + "\n")
+        write_self_test_artifacts(run_dir, source_revision)
+        descriptor = build_descriptor(parse_args(["--run-dir", str(run_dir)]))
+        wiring.require(descriptor.is_file(), "self-test descriptor missing")
+
+    with tempfile.TemporaryDirectory() as tmp:
+        run_dir = Path(tmp) / "run"
+        run_dir.mkdir()
+        write_self_test_artifacts(run_dir, source_revision)
+        path = run_dir / FIELD_ARTIFACTS["G07"]["mrf_responsibility_oracle"]
+        payload = wiring.read_json(path)
+        payload["mrf_responsibility_cases"] = payload["mrf_responsibility_cases"][:-1]
+        wiring.write_json(path, payload)
+        expect_failure(["--run-dir", str(run_dir), "--source-revision", source_revision, "--test", "g07"], "missing cases")
+
+    with tempfile.TemporaryDirectory() as tmp:
+        run_dir = Path(tmp) / "run"
+        run_dir.mkdir()
+        write_self_test_artifacts(run_dir, source_revision)
+        path = run_dir / FIELD_ARTIFACTS["G08"]["disk_full_matrix"]
+        payload = wiring.read_json(path)
+        payload["journal_write_enospc_observed"] = False
+        wiring.write_json(path, payload)
+        expect_failure(["--run-dir", str(run_dir), "--source-revision", source_revision, "--test", "g08"], "journal_write_enospc_observed")
+
+    with tempfile.TemporaryDirectory() as tmp:
+        run_dir = Path(tmp) / "run"
+        run_dir.mkdir()
+        write_self_test_artifacts(run_dir, source_revision)
+        path = run_dir / FIELD_ARTIFACTS["P4"]["mrf_cleanup_gc_soak_evidence"]
+        payload = wiring.read_json(path)
+        payload["duration_seconds"] = 900
+        payload["finished_at"] = payload["started_at"]
+        wiring.write_json(path, payload)
+        expect_failure(["--run-dir", str(run_dir), "--source-revision", source_revision, "--test", "p4"], "two hours")
+
+    with tempfile.TemporaryDirectory() as tmp:
+        run_dir = Path(tmp) / "run"
+        run_dir.mkdir()
+        write_self_test_artifacts(run_dir, source_revision)
+        path = run_dir / FIELD_ARTIFACTS["G08"]["replica_loss_matrix"]
+        payload = wiring.read_json(path)
+        payload["synthetic"] = True
+        wiring.write_json(path, payload)
+        expect_failure(["--run-dir", str(run_dir), "--source-revision", source_revision, "--test", "g08"], "synthetic")
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--run-dir", type=Path)
+    parser.add_argument("--out-file", type=Path)
+    parser.add_argument("--source-revision")
+    parser.add_argument("--test", choices=tuple(SELECTION_GATES), default="all")
+    parser.add_argument("--self-test", action="store_true")
+    args = parser.parse_args(argv)
+    if not args.self_test and args.run_dir is None:
+        parser.error("--run-dir is required unless --self-test is used")
+    return args
+
+
+def main() -> int:
+    try:
+        args = parse_args()
+        if args.self_test:
+            run_self_test()
+            return 0
+        descriptor = build_descriptor(args)
+        print(f"Scanner/Heal MRF release descriptor verified: {descriptor}")
+        return 0
+    except (ValueError, OSError, json.JSONDecodeError, subprocess.SubprocessError) as err:
+        print(f"ERROR: {err}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,296 @@
+#!/usr/bin/env python3
+"""Assemble measured Scanner/Heal G03 scoped ACK release evidence.
+
+This producer consumes one operator-collected measured JSON proof. It packages
+durable root publication, exact scoped ACK identity, peer capability, and
+mixed-peer fallback evidence into the common release-bundle descriptor shape.
+"""
+
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timezone
+import json
+import re
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+import check_test_wiring as wiring
+
+ROOT = Path(__file__).resolve().parents[1]
+G03_FIELDS = (
+    "durable_root_publication_proof",
+    "scoped_ack_request_identity",
+    "participating_peer_capability_snapshot",
+    "mixed_peer_ack_fallback_oracle",
+)
+
+
+def git_head() -> str:
+    return subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
+
+
+def timestamp(value: Any, name: str) -> str:
+    wiring.require(isinstance(value, str) and value.strip(), f"missing {name}")
+    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    wiring.require(parsed.tzinfo is not None, f"{name} must include timezone")
+    return parsed.isoformat().replace("+00:00", "Z")
+
+
+def measured_proof(path: Path, source_revision: str) -> dict[str, Any]:
+    proof = wiring.read_json(path)
+    wiring.require(proof.get("schema") == 1, "proof schema must be 1")
+    wiring.require(proof.get("evidence_type") == "measured", "proof must be measured")
+    for marker in ("fixture", "fixture_only", "dry_run", "synthetic"):
+        wiring.require(proof.get(marker) is not True, f"proof is {marker}")
+    wiring.require(proof.get("source_revision") == source_revision, "proof source revision mismatch")
+    started_at = timestamp(proof.get("started_at"), "proof.started_at")
+    finished_at = timestamp(proof.get("finished_at"), "proof.finished_at")
+    wiring.require(
+        datetime.fromisoformat(finished_at.replace("Z", "+00:00"))
+        >= datetime.fromisoformat(started_at.replace("Z", "+00:00")),
+        "proof timestamps are inverted",
+    )
+    wiring.evidence_string(proof.get("run_id"), "proof.run_id", r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}")
+    wiring.evidence_string(
+        proof.get("measurement_window_id"),
+        "proof.measurement_window_id",
+        r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}",
+    )
+    wiring.require(proof["measurement_window_id"] != proof["run_id"], "proof must separate run/window identities")
+    versions = proof.get("versions")
+    wiring.require(
+        isinstance(versions, list)
+        and len(set(versions)) >= 2
+        and all(isinstance(version, str) and re.fullmatch(r"[0-9a-f]{40}", version) for version in versions),
+        "proof requires two source revisions",
+    )
+    wiring.require(source_revision in versions, "proof versions omit tested source revision")
+    return proof
+
+
+def field_from_proof(proof: dict[str, Any], field: str) -> dict[str, Any]:
+    value = proof.get(field)
+    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",
+        )
+    evidence = {
+        **value,
+        "versions": proof["versions"],
+        "mixed_version_role": wiring.SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES[("G03", field)],
+    }
+    wiring.validate_release_bundle_domain_evidence("G03", field, evidence)
+    return evidence
+
+
+def write_field(out_dir: Path, common: dict[str, Any], field: str, field_evidence: dict[str, Any]) -> dict[str, Any]:
+    evidence = {
+        **field_evidence,
+        **common,
+        "evidence_type": "measured",
+        "summary": field_evidence.get("summary") or f"Measured Scanner/Heal G03 {field} evidence.",
+    }
+    artifact = out_dir / "artifacts" / f"G03-{field}.json"
+    artifact.parent.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "schema": 1,
+        "evidence_type": "measured",
+        "source_revision": evidence["source_revision"],
+        "run_id": evidence["run_id"],
+        "measurement_window_id": evidence["measurement_window_id"],
+        "gate": "G03",
+        "field": field,
+    }
+    for key, value in evidence.items():
+        if key not in {"artifact", "sha256", "artifact_format", "summary", "started_at", "finished_at", "command"}:
+            payload[key] = value
+    wiring.write_json(artifact, payload)
+    evidence["artifact"] = artifact.relative_to(out_dir).as_posix()
+    evidence["sha256"] = wiring.digest(artifact)
+    evidence["artifact_format"] = "json"
+    return evidence
+
+
+def build_descriptor(args: argparse.Namespace) -> Path:
+    out_dir = args.out_dir.resolve()
+    wiring.require(not out_dir.exists(), "output directory must be new")
+    source_revision = args.source_revision or git_head()
+    proof = measured_proof(args.proof_json.resolve(), source_revision)
+    out_dir.mkdir(parents=True)
+    common = {
+        "evidence_type": "measured",
+        "source_revision": source_revision,
+        "run_id": proof["run_id"],
+        "measurement_window_id": proof["measurement_window_id"],
+        "started_at": timestamp(proof["started_at"], "proof.started_at"),
+        "finished_at": timestamp(proof["finished_at"], "proof.finished_at"),
+        "command": [
+            "scripts/run_scanner_heal_scoped_ack_evidence.py",
+            "--proof-json",
+            "<proof-json>",
+        ],
+    }
+    descriptor = out_dir / "release-bundle-scoped-ack.json"
+    gates = {
+        "G03": {
+            "status": "pass",
+            "lane": "mixed-version-rollback",
+            "evidence_type": "measured",
+            "evidence_fields": {
+                field: write_field(out_dir, common, field, field_from_proof(proof, field))
+                for field in G03_FIELDS
+            },
+        },
+    }
+    wiring.write_json(descriptor, {
+        "schema": 1,
+        "evidence": "measured",
+        "source_revision": source_revision,
+        "gates": gates,
+    })
+    subprocess.check_call([
+        sys.executable,
+        str(ROOT / "scripts/check_test_wiring.py"),
+        "--check-scanner-heal-release-bundle-gate",
+        str(descriptor),
+        "G03",
+    ], cwd=ROOT)
+    return descriptor
+
+
+def write_self_test_proof(path: Path, source_revision: str) -> None:
+    now = datetime.now(timezone.utc).replace(microsecond=0)
+    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]}",
+        "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"]),
+        },
+    }
+    wiring.write_json(path, proof)
+
+
+def run_self_test() -> None:
+    import tempfile
+
+    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)
+        descriptor = build_descriptor(parse_args([
+            "--proof-json", str(proof),
+            "--out-dir", str(root / "out"),
+        ]))
+        wiring.require(descriptor.is_file(), "self-test descriptor missing")
+
+    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["scoped_ack_request_identity"]["scoped_ack_cases"].remove("exact-generation")
+        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("missing cases" in str(err), "wrong self-test failure for scoped ACK cases")
+        else:
+            raise ValueError("self-test accepted incomplete scoped ACK 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["durable_root_publication_proof"]["root_readback_observed"] = False
+        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("root_readback_observed" in str(err), "wrong self-test failure for root readback")
+        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["evidence_type"] = "synthetic"
+        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("measured" in str(err), "wrong self-test failure for synthetic proof")
+        else:
+            raise ValueError("self-test accepted synthetic proof")
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--proof-json", type=Path)
+    parser.add_argument("--out-dir", type=Path)
+    parser.add_argument("--source-revision")
+    parser.add_argument("--self-test", action="store_true")
+    args = parser.parse_args(argv)
+    if not args.self_test:
+        if args.proof_json is None:
+            parser.error("--proof-json is required unless --self-test is used")
+        if args.out_dir is None:
+            parser.error("--out-dir is required unless --self-test is used")
+    return args
+
+
+def main() -> int:
+    try:
+        args = parse_args()
+        if args.self_test:
+            run_self_test()
+            return 0
+        descriptor = build_descriptor(args)
+        print(f"Scoped ACK release descriptor verified: {descriptor}")
+        return 0
+    except (ValueError, OSError, json.JSONDecodeError, subprocess.SubprocessError) as err:
+        print(f"ERROR: {err}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -305,104 +305,13 @@ PY

 write_release_descriptor() {
    local source_revision="$1"
-    "$PYTHON_BIN" - "$ROOT" "$RUN_DIR" "$source_revision" "$TEST_SELECTION" <<'PY'
-import hashlib
-import json
-import pathlib
-import sys
-
-root = pathlib.Path(sys.argv[1])
-run_dir = pathlib.Path(sys.argv[2])
-source_revision = sys.argv[3]
-selection = sys.argv[4]
-descriptor = run_dir / "release-bundle-w13.json"
-registry = json.loads((root / ".config/scanner-heal-required-tests.json").read_text())
-requirements = {item["gate"]: item for item in registry["release_requirements"]}
-artifacts = {
-    "G07": {
-        "mrf_responsibility_oracle": run_dir / "g07-mrf-responsibility" / "G07-mrf_responsibility_oracle.json",
-        "commit_boundary_crash_matrix": run_dir / "g07-mrf-responsibility" / "G07-commit_boundary_crash_matrix.json",
-    },
-    "G08": {
-        "mrf_capacity_evidence": run_dir / "g08-mrf-capacity" / "G08-mrf_capacity_evidence.json",
-        "disk_full_matrix": run_dir / "g08-mrf-capacity" / "G08-disk_full_matrix.json",
-        "replica_loss_matrix": run_dir / "g08-mrf-capacity" / "G08-replica_loss_matrix.json",
-    },
-    "P4": {
-        "mrf_scale_measurement": run_dir / "p4-mrf-soak" / "P4-mrf_scale_measurement.json",
-        "mrf_replay_cost_measurement": run_dir / "p4-mrf-soak" / "P4-mrf_replay_cost_measurement.json",
-        "retained_responsibility_evidence": run_dir / "p4-mrf-soak" / "P4-retained_responsibility_evidence.json",
-        "mrf_cleanup_gc_soak_evidence": run_dir / "p4-mrf-soak" / "P4-mrf_cleanup_gc_soak_evidence.json",
-    },
-}
-if selection == "g07":
-    artifacts = {"G07": artifacts["G07"]}
-elif selection == "g08":
-    artifacts = {"G08": artifacts["G08"]}
-elif selection == "p4":
-    artifacts = {"P4": artifacts["P4"]}
-
-mirrors = {
-    ("G07", "mrf_responsibility_oracle"): ("crash_points", "mrf_responsibility_cases", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
-    ("G07", "commit_boundary_crash_matrix"): ("crash_points", "commit_crash_cases", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
-    ("G08", "mrf_capacity_evidence"): ("capacity_cases",),
-    ("G08", "disk_full_matrix"): ("disk_full_cases",),
-    ("G08", "replica_loss_matrix"): ("replica_loss_cases",),
-    ("P4", "mrf_replay_cost_measurement"): ("duration_seconds", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
-    ("P4", "retained_responsibility_evidence"): ("duration_seconds", "retained_responsibility_cases", "retention_window_seconds", "idle_cleanup_observed", "verified_proof_discharge_observed", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
-    ("P4", "mrf_cleanup_gc_soak_evidence"): ("duration_seconds", "cleanup_gc_cases", "verified_idle_gc_observed", "pending_responsibilities_after_gc", "stale_journals_after_gc", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
-}
-
-def digest(path: pathlib.Path) -> str:
-    hasher = hashlib.sha256()
-    with path.open("rb") as source:
-        for chunk in iter(lambda: source.read(1024 * 1024), b""):
-            hasher.update(chunk)
-    return hasher.hexdigest()
-
-def relative_to_descriptor(path: pathlib.Path) -> str:
-    return path.resolve(strict=True).relative_to(descriptor.parent.resolve()).as_posix()
-
-gates: dict[str, object] = {}
-for gate, gate_artifacts in artifacts.items():
-    fields: dict[str, object] = {}
-    for field, artifact in gate_artifacts.items():
-        payload = json.loads(artifact.read_text())
-        if payload.get("source_revision") != source_revision:
-            raise SystemExit(f"{gate}.{field}: source revision does not match this checkout")
-        evidence = {
-            "artifact": relative_to_descriptor(artifact),
-            "sha256": digest(artifact),
-            "evidence_type": "measured",
-            "source_revision": source_revision,
-            "run_id": payload["run_id"],
-            "measurement_window_id": payload["measurement_window_id"],
-            "started_at": payload["started_at"],
-            "finished_at": payload["finished_at"],
-            "command": payload["command"],
-            "artifact_format": "json",
-            "summary": payload["summary"],
-        }
-        for mirror in mirrors.get((gate, field), ()):
-            evidence[mirror] = payload[mirror]
-        if gate == "P4" and field == "mrf_scale_measurement":
-            evidence["duration_seconds"] = payload["duration_seconds"]
-        fields[field] = evidence
-    gates[gate] = {
-        "status": "pass",
-        "lane": requirements[gate]["lane"],
-        "evidence_type": "measured",
-        "evidence_fields": fields,
-    }
-
-descriptor.write_text(json.dumps({
-    "schema": 1,
-    "evidence": "measured",
-    "source_revision": source_revision,
-    "gates": gates,
-}, indent=2, sort_keys=True) + "\n")
-print(descriptor)
-PY
+    local descriptor="$RUN_DIR/release-bundle-w13.json"
+    "$PYTHON_BIN" "$ROOT/scripts/run_scanner_heal_mrf_evidence.py" \
+        --run-dir "$RUN_DIR" \
+        --out-file "$descriptor" \
+        --source-revision "$source_revision" \
+        --test "$TEST_SELECTION" >&2
+    printf '%s\n' "$descriptor"
 }

 run_self_test() {
@@ -423,7 +423,7 @@ for gate, gate_artifacts in artifacts.items():
            "artifact_format": "json",
            "summary": payload["summary"],
        }
-        for mirror in ("crash_points", "durable_intent_cases", "persist_failure_blocks_acceptance"):
+        for mirror in ("crash_points", "durable_intent_cases", "persist_failure_blocks_acceptance", "quota_path_cases"):
            if mirror in payload:
                evidence[mirror] = payload[mirror]
        fields[field] = evidence
@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+RUNNER="$SCRIPT_DIR/run_scanner_heal_authority_evidence.py"
+
+"${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" --self-test
@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+RUNNER="$SCRIPT_DIR/run_scanner_heal_legacy_rollback_evidence.py"
+
+"${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" --self-test
@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+RUNNER="$SCRIPT_DIR/run_scanner_heal_scoped_ack_evidence.py"
+
+"${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" --self-test
@@ -55,3 +55,4 @@ if bash "$RUNNER" --plan-only --test p4 --soak-seconds 10 >/dev/null 2>&1; then
 fi

 bash "$RUNNER" --self-test
+"$PROJECT_ROOT/scripts/python_bin.sh" "$PROJECT_ROOT/scripts/run_scanner_heal_mrf_evidence.py" --self-test