#!/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())