#!/usr/bin/env python3

"""P4 robustness suite runner: crash points, fault injection, concurrency, env faults.

Writes robustness-lane JSONL rows (one per case) through oakbench.ResultsStore

into ``--results`` (default results/robustness, never committed, ADR-0003),

flushing each row to a .partial.jsonl as it completes, and prints a Markdown

classification table. Every case is classified held / violated / unsupported /

harness_error (see README.md). Loopback servers and local fixtures only.

Examples:

  OAK_BENCH_OAK_READER_PYTHON=/path/to/py python3 suites/robustness/run_robustness.py --quick

  python3 suites/robustness/run_robustness.py --subjects oak,git --families crash_time,proxy

"""

from __future__ import annotations

import argparse

import json

import platform

import shutil

import sys

import tempfile

import time

from datetime import datetime, timezone

from pathlib import Path

from typing import Any, Callable, Optional

sys.path.insert(0, str(Path(__file__).resolve().parent))

from robustness import CLASSIFICATIONS, HARNESS_ERROR, HELD, SUITE_VERSION, UNSUPPORTED, VIOLATED  # noqa: E402

from robustness import crashpoints  # noqa: E402

from robustness.crash import (TIME_FRACTIONS, calibrate, run_trial, syscall_fault, syscall_points,  # noqa: E402

                              time_fault)

from robustness.fixtures import FIXTURE_VERSION, build_git, build_oak  # noqa: E402

from robustness.procs import free_port  # noqa: E402

from oakbench.results import ResultsStore  # noqa: E402

from oakbench.rows import SKIP_RETURNCODE  # noqa: E402

from oakbench.subjects import binary_sha256, which_measured_git  # noqa: E402

LANE = "robustness"

FAMILIES = ("crash_time", "crash_syscall", "proxy", "concurrency", "env", "hosted")

CRASH_OPS = ("commit", "push", "pull_ff", "pull_diverged", "clone", "switch", "merge_local", "merge_hosted",

             "mount_finish")

PROXY_OPS = ("push", "pull_ff", "pull_diverged", "clone")

SCENARIO = {

    "crash_time": "crash_time_v1",

    "crash_syscall": "crash_syscall_v1",

    "proxy": "fault_proxy_v1",

}

RETURNCODE = {HELD: 0, VIOLATED: 1, HARNESS_ERROR: 2, UNSUPPORTED: SKIP_RETURNCODE}

MEASUREMENT_SOURCE = "independent_reader_invariants_v1"

UNSUPPORTED_OPS = {

    ("oak", "merge_local"): "oak has no local branch-to-branch merge; `oak merge` is the hosted squash-merge (see merge_hosted)",

    ("oak", "merge_hosted"): "loopback oak serve 0.105.0 answers merge publication with 404; hosted merge needs a local oakspace (hosted_merge_invariants_v1)",

    ("git", "merge_hosted"): "git has no hosted merge; the git differential is merge_local",

    ("oak", "mount_finish"): "oak mount needs macOS FSKit (signed Oak Mount app) or Linux FUSE; not exercised against loopback serve in suite v1",

    ("git", "mount_finish"): "git has no mount",

}

class Runner:

    def __init__(self, args: argparse.Namespace):

        self.args = args

        self.bench_id = args.bench_id or datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")

        self.work = Path(args.workdir or tempfile.mkdtemp(prefix="p4-robustness-")).resolve()

        self.work.mkdir(parents=True, exist_ok=True)

        self.results = Path(args.results).resolve()

        self.evidence = self.results / f"{self.bench_id}.robustness-evidence"

        self.store = ResultsStore(self.results, lane=LANE, filename_suffix="robustness", flush_partial=True)

        self.rows: list[dict] = []

        self.run_index: dict[tuple, int] = {}

        self.retained = 0

        self.bins = {"oak": args.oak_bin, "git": args.git_bin}

        self.instrument = None

        self.instrument_reason = ""

    # rows ----------------------------------------------------------------------

    def emit(self, subject_kind: str, scenario: str, operation: str, family: str, data: dict) -> dict:

        key = (subject_kind, scenario, operation)

        run = self.run_index.get(key, 0)

        self.run_index[key] = run + 1

        classification = data.get("classification", UNSUPPORTED if "unsupported" in data else HARNESS_ERROR)

        binary = self.bins.get(subject_kind)

        row: dict[str, Any] = {

            "bench_id": self.bench_id,

            "suite_version": SUITE_VERSION,

            "scenario": scenario,

            "operation": operation,

            "run": run,

            "subject": "oak_installed" if subject_kind == "oak" else "git",

            "subject_kind": subject_kind,

            "subject_bin": binary,

            "binary_sha256": binary_sha256(Path(binary)) if binary and Path(binary).exists() else None,

            "subject_version": self.versions.get(subject_kind),

            "fault_family": family,

            "fault_point": data.get("fault_point", "n/a"),

            "fault_fired": data.get("fault_fired", data.get("observations", {}).get("fault_fired")),

            "crash_model": "process_kill" if family.startswith("crash") or family == "proxy" else None,

            "classification": classification,

            "returncode": RETURNCODE[classification],

            "elapsed_ms": float(data.get("elapsed_ms", 0.0) or 0.0),

            "invariants": data.get("invariants", {}),

            "violations": data.get("violations", sorted(k for k, v in data.get("invariants", {}).items() if v is False)),

            "observations": data.get("observations", {}),

            "command": data.get("command", []),

            "fixture_version": FIXTURE_VERSION,

            "measurement_source": MEASUREMENT_SOURCE,

            "host_platform": platform.platform(),

        }

        if classification == UNSUPPORTED:

            row.update(skipped=True, skip_reason=data.get("unsupported") or data.get("skip_reason") or "unsupported",

                       measurement_source=None)

            row.pop("measurement_source")

        if data.get("harness_error"):

            row["harness_error"] = data["harness_error"]

        transport = data.get("remote_transport")

        if transport:

            row.update(remote_transport=transport, remote_server=data.get("remote_server"),

                       remote_repo=data.get("remote_repo"))

        if self.instrument is not None and family == "crash_syscall":

            row["crash_instrument"] = self.instrument.provenance

        self.rows.append(row)

        self.store.append_partial(self.bench_id, [row])

        mark = {HELD: ".", VIOLATED: "V", HARNESS_ERROR: "E", UNSUPPORTED: "u"}[classification]

        print(mark, end="", flush=True)

        if classification in (VIOLATED, HARNESS_ERROR):

            print(f"\n  [{classification}] {subject_kind} {scenario} {operation} {row['fault_point']} "

                  f"{row['violations']} {row.get('harness_error', '')}", flush=True)

        return row

    def keep(self, trial, classification: str) -> None:

        trial.stop_server()

        if classification in (VIOLATED, HARNESS_ERROR) and self.retained < self.args.max_evidence:

            self.retained += 1

            dest = self.evidence / f"{self.retained:03d}-{trial.dir.name}"

            dest.parent.mkdir(parents=True, exist_ok=True)

            shutil.move(str(trial.dir), str(dest))

        else:

            shutil.rmtree(trial.dir, ignore_errors=True)

    def transport(self, kind: str, family: str) -> dict:

        if kind == "oak":

            return {"remote_transport": "network", "remote_server": "loopback-oak-serve", "remote_repo": "p4/robust"}

        if family == "proxy":

            return {"remote_transport": "network", "remote_server": "loopback-git-http-backend",

                    "remote_repo": "remote.git"}

        return {"remote_transport": "local_file", "remote_server": "local", "remote_repo": "../srv/remote.git"}

    # families --------------------------------------------------------------------

    def fixture(self, kind: str, op: str, tag: str):

        root = self.work / "fixtures" / f"{tag}-{kind}-{op}"

        if kind == "oak":

            return build_oak(self.bins["oak"], op, root, free_port())

        return build_git(self.bins["git"], op, root, free_port())

    def crash_family(self, kind: str, families: list[str]) -> None:

        ops = [op for op in self.args.ops if op in CRASH_OPS]

        for op in ops:

            if (kind, op) in UNSUPPORTED_OPS or (kind == "oak" and op == "merge_local"):

                for fam in families:

                    self.emit(kind, SCENARIO[fam], op, fam, {"unsupported": UNSUPPORTED_OPS[(kind, op)]})

                continue

            try:

                fx = self.fixture(kind, op, "crash")

                calib = calibrate(fx, self.bins[kind], self.work / "trials", self.instrument)

            except Exception as exc:  # noqa: BLE001

                for fam in families:

                    self.emit(kind, SCENARIO[fam], op, fam, {"classification": HARNESS_ERROR,

                                                             "harness_error": f"fixture/calibration: {exc}"})

                continue

            if not calib.ok:

                for fam in families:

                    self.emit(kind, SCENARIO[fam], op, fam, {"classification": HARNESS_ERROR,

                                                             "harness_error": "calibration: " + calib.detail})

                continue

            extra = self.transport(kind, "crash")

            if "crash_time" in families:

                fractions = TIME_FRACTIONS if not self.args.quick else (0.1, 0.5, 0.9)

                for frac in fractions:

                    delay = max(0.002, calib.elapsed_ms / 1000.0 * frac)

                    row = run_trial(fx, self.bins[kind], self.work / "trials", f"t-{kind}-{op}-{frac}", calib,

                                    time_fault(delay), self.keep)

                    self.emit(kind, "crash_time_v1", op, "crash_time", {**row, **extra})

            if "crash_syscall" in families:

                if self.instrument is None:

                    self.emit(kind, "crash_syscall_v1", op, "crash_syscall", {"unsupported": self.instrument_reason})

                elif calib.boundaries == 0:

                    self.emit(kind, "crash_syscall_v1", op, "crash_syscall", {

                        "unsupported": "no durability boundary observed in the top-level process (git pull delegates "

                                       "to libexec children that the interposer cannot load into)"})

                else:

                    points = syscall_points(calib.boundaries)

                    if self.args.quick:

                        points = points[:: max(1, len(points) // 4)][:4]

                    for n in points:

                        row = run_trial(fx, self.bins[kind], self.work / "trials", f"s-{kind}-{op}-{n}", calib,

                                        syscall_fault(self.instrument, n, calib.boundary_kinds), self.keep)

                        row.setdefault("observations", {})["boundaries_total"] = calib.boundaries

                        self.emit(kind, "crash_syscall_v1", op, "crash_syscall", {**row, **extra})

            shutil.rmtree(fx.root, ignore_errors=True)

    def proxy_family(self, kind: str) -> None:

        from robustness.proxycases import PER_REQUEST, fault_points, plan, record, run_proxy_trial

        for op in [op for op in self.args.ops if op in PROXY_OPS]:

            try:

                fx = self.fixture(kind, op, "proxy")

                calib, keys, methods = record(fx, self.bins[kind], self.work / "trials")

            except Exception as exc:  # noqa: BLE001

                self.emit(kind, "fault_proxy_v1", op, "proxy", {"classification": HARNESS_ERROR,

                                                                 "harness_error": f"record: {exc}"})

                continue

            if not calib.ok:

                self.emit(kind, "fault_proxy_v1", op, "proxy", {"classification": HARNESS_ERROR,

                                                                 "harness_error": "record: " + calib.detail})

                continue

            faults = plan(keys, methods)

            if self.args.quick:

                pts = fault_points(keys)

                keep_keys = {pts[0], pts[-1]} | {p for p in pts if p[0].startswith("POST")}

                faults = [f for f in faults if f.key is None or (f.key, f.occurrence) in keep_keys][:14]

            for index, fault in enumerate(faults):

                row = run_proxy_trial(fx, self.bins[kind], self.work / "trials", f"p-{kind}-{op}-{index}", calib,

                                      fault, self.keep)

                row.setdefault("observations", {})["recorded_requests"] = calib.requests[:60]

                self.emit(kind, "fault_proxy_v1", op, "proxy", {**row, **self.transport(kind, "proxy")})

            shutil.rmtree(fx.root, ignore_errors=True)

    def concurrency_family(self, kind: str) -> None:

        from robustness.concurrency import same_branch, shared_checkout

        from robustness.envfaults import FsyncLoad

        sizes = self.args.clients

        extra = self.transport(kind, "concurrency")

        for n in sizes:

            row = self._guard(lambda: same_branch(kind, self.bins[kind], self.work / f"conc-same-{kind}-{n}", n))

            self.emit(kind, "conc_same_branch_v1", "concurrent.push", "concurrency",

                      {**row, "fault_point": f"clients={n}", **extra})

            row = self._guard(lambda: same_branch(kind, self.bins[kind], self.work / f"conc-dist-{kind}-{n}", n,

                                                  distinct=True))

            self.emit(kind, "conc_distinct_branches_v1", "concurrent.push", "concurrency",

                      {**row, "fault_point": f"clients={n}", **extra})

        n = sizes[0]

        with FsyncLoad(self.work / f"fsyncload-{kind}"):

            row = self._guard(lambda: same_branch(kind, self.bins[kind], self.work / f"conc-load-{kind}", n))

        self.emit(kind, "env_fsync_load_v1", "concurrent.push", "env",

                  {**row, "fault_point": f"clients={n},fsync_writers=4", **extra})

        for what, scenario, op in (("commit", "lock_shared_commit_v1", "concurrent.commit"),

                                   ("push", "lock_shared_push_v1", "concurrent.push"),

                                   ("stale_owner", "lock_stale_owner_v1", "concurrent.commit")):

            row = self._guard(lambda: shared_checkout(kind, self.bins[kind], self.work / f"lock-{what}-{kind}", 8, what))

            self.emit(kind, scenario, op, "concurrency", {**row, "fault_point": "contenders=8",

                                                          **(extra if what == "push" else {})})

        self.emit(kind, "lock_credentials_v1", "credentials.save", "concurrency", {

            "unsupported": ("no non-interactive credential writer in the CLI (oak login is interactive; OAK_API_KEY "

                            "bypasses the store); credential-lock stress stays in oak unit tests (L8 credstress)")

            if kind == "oak" else "git has no credential store under test"})

    def env_family(self, kind: str) -> None:

        from robustness.envfaults import FREE_KIB, disk_full_trial, readonly_trial

        for op in [op for op in ("commit", "pull_ff", "clone") if op in self.args.ops]:

            try:

                fx = self.fixture(kind, op, "env")

                calib = calibrate(fx, self.bins[kind], self.work / "trials", None)

            except Exception as exc:  # noqa: BLE001

                self.emit(kind, "env_disk_full_v1", op, "env", {"classification": HARNESS_ERROR,

                                                                 "harness_error": f"fixture: {exc}"})

                continue

            extra = self.transport(kind, "env")

            for free in (FREE_KIB if not self.args.quick else (0,)):

                row = disk_full_trial(fx, self.bins[kind], self.work / "trials", f"df-{kind}-{op}-{free}", calib, free)

                self.emit(kind, "env_disk_full_v1", op, "env", {**row, **extra})

            if op != "clone":

                for target in ("store", "subdir"):

                    if target == "subdir" and op == "commit":

                        continue

                    row = readonly_trial(fx, self.bins[kind], self.work / "trials", f"ro-{kind}-{op}-{target}", calib,

                                         target)

                    self.emit(kind, "env_readonly_v1", op, "env", {**row, **extra})

            shutil.rmtree(fx.root, ignore_errors=True)

    def hosted_family(self) -> None:

        from robustness.hosted import rows_spec

        for spec in rows_spec():

            self.emit("oak", "hosted_merge_invariants_v1", spec["operation"], "hosted",

                      {"unsupported": spec["skip_reason"], "fault_point": spec["invariant_id"]})

    @staticmethod

    def _guard(fn: Callable[[], dict]) -> dict:

        try:

            return fn()

        except Exception as exc:  # noqa: BLE001

            return {"classification": HARNESS_ERROR, "harness_error": f"{type(exc).__name__}: {exc}"}

    # main ------------------------------------------------------------------------

    def run(self) -> int:

        import subprocess

        self.versions = {}

        for kind in self.args.subjects:

            binary = self.bins.get(kind)

            if not binary or not Path(binary).exists():

                raise SystemExit(f"{kind} binary not found: {binary}")

            self.versions[kind] = subprocess.run([binary, "--version"], capture_output=True, text=True).stdout.strip()

        if "crash_syscall" in self.args.families:

            self.instrument, self.instrument_reason = crashpoints.build(

                self.work / "instrument", self.bins.get("oak") if "oak" in self.args.subjects else None,

                self.bins.get("git") if "git" in self.args.subjects else None)

        started = time.time()

        for kind in self.args.subjects:

            crash = [f for f in ("crash_time", "crash_syscall") if f in self.args.families]

            if crash:

                self.crash_family(kind, crash)

            if "proxy" in self.args.families:

                self.proxy_family(kind)

            if "concurrency" in self.args.families:

                self.concurrency_family(kind)

            if "env" in self.args.families:

                self.env_family(kind)

        if "hosted" in self.args.families:

            self.hosted_family()

        print()

        summary = summarize(self.rows, self.bench_id, time.time() - started)

        raw, summary_path = self.store.write(self.bench_id, self.rows, summary)

        print(summary)

        print(f"rows: {raw}\nsummary: {summary_path}")

        if not self.args.keep_workdir:

            shutil.rmtree(self.work / "trials", ignore_errors=True)

        return 1 if any(r["classification"] == VIOLATED for r in self.rows) and self.args.fail_on_violation else 0

def summarize(rows: list[dict], bench_id: str, seconds: float) -> str:

    table: dict[tuple, dict[str, int]] = {}

    for row in rows:

        key = (row["subject_kind"], row["scenario"], row["operation"])

        cell = table.setdefault(key, {c: 0 for c in CLASSIFICATIONS})

        cell[row["classification"]] += 1

    lines = [f"# Robustness suite {SUITE_VERSION} — {bench_id}", "",

             f"{len(rows)} cases in {seconds / 60:.1f} min.", "",

             "| subject | scenario | operation | held | violated | unsupported | harness_error |",

             "|---|---|---|---:|---:|---:|---:|"]

    for (subject, scenario, operation), cell in sorted(table.items()):

        lines.append(f"| {subject} | {scenario} | {operation} | {cell[HELD]} | {cell[VIOLATED]} | "

                     f"{cell[UNSUPPORTED]} | {cell[HARNESS_ERROR]} |")

    totals = {c: sum(1 for r in rows if r["classification"] == c) for c in CLASSIFICATIONS}

    lines += ["", "Totals: " + ", ".join(f"{k}={v}" for k, v in totals.items()), ""]

    violations: dict[tuple, list[str]] = {}

    for row in rows:

        if row["classification"] == VIOLATED:

            key = (row["subject_kind"], row["scenario"], row["operation"], ",".join(row["violations"]))

            violations.setdefault(key, []).append(row["fault_point"])

    if violations:

        lines += ["## Violations", ""]

        for (subject, scenario, operation, names), points in sorted(violations.items()):

            lines.append(f"- {subject} {scenario} {operation}: {names} ({len(points)}x; e.g. {', '.join(points[:3])})")

    return "\n".join(lines) + "\n"

def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace:

    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)

    parser.add_argument("--subjects", default="oak,git")

    parser.add_argument("--families", default=",".join(FAMILIES))

    parser.add_argument("--ops", default=",".join(CRASH_OPS))

    parser.add_argument("--clients", default="8,16", help="concurrency sizes (comma-separated)")

    parser.add_argument("--oak-bin", default=shutil.which("oak"))

    parser.add_argument("--git-bin", default=None, help="default: real git (xcrun shim resolved)")

    parser.add_argument("--results", default=str(Path(__file__).resolve().parents[2] / "results" / "robustness"))

    parser.add_argument("--workdir", default=None)

    parser.add_argument("--bench-id", default=None)

    parser.add_argument("--quick", action="store_true", help="few points per case (smoke)")

    parser.add_argument("--max-evidence", type=int, default=40, help="violated/harness_error trial dirs retained")

    parser.add_argument("--keep-workdir", action="store_true")

    parser.add_argument("--fail-on-violation", action="store_true")

    parser.add_argument("--measurement-lock", action="store_true",

                        help="hold the benchmark measurement lock for the whole run (serialize with timing campaigns)")

    args = parser.parse_args(argv)

    args.subjects = [s for s in args.subjects.split(",") if s]

    args.families = [f for f in args.families.split(",") if f]

    args.ops = [o for o in args.ops.split(",") if o]

    args.clients = [int(c) for c in str(args.clients).split(",") if c]

    unknown = set(args.families) - set(FAMILIES)

    if unknown or not set(args.subjects) <= {"oak", "git"} or not set(args.ops) <= set(CRASH_OPS):

        parser.error(f"unknown family/subject/op: {sorted(unknown)} {args.subjects} {args.ops}")

    if args.git_bin is None:

        args.git_bin = which_measured_git()

    return args

def main(argv: Optional[list[str]] = None) -> int:

    args = parse_args(argv)

    if not args.measurement_lock:

        # Robustness cases are not timing measurements, and holding the shared

        # lock for a long run would block every other lane (and their tests).

        # Do not run the full suite beside a timing campaign; pass

        # --measurement-lock to serialize behind one instead.

        return Runner(args).run()

    from oakbench.runlock import measurement_lock

    with measurement_lock("robustness-suite"):

        return Runner(args).run()

if __name__ == "__main__":

    raise SystemExit(main())