rustfs/scripts/scanner_abba.py

#!/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 select
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",
    "foreground_pressure_samples", "foreground_pressure_high_samples",
    "heal_mainline_throttle_delayed",
    "heal_lock_wait_p99_ms", "heal_attempts", "heal_attempt_failures",
    "heal_retry_attempts",
)
REPEATABILITY_LIMIT = Decimal("0.05")
P2_WORK_MULTIPLE_LIMIT = Decimal("1.2")
RELEASE_PROFILE_ARTIFACTS = (
    "allocation-profile",
    "flamegraph",
    "rss-samples",
    "save-frequency",
)
RELEASE_FAULT_MODES = (
    "process-restart",
    "process-crash-restart",
)


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 relative_change_or_none(current, baseline, name):
    if decimal_number(baseline, f"{name} baseline") == 0:
        return None
    return relative_change(current, baseline, name)


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:
        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 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")
    validate_release_evidence_manifest(manifest)


def release_evidence_integer(value, name, minimum=1, maximum=1024):
    require(type(value) is int and minimum <= value <= maximum, f"invalid release_evidence.{name}")
    return value


def release_evidence_string(value, name):
    require(isinstance(value, str) and value.strip(), f"missing release_evidence.{name}")
    return value


def release_evidence_bool(value, name):
    require(type(value) is bool, f"invalid release_evidence.{name}")
    return value


def release_evidence_true(value, name):
    release_evidence_bool(value, name)
    require(value is True, f"missing release_evidence.{name}")


def validate_release_evidence_manifest(manifest):
    if manifest["evidence"] != "measured":
        return

    evidence = manifest.get("release_evidence")
    require(isinstance(evidence, dict), "missing release_evidence for measured ABBA")

    topology = evidence.get("topology")
    require(isinstance(topology, dict), "missing release_evidence.topology")
    nodes = release_evidence_integer(topology.get("nodes"), "topology.nodes", 3, 64)
    drives = release_evidence_integer(topology.get("drives_per_node"), "topology.drives_per_node", 1, 64)
    set_size = release_evidence_integer(topology.get("erasure_set_size"), "topology.erasure_set_size", 12, 12)
    data = release_evidence_integer(topology.get("erasure_data_blocks"), "topology.erasure_data_blocks", 8, 8)
    parity = release_evidence_integer(topology.get("erasure_parity_blocks"), "topology.erasure_parity_blocks", 4, 4)
    require(data + parity == set_size, "release_evidence.topology must be EC8+4")
    require(nodes * drives >= set_size, "release_evidence.topology cannot host one EC8+4 set")
    pools = release_evidence_integer(topology.get("pools"), "topology.pools", 1)
    sets_total = release_evidence_integer(topology.get("sets_total"), "topology.sets_total", 1)
    sampled_pools = release_evidence_integer(topology.get("sampled_pools"), "topology.sampled_pools", 2)
    sampled_sets = release_evidence_integer(topology.get("sampled_sets"), "topology.sampled_sets", 2)
    require(sampled_pools <= pools, "release_evidence.topology sampled pools exceed total pools")
    require(sampled_sets <= sets_total, "release_evidence.topology sampled sets exceed total sets")

    distributed = evidence.get("distributed")
    require(isinstance(distributed, dict), "missing release_evidence.distributed")
    endpoints = distributed.get("metrics_endpoints")
    require(isinstance(endpoints, list) and len(endpoints) >= nodes, "missing release_evidence.distributed.metrics_endpoints")
    require(
        all(isinstance(endpoint, str) and endpoint.strip() for endpoint in endpoints)
        and len(set(endpoints)) == len(endpoints),
        "invalid release_evidence.distributed.metrics_endpoints",
    )
    release_evidence_string(distributed.get("failure_domain"), "distributed.failure_domain")
    release_evidence_true(distributed.get("same_window_sampling"), "distributed.same_window_sampling")

    crash = evidence.get("crash_restart")
    require(isinstance(crash, dict), "missing release_evidence.crash_restart")
    fault_modes = crash.get("fault_modes")
    require(
        isinstance(fault_modes, list)
        and all(mode in fault_modes for mode in RELEASE_FAULT_MODES)
        and all(isinstance(mode, str) and mode.strip() for mode in fault_modes),
        "missing release_evidence.crash_restart.fault_modes",
    )
    release_evidence_true(crash.get("unclean_shutdown_marker"), "crash_restart.unclean_shutdown_marker")

    mixed = evidence.get("mixed_version")
    require(isinstance(mixed, dict), "missing release_evidence.mixed_version")
    revisions = mixed.get("participating_revisions")
    require(
        isinstance(revisions, list)
        and len(set(revisions)) >= 2
        and all(isinstance(revision, str) and len(revision) == 40 and all(c in "0123456789abcdef" for c in revision)
                for revision in revisions),
        "invalid release_evidence.mixed_version.participating_revisions",
    )
    for revision in (manifest["baseline"]["revision"], manifest["candidate"]["revision"]):
        require(revision in revisions, "release_evidence.mixed_version omits tested build revision")
    for key in ("reader", "writer", "rollback_payload"):
        require(mixed.get(key) is True, f"missing release_evidence.mixed_version.{key}")

    profile = evidence.get("profile")
    require(isinstance(profile, dict), "missing release_evidence.profile")
    artifacts = profile.get("required_artifacts")
    require(
        isinstance(artifacts, list)
        and all(item in artifacts for item in RELEASE_PROFILE_ARTIFACTS)
        and all(isinstance(item, str) and item.strip() for item in artifacts),
        "missing release_evidence.profile.required_artifacts",
    )
    for key in ("collector_config_sha256", "profiler_config_sha256"):
        require(sha(profile.get(key)), f"invalid release_evidence.profile.{key}")


class OwnedCommand:
    """Keep the session leader unreaped until its group's last signal is sent."""

    def __init__(self, args, log):
        require(sys.platform == "darwin" or hasattr(os, "waitid"), "non-reaping child observation is unavailable")
        self.args, self.status = args, None
        self.queue = select.kqueue() if sys.platform == "darwin" else None
        self.process = None
        read_gate, write_gate = os.pipe()
        try:
            # The shell has already exec'd when Popen returns. Gate the target
            # until kqueue is registered; preexec_fn would deadlock Popen here.
            gate = f'read -r _scanner_gate <&{read_gate} || exit 125; exec {read_gate}<&-; exec "$@"'
            self.process = subprocess.Popen(["bash", "-c", gate, "scanner-abba", *args], pass_fds=(read_gate,),
                                            stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
            if self.queue is not None:
                # Darwin NOTE_EXITSTATUS is not exposed by Python's select constants.
                event = select.kevent(self.process.pid, filter=select.KQ_FILTER_PROC,
                                      flags=select.KQ_EV_ADD | select.KQ_EV_ONESHOT,
                                      fflags=select.KQ_NOTE_EXIT | 0x04000000)
                self.queue.control([event], 0, 0)
            os.write(write_gate, b"\n")
        except BaseException:
            try:
                if self.process is not None:
                    try:
                        self._signal_group(signal.SIGKILL)
                    finally:
                        self.process.wait(timeout=10)
            finally:
                if self.queue is not None:
                    self.queue.close()
            raise
        finally:
            os.close(read_gate)
            os.close(write_gate)

    def wait(self, timeout):
        if self.status is not None:
            return self.status
        deadline = time.monotonic() + timeout
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise subprocess.TimeoutExpired(self.args, timeout)
            if self.queue is not None:
                events = self.queue.control(None, 1, remaining)
                if events:
                    self.status = os.waitstatus_to_exitcode(events[0].data)
                    return self.status
            else:
                result = os.waitid(os.P_PID, self.process.pid, os.WEXITED | os.WNOWAIT | os.WNOHANG)
                if result is not None:
                    self.status = result.si_status if result.si_code == os.CLD_EXITED else -result.si_status
                    return self.status
                time.sleep(min(0.05, remaining))

    def _signal_group(self, sig):
        try:
            os.killpg(self.process.pid, sig)
            return True
        except ProcessLookupError:
            return False

    def finish(self, terminate=False):
        if self.process.returncode is not None:
            return self.process.returncode
        try:
            if terminate:
                try:
                    self._signal_group(signal.SIGTERM)
                    deadline = time.monotonic() + 10
                    while time.monotonic() < deadline and self._signal_group(0):
                        time.sleep(0.05)
                finally:
                    # Keep the PID reserved through the last group signal, even
                    # when the cleanup grace period itself is interrupted.
                    self._signal_group(signal.SIGKILL)
        finally:
            try:
                returncode = self.process.wait(timeout=10)
            finally:
                if self.queue is not None:
                    self.queue.close()
        return returncode


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 = OwnedCommand([str(adapter), action, str(request), str(output)], log)
        try:
            returncode = process.wait(timeout)
            if returncode:
                raise subprocess.CalledProcessError(returncode, [str(adapter), action])
            result = read_json(output)
        except BaseException:
            process.finish(terminate=True)
            raise
        else:
            # Successful prepare may intentionally leave adapter-owned services.
            process.finish()
            return result


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")
    if request["evidence"] == "measured":
        require(result.get("release_evidence") == request["release_evidence"], "release evidence provenance 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["foreground_pressure_samples"] > 0, "zero foreground pressure samples")
    require(metrics["foreground_pressure_high_samples"] <= metrics["foreground_pressure_samples"],
            "foreground pressure high samples exceed samples")
    require(metrics["heal_attempt_failures"] <= metrics["heal_attempts"], "heal failures exceed attempts")
    require(metrics["heal_retry_attempts"] <= metrics["heal_attempts"], "heal retries exceed attempts")
    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 attempt_cost(metrics):
    healed = decimal_number(metrics["healed_objects"], "healed_objects")
    if healed == 0:
        return None
    return ratio(metrics["heal_attempts"], healed, "heal attempt cost")


def pressure_high_ratio(metrics):
    return ratio(metrics["foreground_pressure_high_samples"],
                 metrics["foreground_pressure_samples"], "foreground pressure high samples")


def scanner_cache_cost(metrics):
    walked = decimal_number(metrics["walk_objects"], "walk_objects")
    encoded = decimal_number(metrics["encode_bytes"], "encode_bytes")
    return {
        "clone_bytes_per_walk_object": None if walked == 0 else float(ratio(metrics["cache_clone_bytes"], walked, "clone bytes per walk object")),
        "encode_bytes_per_walk_object": None if walked == 0 else float(ratio(encoded, walked, "encode bytes per walk object")),
        "save_bytes_per_walk_object": None if walked == 0 else float(ratio(metrics["save_bytes"], walked, "save bytes per walk object")),
        "clone_to_encode_byte_ratio": None if encoded == 0 else float(ratio(metrics["cache_clone_bytes"], encoded, "clone to encode bytes")),
        "save_to_encode_byte_amplification": None if encoded == 0 else float(ratio(metrics["save_bytes"], encoded, "save to encode bytes")),
    }


def scanner_cache_cost_change(candidate, baseline):
    changes = {}
    for key in ("cache_clone_bytes", "encode_bytes", "save_bytes"):
        change = relative_change_or_none(candidate[key], baseline[key], key)
        changes[f"{key}_change"] = None if change is None else float(change)
    return changes


def running_heal_pacing(group, baseline, candidate, p99, throughput, noisy):
    if group[0]["scenario"] != "running-heal" or group[0]["comparison"] != "build":
        return None
    baseline_seconds = sum(decimal_number(cell["result"]["elapsed_seconds"], "elapsed_seconds") for cell in (group[0], group[3])) / 2
    candidate_seconds = sum(decimal_number(cell["result"]["elapsed_seconds"], "elapsed_seconds") for cell in (group[1], group[2])) / 2
    baseline_rate = ratio(baseline["heal_attempts"], baseline_seconds, "baseline heal attempt rate")
    candidate_rate = ratio(candidate["heal_attempts"], candidate_seconds, "candidate heal attempt rate")
    rate_change = relative_change_or_none(candidate_rate, baseline_rate, "heal attempt rate")
    candidate_high_ratio = pressure_high_ratio(candidate)
    baseline_delayed = decimal_number(baseline["heal_mainline_throttle_delayed"], "baseline pacing delays")
    delayed = decimal_number(candidate["heal_mainline_throttle_delayed"], "candidate pacing delays")
    pacing_observed = candidate_high_ratio > 0 and delayed > 0
    foreground_improved = p99 < 0 or throughput > 0
    status = (
        "inconclusive"
        if noisy
        else "observed"
        if pacing_observed and foreground_improved
        else "no_measured_benefit"
        if pacing_observed
        else "pending"
    )
    return {
        "status": status,
        "pacing_observed": pacing_observed,
        "candidate_pressure_high_ratio": float(candidate_high_ratio),
        "baseline_delay_events": float(baseline_delayed),
        "candidate_delay_events": float(delayed),
        "baseline_heal_attempts_per_second": float(baseline_rate),
        "candidate_heal_attempts_per_second": float(candidate_rate),
        "heal_attempt_rate_change": None if rate_change is None else float(rate_change),
        "foreground_p99_change": float(p99),
        "foreground_throughput_change": float(throughput),
    }


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 ratio(window["walk_objects"], window["full_walk_objects"], "convergence work")


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)
        p2_report = [None if value is None else float(value) for value in p2]
        attempt_costs = [attempt_cost(cell["result"]["metrics"]) if cell["background"] == "on" else None for cell in group]
        candidate_attempt_costs = [
            value for cell, value in zip(group, attempt_costs) if cell["leg"].startswith("B") and value is not None
        ]
        w10 = running_heal_pacing(group, a, b, p99, throughput, noise)
        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_report,
                            "w22": {
                                "baseline": scanner_cache_cost(a),
                                "candidate": scanner_cache_cost(b),
                                "candidate_vs_baseline": scanner_cache_cost_change(b, a),
                            },
                            "w10": w10,
                            "w10_w11": {
                                "foreground_pressure_high_sample_ratios": [
                                    float(pressure_high_ratio(cell["result"]["metrics"])) for cell in group
                                ],
                                "heal_lock_wait_p99_ms": [
                                    cell["result"]["metrics"]["heal_lock_wait_p99_ms"] for cell in group
                                ],
                                "attempt_cost_per_healed_object": [
                                    None if value is None else float(value) for value in attempt_costs
                                ],
                                "candidate_attempt_cost_per_healed_object": (
                                    None if not candidate_attempt_costs else float(max(candidate_attempt_costs))
                                ),
                            }})
    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")
    expected_metrics_endpoints = None
    if request.get("evidence") == "measured":
        expected_metrics_endpoints = request["release_evidence"]["distributed"]["metrics_endpoints"]
    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 = OwnedCommand(args, log)
        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(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]
            if expected_metrics_endpoints is not None:
                require(endpoints == expected_metrics_endpoints,
                        "collector metrics endpoints do not match release evidence")
            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:
            process.finish(terminate=True)


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]}
                        if manifest["evidence"] == "measured":
                            request["release_evidence"] = manifest["release_evidence"]
                        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())