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/operations/scanner-benchmark-runbook.md

@@ -171,6 +171,22 @@ performance acceptance gate. Run the fake-adapter self-tests with:
 scripts/test_scanner_validation_harness.sh
 ```
 
+For CI summaries, PR evidence tables, and operator handoff, collapse the raw
+matrix into a quiet one-line verdict plus durable JSON/Markdown artifacts:
+
+```bash
+scripts/summarize_scanner_heal_perf.py \
+  --abba-dir /path/to/new-artifacts \
+  --cache-cost-log /path/to/cache-cost-profile.log \
+  --json-out /path/to/new-artifacts/perf-summary.json \
+  --markdown-out /path/to/new-artifacts/perf-summary.md
+```
+
+The command prints only `PASS scanner_heal_perf ...` for measured passing ABBA
+evidence, otherwise `FAIL scanner_heal_perf ...`. The JSON and Markdown outputs
+carry the key p99/throughput/P1/P2/cache-cost fields and artifact provenance
+hashes; raw per-cell logs remain in the original artifact tree for audit.
+
 They cover the complete 120-cell schedule, data isolation, missing builds and
 oracles, zero samples/requests, swallowed request errors, offered-load drift,
 incomplete repairs, missing metrics, noise, and P1/P2/p99 regressions. A real

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/summarize_scanner_heal_perf.py

@@ -0,0 +1,311 @@
+#!/usr/bin/env python3
+"""Summarize Scanner/Heal ABBA and cache-cost profile artifacts quietly."""
+
+from __future__ import annotations
+
+import argparse
+from collections import Counter
+from decimal import Decimal
+import hashlib
+import json
+from pathlib import Path
+import sys
+from typing import Any
+
+MAX_JSON_BYTES = 1024 * 1024
+CACHE_COST_PREFIX = "CACHE_COST "
+PASS_STATES = {"pass"}
+FAIL_STATES = {"fail", "failed"}
+
+
+def require(condition: bool, message: str) -> None:
+    if not condition:
+        raise ValueError(message)
+
+
+def read_json(path: Path) -> dict[str, Any]:
+    require(path.is_file(), f"missing JSON artifact: {path}")
+    require(path.stat().st_size <= MAX_JSON_BYTES, f"oversized JSON artifact: {path}")
+    with path.open(encoding="utf-8") as stream:
+        value = json.load(stream)
+    require(isinstance(value, dict), f"expected JSON object: {path}")
+    return value
+
+
+def write_json(path: Path, value: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(value, indent=2, sort_keys=True, allow_nan=False) + "\n", encoding="utf-8")
+
+
+def digest(path: Path) -> str:
+    with path.open("rb") as stream:
+        if hasattr(hashlib, "file_digest"):
+            return hashlib.file_digest(stream, "sha256").hexdigest()
+        hasher = hashlib.sha256()
+        while chunk := stream.read(1024 * 1024):
+            hasher.update(chunk)
+        return hasher.hexdigest()
+
+
+def number(value: Any, name: str) -> Decimal:
+    require(type(value) in (float, int), f"invalid numeric field: {name}")
+    return Decimal(str(value))
+
+
+def maybe_number(value: Any, name: str) -> Decimal | None:
+    if value is None:
+        return None
+    return number(value, name)
+
+
+def pct(value: Decimal | None) -> str:
+    if value is None:
+        return "pending"
+    return f"{float(value * Decimal('100')):.2f}%"
+
+
+def ratio(value: Decimal | None) -> str:
+    if value is None:
+        return "pending"
+    return f"{float(value):.3f}x"
+
+
+def max_decimal(values: list[Decimal | None]) -> Decimal | None:
+    present = [value for value in values if value is not None]
+    if not present:
+        return None
+    return max(present)
+
+
+def summarize_abba(abba_dir: Path) -> dict[str, Any]:
+    manifest_path = abba_dir / "manifest.json"
+    report_path = abba_dir / "report.json"
+    manifest = read_json(manifest_path)
+    report = read_json(report_path)
+    comparisons = report.get("comparisons")
+    require(isinstance(comparisons, list), "report.comparisons must be a list")
+
+    counts = Counter()
+    p99_regressions: list[Decimal] = []
+    throughput_losses: list[Decimal] = []
+    p1_rows = []
+    p2_values: list[Decimal | None] = []
+    for index, comparison in enumerate(comparisons):
+        require(isinstance(comparison, dict), f"comparison {index} must be an object")
+        state = comparison.get("status")
+        require(isinstance(state, str) and state, f"comparison {index} missing status")
+        counts[state] += 1
+        p99_regressions.append(number(comparison.get("p99_regression"), f"comparison {index} p99_regression"))
+        throughput_change = number(comparison.get("throughput_change"), f"comparison {index} throughput_change")
+        throughput_losses.append(max(Decimal("0"), -throughput_change))
+        p1 = comparison.get("p1")
+        if isinstance(p1, dict):
+            p1_rows.append({
+                "scenario": comparison.get("scenario"),
+                "comparison": comparison.get("comparison"),
+                "round": comparison.get("round"),
+                "required_reduction": float(number(p1.get("required_reduction"), "p1.required_reduction")),
+                "observed_reduction": float(number(p1.get("observed_reduction"), "p1.observed_reduction")),
+                "repeatability_drift": (
+                    None if p1.get("repeatability_drift") is None
+                    else float(number(p1.get("repeatability_drift"), "p1.repeatability_drift"))
+                ),
+            })
+        p2 = comparison.get("p2_post_stop_work_multiples")
+        if isinstance(p2, list):
+            p2_values.extend(maybe_number(value, "p2_post_stop_work_multiple") for value in p2)
+
+    report_state = report.get("status")
+    performance_state = report.get("performance")
+    require(isinstance(report_state, str) and report_state, "report.status missing")
+    require(isinstance(performance_state, str) and performance_state, "report.performance missing")
+    measured = report.get("evidence") == "measured"
+    passed = report_state in PASS_STATES and performance_state in PASS_STATES and measured
+    gate_state = "pass" if passed else "fail"
+    if report_state == "synthetic_validated":
+        reason = "synthetic evidence validates the harness only; measured performance remains pending"
+    elif report_state not in PASS_STATES:
+        reason = f"ABBA report status is {report_state}"
+    elif performance_state not in PASS_STATES:
+        reason = f"performance status is {performance_state}"
+    elif not measured:
+        reason = "measured evidence is required for a performance conclusion"
+    else:
+        reason = "measured ABBA report passed"
+
+    fixed = manifest.get("fixed", {})
+    require(isinstance(fixed, dict), "manifest.fixed must be an object")
+    return {
+        "gate_state": gate_state,
+        "reason": reason,
+        "status": report_state,
+        "performance": performance_state,
+        "evidence": report.get("evidence"),
+        "cells": report.get("cells", 0),
+        "comparisons_total": len(comparisons),
+        "comparison_status_counts": dict(sorted(counts.items())),
+        "worst_p99_regression": None if not p99_regressions else float(max(p99_regressions)),
+        "worst_throughput_loss": None if not throughput_losses else float(max(throughput_losses)),
+        "p2_worst_post_stop_work_multiple": None if max_decimal(p2_values) is None else float(max_decimal(p2_values)),
+        "p1_reductions": p1_rows,
+        "provenance": {
+            "abba_dir": str(abba_dir.resolve()),
+            "manifest_sha256": digest(manifest_path),
+            "report_sha256": digest(report_path),
+            "baseline_revision": manifest.get("baseline", {}).get("revision"),
+            "baseline_sha256": manifest.get("baseline", {}).get("sha256"),
+            "candidate_revision": manifest.get("candidate", {}).get("revision"),
+            "candidate_sha256": manifest.get("candidate", {}).get("sha256"),
+            "adapter_sha256": manifest.get("adapter_sha256"),
+            "collector_sha256": manifest.get("collector_sha256"),
+            "config_sha256": fixed.get("config_sha256"),
+            "dataset_sha256": fixed.get("dataset_sha256"),
+            "release_flags": fixed.get("release_flags"),
+            "durability": fixed.get("durability"),
+            "topology": fixed.get("topology"),
+            "offered_load_ops": fixed.get("offered_load_ops"),
+        },
+    }
+
+
+def cache_cost_records(path: Path) -> list[dict[str, Any]]:
+    require(path.is_file(), f"missing cache-cost log: {path}")
+    records = []
+    with path.open(encoding="utf-8", errors="replace") as stream:
+        for line_no, line in enumerate(stream, start=1):
+            if CACHE_COST_PREFIX not in line:
+                continue
+            payload = line.split(CACHE_COST_PREFIX, 1)[1].strip()
+            value = json.loads(payload)
+            require(isinstance(value, dict), f"cache-cost line {line_no} is not a JSON object")
+            require(value.get("schema") == 1, f"cache-cost line {line_no} has unsupported schema")
+            records.append(value)
+    require(records, f"no {CACHE_COST_PREFIX.strip()} records found in {path}")
+    return records
+
+
+def summarize_cache_cost(path: Path) -> dict[str, Any]:
+    records = cache_cost_records(path)
+    scenarios = Counter()
+    max_wire = Decimal("0")
+    max_save_amp = Decimal("0")
+    max_clone_ns = Decimal("0")
+    max_encode_ns = Decimal("0")
+    max_save_ns = Decimal("0")
+    build_sources = set()
+    for index, record in enumerate(records):
+        scenarios[str(record.get("scenario"))] += 1
+        wire = number(record.get("cache_wire_bytes"), f"cache_cost {index} cache_wire_bytes")
+        save_body = number(record.get("save_body_bytes_per_sample"), f"cache_cost {index} save_body_bytes_per_sample")
+        require(wire > 0, f"cache_cost {index} has zero wire bytes")
+        max_wire = max(max_wire, wire)
+        max_save_amp = max(max_save_amp, save_body / wire)
+        for field, target in (("clone", "max_clone_ns"), ("encode", "max_encode_ns"), ("save_inclusive", "max_save_ns")):
+            quantiles = record.get(field)
+            require(isinstance(quantiles, dict), f"cache_cost {index} missing {field} quantiles")
+            value = number(quantiles.get("max_ns"), f"cache_cost {index} {field}.max_ns")
+            if target == "max_clone_ns":
+                max_clone_ns = max(max_clone_ns, value)
+            elif target == "max_encode_ns":
+                max_encode_ns = max(max_encode_ns, value)
+            else:
+                max_save_ns = max(max_save_ns, value)
+        build = record.get("build", {})
+        require(isinstance(build, dict), f"cache_cost {index} build must be an object")
+        build_sources.add((build.get("source_revision"), build.get("source_tree"), build.get("test_opt_level_override")))
+    return {
+        "records": len(records),
+        "scenario_counts": dict(sorted(scenarios.items())),
+        "max_cache_wire_bytes": int(max_wire),
+        "max_save_body_amplification": float(max_save_amp),
+        "max_clone_ns": int(max_clone_ns),
+        "max_encode_ns": int(max_encode_ns),
+        "max_save_inclusive_ns": int(max_save_ns),
+        "build_sources": [
+            {"source_revision": revision, "source_tree": tree, "test_opt_level_override": opt}
+            for revision, tree, opt in sorted(build_sources, key=lambda item: tuple("" if part is None else str(part) for part in item))
+        ],
+        "provenance": {
+            "cache_cost_log": str(path.resolve()),
+            "cache_cost_log_sha256": digest(path),
+        },
+    }
+
+
+def markdown(summary: dict[str, Any]) -> str:
+    abba = summary["abba"]
+    p2 = None if abba["p2_worst_post_stop_work_multiple"] is None else Decimal(str(abba["p2_worst_post_stop_work_multiple"]))
+    p99 = None if abba["worst_p99_regression"] is None else Decimal(str(abba["worst_p99_regression"]))
+    throughput = None if abba["worst_throughput_loss"] is None else Decimal(str(abba["worst_throughput_loss"]))
+    lines = [
+        f"# Scanner/Heal Performance Summary",
+        "",
+        f"- verdict: {summary['verdict']}",
+        f"- reason: {summary['reason']}",
+        f"- abba: status={abba['status']} performance={abba['performance']} evidence={abba['evidence']} cells={abba['cells']} comparisons={abba['comparisons_total']}",
+        f"- worst_p99_regression: {pct(p99)}",
+        f"- worst_throughput_loss: {pct(throughput)}",
+        f"- p2_worst_post_stop_work_multiple: {ratio(p2)}",
+    ]
+    if summary.get("cache_cost") is not None:
+        cache = summary["cache_cost"]
+        lines.extend([
+            f"- cache_cost_records: {cache['records']}",
+            f"- max_cache_wire_bytes: {cache['max_cache_wire_bytes']}",
+            f"- max_save_body_amplification: {cache['max_save_body_amplification']:.3f}x",
+            f"- max_clone_ns: {cache['max_clone_ns']}",
+            f"- max_encode_ns: {cache['max_encode_ns']}",
+            f"- max_save_inclusive_ns: {cache['max_save_inclusive_ns']}",
+        ])
+    lines.extend([
+        "",
+        "## Provenance",
+        "",
+    ])
+    for key, value in abba["provenance"].items():
+        lines.append(f"- {key}: {value}")
+    if summary.get("cache_cost") is not None:
+        for key, value in summary["cache_cost"]["provenance"].items():
+            lines.append(f"- {key}: {value}")
+    return "\n".join(lines) + "\n"
+
+
+def build_summary(args: argparse.Namespace) -> dict[str, Any]:
+    abba = summarize_abba(args.abba_dir)
+    cache = summarize_cache_cost(args.cache_cost_log) if args.cache_cost_log else None
+    if args.require_cache_cost and cache is None:
+        raise ValueError("cache-cost profile log is required")
+    verdict = "PASS" if abba["gate_state"] == "pass" else "FAIL"
+    reason = abba["reason"]
+    return {"schema": 1, "verdict": verdict, "reason": reason, "abba": abba, "cache_cost": cache}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--abba-dir", type=Path, required=True, help="Directory containing manifest.json and report.json")
+    parser.add_argument("--cache-cost-log", type=Path, help="Rust test output containing CACHE_COST JSON lines")
+    parser.add_argument("--require-cache-cost", action="store_true", help="Fail when --cache-cost-log is missing")
+    parser.add_argument("--json-out", type=Path, help="Write the normalized summary JSON artifact")
+    parser.add_argument("--markdown-out", type=Path, help="Write a compact Markdown summary artifact")
+    args = parser.parse_args()
+    try:
+        summary = build_summary(args)
+        if args.json_out:
+            write_json(args.json_out, summary)
+        if args.markdown_out:
+            args.markdown_out.parent.mkdir(parents=True, exist_ok=True)
+            args.markdown_out.write_text(markdown(summary), encoding="utf-8")
+        abba = summary["abba"]
+        print(
+            f"{summary['verdict']} scanner_heal_perf "
+            f"status={abba['status']} performance={abba['performance']} evidence={abba['evidence']} "
+            f"comparisons={abba['comparisons_total']} reason={summary['reason']}"
+        )
+        return 0 if summary["verdict"] == "PASS" else 1
+    except (ValueError, OSError, json.JSONDecodeError) as error:
+        print(f"FAIL scanner_heal_perf error={error}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

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()

scripts/test_summarize_scanner_heal_perf.py

@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import contextlib
+import hashlib
+import io
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import summarize_scanner_heal_perf as summary
+
+
+def sha(path: Path) -> str:
+    with path.open("rb") as stream:
+        hasher = hashlib.sha256()
+        while chunk := stream.read(1024 * 1024):
+            hasher.update(chunk)
+        return hasher.hexdigest()
+
+
+class ScannerHealPerfSummaryTest(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.abba = self.root / "abba"
+        self.abba.mkdir()
+        self.manifest = {
+            "fixed": {
+                "config_sha256": "1" * 64,
+                "dataset_sha256": "2" * 64,
+                "release_flags": "--profile production",
+                "durability": "drive-sync=on",
+                "topology": "EC8+4",
+                "offered_load_ops": 100,
+            },
+            "baseline": {"revision": "a" * 40, "sha256": "3" * 64},
+            "candidate": {"revision": "b" * 40, "sha256": "4" * 64},
+            "adapter_sha256": "5" * 64,
+            "collector_sha256": "6" * 64,
+        }
+        self.comparison = {
+            "scenario": "cold-hot",
+            "comparison": "build",
+            "round": 1,
+            "status": "pass",
+            "p99_regression": 0.02,
+            "throughput_change": -0.01,
+            "p1": {"required_reduction": 0.8, "observed_reduction": 0.82, "repeatability_drift": 0.01},
+            "p2_post_stop_work_multiples": [None, 1.1, 1.0, None],
+        }
+        self.report = {
+            "status": "pass",
+            "performance": "pass",
+            "evidence": "measured",
+            "cells": 120,
+            "comparisons": [self.comparison],
+        }
+        self.write_inputs()
+
+    def write_inputs(self):
+        (self.abba / "manifest.json").write_text(json.dumps(self.manifest), encoding="utf-8")
+        (self.abba / "report.json").write_text(json.dumps(self.report), encoding="utf-8")
+
+    def test_measured_pass_writes_quiet_artifacts(self):
+        cache_log = self.root / "cache.log"
+        cache_log.write_text(
+            "compiler noise\nCACHE_COST "
+            + json.dumps({
+                "schema": 1,
+                "scenario": "small_dirty",
+                "cache_wire_bytes": 100,
+                "save_body_bytes_per_sample": 200,
+                "clone": {"max_ns": 10},
+                "encode": {"max_ns": 20},
+                "save_inclusive": {"max_ns": 30},
+                "build": {"source_revision": "abc", "source_tree": "clean", "test_opt_level_override": "0"},
+            })
+            + "\nmore noise\n",
+            encoding="utf-8",
+        )
+        args = type("Args", (), {
+            "abba_dir": self.abba,
+            "cache_cost_log": cache_log,
+            "require_cache_cost": False,
+            "json_out": None,
+            "markdown_out": None,
+        })
+        result = summary.build_summary(args)
+        self.assertEqual(result["verdict"], "PASS")
+        self.assertEqual(result["abba"]["provenance"]["manifest_sha256"], sha(self.abba / "manifest.json"))
+        self.assertEqual(result["cache_cost"]["max_save_body_amplification"], 2.0)
+
+    def test_synthetic_report_fails_as_performance_conclusion(self):
+        self.report.update(status="synthetic_validated", performance="pending", evidence="synthetic")
+        self.write_inputs()
+        args = type("Args", (), {
+            "abba_dir": self.abba,
+            "cache_cost_log": None,
+            "require_cache_cost": False,
+            "json_out": None,
+            "markdown_out": None,
+        })
+        result = summary.build_summary(args)
+        self.assertEqual(result["verdict"], "FAIL")
+        self.assertIn("synthetic evidence", result["reason"])
+
+    def test_requires_cache_profile_when_requested(self):
+        args = type("Args", (), {
+            "abba_dir": self.abba,
+            "cache_cost_log": None,
+            "require_cache_cost": True,
+            "json_out": None,
+            "markdown_out": None,
+        })
+        with self.assertRaisesRegex(ValueError, "cache-cost profile log is required"):
+            summary.build_summary(args)
+
+    def test_cli_prints_one_line_and_exits_nonzero_for_pending_performance(self):
+        self.report.update(status="inconclusive", performance="inconclusive")
+        self.write_inputs()
+        stdout = io.StringIO()
+        stderr = io.StringIO()
+        argv = [
+            "summarize_scanner_heal_perf.py",
+            "--abba-dir",
+            str(self.abba),
+            "--json-out",
+            str(self.root / "summary.json"),
+            "--markdown-out",
+            str(self.root / "summary.md"),
+        ]
+        with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
+            with mock.patch.object(sys, "argv", argv):
+                code = summary.main()
+        self.assertEqual(code, 1)
+        self.assertEqual(stdout.getvalue().count("\n"), 1)
+        self.assertTrue(stdout.getvalue().startswith("FAIL scanner_heal_perf "))
+        self.assertEqual(stderr.getvalue(), "")
+        self.assertEqual(summary.read_json(self.root / "summary.json")["verdict"], "FAIL")
+        self.assertIn("worst_p99_regression", (self.root / "summary.md").read_text(encoding="utf-8"))
+
+    def test_invalid_cache_cost_lines_fail_closed(self):
+        cache_log = self.root / "cache.log"
+        cache_log.write_text("CACHE_COST {\"schema\": 2}\n", encoding="utf-8")
+        with self.assertRaisesRegex(ValueError, "unsupported schema"):
+            summary.summarize_cache_cost(cache_log)
+
+    def test_script_entrypoint_is_quiet(self):
+        script = Path(__file__).with_name("summarize_scanner_heal_perf.py")
+        process = subprocess.run(
+            [sys.executable, str(script), "--abba-dir", str(self.abba)],
+            check=False,
+            text=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        self.assertEqual(process.returncode, 0, process.stderr)
+        self.assertEqual(process.stdout.count("\n"), 1)
+        self.assertEqual(process.stderr, "")
+
+
+if __name__ == "__main__":
+    unittest.main()