test(scanner): add bounded ABBA validation harness (#7181)

* chore(deps): refresh SDKs and pin clock skew regression coverage Refresh compatible dependencies for Scanner/Heal V2 batch 1 and verify the production S3 retry/signing path with a deterministic clock. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(scanner): add bounded ABBA validation harness Refs rustfs/backlog#2266 and rustfs/backlog#2240. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(scanner): harden ABBA threshold evaluation Handle exact threshold comparisons without binary floating point boundary drift and mark unstable P1 walk controls inconclusive. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(scanner): reject invalid ABBA evidence and boundary drift --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com> Co-authored-by: overtrue <anzhengchao@gmail.com>
2026-09-06 12:09:12 +00:00 · 2026-09-06 07:27:46 +08:00
parent 159dc13548
commit 081a8b61d8
7 changed files with 796 additions and 0 deletions
@@ -54,6 +54,8 @@ their issue closes.
 | `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` |
 | `test_scanner_validation_harness.sh` | dev-tool | Self-test for the scanner validation harness | — |
+| `scanner_abba.py` | dev-tool | Scanner/heal ABBA orchestration and evidence gates via `run_scanner_validation_harness.sh --abba` | `docs/operations/scanner-benchmark-runbook.md` |
+| `test_scanner_abba.py` | dev-tool | Synthetic ABBA adapter and failure-path tests | `test_scanner_validation_harness.sh` |
 | `test_build_rustfs_options.sh` | dev-tool | Shell test for rustfs build-option wiring | `make test` (script-tests) |
 | `test_entrypoint_credentials.sh` | dev-tool | Container entrypoint credential-handling test | `make test` (script-tests) |
 | `test_helm_chart_version.sh` | dev-tool | Test for `helm_chart_version.sh` | — |
@@ -1,6 +1,12 @@
 #!/usr/bin/env bash
 set -euo pipefail

+if [[ "${1:-}" == "--abba" ]]; then
+  shift
+  SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+  exec "$SCRIPT_DIR/python_bin.sh" "$SCRIPT_DIR/scanner_abba.py" "$@"
+fi
+
 ALIAS=""
 ENDPOINT=""
 ACCESS_KEY="${RUSTFS_ACCESS_KEY:-}"
@@ -23,6 +29,7 @@ TELEMETRY_PIDS=()
 usage() {
  cat <<'USAGE'
 Usage:
+  scripts/run_scanner_validation_harness.sh --abba --help
  scripts/run_scanner_validation_harness.sh --alias <admin-alias> \
    --endpoint <url> [options]

@@ -0,0 +1,384 @@
+#!/usr/bin/env python3
+"""Run isolated scanner/heal ABBA cells through a deployment-specific adapter."""
+
+import argparse
+from decimal import Decimal
+import hashlib
+import json
+import math
+import os
+from pathlib import Path
+import shutil
+import signal
+import subprocess
+import sys
+import time
+
+SCENARIOS = ("cold-hot", "fresh-hot", "multi-hot-new", "running-heal", "mrf-replay")
+LEGS = ("A1", "B1", "B2", "A2")
+MAX_JSON_BYTES = 1024 * 1024
+METRICS = (
+    "p99_ms", "throughput_ops", "rss_bytes", "cpu_seconds", "iops", "rpc_count",
+    "cache_clone_bytes", "encode_bytes", "save_bytes", "oldest_age_seconds",
+    "walk_objects", "cold_walk_objects", "healed_objects", "errors", "requests",
+)
+REPEATABILITY_LIMIT = Decimal("0.05")
+P2_WORK_MULTIPLE_LIMIT = Decimal("1.2")
+
+
+def require(condition, message):
+    if not condition:
+        raise ValueError(message)
+
+
+def number(value, name, minimum=0):
+    require(type(value) in (float, int) and math.isfinite(value) and value >= minimum,
+            f"invalid {name}")
+    return value
+
+
+def decimal_number(value, name, minimum=0):
+    if isinstance(value, Decimal):
+        require(value.is_finite() and value >= Decimal(str(minimum)), f"invalid {name}")
+        return value
+    number(value, name, minimum)
+    return Decimal(str(value))
+
+
+def ratio(numerator, denominator, name):
+    denominator = decimal_number(denominator, f"{name} denominator")
+    require(denominator > 0, f"invalid {name} denominator")
+    return decimal_number(numerator, name) / denominator
+
+
+def relative_change(current, baseline, name):
+    return ratio(current, baseline, name) - Decimal("1")
+
+
+def repeatability_change(first, second, name):
+    first = decimal_number(first, name)
+    second = decimal_number(second, name)
+    if first == 0 and second == 0:
+        return Decimal("0")
+    if first == 0 or second == 0:
+        return Decimal("Infinity")
+    return abs(second / first - Decimal("1"))
+
+
+def report_number(value):
+    return None if value.is_infinite() else float(value)
+
+
+def digest(path):
+    with Path(path).open("rb") as stream:
+        return hashlib.file_digest(stream, "sha256").hexdigest()
+
+
+def read_json(path):
+    require(path.stat().st_size <= MAX_JSON_BYTES, f"oversized JSON: {path.name}")
+    with path.open() as stream:
+        value = json.load(stream)
+    require(isinstance(value, dict), f"expected JSON object: {path.name}")
+    return value
+
+
+def write_json(path, value):
+    data = json.dumps(value, indent=2, allow_nan=False) + "\n"
+    require(len(data.encode()) <= MAX_JSON_BYTES, "oversized result")
+    path.write_text(data)
+
+
+def sha(value):
+    return isinstance(value, str) and len(value) == 64 and all(c in "0123456789abcdef" for c in value)
+
+
+def validate_manifest(manifest):
+    require(manifest.get("schema") == 1, "unsupported manifest schema")
+    require(manifest.get("evidence") in ("synthetic", "measured"), "missing evidence type")
+    fixed = manifest["fixed"]
+    for key in ("config_sha256", "dataset_sha256"):
+        require(sha(fixed.get(key)), f"invalid fixed.{key}")
+    for key in ("release_flags", "durability", "disk_type", "cache_state", "load_command", "resource_isolation"):
+        require(isinstance(fixed.get(key), str) and fixed[key].strip(), f"missing fixed.{key}")
+    require(fixed.get("topology") == "EC8+4", "formal matrix requires EC8+4")
+    number(fixed.get("offered_load_ops"), "offered load", 1)
+    require(type(manifest.get("rounds")) is int and 3 <= manifest["rounds"] <= 10,
+            "rounds must be 3..10")
+    minimum = 900 if manifest["evidence"] == "measured" else 1
+    require(type(manifest.get("duration_seconds")) is int and
+            minimum <= manifest["duration_seconds"] <= 86400, "invalid duration_seconds")
+    number(manifest.get("min_free_bytes"), "min_free_bytes", 1)
+    for phase in ("baseline", "candidate"):
+        build = manifest[phase]
+        path = Path(build["binary"]).resolve(strict=True)
+        require(path.is_file() and os.access(path, os.X_OK), f"missing executable {phase} build")
+        require(sha(build.get("sha256")) and digest(path) == build["sha256"], f"{phase} binary hash mismatch")
+        require(isinstance(build.get("revision"), str) and len(build["revision"]) == 40 and
+                all(c in "0123456789abcdef" for c in build["revision"]), f"invalid {phase} revision")
+        build["binary"] = str(path)
+    for scenario in SCENARIOS:
+        expected = manifest["oracles"][scenario]
+        for key in ("objects", "versions", "bytes"):
+            require(type(expected.get(key)) is int and expected[key] > 0, f"missing {scenario} oracle {key}")
+        require(sha(expected.get("sha256")), f"missing {scenario} content/version digest")
+        number(manifest["expected_healed_objects"].get(scenario), f"{scenario} expected repairs")
+        if scenario in ("running-heal", "mrf-replay"):
+            require(manifest["expected_healed_objects"][scenario] > 0, f"{scenario} requires repairs")
+
+
+def invoke(adapter, action, request, timeout):
+    """The adapter writes bounded JSON separately; stderr/stdout remain raw evidence."""
+    output = request.parent / f"{action}.json"
+    with (request.parent / f"{action}.log").open("wb") as log:
+        process = subprocess.Popen([str(adapter), action, str(request), str(output)],
+                                   stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
+        try:
+            returncode = process.wait(timeout=timeout)
+            if returncode:
+                raise subprocess.CalledProcessError(returncode, [str(adapter), action])
+        finally:
+            if process.poll() != 0:
+                try:
+                    os.killpg(process.pid, signal.SIGTERM)
+                except ProcessLookupError:
+                    pass
+                try:
+                    process.wait(timeout=10)
+                except subprocess.TimeoutExpired:
+                    os.killpg(process.pid, signal.SIGKILL)
+                    process.wait()
+    return read_json(output)
+
+
+def validate_result(result, request, expected):
+    require(result.get("evidence") == request["evidence"], "adapter evidence type mismatch")
+    require(result.get("fixed") == request["fixed"], "offered load/config/cache/durability drift")
+    require(result.get("build") == request["build"], "deployed build provenance mismatch")
+    require(result.get("data_dir") == request["data_dir"], "adapter data isolation mismatch")
+    require(result.get("background") == request["background"], "background mode mismatch")
+    require(type(result.get("sample_count")) is int and 1 <= result["sample_count"] <= 3600,
+            "sample_count must be 1..3600")
+    number(result.get("elapsed_seconds"), "elapsed_seconds", request["duration_seconds"])
+    metrics = result["metrics"]
+    for key in METRICS:
+        number(metrics.get(key), key)
+    for key in ("requests", "p99_ms", "throughput_ops"):
+        require(metrics[key] > 0, f"zero {key}")
+    require(metrics["errors"] == 0, "workload request errors")
+    require(metrics["cold_walk_objects"] <= metrics["walk_objects"], "cold walk exceeds total walk")
+    require(result.get("oracle") == expected, "object/version/byte oracle mismatch")
+    if request["background"] == "on":
+        require(metrics["walk_objects"] > 0, "zero background walk")
+        require(metrics["healed_objects"] == request["expected_healed_objects"], "incomplete repair oracle")
+        if request["scenario"] in ("running-heal", "mrf-replay"):
+            require(metrics["healed_objects"] > 0, "zero completed repairs")
+    return result
+
+
+def convergence(result):
+    window = result.get("convergence")
+    if not window or window.get("writes_stopped") is not True or window.get("last_mutation_observed") is not True or window.get("first_complete_publication") is not True:
+        return None
+    for key in ("last_mutation_time", "last_mutation_observed_time", "writes_stopped_time", "window_start", "window_end", "walk_objects", "full_walk_objects", "budget_available_seconds"):
+        number(window.get(key), f"convergence.{key}")
+    require(window["last_mutation_time"] <= window["writes_stopped_time"] <= window["window_start"] < window["window_end"],
+            "invalid post-mutation convergence window")
+    require(window["last_mutation_time"] <= window["last_mutation_observed_time"] <= window["window_start"],
+            "convergence started before last mutation was observed")
+    require(window["full_walk_objects"] > 0, "zero full walk reference")
+    require(0 < window["budget_available_seconds"] <= window["window_end"] - window["window_start"],
+            "invalid convergence budget window")
+    return window["walk_objects"] / window["full_walk_objects"]
+
+
+def evaluate(cells):
+    comparisons = []
+    inconclusive = False
+    failed = False
+    for offset in range(0, len(cells), 4):
+        group = cells[offset:offset + 4]
+        require([cell["leg"] for cell in group] == list(LEGS), "incomplete ABBA group")
+        a1, b1, b2, a2 = (cell["result"]["metrics"] for cell in group)
+        control = group[0]["comparison"] == "background"
+        drift = max(abs(relative_change(a2[k], a1[k], k)) for k in ("p99_ms", "throughput_ops"))
+        repeat_drift = max(abs(relative_change(b2[k], b1[k], k)) for k in ("p99_ms", "throughput_ops"))
+        noise = max(drift, repeat_drift) > REPEATABILITY_LIMIT
+        a = {key: (decimal_number(a1[key], key) + decimal_number(a2[key], key)) / Decimal("2") for key in METRICS}
+        b = {key: (decimal_number(b1[key], key) + decimal_number(b2[key], key)) / Decimal("2") for key in METRICS}
+        p99 = relative_change(b["p99_ms"], a["p99_ms"], "p99_ms")
+        throughput = relative_change(b["throughput_ops"], a["throughput_ops"], "throughput_ops")
+        thresholds = {"p99_regression": Decimal("0.10") if control else Decimal("0.05"),
+                      "throughput_loss": Decimal("0.05") if control else Decimal("0.03")}
+        passed = p99 <= thresholds["p99_regression"] and throughput >= -thresholds["throughput_loss"]
+        p1 = None
+        work_drift = None
+        if not control:
+            if group[0]["scenario"] == "cold-hot":
+                require(a["cold_walk_objects"] > 0, "cold-hot baseline has no cold walk samples")
+            work_drift = max(repeatability_change(a1[key], a2[key], key) for key in ("walk_objects", "cold_walk_objects"))
+            work_drift = max(work_drift, *(repeatability_change(b1[key], b2[key], key) for key in ("walk_objects", "cold_walk_objects")))
+            noise |= work_drift > REPEATABILITY_LIMIT
+            required = ratio(a["cold_walk_objects"], a["walk_objects"], "cold walk baseline") * Decimal("0.80")
+            reduction = Decimal("1") - ratio(b["walk_objects"], a["walk_objects"], "walk reduction")
+            p1 = {"required_reduction": float(required), "observed_reduction": float(reduction),
+                  "repeatability_drift": report_number(work_drift)}
+            if group[0]["scenario"] == "cold-hot":
+                # Compare counts before division can round repeating decimal ratios.
+                passed &= a["walk_objects"] - b["walk_objects"] >= a["cold_walk_objects"] * Decimal("0.80")
+        p2 = [convergence(cell["result"]) if cell["background"] == "on" else None for cell in group]
+        candidate_p2 = [value for cell, value in zip(group, p2) if cell["leg"].startswith("B")]
+        p2_pending = any(value is None for value in candidate_p2)
+        passed &= all(ratio(value, 1, "p2 work multiple") <= P2_WORK_MULTIPLE_LIMIT for value in candidate_p2 if value is not None)
+        inconclusive |= noise or p2_pending
+        if not noise and not passed:
+            failed = True
+        comparisons.append({"scenario": group[0]["scenario"], "comparison": group[0]["comparison"],
+                            "round": group[0]["round"], "status": "inconclusive" if noise else ("fail" if not passed else "inconclusive" if p2_pending else "pass"),
+                            "a2_a1_drift": report_number(drift), "b2_b1_drift": report_number(repeat_drift),
+                            "p99_regression": float(p99), "throughput_change": float(throughput),
+                            "thresholds": {key: float(value) for key, value in thresholds.items()},
+                            "p1": p1, "p2_max_work_multiple": float(P2_WORK_MULTIPLE_LIMIT),
+                            "p2_post_stop_work_multiples": p2})
+    return ("fail" if failed else "inconclusive" if inconclusive else "pass"), comparisons
+
+
+def collect_live(prepared, request, request_path, adapter):
+    collector = Path(__file__).with_name("run_scanner_validation_harness.sh")
+    # Only allow connection fields here; the runner owns cadence and output paths.
+    connection = prepared["collector"]
+    require(set(connection) == {"alias", "endpoint", "metrics_endpoints"}, "invalid collector connection")
+    require(all(isinstance(value, str) and value for value in connection.values()), "missing collector endpoint")
+    output = request_path.parent / "telemetry"
+    args = ["bash", str(collector), "--alias", connection["alias"], "--endpoint", connection["endpoint"],
+            "--metrics-endpoints", connection["metrics_endpoints"], "--deployment", "distributed",
+            "--samples", str(request["duration_seconds"] // 60 + 1), "--interval-secs", "60",
+            "--out-dir", str(output)]
+    with (request_path.parent / "collector.log").open("wb") as log:
+        process = subprocess.Popen(args, stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
+        try:
+            started = time.monotonic()
+            result = invoke(adapter, "measure", request_path, request["duration_seconds"] + 300)
+            require(time.monotonic() - started >= request["duration_seconds"], "measurement ended before required window")
+            require(process.wait(timeout=120) == 0, "scanner collector failed")
+            require(output.joinpath("scanner-summary.csv").stat().st_size > 0, "missing collector samples")
+            samples = list((output / "status").glob("scanner-status.*.json"))
+            require(len(samples) == request["duration_seconds"] // 60 + 1, "missing scanner samples")
+            for sample in samples:
+                status = read_json(sample)
+                require(isinstance(status.get("metrics"), dict) and status["metrics"], "invalid scanner status response")
+            heals = list((output / "heal").glob("background-heal-status.*.json"))
+            require(bool(heals), "missing heal samples")
+            for sample in heals:
+                status = read_json(sample)
+                require(isinstance(status.get("healOperations"), dict) and status["healOperations"], "invalid heal status response")
+            metrics = list((output / "metrics").glob("admin-metrics.*.ndjson"))
+            endpoints = [endpoint for endpoint in connection["metrics_endpoints"].split(",") if endpoint]
+            require(metrics and len(metrics) == len(endpoints) * len(samples), "missing distributed metrics samples")
+            for sample in metrics:
+                # The collector requests n=1, so each file contains one final JSON record.
+                status = read_json(sample)
+                require(status.get("errors") == [], "distributed metrics errors")
+                require(status.get("final") is True, "incomplete distributed metrics")
+                hosts = status.get("by_host")
+                require(isinstance(hosts, dict) and hosts, "missing by-host metrics")
+                for host in hosts.values():
+                    require(isinstance(host, dict) and isinstance(host.get("scanner"), dict) and host["scanner"],
+                            "missing per-host scanner metrics")
+            return result
+        finally:
+            # Stop telemetry children as well when measurement fails or times out.
+            try:
+                os.killpg(process.pid, signal.SIGTERM)
+            except ProcessLookupError:
+                pass
+            try:
+                process.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                os.killpg(process.pid, signal.SIGKILL)
+                process.wait()
+
+
+def run(manifest, adapter, output, data_root):
+    validate_manifest(manifest)
+    require(adapter.is_file() and os.access(adapter, os.X_OK), "missing executable adapter")
+    require(not output.exists() and not data_root.exists(), "output/data root must be new; existing data is preserved")
+    require(output != data_root and output not in data_root.parents and data_root not in output.parents,
+            "output and data roots must not overlap")
+    output.mkdir(parents=True)
+    data_root.mkdir(parents=True)
+    require(shutil.disk_usage(data_root).free >= manifest["min_free_bytes"], "insufficient free disk space")
+    manifest["adapter_sha256"] = digest(adapter)
+    manifest["collector_sha256"] = digest(Path(__file__).with_name("run_scanner_validation_harness.sh"))
+    write_json(output / "manifest.json", manifest)
+    cells = []
+    write_json(output / "report.json", {"status": "incomplete", "performance": "pending"})
+    try:
+        for scenario in SCENARIOS:
+            for comparison in ("build", "background"):
+                for round_id in range(1, manifest["rounds"] + 1):
+                    for leg in LEGS:
+                        phase = "baseline" if comparison == "build" and leg.startswith("A") else "candidate"
+                        background = "off" if comparison == "background" and leg.startswith("A") else "on"
+                        name = f"{scenario}-{comparison}-{round_id}-{leg}"
+                        cell_dir = output / name
+                        cell_dir.mkdir()
+                        data_dir = data_root / name
+                        data_dir.mkdir()
+                        request = {"schema": 1, "scenario": scenario, "comparison": comparison, "round": round_id,
+                                   "leg": leg, "background": background, "build": manifest[phase],
+                                   "evidence": manifest["evidence"], "fixed": manifest["fixed"],
+                                   "duration_seconds": manifest["duration_seconds"], "data_dir": str(data_dir),
+                                   "expected_healed_objects": manifest["expected_healed_objects"][scenario],
+                                   "expected_oracle": manifest["oracles"][scenario]}
+                        require(digest(Path(request["build"]["binary"])) == request["build"]["sha256"], "binary changed during run")
+                        require(digest(adapter) == manifest["adapter_sha256"], "adapter changed during run")
+                        require(shutil.disk_usage(data_root).free >= manifest["min_free_bytes"], "insufficient free disk space")
+                        request_path = cell_dir / "request.json"
+                        write_json(request_path, request)
+                        print(name, flush=True)
+                        try:
+                            prepared = invoke(adapter, "prepare", request_path, 300)
+                            require(prepared.get("ready") is True, "deployment not ready")
+                            if manifest["evidence"] == "measured":
+                                result = collect_live(prepared, request, request_path, adapter)
+                            else:
+                                result = invoke(adapter, "measure", request_path, 300)
+                            # An independent operation must enumerate all object versions and bytes.
+                            oracle = invoke(adapter, "oracle", request_path, 300)
+                            require(oracle.get("complete") is True and oracle.get("errors") == 0, "correctness oracle failed")
+                            require(type(oracle.get("errors")) is int, "invalid oracle error count")
+                            result["oracle"] = oracle["actual"]
+                            validate_result(result, request, request["expected_oracle"])
+                            cells.append({**request, "result": result})
+                        finally:
+                            stopped = invoke(adapter, "stop", request_path, 300)
+                            require(stopped.get("stopped") is True, "adapter failed to stop deployment")
+        status, comparisons = evaluate(cells)
+        synthetic = manifest["evidence"] == "synthetic"
+        report = {"status": "synthetic_validated" if synthetic and status == "pass" else status,
+                  "evidence": manifest["evidence"], "performance": "pending" if synthetic else status,
+                  "cells": len(cells), "comparisons": comparisons}
+        write_json(output / "report.json", report)
+        return 0 if status == "pass" else 3 if status == "inconclusive" else 1
+    except (ValueError, KeyError, OSError, subprocess.SubprocessError) as error:
+        write_json(output / "report.json", {"status": "failed", "performance": "pending",
+                                            "completed_cells": len(cells), "error": str(error)})
+        raise
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--manifest", type=Path, required=True)
+    parser.add_argument("--adapter", type=Path, required=True)
+    parser.add_argument("--out-dir", type=Path, required=True)
+    parser.add_argument("--data-root", type=Path, required=True)
+    args = parser.parse_args()
+    try:
+        return run(read_json(args.manifest), args.adapter.resolve(), args.out_dir.resolve(), args.data_root.resolve())
+    except (ValueError, KeyError, OSError, subprocess.SubprocessError) as error:
+        print(f"ERROR: {error}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,264 @@
+#!/usr/bin/env python3
+"""Synthetic adapter and failure-propagation tests; never start a RustFS server."""
+
+import contextlib
+import copy
+import io
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import Mock, patch
+
+import scanner_abba as harness
+
+
+def fake_adapter():
+    action, request_path, output_path = sys.argv[1:]
+    request = harness.read_json(Path(request_path))
+    fault = os.environ.get("SCANNER_ABBA_TEST_FAULT", "")
+    if action == "prepare":
+        result = {"ready": True}
+    elif action == "stop":
+        result = {"stopped": True}
+    elif action == "oracle":
+        if fault == "oracle-exit":
+            return 42
+        if fault == "missing-oracle":
+            return 0
+        result = {"complete": True, "errors": 0, "actual": request["expected_oracle"]}
+        if fault == "oracle-mismatch":
+            result["actual"]["bytes"] += 1
+    else:
+        if fault == "measure-exit":
+            return 42
+        result = {key: request[key] for key in ("evidence", "fixed", "build", "data_dir", "background")}
+        result.update({"sample_count": 10, "elapsed_seconds": request["duration_seconds"],
+                       "metrics": dict.fromkeys(harness.METRICS, 10)})
+        baseline = request["comparison"] == "build" and request["leg"].startswith("A")
+        result["metrics"].update(p99_ms=10, throughput_ops=100, errors=0, requests=100,
+                                 walk_objects=100 if baseline else 20, cold_walk_objects=100 if baseline else 0,
+                                 healed_objects=request["expected_healed_objects"])
+        result["convergence"] = {"writes_stopped": True, "last_mutation_observed": True,
+                                 "first_complete_publication": True, "last_mutation_time": 1,
+                                 "last_mutation_observed_time": 2,
+                                 "writes_stopped_time": 2, "window_start": 2, "window_end": 3,
+                                 "budget_available_seconds": 1, "walk_objects": 110, "full_walk_objects": 100}
+        if fault == "zero-samples":
+            result["sample_count"] = 0
+        elif fault == "request-errors":
+            result["metrics"]["errors"] = 1
+        elif fault == "load-drift":
+            result["fixed"]["offered_load_ops"] += 1
+        elif fault == "noise" and request["leg"] == "A2":
+            result["metrics"]["p99_ms"] = 20
+        elif fault == "zero-requests":
+            result["metrics"]["requests"] = 0
+        elif fault == "no-publication":
+            result["convergence"]["first_complete_publication"] = False
+        elif fault == "p2-regression":
+            result["convergence"]["walk_objects"] = 121
+        elif fault == "latency-regression" and request["leg"].startswith("B"):
+            result["metrics"]["p99_ms"] = 12
+        elif fault == "exact-thresholds" and request["leg"].startswith("B"):
+            result["metrics"].update(p99_ms=10.5, throughput_ops=97)
+        elif fault == "just-over-threshold" and request["comparison"] == "build" and request["leg"].startswith("B"):
+            result["metrics"]["p99_ms"] = 10.500001
+        elif fault == "p1-regression" and not baseline:
+            result["metrics"]["walk_objects"] = 30
+        elif fault in ("p1-exact-fraction", "p1-over-fraction"):
+            result["metrics"].update(walk_objects=9 if baseline else 5 + (fault == "p1-over-fraction"),
+                                     cold_walk_objects=5 if baseline else 0)
+        elif fault == "unstable-p1-control" and request["comparison"] == "build":
+            if request["leg"] == "A1":
+                result["metrics"].update(walk_objects=1000, cold_walk_objects=1000)
+            elif request["leg"] == "A2":
+                result["metrics"].update(walk_objects=10, cold_walk_objects=10)
+            elif request["leg"].startswith("B"):
+                result["metrics"].update(walk_objects=100, cold_walk_objects=0)
+        elif fault == "missing-metric":
+            del result["metrics"]["save_bytes"]
+        elif fault == "incomplete-repair":
+            result["metrics"]["healed_objects"] = 0
+    harness.write_json(Path(output_path), result)
+    return 0
+
+
+class ScannerAbbaTest(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.binary = Path(sys.executable).resolve()
+        self.adapter = Path(__file__).resolve()
+        self.manifest = {
+            "schema": 1, "evidence": "synthetic", "rounds": 3, "duration_seconds": 1, "min_free_bytes": 1,
+            "fixed": {"config_sha256": "1" * 64, "dataset_sha256": "2" * 64,
+                      "release_flags": "--release", "durability": "drive-sync=on",
+                      "disk_type": "synthetic", "cache_state": "cold", "load_command": "fake",
+                      "topology": "EC8+4", "offered_load_ops": 100, "resource_isolation": "synthetic"},
+            "oracles": {s: {"objects": 10, "versions": 20, "bytes": 30, "sha256": "3" * 64} for s in harness.SCENARIOS},
+            "expected_healed_objects": {s: 10 for s in harness.SCENARIOS},
+        }
+        build = {"binary": str(self.binary), "sha256": harness.digest(self.binary), "revision": "a" * 40}
+        self.manifest.update(baseline=build.copy(), candidate=build.copy())
+
+    def run_harness(self, fault=""):
+        with patch.dict(os.environ, {"SCANNER_ABBA_TEST_FAULT": fault}), contextlib.redirect_stdout(io.StringIO()):
+            return harness.run(copy.deepcopy(self.manifest), self.adapter, self.root / "out", self.root / "data")
+
+    def test_complete_synthetic_matrix_is_not_performance_evidence(self):
+        self.assertEqual(self.run_harness(), 0)
+        report = harness.read_json(self.root / "out/report.json")
+        self.assertEqual((report["status"], report["performance"], report["cells"]), ("synthetic_validated", "pending", 120))
+        requests = [harness.read_json(path) for path in (self.root / "out").glob("*/request.json")]
+        self.assertEqual(len({r["data_dir"] for r in requests}), 120)
+        for scenario in harness.SCENARIOS:
+            for comparison in ("build", "background"):
+                for round_id in (1, 2, 3):
+                    legs = [r for r in requests if (r["scenario"], r["comparison"], r["round"]) == (scenario, comparison, round_id)]
+                    self.assertEqual({r["leg"] for r in legs}, set(harness.LEGS))
+        self.assertTrue(all(c["p2_max_work_multiple"] == 1.2 for c in report["comparisons"]))
+
+    def test_fail_closed_adapter_and_data_errors(self):
+        for fault in ("measure-exit", "oracle-exit", "missing-oracle", "oracle-mismatch", "zero-samples",
+                      "zero-requests", "request-errors", "load-drift", "missing-metric", "incomplete-repair"):
+            with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory:
+                self.root = Path(directory)
+                with self.assertRaises((ValueError, OSError, subprocess.SubprocessError)):
+                    self.run_harness(fault)
+                report = harness.read_json(self.root / "out/report.json")
+                self.assertEqual(report["status"], "failed")
+                self.assertTrue(list((self.root / "out").glob("*/stop.json")))
+
+    def test_noise_is_inconclusive_and_nonzero(self):
+        with patch.object(harness, "SCENARIOS", ("cold-hot",)):
+            self.assertEqual(self.run_harness("noise"), 3)
+        self.assertEqual(harness.read_json(self.root / "out/report.json")["status"], "inconclusive")
+
+    def test_missing_first_publication_is_inconclusive(self):
+        with patch.object(harness, "SCENARIOS", ("cold-hot",)):
+            self.assertEqual(self.run_harness("no-publication"), 3)
+
+    def test_performance_regressions_fail(self):
+        for fault in ("p1-regression", "p2-regression", "latency-regression"):
+            with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory:
+                self.root = Path(directory)
+                with patch.object(harness, "SCENARIOS", ("cold-hot",)):
+                    self.assertEqual(self.run_harness(fault), 1)
+
+    def test_exact_threshold_boundaries_pass(self):
+        with patch.object(harness, "SCENARIOS", ("cold-hot",)):
+            self.assertEqual(self.run_harness("exact-thresholds"), 0)
+
+    def test_just_over_threshold_fails(self):
+        with patch.object(harness, "SCENARIOS", ("cold-hot",)):
+            self.assertEqual(self.run_harness("just-over-threshold"), 1)
+
+    def test_p1_fractional_boundary(self):
+        for fault, expected in (("p1-exact-fraction", 0), ("p1-over-fraction", 1)):
+            with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory:
+                self.root = Path(directory)
+                with patch.object(harness, "SCENARIOS", ("cold-hot",)):
+                    self.assertEqual(self.run_harness(fault), expected)
+
+    def test_live_collector_rejects_missing_or_failed_node_metrics(self):
+        telemetry = self.root / "telemetry"
+        for name in ("status", "heal", "metrics"):
+            (telemetry / name).mkdir(parents=True)
+        (telemetry / "scanner-summary.csv").write_text("timestamp\n")
+        valid = {"errors": [], "final": True, "by_host": {"node-b:9000": {"scanner": {"objects": 10}}}}
+        for index in range(16):
+            harness.write_json(telemetry / f"status/scanner-status.{index}.json", {"metrics": {"objects": 10}})
+            for node in ("node-a", "node-b"):
+                harness.write_json(telemetry / f"heal/background-heal-status.{node}.{index}.json",
+                                   {"healOperations": {"queueLength": 0}})
+                harness.write_json(telemetry / f"metrics/admin-metrics.{node}.{index}.ndjson",
+                                   {**valid, "by_host": {f"{node}:9000": {"scanner": {"objects": 10}}}})
+        sample = telemetry / "metrics/admin-metrics.node-b.15.ndjson"
+        prepared = {"collector": {"alias": "test", "endpoint": "http://node-a:9000",
+                                  "metrics_endpoints": "http://node-a:9000,http://node-b:9000"}}
+        cases = (
+            ("valid", valid, None),
+            ("missing", None, "missing distributed metrics samples"),
+            ("empty", "", "Expecting value"),
+            ("http-error", {"Code": "AccessDenied"}, "distributed metrics errors"),
+            ("partial-error", {**valid, "errors": ["node unavailable"]}, "distributed metrics errors"),
+            ("unfinished", {**valid, "final": False}, "incomplete distributed metrics"),
+            ("missing-host", {**valid, "by_host": {}}, "missing by-host metrics"),
+            ("missing-scanner", {**valid, "by_host": {"node-a:9000": {}}}, "missing per-host scanner metrics"),
+            ("collector-exit", valid, "scanner collector failed"),
+        )
+        for name, payload, error in cases:
+            with self.subTest(fault=name):
+                if payload is None:
+                    sample.unlink()
+                elif isinstance(payload, str):
+                    sample.write_text(payload)
+                else:
+                    harness.write_json(sample, payload)
+                process = Mock(pid=123, wait=Mock(return_value=1 if name == "collector-exit" else 0))
+                with patch.object(harness.subprocess, "Popen", return_value=process), \
+                        patch.object(harness, "invoke", return_value={"sample_count": 10}), \
+                        patch.object(harness.time, "monotonic", side_effect=(0, 900)), \
+                        patch.object(harness.os, "killpg"):
+                    if error:
+                        with self.assertRaisesRegex(ValueError, error):
+                            harness.collect_live(prepared, {"duration_seconds": 900}, self.root / "request.json", self.adapter)
+                    else:
+                        self.assertEqual(harness.collect_live(prepared, {"duration_seconds": 900},
+                                                             self.root / "request.json", self.adapter), {"sample_count": 10})
+
+    def test_unstable_p1_work_control_is_inconclusive(self):
+        with patch.object(harness, "SCENARIOS", ("cold-hot",)):
+            self.assertEqual(self.run_harness("unstable-p1-control"), 3)
+        comparison = harness.read_json(self.root / "out/report.json")["comparisons"][0]
+        self.assertEqual(comparison["status"], "inconclusive")
+        self.assertGreater(comparison["p1"]["repeatability_drift"], 0.05)
+
+    def test_manifest_rejects_missing_build_or_oracle(self):
+        for section, key in (("baseline", "binary"), ("oracles", "cold-hot")):
+            manifest = copy.deepcopy(self.manifest)
+            del manifest[section][key]
+            with self.subTest(section=section), self.assertRaises((ValueError, KeyError)):
+                harness.validate_manifest(manifest)
+
+    def test_short_measured_window_and_fewer_rounds_rejected(self):
+        self.manifest["evidence"] = "measured"
+        with self.assertRaisesRegex(ValueError, "duration_seconds"):
+            harness.validate_manifest(self.manifest)
+        self.manifest["duration_seconds"] = 900
+        self.manifest["rounds"] = 2
+        with self.assertRaisesRegex(ValueError, "rounds"):
+            harness.validate_manifest(self.manifest)
+
+    def test_existing_data_preserved(self):
+        (self.root / "data").mkdir()
+        marker = self.root / "data/keep"
+        marker.write_text("existing")
+        with self.assertRaisesRegex(ValueError, "preserved"):
+            self.run_harness()
+        self.assertEqual(marker.read_text(), "existing")
+
+    def test_invalid_or_live_write_window_does_not_claim_p2(self):
+        self.assertIsNone(harness.convergence({"convergence": {"writes_stopped": False}}))
+        with self.assertRaises(ValueError):
+            harness.convergence({"convergence": {"writes_stopped": True, "last_mutation_observed": True,
+                                                 "first_complete_publication": True}})
+
+    def test_nan_and_oversized_samples_rejected(self):
+        with self.assertRaises(ValueError):
+            harness.number(float("nan"), "latency")
+        path = self.root / "oversized.json"
+        path.write_bytes(b" " * (harness.MAX_JSON_BYTES + 1))
+        with self.assertRaisesRegex(ValueError, "oversized"):
+            harness.read_json(path)
+
+
+if __name__ == "__main__":
+    if len(sys.argv) == 4 and sys.argv[1] in ("prepare", "measure", "oracle", "stop"):
+        sys.exit(fake_adapter())
+    unittest.main()
@@ -312,3 +312,5 @@ if PATH="$BIN_DIR:$PATH" "$SCRIPT" --secret-key rustfsadmin >"$secret_arg_log" 2
 fi

 grep -q -- 'unknown arg: --secret-key' "$secret_arg_log"
+
+"$ROOT_DIR/scripts/python_bin.sh" "$ROOT_DIR/scripts/test_scanner_abba.py"