clay-borg/tools/mutation-check.py

#!/usr/bin/env python3
"""M-D1-MUT: does anything fail when an acceptance row's property is false?

CB-WP-0005 T02, per ADR-0005 §1. Every other instrument in this project
counts **names** — M-D1-COV counts `covers:` tags, M-D1-LNK counts rule-ID
strings. Both answer "is this rule mentioned?" Neither answers "does
anything fail if it is violated?"

Four of the seven defects found by CB-RES-0004 were named in the source
and inert. Two were mutation-proven: AM-7's `hash-identical` clause
survives folding a 100k-event log from an unrelated genesis state, and
K9's `through` field survives `Snapshot::take` discarding it.

So: for each acceptance row in `specs/GameKernel.md` §5, invert the
property and require the verifying command to go **red**.

    adapted:mutation-testing — the denominator is acceptance rows, not
    source lines. Exhaustive tools (cargo-mutants, Stryker, PIT) score a
    kill rate over lines; that is slow and measures something we do not
    claim. Our claims are the AM-* rows, so those are the population.

## Verdicts

    red          baseline green, mutant red — the row is enforced
    SURVIVED     baseline green, mutant ALSO green — the row asserts
                 nothing; this is the defect being hunted
    unmutatable  no property to invert, with a stated reason — **counts
                 against** the metric (ADR-0005 §1). A row nobody can
                 invert asserts nothing.
    inconclusive baseline already red, so "mutant red" would prove nothing

## Positive controls

A mutation harness that silently fails to apply its mutation reports every
row as green-baseline/green-mutant and looks thorough. So each run
asserts, per row: the file content actually changed; the baseline command
passed **before** mutating; and the file was restored afterwards. This
project has seven harness-does-nothing instances and one was found in
`rule-coverage.py` an hour ago.

Usage:
  python3 tools/mutation-check.py            # run every row
  python3 tools/mutation-check.py --row AM-1
  python3 tools/mutation-check.py --self-test
"""
import argparse
import os
import subprocess
import sys

from repo import ROOT, cargo_env, enter_root

CARGO = ["cargo"]


class Row:
    """One acceptance row and the mutation that must break it."""

    def __init__(self, id, claim, verify=None, mutate=None, unmutatable=None,
                 clauses=None, expect=None):
        self.id = id
        self.claim = claim
        self.verify = verify          # command that must be green, then red
        self.mutate = mutate          # (relpath, old, new)
        self.unmutatable = unmutatable
        # CB-WP-0005 T08 / the FA class: a mutation is only evidence if it
        # fails FOR ITS STATED REASON. A no-op mutation produces no failure
        # at all, and would otherwise be scored as a genuine SURVIVED —
        # publishing the claim that working code is broken. When `expect`
        # is set, the mutant's output must contain it.
        self.expect = expect
        # A row with several stated clauses is red only if EVERY clause has
        # a mutation that goes red. AM-7 is the reason this exists.
        self.clauses = clauses or []


def rows():
    """The acceptance table, with a mutation or a stated reason for none.

    Ordered as in specs/GameKernel.md §5. AM-4 splits into a/b/c, so there
    are **14** rows, not the twelve ADR-0005 asserted — see the
    correction note in the report.
    """
    py = [sys.executable]
    return [
        Row("AM-1", "100% of GR-rules covered by >=1 passing scenario",
            verify=py + ["tools/rule-coverage.py"],
            mutate=("scenarios/ground/gr-r06-round-resolve.yaml",
                    "GR-R06, ", "")),

        # A PROPERTY mutation: inflate the implementation past the ratio.
        # 58 rules x 40 = 2,320; the impl is 1,575, so ~800 lines of filler
        # crosses it. Raising the threshold instead would only prove the
        # comparison runs.
        Row("AM-2", "<= 40 spec lines per rule in games/ground",
            verify=py + ["tools/size-metrics.py"],
            mutate=("games/ground/src/lib.rs",
                    "impl Aggregate for GroundState {",
                    "fn _am2_filler() {\n" + "    let _x = 0;\n" * 800
                    + "}\n\nimpl Aggregate for GroundState {"),
            expect="FAIL target <= 40"),

        Row("AM-3", "synthetic workload definition <= 50 LOC",
            unmutatable="BLOCKED, not uninstrumented — the distinction "
                        "matters. `make size-metrics` ships the measurement "
                        "(a marker-delimited region) and reports the row "
                        "blocked, because the artifact it measures has never "
                        "been built: games/ contains only ground, and "
                        "benches/synthetic.rs drives GROUND rather than "
                        "defining a synthetic game. The baseline is a "
                        "declarative 3p commit/reveal game object (~36 LOC, "
                        "boardgame.io); measuring GROUND's 1,575 impl lines "
                        "against it would compare two different games and "
                        "call the difference a D1 result. Still counts "
                        "against M-D1-MUT (ADR-0005 §1) — a row that cannot "
                        "fail asserts nothing, however good the reason."),

        Row("AM-4a", "third-party LOC, shipped runtime <= 250,000",
            verify=py + ["tools/dep-weight.py"],
            mutate=("tools/dep-weight.py",
                    '"shipped-runtime": 250_000,', '"shipped-runtime": 1_000,')),

        Row("AM-4b", "third-party LOC, dev toolchain <= 350,000",
            verify=py + ["tools/dep-weight.py"],
            mutate=("tools/dep-weight.py",
                    '"dev-toolchain": 350_000,', '"dev-toolchain": 1_000,')),

        # Deliberately RETAINED in this denominator after its withdrawal
        # from the acceptance table. Dropping it would move M-D1-MUT from
        # 7/14 to 7/13 without enforcing anything — a score improved by
        # deleting the question.
        Row("AM-4c", "own source per third-party 100k lines",
            unmutatable="WITHDRAWN from the acceptance table 2026-08-01 "
                        "(CB-WP-0006 T04, GameKernel §5a) and retained here "
                        "as a diagnostic. The ratio has no monotone better "
                        "direction — rising can mean owning semantics or "
                        "reimplementing what should be assimilated; falling "
                        "can mean leverage or dependency bloat — so it "
                        "cannot carry a threshold. It is also redundant: "
                        "AM-4a/AM-4b bound the denominator and AM-2 bounds "
                        "own-source density. Kept in this denominator on "
                        "purpose, so the metric cannot be improved by "
                        "deleting rows."),

        Row("AM-5", "clean release build <= 60 s",
            unmutatable="now RECORDED (`make build-time`) and **met**: "
                        "37.3 s dev toolchain, 41.2 s shipped runtime, best "
                        "of 3 on a quiet machine, against a 60 s target on "
                        "bnt-lap001 (1.6x headroom). An earlier reading of "
                        "87.0 s was taken under contention and was wrong. "
                        "Still counts "
                        "against M-D1-MUT because GameKernel §5 declares the "
                        "row `recorded not gated` — a row that cannot fail "
                        "asserts nothing, and promoting it is a spec change "
                        "needing an ADR. The breach is a maintainer decision, "
                        "not something this tool should resolve by gating "
                        "itself."),

        Row("AM-6", ">= 100,000 applied events/s",
            # Release: the gate runs where headroom is ~24x. In debug the
            # test only asserts a sanity floor, so a debug mutation run
            # would need a far larger slowdown to register.
            verify=CARGO + ["test", "--release", "-p", "games-ground",
                            "--all-features", "am6_throughput", "--",
                            "--ignored", "--test-threads=1"],
            # A PROPERTY mutation, not a threshold tweak: slow the fold hot
            # path and require the gate to notice. Raising the target
            # instead would only prove the comparison runs.
            mutate=("games/ground/src/lib.rs",
                    "    fn fold(&mut self, event: &Self::Event) {\n"
                    "        match event {",
                    "    fn fold(&mut self, event: &Self::Event) {\n"
                    "        for _ in 0..100_000 { std::hint::black_box(0u8); }\n"
                    "        match event {"),
            expect="AM-6 UNMET"),

        Row("AM-7", "scaling >= 0.9x, and replay of 100k events <= 5 s, "
                    "hash-identical",
            verify=CARGO + ["test", "-p", "games-ground", "--all-features",
                            "replay_100k"],
            mutate=("games/ground/src/lib.rs",
                    'assert!(elapsed.as_secs_f64() < 5.0,',
                    'assert!(elapsed.as_secs_f64() < 0.0,'),
            clauses=[
                ("timing <= 5 s", True,
                 "the elapsed assert is live; tightening it to 0.0 s goes red"),
                ("hash-identical", True,
                 "RE-EARNED (CB-WP-0006 T06): the probe now replays each "
                 "per-game segment from its own genesis and asserts its "
                 "recorded hash. Folding a segment from the wrong seed fails."),
                ("scaling >= 0.9x", False,
                 "no code computes the ratio of throughput @100k to @5k or "
                 "compares it to 0.9; Criterion reports both and nothing "
                 "relates them."),
            ]),

        Row("AM-8", "determinism: same-seed replays bit-identical",
            verify=CARGO + ["run", "-q", "-p", "cb-sim", "--",
                            "scenarios/ground/gr-r06-round-resolve.yaml"],
            mutate=("crates/cb-kernel/src/rng.rs",
                    "Self(rand_chacha::ChaCha12Rng::seed_from_u64(seed.0))",
                    "{ static N: std::sync::atomic::AtomicU64 = "
                    "std::sync::atomic::AtomicU64::new(0); "
                    "let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed); "
                    "Self(rand_chacha::ChaCha12Rng::seed_from_u64(seed.0 + n)) }"),
            clauses=[
                ("N=10 same-seed replays", False,
                 "the runner executes each scenario TWICE (K8), not ten "
                 "times; the spec says N=10. The double-run is live and the "
                 "count is not."),
                ("HashMap deny lint clean", True,
                 "clippy.toml denies HashMap/HashSet and `make check` runs "
                 "with -D warnings"),
            ]),

        # A PROPERTY mutation: make the workload actually use memory.
        Row("AM-9", "peak RSS <= 64 MB",
            verify=py + ["tools/runtime-metrics.py", "--fast"],
            mutate=("games/ground/src/lib.rs",
                    "    fn replay_100k_events_is_linear_and_fast() {",
                    "    fn replay_100k_events_is_linear_and_fast() {\n"
                    "        let hog: Vec<u8> = vec![7u8; 300_000_000];\n"
                    "        std::hint::black_box(&hog);"),
            expect="FAIL target <= 64 MB"),

        Row("AM-10", "0 foreign types in cb-*-api-visible signatures",
            unmutatable="the population is empty — there is no `cb-*-api` "
                        "crate, so the claim is true over nothing. What is "
                        "measured instead is a clippy deny of "
                        "HashMap/HashSet whose stated reason cites K6 "
                        "(determinism), reported under a D4 leak row. "
                        "Withdrawn by ADR-0005 §4."),

        # A PROPERTY mutation: break ONE impl and require the SHARED suite
        # to fail. That is what M-D4-SWAP claims — that either impl can be
        # substituted for the other — and it is exactly what two separate
        # per-impl tests could never demonstrate.
        Row("AM-11", "null + reference impls passing ONE conformance suite",
            verify=CARGO + ["test", "-p", "cb-kernel", "-p", "cb-events",
                            "conformance"],
            mutate=("crates/cb-kernel/src/rng.rs",
                    "    fn draw(&mut self, _bound: u32) -> u32 {\n        0\n    }",
                    "    fn draw(&mut self, _bound: u32) -> u32 {\n"
                    "        _bound\n    }"),
            expect="outside 0.."),

        Row("AM-12", "tokens and USD recorded per task",
            verify=py + ["tools/cb-cost.py", "--self-test"],
            mutate=("tools/cb-cost.py",
                    'toks["output"] = max(t["output"] for t in per_row)',
                    'toks["output"] = per_row[0]["output"]')),
    ]


def run(cmd, timeout=900):
    """(ok, tail) for one verifying command, run at the repo root."""
    try:
        r = subprocess.run(cmd, cwd=ROOT, env=cargo_env(),
                           capture_output=True, text=True, timeout=timeout)
    except subprocess.TimeoutExpired:
        return False, "TIMEOUT", "TIMEOUT"
    out = (r.stdout + r.stderr).strip()
    tail = out.splitlines()
    return r.returncode == 0, (tail[-1][:70] if tail else ""), out


def check_row(row):
    """(verdict, detail). Restores the tree even when the command fails."""
    if row.unmutatable:
        return "unmutatable", row.unmutatable

    path = os.path.join(ROOT, row.mutate[0])
    original = open(path).read()
    old, new = row.mutate[1], row.mutate[2]

    # Positive control 1: the mutation must be applicable at all. A
    # find-string that no longer matches would otherwise mutate nothing
    # and report the baseline result as the mutant result.
    if original.count(old) < 1:
        return "HARNESS-BROKEN", (
            f"mutation target not found in {row.mutate[0]}: {old!r}")

    # Positive control 2: the baseline must be green, or "mutant red"
    # proves nothing.
    base_ok, base_tail, base_out = run(row.verify)
    if not base_ok:
        return "inconclusive", f"baseline already red: {base_tail}"

    # CB-WP-0006 T08: the FA guard is only a guard if its `expect` string
    # cannot appear in PASSING output. An expect of "AM-2" would match the
    # normal report and accept any failure at all — which is how the guard
    # goes vacuous without anyone noticing. My first attempt on AM-2 did
    # exactly that.
    if row.expect and row.expect in base_out:
        return "EXPECT-VACUOUS", (
            f"expect string {row.expect!r} appears in PASSING output, so it "
            f"would accept any failure — the FA guard is inert for this row")

    try:
        mutated = original.replace(old, new, 1)
        # Positive control 3: the file content must actually differ.
        if mutated == original:
            return "HARNESS-BROKEN", "replace produced no change"
        open(path, "w").write(mutated)
        if open(path).read() == original:
            return "HARNESS-BROKEN", "write did not take effect"

        mut_ok, mut_tail, mut_out = run(row.verify)
    finally:
        open(path, "w").write(original)

    # Positive control 4: restoration must have worked, or every later
    # row runs against a corrupted tree.
    if open(path).read() != original:
        return "HARNESS-BROKEN", f"failed to restore {row.mutate[0]}"

    if mut_ok:
        return "SURVIVED", "mutant is green — this row asserts nothing"
    if row.expect and row.expect not in mut_out:
        # The FA guard. The mutant went red, but not for the reason
        # claimed — a compile error, a panic elsewhere, an unrelated
        # assertion. Scoring that as `red` would credit the row with an
        # assertion it does not have.
        return "WRONG-REASON", (
            f"mutant failed, but its output does not contain {row.expect!r} — "
            f"this is not evidence the row is enforced")
    return "red", mut_tail or "verifier failed as required"


def report(only=None):
    rs = [r for r in rows() if not only or r.id == only]
    print("M-D1-MUT — does anything fail when the property is false?")
    print(f"  {len(rows())} acceptance rows in specs/GameKernel.md §5")
    print("  NOTE: ADR-0005 and CB-WP-0005 both say 'twelve'. There are "
          "**14** — AM-4\n        splits into a/b/c. The 9-of-12 prediction "
          "(75%) is evaluated below\n        on the same basis: >=10 of 14.\n")

    tally = {}
    broken = []
    for r in rs:
        verdict, detail = check_row(r)
        # A multi-clause row is red only if every clause is enforced.
        if verdict == "red" and r.clauses:
            unmet = [c for c in r.clauses if not c[1]]
            if unmet:
                verdict = "PARTIAL"
                detail = (f"{len(r.clauses) - len(unmet)}/{len(r.clauses)} "
                          f"clauses enforced")
        tally[verdict] = tally.get(verdict, 0) + 1
        if verdict in ("HARNESS-BROKEN", "EXPECT-VACUOUS"):
            broken.append(r.id)
        mark = {"red": "red        ", "SURVIVED": "SURVIVED   ",
                "unmutatable": "unmutatable", "inconclusive": "inconclusive",
                "PARTIAL": "PARTIAL    ",
                "HARNESS-BROKEN": "BROKEN     ",
                "EXPECT-VACUOUS": "EXPECT-VOID",
                "WRONG-REASON": "WRONG-REASON"}[verdict]
        print(f"  [{mark}] {r.id:<6} {r.claim[:52]}")
        if detail:
            for line in _wrap(detail, 66):
                print(f"                       {line}")
        for name, enforced, why in r.clauses:
            print(f"                       - {'red ' if enforced else 'NONE'} "
                  f"{name}: {why[:60]}")

    if broken:
        # The harness asserting it did the work it reports.
        print(f"\nERROR — harness broken on {', '.join(broken)}; "
              f"no verdict is valid", file=sys.stderr)
        return 1

    red = tally.get("red", 0)
    total = len(rs)
    withdrawn = sum(1 for r in rs if "WITHDRAWN" in (r.unmutatable or ""))
    print(f"\n  M-D1-MUT: {red}/{total} rows enforced")
    if withdrawn:
        print(f"    ({withdrawn} withdrawn row(s) retained in the denominator "
              f"on purpose — a score\n     improved by deleting the question "
              f"is not an improvement)")
    for k in ("PARTIAL", "SURVIVED", "WRONG-REASON", "EXPECT-VACUOUS",
              "unmutatable", "inconclusive"):
        if tally.get(k):
            print(f"    {k:<13} {tally[k]}")
    if only:
        return 0
    print(f"\n  prediction was >=10 of 14 (the ADR's 9-of-12, 75%). "
          f"Measured {red}: "
          f"{'MET' if red >= 10 else 'UNMET'}")
    if red <= 3:
        print("  CB-WP-0005 stop condition: at or below 3 of 12-equivalent, "
              "the pass is\n  under-scoped and must re-plan before Phase C.")
    return 0


def _wrap(text, width):
    words, line, out = text.split(), "", []
    for w in words:
        if len(line) + len(w) + 1 > width:
            out.append(line)
            line = w
        else:
            line = f"{line} {w}".strip()
    if line:
        out.append(line)
    return out


def self_test():
    """Each assertion pins a failure this harness must detect."""
    results = []

    def check(name, ok, detail=""):
        results.append((name, ok, detail))

    rs = rows()
    check("the acceptance table is enumerated", len(rs) == 14, f"{len(rs)} rows")
    check("every row has a mutation or a stated reason",
          all(r.mutate or r.unmutatable for r in rs))
    unmut = [r for r in rs if r.unmutatable]
    check("no unmutatable row has a token reason",
          bool(unmut) and all(len(r.unmutatable) > 40 for r in unmut),
          f"{len(unmut)} unmutatable; a bare 'unmutatable' is how a row "
          f"escapes the metric")
    check("every mutation target file exists",
          all(os.path.isfile(os.path.join(ROOT, r.mutate[0]))
              for r in rs if r.mutate))
    # The control that matters: a mutation whose find-string no longer
    # matches must be reported BROKEN, not silently skipped.
    stale = Row("AM-X", "fixture", verify=[sys.executable, "-c", "pass"],
                mutate=("Makefile", "this-string-does-not-exist", "x"))
    v, _ = check_row(stale)
    check("a stale mutation target is reported broken, not skipped",
          v == "HARNESS-BROKEN", v)
    # And a mutation that does apply must be detected as applying.
    live = Row("AM-Y", "fixture",
               verify=[sys.executable, "-c",
                       "import sys,os; sys.exit(0 if 'ZZMARKER' "
                       "not in open('Makefile').read() else 1)"],
               mutate=("Makefile", "PY := python3", "PY := python3 # ZZMARKER"))
    v2, _ = check_row(live)
    check("an applied mutation that breaks the verifier reports red",
          v2 == "red", v2)
    check("the tree is restored after a mutation run",
          "ZZMARKER" not in open(os.path.join(ROOT, "Makefile")).read())
    # The FA guard: a mutant that fails for the WRONG reason must not be
    # scored as red. Without this, a mutation that merely fails to compile
    # would credit its row with an assertion it does not have.
    wrong = Row("AM-W", "fixture",
                verify=[sys.executable, "-c",
                        "import sys; sys.exit(0 if 'ZZW' not in "
                        "open('Makefile').read() else 7)"],
                mutate=("Makefile", "PY := python3", "PY := python3 # ZZW"),
                expect="a message the verifier never prints")
    v4, _ = check_row(wrong)
    check("a mutant failing for the wrong reason is not scored red",
          v4 == "WRONG-REASON", v4)

    # A verifier that is already red must not be scored.
    dead = Row("AM-Z", "fixture", verify=[sys.executable, "-c", "raise SystemExit(3)"],
               mutate=("Makefile", "PY := python3", "PY := python3 "))
    v3, _ = check_row(dead)
    check("a red baseline is inconclusive, not red", v3 == "inconclusive", v3)

    print("mutation-check self-test (positive control)")
    ok = True
    for name, passed, det in results:
        print(f"  [{'ok ' if passed else 'FAIL'}] {name}"
              + (f"  — {det}" if det else ""))
        ok &= passed
    return 0 if ok else 1


def main():
    enter_root()
    ap = argparse.ArgumentParser()
    ap.add_argument("--row")
    ap.add_argument("--self-test", action="store_true")
    args = ap.parse_args()
    if args.self_test:
        return self_test()
    return report(args.row)


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