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
from datetime import datetime, timezone

from scanner_client_trace import analyze as reconcile_client_ledger

SCENARIOS = ("cold-hot", "fresh-hot", "multi-hot-new", "running-heal", "mrf-replay")
LEGS = ("A1", "B1", "B2", "A2")
MAX_JSON_BYTES = 1024 * 1024
METRICS = (
    "p95_ms", "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",
    "heal_start_p95_ms", "heal_duplicate_task_count", "heal_lock_hold_p95_ms",
)
REPEATABILITY_LIMIT = Decimal("0.05")
P2_WORK_MULTIPLE_LIMIT = Decimal("1.2")
W11_RSS_GROWTH_LIMIT = Decimal("0.05")
RELEASE_PROFILE_ARTIFACTS = (
    "allocation-profile",
    "flamegraph",
    "rss-samples",
    "save-frequency",
)
MIN_MEASURED_RELEASE_DURATION_SECONDS = 7200
RELEASE_FAULT_MODES = (
    "process-restart",
    "process-crash-restart",
)
RELEASE_SCHEDULER_BOUNDS = (
    "admission-retry-idempotency",
    "deadline-budget",
    "lock-hold-bound",
    "minimum-progress",
)


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, "oversized JSON input")
    with path.open() as stream:
        value = json.load(stream)
    require(isinstance(value, dict), "expected JSON object")
    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 measured_provenance_verified(_output):
    # Collector snapshots and adapter echoes do not yet establish a common
    # source window or exact client percentiles. Keep acceptance closed until
    # the original observations can be independently verified.
    return False


def safe_error_message(error):
    if isinstance(error, subprocess.TimeoutExpired):
        timeout = error.timeout
        return f"command timed out after {timeout} seconds" if type(timeout) in (int, float) else "command timed out"
    if isinstance(error, subprocess.CalledProcessError):
        return f"command exited with status {error.returncode}" if type(error.returncode) is int else "command failed"
    if isinstance(error, subprocess.SubprocessError):
        return type(error).__name__
    if isinstance(error, OSError):
        return f"{type(error).__name__}: {os.strerror(error.errno)}" if type(error.errno) is int else type(error).__name__
    return str(error) or type(error).__name__


def add_safe_error_note(error, note, related):
    error.add_note(note)
    error._scanner_safe_notes = [*getattr(error, "_scanner_safe_notes", []), note]
    error._scanner_related_errors = [*getattr(error, "_scanner_related_errors", []), related]


def error_chain(error):
    messages = []
    seen = set()
    pending = [error]
    while pending and len(seen) < 16:
        error = pending.pop(0)
        if id(error) in seen:
            continue
        seen.add(id(error))
        parts = [safe_error_message(error), *getattr(error, "_scanner_safe_notes", [])[:4]]
        messages.extend(part[:2048] for part in parts if part)
        related = getattr(error, "_scanner_related_errors", [])
        if isinstance(related, (list, tuple)):
            pending.extend(item for item in related[:4] if isinstance(item, BaseException))
        next_error = error.__cause__ or (None if error.__suppress_context__ else error.__context__)
        if next_error is not None:
            pending.append(next_error)
    return "; ".join(messages)


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 = MIN_MEASURED_RELEASE_DURATION_SECONDS 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 release_evidence_exact_strings(value, expected, name):
    require(isinstance(value, list) and all(isinstance(item, str) and item.strip() for item in value),
            f"invalid release_evidence.{name}")
    observed = set(value)
    require(len(observed) == len(value), f"duplicate release_evidence.{name}")
    missing = sorted(set(expected) - observed)
    require(not missing, f"missing release_evidence.{name}: {', '.join(missing)}")
    unknown = sorted(observed - set(expected))
    require(not unknown, f"unknown release_evidence.{name} entries: {len(unknown)}")
    return value


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

    scheduler = evidence.get("scheduler")
    require(isinstance(scheduler, dict), "missing release_evidence.scheduler")
    release_evidence_exact_strings(scheduler.get("bounds"), RELEASE_SCHEDULER_BOUNDS, "scheduler.bounds")
    release_evidence_integer(scheduler.get("max_deferred_items"), "scheduler.max_deferred_items", 1, 2**31 - 1)
    release_evidence_integer(scheduler.get("max_deferred_bytes"), "scheduler.max_deferred_bytes", 1, 2**63 - 1)
    release_evidence_integer(scheduler.get("max_retry_age_seconds"), "scheduler.max_retry_age_seconds", 1, 86400)
    release_evidence_true(scheduler.get("duplicate_task_bound_observed"), "scheduler.duplicate_task_bound_observed")

    crash = evidence.get("crash_restart")
    require(isinstance(crash, dict), "missing release_evidence.crash_restart")
    release_evidence_exact_strings(crash.get("fault_modes"), RELEASE_FAULT_MODES, "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")
    baseline_revision = manifest["baseline"]["revision"]
    candidate_revision = manifest["candidate"]["revision"]
    require(baseline_revision != candidate_revision,
            "release_evidence.mixed_version requires distinct baseline and candidate revisions")
    require(manifest["baseline"]["sha256"] != manifest["candidate"]["sha256"],
            "release_evidence.mixed_version requires distinct baseline and candidate binaries")
    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 (baseline_revision, 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")
    release_evidence_exact_strings(profile.get("required_artifacts"), RELEASE_PROFILE_ARTIFACTS,
                                   "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):
        if self.process.returncode is not None:
            return False
        try:
            os.killpg(self.process.pid, sig)
            return True
        except ProcessLookupError:
            return False
        except PermissionError:
            if sys.platform != "darwin":
                raise
            # Darwin rejects signals to zombie-only groups. The unreaped
            # session leader reserves this PGID; require its zombie plus no
            # live members before treating EPERM as completed cleanup.
            states = subprocess.check_output(["ps", "-axo", "pid=,pgid=,stat="], text=True, timeout=1)
            group = [(pid, state) for pid, pgid, state in (line.split() for line in states.splitlines())
                     if pgid == str(self.process.pid)]
            if not any(pid == str(self.process.pid) and state.startswith("Z") for pid, state in group) \
                    or any(not state.startswith("Z") for _, state in group):
                raise
            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_client_request_samples(result, request, artifact_dir):
    reference = result.get("client_request_samples")
    require(isinstance(reference, dict), "missing client request samples")
    require(artifact_dir is not None, "client request samples require a cell artifact directory")
    observed = reconcile_client_ledger(Path(artifact_dir), reference, request, result.get("stats"), result.get("cancelled"))
    require(observed["sample_complete"] is True and observed["offered_load_complete"] is True
            and observed["p99_ns"] is not None, "client schedule or percentile evidence is incomplete")
    require(observed["errors"] == 0 and observed["http_503"] == 0, "client request errors or status unavailable")
    require(result.get("client_observation") == observed, "client observation differs from original ledgers")
    require(result.get("elapsed_seconds") == observed["elapsed_seconds"], "client request elapsed time mismatch")
    for key in ("requests", "errors", "p95_ms", "p99_ms", "throughput_ops"):
        require(result["metrics"][key] == observed[key], f"client request {key} mismatch")
    return observed


def validate_result(result, request, expected, artifact_dir=None):
    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["heal_duplicate_task_count"] == 0, "duplicate heal task admission")
    require(metrics["heal_start_p95_ms"] > 0, "zero heal start p95")
    require(metrics["heal_lock_hold_p95_ms"] > 0, "zero heal lock hold p95")
    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")
    if request["evidence"] == "measured":
        validate_client_request_samples(result, request, artifact_dir)
    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 bounded_retry_window(group, baseline, candidate, p99, throughput, noisy, candidate_attempt_costs):
    if group[0]["scenario"] != "running-heal" or group[0]["comparison"] != "build":
        return {"status": "not_applicable"}

    rss_growth = relative_change_or_none(candidate["rss_bytes"], baseline["rss_bytes"], "rss_bytes")
    lock_wait_change = relative_change_or_none(
        candidate["heal_lock_wait_p99_ms"], baseline["heal_lock_wait_p99_ms"], "heal lock wait p99",
    )
    latency_improved = p99 < 0 or throughput > 0
    lock_wait_improved = lock_wait_change is not None and lock_wait_change < 0
    rss_within_limit = rss_growth is not None and rss_growth <= W11_RSS_GROWTH_LIMIT
    attempt_cost_available = bool(candidate_attempt_costs)
    status = (
        "inconclusive"
        if noisy
        else "observed"
        if latency_improved and lock_wait_improved and rss_within_limit and attempt_cost_available
        else "rss_regression"
        if latency_improved and lock_wait_improved and not rss_within_limit
        else "no_measured_benefit"
        if attempt_cost_available
        else "pending"
    )
    return {
        "status": status,
        "rss_growth_limit": float(W11_RSS_GROWTH_LIMIT),
        "rss_growth": None if rss_growth is None else float(rss_growth),
        "rss_within_limit": rss_within_limit,
        "baseline_rss_bytes": float(baseline["rss_bytes"]),
        "candidate_rss_bytes": float(candidate["rss_bytes"]),
        "baseline_heal_lock_wait_p99_ms": float(baseline["heal_lock_wait_p99_ms"]),
        "candidate_heal_lock_wait_p99_ms": float(candidate["heal_lock_wait_p99_ms"]),
        "heal_lock_wait_p99_change": None if lock_wait_change is None else float(lock_wait_change),
        "healthy_page_latency_observed": latency_improved,
        "foreground_p99_change": float(p99),
        "foreground_throughput_change": float(throughput),
        "candidate_attempt_cost_per_healed_object": (
            None if not candidate_attempt_costs else float(max(candidate_attempt_costs))
        ),
    }


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 comparison_metrics(cell):
    metrics = cell["result"]["metrics"].copy()
    if cell["evidence"] == "measured":
        observation = cell["result"].get("client_observation")
        require(isinstance(observation, dict), "missing measured client observation")
        for percentile in (95, 99):
            key = f"p{percentile}_ns"
            require(type(observation.get(key)) is int and observation[key] >= 0, f"invalid measured {key}")
            metrics[f"p{percentile}_ms"] = Decimal(observation[key]) / Decimal(1_000_000)
    return metrics


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 = (comparison_metrics(cell) 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),
                  "baseline_walk_objects": int(a["walk_objects"]),
                  "baseline_cold_walk_objects": int(a["cold_walk_objects"]),
                  "candidate_walk_objects": int(b["walk_objects"]),
                  "candidate_cold_walk_objects": int(b["cold_walk_objects"])}
            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)
        w11 = bounded_retry_window(group, a, b, p99, throughput, noise, candidate_attempt_costs)
        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),
                            # Four disjoint legs do not supply one combined percentile.
                            "foreground_p95_ms": None, "foreground_p99_ms": None,
                            "foreground_percentiles_ms_by_leg": {
                                cell["leg"]: {"p95": float(metrics["p95_ms"]), "p99": float(metrics["p99_ms"])}
                                for cell, metrics in zip(group, (a1, b1, b2, a2))
                            },
                            "throughput_ops": float(min(a["throughput_ops"], b["throughput_ops"])),
                            "error_rate": float(max(a["errors"] / a["requests"], b["errors"] / b["requests"])),
                            "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,
                            "w11": w11,
                            "w10_w11": {
                                "foreground_pressure_samples": [
                                    cell["result"]["metrics"]["foreground_pressure_samples"] for cell in group
                                ],
                                "foreground_pressure_high_samples": [
                                    cell["result"]["metrics"]["foreground_pressure_high_samples"] for cell in group
                                ],
                                "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))
                                ),
                            },
                            "w09": {
                                "heal_start_p95_ms": [
                                    cell["result"]["metrics"]["heal_start_p95_ms"] for cell in group
                                ],
                                "heal_duplicate_task_count": [
                                    cell["result"]["metrics"]["heal_duplicate_task_count"] for cell in group
                                ],
                                "heal_lock_hold_p95_ms": [
                                    cell["result"]["metrics"]["heal_lock_hold_p95_ms"] for cell in group
                                ],
                            }})
    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.get("collector")
    require(isinstance(connection, dict) and set(connection) == {"alias", "endpoint", "metrics_endpoints"},
            "invalid collector connection")
    require(all(isinstance(value, str) and value for value in connection.values()), "missing collector endpoint")
    require(os.environ.get("RUSTFS_ACCESS_KEY"), "collector requires RUSTFS_ACCESS_KEY")
    require(os.environ.get("RUSTFS_SECRET_KEY"), "collector requires RUSTFS_SECRET_KEY")
    expected_metrics_endpoints = None
    if request.get("evidence") == "measured":
        expected_metrics_endpoints = request["release_evidence"]["distributed"]["metrics_endpoints"]
    output = request_path.parent / "telemetry"
    interval_seconds = 60
    sample_count = request["duration_seconds"] // interval_seconds + 1
    collector_timeout_seconds = (sample_count - 1) * interval_seconds + 300
    args = ["bash", str(collector), "--alias", connection["alias"], "--endpoint", connection["endpoint"],
            "--metrics-endpoints", connection["metrics_endpoints"], "--deployment", "distributed",
            "--samples", str(sample_count), "--interval-secs", str(interval_seconds),
            "--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(collector_timeout_seconds) == 0,
                    f"scanner collector failed within {collector_timeout_seconds}s timeout")
            require(output.joinpath("scanner-summary.csv").stat().st_size > 0, "missing collector samples")
            samples = list((output / "status").glob("scanner-status.*.json"))
            require(len(samples) == sample_count, "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"))
    started_at = datetime.now(timezone.utc).replace(microsecond=0)
    manifest["started_at"] = started_at.isoformat().replace("+00:00", "Z")
    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(mode=0o700)
                        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)
                        primary_error = None
                        try:
                            prepared = invoke(adapter, "prepare", request_path, 300)
                            require(prepared.get("ready") is True, "deployment not ready")
                            if manifest["evidence"] != "synthetic":
                                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"], cell_dir)
                            cells.append({**request, "result": result})
                        except BaseException as error:
                            primary_error = error
                            raise
                        finally:
                            try:
                                stopped = invoke(adapter, "stop", request_path, 300)
                                require(stopped.get("stopped") is True, "adapter failed to stop deployment")
                            except BaseException as stop_error:
                                if primary_error is not None:
                                    if not isinstance(stop_error, Exception):
                                        add_safe_error_note(stop_error, "prior operation failed", primary_error)
                                        raise stop_error from None
                                    if not isinstance(primary_error, Exception):
                                        prior = primary_error.__cause__ or (
                                            None if primary_error.__suppress_context__ else primary_error.__context__
                                        )
                                        if prior is not None:
                                            add_safe_error_note(primary_error, "prior operation failed", prior)
                                        add_safe_error_note(primary_error, "adapter stop failed", stop_error)
                                        raise primary_error from None
                                    combined = ValueError(
                                        f"{safe_error_message(primary_error)}; adapter stop failed: {safe_error_message(stop_error)}"
                                    )
                                    combined._scanner_related_errors = [primary_error, stop_error]
                                    raise combined from primary_error
                                raise
        status, comparisons = evaluate(cells)
        synthetic = manifest["evidence"] == "synthetic"
        verified = manifest["evidence"] == "measured" and measured_provenance_verified(output)
        if not synthetic and not verified:
            # The adapter currently echoes the release manifest without proving
            # source windows or client latency samples. Keep results diagnostic
            # until those observations can be independently verified.
            for comparison in comparisons:
                comparison["measurement_provenance"] = "unverified"
                if comparison["status"] == "pass":
                    comparison["status"] = "inconclusive"
            if status == "pass":
                status = "inconclusive"
        finished_at = datetime.now(timezone.utc).replace(microsecond=0)
        report = {"status": "synthetic_validated" if synthetic and status == "pass" else status,
                  "evidence": manifest["evidence"], "performance": "pending" if synthetic or status == "inconclusive" else status,
                  "cells": len(cells), "comparisons": comparisons,
                  "started_at": manifest["started_at"], "finished_at": finished_at.isoformat().replace("+00:00", "Z")}
        if not synthetic:
            report["measurement_provenance"] = "verified" if verified else "unverified"
        write_json(output / "report.json", report)
        return 0 if status == "pass" else 3 if status == "inconclusive" else 1
    except BaseException as error:
        finished_at = datetime.now(timezone.utc).replace(microsecond=0)
        error_text = error_chain(error)
        report = {"status": "failed", "performance": "pending", "evidence": manifest["evidence"],
                  "completed_cells": len(cells), "error": error_text, "started_at": manifest.get("started_at"),
                  "finished_at": finished_at.isoformat().replace("+00:00", "Z")}
        if manifest["evidence"] != "synthetic":
            report["measurement_provenance"] = "unverified"
        write_json(output / "report.json", report)
        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 KeyboardInterrupt as error:
        print(f"ERROR: {error_chain(error)}", file=sys.stderr)
        return 130
    except Exception as error:
        print(f"ERROR: {error_chain(error)}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())