#!/usr/bin/env python3
"""q65_rc_probe.py -- ask a committed verifier, by execution, whether its exit
code can move at all (Q65, volt lane, 2026-10-25, cycle 6).

WHY THIS EXISTS
  Cycle 5 answered "does a verifier have a failure path" statically: an AST scan
  for `raise` / `assert` / non-zero `exit` / `unittest.main` found 15 of 106
  committed verifiers with none, 14 of them carrying a PASS row in the runner
  ledger. Static presence is a proxy, and it has two blind spots the tool said
  so about itself: an exit path inside a `try/except` that swallows it still
  passes the scan, and a present exit path can be unreachable from any verdict.

  This is the dynamic question instead, and it does not read the source for
  intent. It perturbs the verifier's OWN decisions and watches the process:
  flip one comparison operator at a time, re-run the file, and see whether the
  exit code changes. If a verifier's rc is unchanged after its comparisons have
  been inverted, its rc is not connected to what it measured -- no input could
  have made that file say no. That is a stronger statement than "no raise in the
  AST", because it is made about a process that actually ran.

  The mutation is of the verifier, not of the subject. Conventionally, mutation
  testing mutates the SUBJECT and asks whether the test still passes (a killed
  mutant means the test discriminates). Here the subject stays exactly as it is
  and the ASSERTION is mutated, which inverts the question into the one Q65
  needs: does an inverted assertion reach the exit code? A verifier whose rc is
  stable under its own inverted assertions has no verdict channel, whatever its
  source looks like.

WHAT IT MEASURES -- one row per probed file, one class per row
  sensitive       baseline rc 0, and >=1 VALID mutant returned a different rc.
                  The rc is wired to a comparison; the file can fail.
  inert           baseline rc 0, >=1 valid mutant, every valid mutant returned
                  the baseline rc, AND every mutable comparison site in the file
                  was tried. The rc does not respond to any of the file's own
                  comparisons. This is the finding.
                  It is a statement about COMPARISONS, not about the world: a
                  verifier whose rc is `all(exists and is_dir ...)` over
                  `os.path.isdir` is comparison-inert and can still fail on a
                  falsified subject (measured: spark/verify_hermes_path.py with
                  one required path renamed -> rc 1). 3 of the 19 comparison-inert
                  rows at 7cd4dd5 are of that kind; the other 16 cannot fail at
                  all, 15 of them static no-verdict printers.
  inert-partial   the same, but the file has more sites than `--exhaustive-cap`
                  allowed, so the negative is bounded: nothing moved in the
                  first N sites, and sites N+1.. were never tried. Explicitly
                  not the same claim as `inert` -- see WHY THE CLASSES ARE SPLIT.
  no-comparison   baseline rc 0 and no single-operator comparison to mutate.
                  Nothing in the file is a decision the rc could carry.
  unprobed-baseline  the unmutated file already returned a non-zero rc: there is
                  no pass state to defend, so the question does not apply.
  unprobed-timeout   the unmutated file did not finish inside its window.
  unprobed-mutant    every mutant tried was invalid (see below), so nothing was
                  measured. Not a finding either way.
  unprobed-window    the verifier's own measurement window (`q65_row_windows.tsv`,
                  shared with the runner) exceeds `--skip-over`, so it was not
                  mutated. A 30-minute live-HTTP verifier is not inverted three
                  times inside a sweep; its rc is the runner's row.
  unprobed-error     the sweep could not touch the file (the work copy lost it
                  mid-run). Recorded as a row with its message, so a pruned
                  scratch directory leaves a readable ledger instead of a
                  traceback halfway through the population.

  A MUTANT IS VALID when its run (a) compiled and (b) finished inside the window
  and (c) did NOT print a traceback. Flipping `==` to `!=` can turn a type check
  into a TypeError; an rc that moves because the mutant crashed says nothing
  about whether a verdict reaches the exit code. Invalid mutants are counted and
  discarded, and a file left with none is `unprobed-mutant`, not `inert`.

HOW MUCH IS TRIED -- deterministic spread, so the figure is reproducible
  `--max-mutants` (default 3) sites are taken from the sorted list of mutable
  comparison line numbers by even spacing, so a 76-site file is probed near its
  beginning, middle and end rather than at its first three lines. No randomness:
  the same rev and the same `--max-mutants` select the same sites.

  If those first sites all return the baseline rc, the sample is NOT the
  verdict: every remaining site is tried too, up to `--exhaustive-cap`
  (default 40) per file; 0 means no cap. Only a file whose whole site list was
  tried can be called `inert`; a file stopped by the cap is `inert-partial`.

WHY THE CLASSES ARE SPLIT -- a measured false positive, not a hypothetical
  Cycle 6's first sweep wrote `inert` for `forge/verify_chunker_refactor.py`
  from a 3-of-11 sample: sites 58, 105, 150 (one of which crashed). Swept
  exhaustively in a scratch copy, that file's rc moves at 3 of its 11 sites
  (L123 -> 1, L139 -> 1, L142 -> 2) and the sample happened to hit none of
  them. The file can fail; the sample said it could not. `inert` is therefore
  a statement about every comparison in the file or it is not made.

WHERE IT WRITES
  `--emit` mutates a COPY, never this worktree. The copy is a detached `git
  worktree` at `--rev` -- a real checkout, not an export, because a verifier
  that reads git plumbing returns a non-zero rc in a bare export and that
  non-zero would be the copy's defect graded as the verifier's baseline. The
  tool refuses (rc 2) a `--work` equal to the repository root -- a gate that
  edits the tree it grades is the defect reply 93 §1 already measured once.
  Every mutant is written, run, and restored inside a `try/finally`; the file
  hash is re-checked after each file and a mismatch is rc 1, not a silent pass.

  `--emit` writes the TSV (`--tsv`) and appends progress as it goes, so a sweep
  that is interrupted leaves a readable partial ledger rather than nothing.

  `--check` is the fast half and what the lane's CHECK runs: it re-reads the TSV,
  binds every row to the blob at `--rev` (a row graded a file that has since
  changed is STALE), re-derives the class counts from the rows, and faults on a
  row that contradicts its own class (a `sensitive` row with no moving line, an
  `inert` row with no valid mutant, a `no-comparison` row that lists sites).
  It does NOT require population coverage: that is the census's job
  (`q63_verify_scan.py`) and the runner's, and duplicating it here would make
  two gates red for one added file. It prints the coverage gap instead.

  rc 0  the ledger is internally consistent and content-bound
  rc 1  a fault it graded: STALE, MISSING, an unknown class, or a row that
        contradicts its class
  rc 2  it could not grade at all (no TSV, not a git repo, --work is the tree)
"""
import argparse
import ast
import hashlib
import json
import os
import subprocess
import sys
import tempfile
import time

HERE = os.path.dirname(os.path.abspath(__file__))
DEFAULT_TSV = os.path.join(HERE, "q65_rc_probe.tsv")
DEFAULT_REPO = os.path.abspath(os.path.join(HERE, "..", ".."))
SCRATCH_ROOT = "/root/.hermes/profiles/volt/cache/scratch"

CLASSES = ("sensitive", "inert", "inert-partial", "no-comparison",
           "unprobed-baseline", "unprobed-timeout", "unprobed-mutant",
           "unprobed-window", "unprobed-error")

FLIP = {
    ast.Eq: ast.NotEq, ast.NotEq: ast.Eq,
    ast.Lt: ast.Gt, ast.Gt: ast.Lt,
    ast.LtE: ast.GtE, ast.GtE: ast.LtE,
    ast.In: ast.NotIn, ast.NotIn: ast.In,
    ast.Is: ast.IsNot, ast.IsNot: ast.Is,
}

TRACEBACK = "Traceback (most recent call last)"


# ---------------------------------------------------------------- source side
def mutable_lines(src):
    """Line numbers of single-operator comparisons that FLIP can invert."""
    try:
        tree = ast.parse(src)
    except SyntaxError:
        return []
    lines = set()
    for node in ast.walk(tree):
        if (isinstance(node, ast.Compare) and len(node.ops) == 1
                and type(node.ops[0]) in FLIP):
            lines.add(node.lineno)
    return sorted(lines)


def spread(picks, n):
    """Even-spaced deterministic subset of a sorted list, n or fewer."""
    if n <= 0 or not picks:
        return []
    if len(picks) <= n:
        return list(picks)
    if n == 1:
        return [picks[0]]
    step = (len(picks) - 1) / float(n - 1)
    out = []
    for i in range(n):
        idx = int(round(i * step))
        if picks[idx] not in out:
            out.append(picks[idx])
    return out


def mutate(src, line):
    """Return source with the comparison at `line` inverted, or None."""
    tree = ast.parse(src)
    hits = [n for n in ast.walk(tree)
            if isinstance(n, ast.Compare) and n.lineno == line
            and len(n.ops) == 1 and type(n.ops[0]) in FLIP]
    if not hits:
        return None
    hits[0].ops[0] = FLIP[type(hits[0].ops[0])]()
    return ast.unparse(ast.fix_missing_locations(tree)) + "\n"


# ---------------------------------------------------------------- process side
def run_file(path, timeout):
    """Run one verifier the way the ledger does: cwd is the file's directory."""
    t0 = time.time()
    try:
        p = subprocess.run([sys.executable, os.path.basename(path)],
                           cwd=os.path.dirname(path) or ".",
                           capture_output=True, text=True, timeout=timeout)
        return {"rc": p.returncode, "traceback": TRACEBACK in p.stderr,
                "seconds": round(time.time() - t0, 2)}
    except subprocess.TimeoutExpired:
        return {"rc": "TIMEOUT", "traceback": False,
                "seconds": round(time.time() - t0, 2)}


def read_windows(path):
    """path -> seconds, from the runner's per-row window table (may be absent)."""
    table = {}
    if not path or not os.path.isfile(path):
        return table
    with open(path, encoding="utf-8") as fh:
        for line in fh:
            line = line.rstrip("\n")
            if not line or line.startswith("#"):
                continue
            cells = line.split("\t")
            if len(cells) >= 2:
                try:
                    table[cells[0]] = float(cells[1])
                except ValueError:
                    continue
    return table


def probe(path, max_mutants, timeout, log=None, exhaustive_cap=40):
    """Baseline, up to max_mutants sample sites, then every remaining site.

    The sample is the cheap question; the exhaustive pass is what licenses the
    word `inert`. A file stopped by `exhaustive_cap` is inert-partial.
    """
    with open(path, encoding="utf-8") as fh:
        src = fh.read()
    before = hashlib.sha256(src.encode()).hexdigest()
    sites = mutable_lines(src)
    base = run_file(path, timeout)
    row = {"path": path, "baseline_rc": base["rc"], "sites": len(sites),
           "tried": 0, "valid": 0, "moved_at": "-", "seconds": base["seconds"],
           "cls": None, "note": ""}

    if base["rc"] == "TIMEOUT":
        row["cls"] = "unprobed-timeout"
        return row
    if base["rc"] != 0:
        row["cls"] = "unprobed-baseline"
        row["note"] = "baseline rc %s" % base["rc"]
        return row
    if not sites:
        row["cls"] = "no-comparison"
        return row

    tries = spread(sites, max_mutants)
    if len(tries) < len(sites):
        # The negative half of the class needs every site, not the sample.
        rest = [s for s in sites if s not in tries]
        cap = len(rest) if exhaustive_cap == 0 else max(0, exhaustive_cap - len(tries))
        tries += spread(rest, cap)
    for line in tries:
        mutant = mutate(src, line)
        if mutant is None:
            continue
        row["tried"] += 1
        with open(path, "w", encoding="utf-8") as fh:
            fh.write(mutant)
        try:
            got = run_file(path, timeout)
        finally:
            with open(path, "w", encoding="utf-8") as fh:
                fh.write(src)
        row["seconds"] += got["seconds"]
        valid = got["rc"] != "TIMEOUT" and not got["traceback"]
        if not valid:
            if log:
                log("    L%d invalid mutant (%s)" % (line, got["rc"]))
            continue
        row["valid"] += 1
        if log:
            log("    L%d valid mutant rc=%s" % (line, got["rc"]))
        if got["rc"] != base["rc"]:
            row["cls"] = "sensitive"
            row["moved_at"] = line
            break

    if row["cls"] != "sensitive":
        if not row["valid"]:
            row["cls"] = "unprobed-mutant"
        elif row["tried"] >= len(sites):
            row["cls"] = "inert"
        else:
            row["cls"] = "inert-partial"
            row["note"] = "%d of %d sites tried (cap %d)" % (
                row["tried"], len(sites), exhaustive_cap)
    after = hashlib.sha256(open(path, encoding="utf-8").read().encode()).hexdigest()
    if after != before:
        raise RuntimeError("restore failed for %s" % path)
    return row


# ---------------------------------------------------------------- git plumbing
def tree_blobs(repo, rev):
    out = subprocess.run(["git", "-C", repo, "ls-tree", "-r", rev],
                         capture_output=True, text=True)
    if out.returncode != 0:
        raise SystemExit("q65_rc_probe: cannot read tree at %s: %s"
                         % (rev, out.stderr.strip()))
    blobs = {}
    for line in out.stdout.splitlines():
        meta, path = line.split("\t", 1)
        parts = meta.split()
        if len(parts) >= 3 and parts[1] == "blob":
            blobs[path] = parts[2]
    return blobs


def population(blobs):
    return sorted(p for p in blobs
                  if os.path.basename(p).startswith("verify_")
                  and p.endswith(".py")
                  and p.startswith("wizard-work/"))


def strip_basename(path):
    return path


# ---------------------------------------------------------------------- modes
def read_tsv(path):
    rows, head = [], {}
    with open(path, encoding="utf-8") as fh:
        for line in fh:
            line = line.rstrip("\n")
            if not line.strip():
                continue
            if line.startswith("#"):
                for key in ("rev", "rows", "committed"):
                    if line.startswith("# %s " % key):
                        head[key] = line.split(None, 2)[2].split()[0]
                continue
            cells = line.split("\t")
            if len(cells) != 9:
                raise ValueError("row has %d cells, want 9: %r" % (len(cells), line))
            rows.append(dict(zip(
                ("cls", "path", "baseline_rc", "sites", "tried", "valid",
                 "moved_at", "blob", "seconds"), cells)))
    return head, rows


def write_header(fh, rev, rows, committed):
    fh.write("# Q65 rc-sensitivity ledger -- one row per PROBED committed verifier.\n")
    fh.write("# Written by q65_rc_probe.py --emit. A row states a measurement, not an\n")
    fh.write("# exemption: --check faults on a row that contradicts its class or whose\n")
    fh.write("# file has changed since (STALE).\n")
    fh.write("# columns: class, path, baseline_rc, sites, tried, valid, moved_at, blob12, seconds\n")
    fh.write("# rev %s\n" % rev)
    fh.write("# rows %d\n" % rows)
    fh.write("# committed %d\n" % committed)


def do_emit(args):
    repo = os.path.abspath(args.repo)
    rev = subprocess.run(["git", "-C", repo, "rev-parse", args.rev],
                         capture_output=True, text=True)
    if rev.returncode != 0:
        print("q65_rc_probe: not a git tree at %s" % repo, file=sys.stderr)
        return 2
    rev = rev.stdout.strip()
    blobs = tree_blobs(repo, rev)
    pop = population(blobs)
    windows = read_windows(args.windows)
    prev_rows = []
    if args.merge and os.path.isfile(args.tsv):
        try:
            _, prev_rows = read_tsv(args.tsv)
        except ValueError:
            prev_rows = []
        print("q65_rc_probe: merging into %d existing row(s) from %s"
              % (len(prev_rows), args.tsv))
    if args.only:
        pop = [p for p in pop if args.only in p]
    if args.limit:
        pop = pop[:args.limit]

    if args.work:
        work = os.path.abspath(args.work)
        if work == os.path.realpath(repo) or work == repo:
            print("q65_rc_probe: --work is the repository itself; refusing to "
                  "mutate the tree this gate grades", file=sys.stderr)
            return 2
    else:
        work = os.path.join(SCRATCH_ROOT, "rcprobe-%d" % int(time.time()))
    if not os.path.isdir(work):
        os.makedirs(work)
    if not os.path.isdir(os.path.join(work, ".git")) and not os.path.exists(
            os.path.join(work, ".git")):
        # A real git worktree, not an export: a verifier that reads git plumbing
        # (rev-parse, ls-tree, log) returns a non-zero rc in a bare export, and a
        # non-zero BASELINE is a row with no pass state to defend -- the sweep
        # would grade its own copy's defect instead of the verifier.
        add = subprocess.run(["git", "-C", repo, "worktree", "add", "--detach",
                              work, rev], capture_output=True, text=True)
        if add.returncode != 0:
            print("q65_rc_probe: cannot add a worktree at %s: %s"
                  % (rev[:12], add.stderr.strip()), file=sys.stderr)
            return 2

    print("q65_rc_probe -- rev %s, %d population file(s), work copy %s"
          % (rev[:12], len(pop), work))
    rows = []
    for i, rel in enumerate(pop, 1):
        path = os.path.join(work, rel)
        if not os.path.isfile(path):
            rows.append({"cls": "unprobed-mutant", "path": rel, "baseline_rc": "-",
                         "sites": "-", "tried": 0, "valid": 0, "moved_at": "-",
                         "blob": blobs.get(rel, "-")[:12], "seconds": 0.0,
                         "note": "absent from the work copy"})
            continue
        print("[%d/%d] %s" % (i, len(pop), rel))
        window = windows.get(rel, args.timeout)
        if window > args.skip_over:
            row = {"cls": "unprobed-window", "path": rel, "baseline_rc": "-",
                   "sites": "-", "tried": 0, "valid": 0, "moved_at": "-",
                   "seconds": 0.0, "note": "%ss window" % int(window)}
            row["blob"] = blobs.get(rel, "-")[:12]
            rows.append(row)
            print("   -> %-17s window %ss exceeds --skip-over %ss"
                  % (row["cls"], int(window), int(args.skip_over)))
            continue
        try:
            row = probe(path, args.max_mutants, window, log=lambda m: print(m),
                        exhaustive_cap=getattr(args, "exhaustive_cap", 40))
        except (OSError, RuntimeError) as exc:
            # A work copy that lost a file mid-sweep (a scratch pruner, a full
            # disk) must be a row, not a traceback: the sweep's ledger stays
            # readable and the row says what happened.
            row = {"cls": "unprobed-error", "path": rel, "baseline_rc": "-",
                   "sites": "-", "tried": 0, "valid": 0, "moved_at": "-",
                   "seconds": 0.0, "note": str(exc)[:120]}
            row["blob"] = blobs.get(rel, "-")[:12]
            rows.append(row)
            print("   -> %-17s %s" % (row["cls"], str(exc)[:90]))
            continue
        row["blob"] = blobs.get(rel, "-")[:12]
        row["path"] = rel
        rows.append(row)
        print("   -> %-17s rc=%s sites=%s tried=%s valid=%s moved_at=%s %ss"
              % (row["cls"], row["baseline_rc"], row["sites"], row["tried"],
                 row["valid"], row["moved_at"], row["seconds"]))
        with open(args.tsv, "w", encoding="utf-8") as fh:
            write_header(fh, rev, len(rows), len(pop))
            for r in rows:
                fh.write("\t".join(str(r.get(k, "-")) for k in
                                   ("cls", "path", "baseline_rc", "sites", "tried",
                                    "valid", "moved_at", "blob", "seconds")) + "\n")

    with open(args.tsv, "w", encoding="utf-8") as fh:
        merged = rows
        if args.merge and prev_rows:
            by_path = {r["path"]: r for r in prev_rows}
            for r in rows:
                by_path[r["path"]] = r
            merged = [by_path[k] for k in sorted(by_path)]
        write_header(fh, rev, len(merged), len(population(tree_blobs(repo, rev))))
        for r in merged:
            fh.write("\t".join(str(r.get(k, "-")) for k in
                               ("cls", "path", "baseline_rc", "sites", "tried",
                                "valid", "moved_at", "blob", "seconds")) + "\n")
    print("wrote %s (%d rows)" % (args.tsv, len(merged)))
    print("work copy left at %s" % work)
    return 0


def do_check(args):
    if not os.path.isfile(args.tsv):
        print("q65_rc_probe: no ledger at %s -- cannot grade" % args.tsv,
              file=sys.stderr)
        return 2
    try:
        head, rows = read_tsv(args.tsv)
    except ValueError as exc:
        print("q65_rc_probe: %s" % exc, file=sys.stderr)
        return 1
    repo = os.path.abspath(args.repo)
    try:
        blobs = tree_blobs(repo, args.rev)
    except SystemExit as exc:
        print(str(exc), file=sys.stderr)
        return 2

    faults = []
    counts = {c: 0 for c in CLASSES}
    for r in rows:
        if r["cls"] not in CLASSES:
            faults.append("UNKNOWN_CLASS %s: %r" % (r["path"], r["cls"]))
            continue
        counts[r["cls"]] += 1
        if r["path"] not in blobs:
            faults.append("MISSING %s: row in the ledger, absent at %s" % (r["path"], args.rev[:12]))
            continue
        if not blobs[r["path"]].startswith(r["blob"]):
            faults.append("STALE %s: row graded blob %s, tree holds %s"
                          % (r["path"], r["blob"], blobs[r["path"]][:12]))
        try:
            moved = int(r["moved_at"]) if r["moved_at"] not in ("-", "") else None
            tried, valid = int(r["tried"]), int(r["valid"])
            sites = int(r["sites"]) if r["sites"] not in ("-", "") else None
        except ValueError:
            faults.append("BAD_CELL %s: non-numeric tried/valid/moved_at" % r["path"])
            continue
        if r["cls"] == "sensitive" and not moved:
            faults.append("CONTRADICTION %s: sensitive with no moving line" % r["path"])
        if r["cls"] == "inert" and not (valid >= 1 and sites is not None
                                        and tried >= sites and moved is None):
            faults.append("CONTRADICTION %s: inert with valid=%d tried=%s sites=%s moved=%s"
                          % (r["path"], valid, tried, sites, moved))
        if r["cls"] == "inert-partial" and not (valid >= 1 and sites is not None
                                               and tried < sites and moved is None):
            faults.append("CONTRADICTION %s: inert-partial with valid=%d tried=%s "
                          "sites=%s moved=%s"
                          % (r["path"], valid, tried, sites, moved))
        if r["cls"] == "no-comparison" and sites:
            faults.append("CONTRADICTION %s: no-comparison but lists %d site(s)"
                          % (r["path"], sites))
        if r["cls"] == "unprobed-mutant" and valid:
            faults.append("CONTRADICTION %s: unprobed-mutant with %d valid mutant(s)"
                          % (r["path"], valid))

    committed = population(blobs)
    probed = {r["path"] for r in rows}
    print("q65_rc_probe -- rev %s, %d row(s), %d committed verifier(s)"
          % (args.rev[:12], len(rows), len(committed)))
    for c in CLASSES:
        print("  %-18s %d" % (c, counts[c]))
    print("  rc-sensitive       %d of %d probed of %d committed"
          % (counts["sensitive"], counts["sensitive"] + counts["inert"]
             + counts["inert-partial"] + counts["no-comparison"], len(committed)))
    gap = [p for p in committed if p not in probed]
    if gap:
        print("  not probed         %d (population coverage is the census's leg, "
              "not this one)" % len(gap))
    if faults:
        for f in faults:
            print("  FAULT %s" % f)
        print("q65_rc_probe: %d fault(s)" % len(faults))
        return 1
    print("q65_rc_probe: 0 fault(s) -- every row content-bound and consistent "
          "with its class")
    return 0


# -------------------------------------------------------------------- selftest
def selftest():
    tmp = tempfile.mkdtemp(prefix="rcprobe-selftest-", dir=SCRATCH_ROOT)
    cells = []

    def cell(name, ok, detail=""):
        cells.append((name, bool(ok), detail))

    def mk(name, body):
        d = os.path.join(tmp, name)
        os.makedirs(d, exist_ok=True)
        p = os.path.join(d, "verify_x.py")
        with open(p, "w", encoding="utf-8") as fh:
            fh.write(body)
        return p

    # c1 sensitive: the inverted assertion reaches sys.exit
    p = mk("sensitive", "import sys\nx = 2\nif x != 2:\n    sys.exit(0)\n"
                        "if x == 2:\n    print('ok')\nelse:\n    sys.exit(1)\n")
    r = probe(p, 3, 20)
    cell("c1 an inverted assertion that reaches sys.exit is sensitive",
         r["cls"] == "sensitive" and r["moved_at"], str(r))

    # c2 inert: comparisons whose result is only printed
    p = mk("inert", "x = 2\ny = 3\nprint('x==2', x == 2)\nprint('y<x', y < x)\n")
    r = probe(p, 3, 20)
    cell("c2 comparisons that only print leave the rc unmoved (inert)",
         r["cls"] == "inert" and r["valid"] >= 1, str(r))

    # c3 no comparison at all
    p = mk("nocomp", "print('all checks passed')\n")
    r = probe(p, 3, 20)
    cell("c3 a file with no comparison is no-comparison",
         r["cls"] == "no-comparison", str(r))

    # c4 baseline non-zero
    p = mk("baseline", "import sys\nsys.exit(2)\n")
    r = probe(p, 3, 20)
    cell("c4 a non-zero baseline is unprobed-baseline, not a pass",
         r["cls"] == "unprobed-baseline", str(r))

    # c5 every mutant invalid (the flip reaches an unreachable branch and crashes)
    p = mk("invalid", "x = None\nif x == None:\n    print('none')\nelse:\n"
                      "    print(x.attr)\n")
    r = probe(p, 3, 20)
    cell("c5 a mutant that crashes is invalid, and no valid mutant is not inert",
         r["cls"] == "unprobed-mutant" and r["tried"] == 1 and r["valid"] == 0, str(r))
    cell("c5b the crashed-mutant file was restored byte for byte",
         open(p).read() == "x = None\nif x == None:\n    print('none')\nelse:\n"
                           "    print(x.attr)\n")

    # c5c one invalid mutant beside a valid one: counted, discarded, still graded
    p = mk("mixed", "x = None\ny = 1\nif x == None:\n    print('none')\nelse:\n"
                    "    print(x.attr)\nif y == 1:\n    print('y')\n")
    r = probe(p, 3, 20)
    cell("c5c an invalid mutant is counted and discarded, not counted as a test",
         r["cls"] == "inert" and r["tried"] == 2 and r["valid"] == 1, str(r))

    # c5d the sample is not the verdict: a moving site outside the 3-site
    #     even-spaced sample is found by the exhaustive pass
    body5d = ("import sys\n"
              "x1 = 1\n"
              "x2 = 2\n"
              "x3 = 3\n"
              "x4 = 4\n"
              "x5 = 5\n"
              "print(x1 == 1)\n"
              "if x2 == 2:\n"
              "    sys.exit(0)\n"
              "print(x3 == 3)\n"
              "print(x4 == 4)\n"
              "print(x5 == 5)\n"
              "sys.exit(1)\n")
    p = mk("outside_sample", body5d)
    r = probe(p, 3, 20)
    cell("c5d a moving site the 3-site sample misses is still found",
         r["cls"] == "sensitive" and r["moved_at"] == 8 and r["sites"] == 5,
         str(r))
    cell("c5d2 the sample really did miss it (site 8 is not in spread(sites,3))",
         8 not in spread(mutable_lines(body5d), 3),
         str(spread(mutable_lines(body5d), 3)))

    # c5e the same file under a cap that stops before the moving site is
    #     inert-partial, not inert and not sensitive
    p = mk("capped", body5d)
    r = probe(p, 3, 20, exhaustive_cap=3)
    cell("c5e a cap that stops before the moving site is inert-partial",
         r["cls"] == "inert-partial" and r["tried"] == 3 and r["sites"] == 5,
         str(r))

    # c6 restore on a sensitive file
    p = mk("restore", "import sys\nx = 1\nif x == 1:\n    sys.exit(0)\n")
    before = open(p, "rb").read()
    probe(p, 3, 20)
    cell("c6 a probed file is restored byte for byte", open(p, "rb").read() == before)

    # c7 spread is deterministic and even
    picks = list(range(0, 100))
    cell("c7 spread picks evenly and deterministically",
         spread(picks, 3) == spread(picks, 3) and len(spread(picks, 3)) == 3
         and spread(picks, 3)[0] == 0 and spread(picks, 3)[-1] == 99,
         str(spread(picks, 3)))
    cell("c7b spread never invents a site", all(s in picks for s in spread(picks, 7)))

    # c8 tsv round-trip
    tsv = os.path.join(tmp, "led.tsv")
    rows = [
        {"cls": "sensitive", "path": "a/verify_a.py", "baseline_rc": 0, "sites": 4,
         "tried": 2, "valid": 2, "moved_at": 12, "blob": "abc123", "seconds": 0.1},
        {"cls": "inert", "path": "a/verify_b.py", "baseline_rc": 0, "sites": 1,
         "tried": 1, "valid": 1, "moved_at": "-", "blob": "def456", "seconds": 0.2},
    ]
    with open(tsv, "w") as fh:
        write_header(fh, "f" * 40, len(rows), 2)
        for r in rows:
            fh.write("\t".join(str(r[k]) for k in ("cls", "path", "baseline_rc",
                                                   "sites", "tried", "valid",
                                                   "moved_at", "blob", "seconds")) + "\n")
    head, back = read_tsv(tsv)
    cell("c8 the ledger round-trips its rows and its header",
         len(back) == 2 and head.get("rows") == "2" and back[0]["moved_at"] == "12",
         str(head))

    # c9 a short row is refused, not silently skipped
    bad = os.path.join(tmp, "short.tsv")
    open(bad, "w").write("# rev " + "f" * 40 + "\n# rows 1\n# committed 1\n"
                         "inert\ta/verify_a.py\t0\t1\n")
    try:
        read_tsv(bad)
        cell("c9 a short row is refused", False)
    except ValueError:
        cell("c9 a short row is refused", True)

    # c10 --check refuses a --work that is the tree
    ns = argparse.Namespace(repo=DEFAULT_REPO, rev="HEAD", tsv=DEFAULT_TSV,
                            only=None, limit=0, work=os.path.realpath(DEFAULT_REPO),
                            max_mutants=1, timeout=1,
                            windows=os.path.join(HERE, "q65_row_windows.tsv"),
                            skip_over=300.0, merge=False)
    cell("c10 --emit refuses --work equal to the repo root", do_emit(ns) == 2)

    # c12 the window table parses, and a malformed row is skipped not fatal
    wt = os.path.join(tmp, "win.tsv")
    open(wt, "w").write("# path\ts\twhy\na/verify_a.py\t1800\tlive HTTP\n"
                        "b/verify_b.py\tnotanumber\tbroken\n")
    w = read_windows(wt)
    cell("c12 the window table parses a path and skips a malformed row",
         w.get("a/verify_a.py") == 1800.0 and "b/verify_b.py" not in w, str(w))

    # c13 a verifier over the skip-over is unprobed-window, and the emit still succeeds
    one = os.path.join(tmp, "one.tsv")
    ns3 = argparse.Namespace(repo=DEFAULT_REPO, rev="HEAD", tsv=one,
                             only="wizard-work/sphinx/archive-index-2026-10-23/verify_index.py",
                             limit=0, work=os.path.join(tmp, "work"),
                             max_mutants=1, timeout=1,
                             windows=os.path.join(HERE, "q65_row_windows.tsv"),
                             skip_over=1.0, merge=False)
    rc = do_emit(ns3)
    head2, rows2 = read_tsv(one)
    cell("c13 a file whose window exceeds --skip-over is unprobed-window, rc 0",
         rc == 0 and len(rows2) == 1 and rows2[0]["cls"] == "unprobed-window",
         str(rows2))

    # c15 --merge keeps a row this run did not probe
    ns4 = argparse.Namespace(repo=DEFAULT_REPO, rev="HEAD", tsv=one,
                             only="wizard-work/spark/verify_sum.py",
                             limit=0, work=os.path.join(tmp, "work"),
                             max_mutants=1, timeout=30,
                             windows=os.path.join(HERE, "q65_row_windows.tsv"),
                             skip_over=300.0, merge=True)
    rc = do_emit(ns4)
    head3, rows3 = read_tsv(one)
    kept = [r for r in rows3 if r["path"].endswith("archive-index-2026-10-23/verify_index.py")]
    fresh = [r for r in rows3 if r["path"].endswith("spark/verify_sum.py")]
    cell("c15 --merge keeps the unprobed row beside the new one",
         rc == 0 and len(rows3) == 2 and kept and fresh, str(rows3))

    # c14 a file the sweep cannot read raises (do_emit turns it into a row)
    try:
        probe(os.path.join(tmp, "absent", "verify_x.py"), 1, 5)
        cell("c14 an unreadable file raises OSError, caught into a row", False)
    except OSError:
        cell("c14 an unreadable file raises OSError, caught into a row", True)

    # c16 --check enforces the coverage half of the inert claim
    blobs_now = tree_blobs(DEFAULT_REPO, "HEAD")
    real = sorted(p for p in blobs_now
                  if p.startswith("wizard-work/")
                  and os.path.basename(p).startswith("verify_"))[0]
    b12 = blobs_now[real][:12]

    def _led(cls, tried, sites):
        f = os.path.join(tmp, "chk-%s-%d-%d.tsv" % (cls, tried, sites))
        with open(f, "w", encoding="utf-8") as fh:
            write_header(fh, "HEAD", 1, 1)
            fh.write("\t".join([cls, real, "0", str(sites), str(tried),
                                "1", "-", b12, "0.1"]) + "\n")
        return do_check(argparse.Namespace(repo=DEFAULT_REPO, rev="HEAD", tsv=f))

    cell("c16 inert with sites untried is a contradiction (rc 1)",
         _led("inert", 2, 5) == 1)
    cell("c16b inert with every site tried is consistent (rc 0)",
         _led("inert", 5, 5) == 0)
    cell("c16c inert-partial with sites untried is consistent (rc 0)",
         _led("inert-partial", 2, 5) == 0)
    cell("c16d inert-partial with every site tried is a contradiction (rc 1)",
         _led("inert-partial", 5, 5) == 1)

    # c11 --check on a missing ledger is rc 2 (cannot grade), not rc 0
    ns2 = argparse.Namespace(repo=DEFAULT_REPO, rev="HEAD",
                             tsv=os.path.join(tmp, "nope.tsv"))
    cell("c11 a missing ledger is rc 2, not a pass", do_check(ns2) == 2)

    ok = sum(1 for _, o, _ in cells if o)
    for name, o, detail in cells:
        print("%s %s%s" % ("ok  " if o else "FAIL", name,
                           "" if o else "   [%s]" % detail))
    print("selftest: %d cells, %d passed, %d failed" % (len(cells), ok, len(cells) - ok))
    return 0 if ok == len(cells) else 1


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--repo", default=DEFAULT_REPO)
    ap.add_argument("--rev", default="HEAD")
    ap.add_argument("--tsv", default=DEFAULT_TSV)
    ap.add_argument("--work", help="copy dir to mutate (default: fresh scratch dir)")
    ap.add_argument("--max-mutants", type=int, default=3)
    ap.add_argument("--exhaustive-cap", type=int, default=40,
                    help="sites per file tried before an inert verdict is "
                         "downgraded to inert-partial; 0 = no cap")
    ap.add_argument("--timeout", type=float, default=90.0,
                    help="window for a file the window table does not name")
    ap.add_argument("--windows", default=os.path.join(HERE, "q65_row_windows.tsv"),
                    help="per-path measurement windows, shared with the runner")
    ap.add_argument("--skip-over", type=float, default=300.0,
                    help="a verifier whose window exceeds this is not mutated; "
                         "grading it is the runner's row, not this sweep's")
    ap.add_argument("--only", help="with --emit: only paths containing this substring")
    ap.add_argument("--limit", type=int, default=0)
    ap.add_argument("--emit", action="store_true")
    ap.add_argument("--merge", action="store_true",
                    help="with --emit: keep rows for paths this run did not probe")
    ap.add_argument("--check", action="store_true")
    ap.add_argument("--selftest", action="store_true")
    args = ap.parse_args()
    if args.selftest:
        return selftest()
    if args.check:
        return do_check(args)
    if args.emit:
        return do_emit(args)
    ap.print_help()
    return 2


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