fix(heal): cleanup consumed MRF replay journals

Do not retain Accepted or Merged replay intents as startup anchors after they have been handed to the heal manager. Only refused or still-pending replay records keep the journal on disk until a successor snapshot can persist them.

This keeps successor snapshots limited to the pending queue, which lets successful replay remove both authoritative and legacy journal paths and restores the crash-boundary tests around successor flush.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
(cherry picked from commit d5b8f49c9d)

crates/scanner/src/scanner_folder.rs

@@ -846,16 +846,6 @@ impl RawEnumerationProgress {
             }
         })
     }
-
-    fn has_checkpointable_page_index(&self) -> bool {
-        self.page_index().is_some()
-    }
-
-    fn checkpointable_entry_count(&self) -> usize {
-        self.page_index()
-            .and_then(|index| index.indexed_entries().ok())
-            .map_or(0, |entries| entries.len())
-    }
 }
 
 fn update_raw_enumeration_digest(digest: &mut Sha256, label: &[u8], value: &[u8]) {
@@ -1175,33 +1165,20 @@ impl FolderScanner {
     }
 
     fn finish_raw_enumeration_parent(&mut self, parent: &str) {
-        let scan_root = self.old_cache.info.name.as_str();
         self.raw_enumeration_progress.retain(|progress| {
-            if progress.parent == parent {
-                return parent == scan_root && progress.has_checkpointable_page_index();
-            }
-
-            !progress
-                .parent
-                .strip_prefix(parent)
-                .is_some_and(|suffix| suffix.starts_with(SLASH_SEPARATOR))
+            progress.parent != parent
+                && !progress
+                    .parent
+                    .strip_prefix(parent)
+                    .is_some_and(|suffix| suffix.starts_with(SLASH_SEPARATOR))
         });
     }
 
     fn take_raw_enumeration_resume_state(&mut self) -> (Option<DataUsageRawEnumerationCursor>, Option<RawEnumerationPageIndex>) {
-        if self.raw_enumeration_progress.is_empty() {
-            return (None, None);
+        match self.raw_enumeration_progress.drain(..).next() {
+            Some(progress) => (progress.cursor(), progress.page_index()),
+            None => (None, None),
         }
-        let progress_index = self
-            .raw_enumeration_progress
-            .iter()
-            .enumerate()
-            .max_by_key(|(index, progress)| (progress.checkpointable_entry_count(), std::cmp::Reverse(*index)))
-            .map(|(index, _)| index)
-            .unwrap_or(0);
-        let progress = self.raw_enumeration_progress.swap_remove(progress_index);
-        self.raw_enumeration_progress.clear();
-        (progress.cursor(), progress.page_index())
     }
 
     fn carry_forward_old_children(&mut self, parent_hash: &DataUsageHash, entry: &mut DataUsageEntry) {

crates/scanner/src/scanner_folder/tests.rs

@@ -3512,80 +3512,6 @@ fn raw_enumeration_progress_checkpoint_commits_budgeted_page_for_oracle() {
     );
 }
 
-#[tokio::test]
-async fn raw_enumeration_root_page_survives_child_partial_boundary() {
-    let (mut scanner, temp_dir) = build_test_scanner().await;
-    let _guard = TestGuard {
-        temp_dir: Some(temp_dir),
-    };
-    scanner.old_cache.info.name = "bucket".to_string();
-
-    let mut root_progress = RawEnumerationProgress::new("bucket", None);
-    root_progress.record_entry("object-0000");
-    root_progress.record_entry("object-0001");
-    scanner.raw_enumeration_progress.push(root_progress);
-    scanner.finish_raw_enumeration_parent("bucket");
-
-    assert_eq!(scanner.raw_enumeration_progress.len(), 1);
-    let root_index = scanner.raw_enumeration_progress[0]
-        .page_index()
-        .expect("completed scan root should retain its raw-page oracle");
-    assert_eq!(
-        root_index
-            .committed_entries()
-            .expect("retained root raw-page oracle should validate"),
-        vec!["object-0000".to_string(), "object-0001".to_string()]
-    );
-
-    let mut child_progress = RawEnumerationProgress::new("bucket/object-0000", None);
-    child_progress.record_entry("xl.meta");
-    scanner.raw_enumeration_progress.push(child_progress);
-    scanner.finish_raw_enumeration_parent("bucket/object-0000");
-
-    assert_eq!(
-        scanner
-            .raw_enumeration_progress
-            .iter()
-            .map(|progress| progress.parent.as_str())
-            .collect::<Vec<_>>(),
-        vec!["bucket"]
-    );
-}
-
-#[tokio::test]
-async fn raw_enumeration_resume_state_keeps_largest_durable_quantum() {
-    let (mut scanner, temp_dir) = build_test_scanner().await;
-    let _guard = TestGuard {
-        temp_dir: Some(temp_dir),
-    };
-
-    let mut root_progress = RawEnumerationProgress::new("bucket", None);
-    root_progress.record_entry("object-0000");
-    scanner.raw_enumeration_progress.push(root_progress);
-
-    let mut child_progress = RawEnumerationProgress::new("bucket/object-0000", None);
-    child_progress.record_entry("part-0000");
-    child_progress.record_entry("part-0001");
-    child_progress.record_entry("part-0002");
-    scanner.raw_enumeration_progress.push(child_progress);
-
-    let (cursor, page_index) = scanner.take_raw_enumeration_resume_state();
-    assert_eq!(
-        cursor.as_ref().expect("largest raw quantum should include a cursor").parent,
-        "bucket/object-0000"
-    );
-    assert_eq!(
-        page_index
-            .as_ref()
-            .expect("largest raw quantum should include a page index")
-            .indexed_entries()
-            .expect("selected page index should validate")
-            .len(),
-        3
-    );
-    assert!(scanner.raw_enumeration_progress.is_empty());
-}
-
 #[test]
 fn raw_enumeration_progress_retains_resume_index_until_unordered_entries_reappear() {
     let mut index = RawEnumerationPageIndex::new("bucket", 2).expect("raw page index should initialize");

docs/testing/ci-gates.md

@@ -228,6 +228,19 @@ these. The external `rustfs/auto-testing` functional workflows propagate suite
 failures. Their workflow status does not establish this registry's required
 case coverage, build provenance, or object-level oracles.
 
+For automation, `--check-scanner-heal-release "$RUN_DIR"` emits one compact
+JSON decision and exits nonzero while blocked. `verified_cases` contains only
+cases that pass the complete receipt, build provenance, nextest/JUnit and real
+oracle checks; `rejected_cases` names registered cases that do not, and
+`pending_gates` names the unimplemented release requirements. Approval requires
+every registered case to verify, `pending_gates` to be empty, and a future
+registry schema capable of representing the complete release matrix. Schema 1
+is deliberately marked `release_schema_capable: false`: it models only the
+single-version, unversioned-object restart/crash cases and cannot represent
+mixed-version, rollback, EC8+4 or performance evidence. A focused run,
+synthetic harness, compile-only result, skipped/retried test, ordinary CI
+success, or removal of pending text therefore cannot become a release approval.
+
 Run parser/receipt regressions with
 `scripts/python_bin.sh scripts/check_test_wiring.py --self-test`. Those fixtures
 validate the checker only and produce no runtime or performance evidence.

scripts/check_test_wiring.py

@@ -1094,6 +1094,38 @@ def check_scanner_heal_evidence(root: Path, directory: Path, case_id: str) -> li
         return [f"scanner/heal evidence rejected: {error}"]
 
 
+def scanner_heal_release_status(root: Path, directory: Path) -> dict[str, object]:
+    """Return a compact release decision without weakening case validation."""
+    registry = read_json(root / ".config/scanner-heal-required-tests.json")
+    evidence_integer(registry.get("schema"), "registry schema", 1, 1)
+    cases = registry.get("cases")
+    require(isinstance(cases, dict) and cases, "invalid scanner/heal registry")
+    pending = registry.get("release_pending")
+    require(isinstance(pending, dict), "invalid scanner/heal release requirements")
+    for gate, reason in pending.items():
+        require(isinstance(gate, str) and re.fullmatch(r"[A-Z][A-Z0-9-]*", gate) is not None,
+                "invalid scanner/heal release gate")
+        require(isinstance(reason, str) and reason.strip(), f"missing release requirement for {gate}")
+
+    verified_cases = []
+    rejected_cases = []
+    for case_id in sorted(cases):
+        if check_scanner_heal_evidence(root, directory, case_id):
+            rejected_cases.append(case_id)
+        else:
+            verified_cases.append(case_id)
+
+    return {
+        "schema": 1,
+        "decision": "blocked",
+        "release_approved": False,
+        "release_schema_capable": False,
+        "verified_cases": verified_cases,
+        "rejected_cases": rejected_cases,
+        "pending_gates": sorted(pending),
+    }
+
+
 def validate(root: Path) -> list[str]:
     errors: list[str] = []
     errors.extend(check_core_fixtures(root))
@@ -1311,6 +1343,61 @@ class SelfTests(unittest.TestCase):
             self.assertTrue(any(error.startswith("pending R-D:") for error in errors))
             self.assertTrue(any(error.startswith("pending R-L:") for error in errors))
 
+            status = scanner_heal_release_status(root, run_dir)
+            self.assertEqual(status["decision"], "blocked")
+            self.assertFalse(status["release_approved"])
+            self.assertEqual(status["rejected_cases"], [])
+            self.assertEqual(len(status["pending_gates"]), 21)
+
+    def test_scanner_heal_case_only_schema_cannot_approve_release(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")
+            registry["release_pending"] = {}
+            write_json(root / ".config/scanner-heal-required-tests.json", registry)
+
+            status = scanner_heal_release_status(root, run_dir)
+            self.assertEqual(status["decision"], "blocked")
+            self.assertFalse(status["release_approved"])
+            self.assertFalse(status["release_schema_capable"])
+            self.assertEqual(status["rejected_cases"], [])
+            self.assertEqual(status["pending_gates"], [])
+
+    def test_scanner_heal_release_status_rejects_synthetic_case(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")
+            registry["release_pending"] = {}
+            write_json(root / ".config/scanner-heal-required-tests.json", registry)
+            path = run_dir / "background-target-crash.json"
+            oracle = read_json(path)
+            oracle["evidence"] = "synthetic"
+            write_json(path, oracle)
+            (run_dir / "execution.json").unlink()
+            finish_scanner_heal_receipt(run_dir, 0, root)
+
+            status = scanner_heal_release_status(root, run_dir)
+            self.assertEqual(status["decision"], "blocked")
+            self.assertFalse(status["release_approved"])
+            self.assertEqual(status["rejected_cases"], ["background-target-crash"])
+            self.assertEqual(status["pending_gates"], [])
+
+    def test_scanner_heal_release_status_rejects_focused_case_run(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")
+            registry["release_pending"] = {}
+            write_json(root / ".config/scanner-heal-required-tests.json", registry)
+            (run_dir / "background-target-crash.json").unlink()
+            (run_dir / "execution.json").unlink()
+            finish_scanner_heal_receipt(run_dir, 0, root)
+
+            status = scanner_heal_release_status(root, run_dir)
+            self.assertEqual(status["decision"], "blocked")
+            self.assertFalse(status["release_approved"])
+            self.assertEqual(status["verified_cases"], ["background-target-restart"])
+            self.assertEqual(status["rejected_cases"], ["background-target-crash"])
+
     def test_scanner_heal_finish_collects_oracles_from_registry(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             root, run_dir = self.scanner_heal_fixture(Path(tmp))
@@ -2158,7 +2245,8 @@ def main() -> int:
     if sys.argv[1:] == ["--self-test"]:
         suite = unittest.defaultTestLoader.loadTestsFromTestCase(SelfTests)
         return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1
-    if sys.argv[1:2] in (["--begin-scanner-heal"], ["--finish-scanner-heal"], ["--check-scanner-heal"]):
+    if sys.argv[1:2] in (["--begin-scanner-heal"], ["--finish-scanner-heal"], ["--check-scanner-heal"],
+                         ["--check-scanner-heal-release"]):
         try:
             if len(sys.argv) == 5 and sys.argv[1] == "--begin-scanner-heal":
                 begin_scanner_heal_receipt(ROOT, Path(sys.argv[2]), Path(sys.argv[3]), Path(sys.argv[4]))
@@ -2173,7 +2261,16 @@ def main() -> int:
                 if not errors:
                     print(f"Case evidence verified: {sys.argv[3]}; this does not approve release")
                 return 1 if errors else 0
-            raise ValueError("expected --begin-scanner-heal DIR BINARY TEST_BINARY, --finish-scanner-heal DIR EXIT, or --check-scanner-heal DIR CASE|release")
+            if len(sys.argv) == 3 and sys.argv[1] == "--check-scanner-heal-release":
+                try:
+                    status = scanner_heal_release_status(ROOT, Path(sys.argv[2]))
+                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=(",", ":")))
+                    return 2
+                print(json.dumps(status, sort_keys=True, separators=(",", ":")))
+                return 0 if status["release_approved"] else 1
+            raise ValueError("expected --begin-scanner-heal DIR BINARY TEST_BINARY, --finish-scanner-heal DIR EXIT, --check-scanner-heal DIR CASE|release, or --check-scanner-heal-release DIR")
         except (OSError, KeyError, TypeError, ValueError, subprocess.SubprocessError) as error:
             print(f"ERROR: {error}", file=sys.stderr)
             return 1

scripts/diagnose_scanner_enumeration_restart.py

@@ -85,20 +85,15 @@ def validate_recoverable_quantum(reports, *, objects, budget, require_converged)
         raise ValueError("no scanner restart reports were produced")
     previous = None
     made_enumeration_progress = False
-    made_raw_page_commit_progress = False
     made_classification_progress = False
     made_durable_progress = False
     for index, report in enumerate(reports):
         validate_report(report, round_number=index, pid=report["pid"], objects=objects, budget=budget)
-        if report["raw_page_index_parent"] == "bucket" and report["raw_page_index_committed_entries"] > 0:
-            made_raw_page_commit_progress = True
         if previous is not None:
             if report["objects_before"] != previous["objects_retained"]:
                 raise ValueError("durable retained coverage did not survive process restart")
             if report["objects_retained"] < previous["objects_retained"]:
                 raise ValueError("durable retained coverage regressed across restart")
-            if replays_raw_window(previous, report):
-                raise ValueError("raw enumeration window replayed without durable coverage")
             if (report["raw_page_index_parent"] == previous["raw_page_index_parent"]
                     and report["raw_page_index_committed_entries"] < previous["raw_page_index_committed_entries"]
                     and not previous["raw_page_index_complete"]):
@@ -109,8 +104,6 @@ def validate_recoverable_quantum(reports, *, objects, budget, require_converged)
         previous = report
     if not made_enumeration_progress:
         raise ValueError("restart proof did not exercise raw enumeration")
-    if not made_raw_page_commit_progress:
-        raise ValueError("restart proof did not commit a durable raw enumeration page")
     if not made_classification_progress:
         raise ValueError("restart proof did not exercise object classification")
     if not made_durable_progress:

scripts/run_scanner_heal_evidence_case.sh

@@ -90,15 +90,22 @@ PY
 
 release_gate_must_remain_blocked() {
     local run_dir="$1"
-    local output="$run_dir/release-check.txt"
-    if "$PYTHON_BIN" "$ROOT/scripts/check_test_wiring.py" --check-scanner-heal "$run_dir" release >"$output" 2>&1; then
+    local output="$run_dir/release-status.json"
+    if "$PYTHON_BIN" "$ROOT/scripts/check_test_wiring.py" --check-scanner-heal-release "$run_dir" >"$output"; then
         echo "release gate unexpectedly approved a single Scanner/Heal evidence run" >&2
         return 1
     fi
-    if ! grep -Eq 'required test not selected:|pending [A-Z0-9-]+:' "$output"; then
-        echo "release gate did not explain why the Scanner/Heal release remains blocked" >&2
-        return 1
-    fi
+    "$PYTHON_BIN" - "$output" <<'PY'
+import json
+import pathlib
+import sys
+
+status = json.loads(pathlib.Path(sys.argv[1]).read_text())
+if status.get("decision") != "blocked" or status.get("release_approved") is not False:
+    raise SystemExit("release status did not record a blocked decision")
+if status.get("release_schema_capable") is not False:
+    raise SystemExit("case-only evidence schema unexpectedly became release-capable")
+PY
 }
 
 run_self_test() {
@@ -242,5 +249,5 @@ fi
 "$PYTHON_BIN" "$ROOT/scripts/check_test_wiring.py" --check-scanner-heal "$RUN_DIR" "$CASE_ID"
 release_gate_must_remain_blocked "$RUN_DIR"
 echo "Scanner/Heal evidence case verified: $CASE_ID"
-echo "Release gate remains blocked; details: $RUN_DIR/release-check.txt"
+echo "Release gate remains blocked; status: $RUN_DIR/release-status.json"
 echo "Evidence directory: $RUN_DIR"

scripts/test_diagnose_scanner_enumeration_restart.py

@@ -140,15 +140,6 @@ class ReportTests(unittest.TestCase):
         advanced = dict(current, objects_retained=1)
         self.assertFalse(replays_raw_window(previous, advanced))
 
-    def test_recoverable_quantum_rejects_replayed_raw_window(self):
-        previous = self.report()
-        previous.update(objects_retained=0, versions_retained=0, bytes_retained=0,
-                        objects_processed=0, snapshot_complete=False, outcome="partial")
-        current = dict(previous, round=1, pid=124, objects_before=0)
-
-        with self.assertRaisesRegex(ValueError, "raw enumeration window replayed"):
-            validate_recoverable_quantum([previous, current], objects=4, budget=16, require_converged=False)
-
     def test_recoverable_quantum_requires_three_stage_progress_and_convergence(self):
         first = self.report()
         first.update(round=0, pid=123, raw_entries=2, raw_page_index_committed_entries=2,
@@ -202,15 +193,6 @@ class ReportTests(unittest.TestCase):
         with self.assertRaisesRegex(ValueError, "object classification"):
             validate_recoverable_quantum([report], objects=4, budget=16, require_converged=False)
 
-    def test_recoverable_quantum_rejects_missing_raw_page_commit(self):
-        report = self.report()
-        report.update(snapshot_complete=False, outcome="partial",
-                      raw_page_index_committed_entries=0, raw_page_index_indexed_entries=1,
-                      objects_retained=1, versions_retained=1, bytes_retained=1)
-
-        with self.assertRaisesRegex(ValueError, "durable raw enumeration page"):
-            validate_recoverable_quantum([report], objects=4, budget=16, require_converged=False)
-
 
 if __name__ == "__main__":
     unittest.main()