troupe-standup/wizard-work/sphinx/archive-lcmm-token-ledger-2026-10-26/mutation_battery.py

#!/usr/bin/env python3
"""Falsifiability harness: the verifier must FAIL on a mutated ledger.

Copies the ledger to a scratch dir, applies each mutation, runs verify_ledger.py
as a subprocess and asserts a non-zero exit. Also asserts the pristine copy
passes. Exit 0 = every mutation was caught (and the control held).

Two things changed after queue item Q72:
  * the scratch tree is taken from --scratch, else $TMPDIR, else the system
    temp dir, and is removed on the way out (--keep leaves it). The first
    version hard-coded an absolute path under the author's own home, so on a
    host without that directory mkdtemp raised FileNotFoundError before grading
    anything, and it left its trees behind where it did run;
  * a mutation can name the check that must catch it. Without that, a
    manifest-hash failure (any edit to ledger.json invalidates manifest.json)
    masks whether the leg under test is awake at all -- every mutation would
    look caught for the wrong reason.
"""
import json
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parent
LEDGER = ROOT / "ledger.json"


def load(root):
    return json.loads((root / "ledger.json").read_text())


def save(root, L):
    (root / "ledger.json").write_text(json.dumps(L, indent=1, ensure_ascii=False) + "\n")


def run_verify(root):
    p = subprocess.run([sys.executable, str(ROOT / "verify_ledger.py"), str(root)],
                       capture_output=True, text=True)
    verdict = [l for l in p.stdout.splitlines() if l.startswith("VERDICT:")]
    failed = [l.strip()[7:].split(" -- ")[0] for l in p.stdout.splitlines() if l.strip().startswith("[FAIL]")]
    return p.returncode, (verdict[-1] if verdict else "?"), failed


def save_roles(root, R):
    (root / "roles-expected.json").write_text(json.dumps(R, indent=1, ensure_ascii=False) + "\n")


MUTATIONS = []


def mutation(name, expect=None):
    def deco(fn):
        MUTATIONS.append((name, fn, expect))
        return fn
    return deco


@mutation("M1 context_tokens inflated by 1", "E:")
def m1(root):
    L = load(root)
    e = next(x for x in L["entries"] if x["claim_role"] == "model_context_window")
    e["context_tokens"] += 1
    save(root, L)


@mutation("M2 title word altered", "F:")
def m2(root):
    L = load(root)
    t = L["entries"][0]["title"]
    L["entries"][0]["title"] = t.replace(t.split(" ")[0], t.split(" ")[0] + "x", 1)
    save(root, L)


@mutation("M3 claim sentence edited (token -> token!)", "D:")
def m3(root):
    L = load(root)
    e = L["entries"][0]
    e["claim_sentence"] = e["claim_sentence"].replace("tokens", "tokens!", 1)
    save(root, L)


@mutation("M4 pinned source sha256 flipped", "C:")
def m4(root):
    L = load(root)
    h = L["entries"][0]["sources"][0]["sha256"]
    L["entries"][0]["sources"][0]["sha256"] = ("0" if h[0] != "0" else "1") + h[1:]
    save(root, L)


@mutation("M5 published date shifted", "F:")
def m5(root):
    L = load(root)
    L["entries"][0]["published"] = "1999-01-01T00:00:00Z"
    save(root, L)


@mutation("M6 primary_category swapped", "F:")
def m6(root):
    L = load(root)
    L["entries"][0]["primary_category"] = "cs.XX"
    save(root, L)


@mutation("M7 entry dropped (n_entries kept in sync)", "P6")
def m7(root):
    L = load(root)
    L["entries"] = L["entries"][:-1]
    L["n_entries"] = len(L["entries"])
    save(root, L)


@mutation("M8 bogus claim_role", "B3")
def m8(root):
    L = load(root)
    L["entries"][0]["claim_role"] = "magic_number"
    save(root, L)


@mutation("M9 REPORT.md citation removed", "H1")
def m9(root):
    rep = root / "REPORT.md"
    L = load(root)
    aid = L["entries"][0]["arxiv_id"]
    rep.write_text(rep.read_text().replace(aid, "0000.00000"))


@mutation("M10 pinned abs page bytes tampered", "C:")
def m10(root):
    L = load(root)
    p = root / L["entries"][0]["sources"][1]["path"]
    b = bytearray(p.read_bytes())
    b[200] ^= 0x20
    p.write_bytes(bytes(b))


@mutation("M11 manifest entry deleted", "I1")
def m11(root):
    m = json.loads((root / "manifest.json").read_text())
    k = next(iter(m))
    m.pop(k)
    (root / "manifest.json").write_text(json.dumps(m, indent=1) + "\n")


@mutation("M12 manifest hash stale", "I2")
def m12(root):
    m = json.loads((root / "manifest.json").read_text())
    k = next(iter(m))
    m[k] = "0" * 64
    (root / "manifest.json").write_text(json.dumps(m, indent=1) + "\n")


@mutation("M13 claim_note emptied (honesty field)", "B6")
def m13(root):
    L = load(root)
    L["entries"][0]["claim_note"] = ""
    save(root, L)


@mutation("M14 three entries dropped, n_entries + manifest recomputed (the Q72 mutation)", "P6")
def m14(root):
    drop = {"2608.08630v1", "2610.15542v1", "2607.14548v1"}
    L = load(root)
    L["entries"] = [e for e in L["entries"] if e["arxiv_id"] not in drop]
    L["n_entries"] = len(L["entries"])
    L["api_query"] = "https://export.arxiv.org/api/query?id_list=" + ",".join(e["arxiv_id"] for e in L["entries"])
    save(root, L)
    rep = root / "REPORT.md"
    rep.write_text("\n".join(l for l in rep.read_text().splitlines() if not any(d in l for d in drop)) + "\n")
    m = json.loads((root / "manifest.json").read_text())
    m["ledger.json"] = __import__("hashlib").sha256((root / "ledger.json").read_bytes()).hexdigest()
    m["REPORT.md"] = __import__("hashlib").sha256((root / "REPORT.md").read_bytes()).hexdigest()
    (root / "manifest.json").write_text(json.dumps(m, indent=1) + "\n")


@mutation("M15 claim_role relabelled to model_context_window on 2507.02001v1 (the Q72 mutation)", "B8")
def m15(root):
    L = load(root)
    e = next(x for x in L["entries"] if x["arxiv_id"] == "2507.02001v1")
    e["claim_role"] = "model_context_window"
    e["claim_note"] = "Model context window is 32K tokens."
    save(root, L)


@mutation("M16 excluded candidate's reason blanked", "P7")
def m16(root):
    s = json.loads((root / "scan-candidates.json").read_text())
    ex = next(c for c in s["candidates"] if c["decision"] == "excluded")
    ex["reason"] = "   "
    (root / "scan-candidates.json").write_text(json.dumps(s, indent=1, ensure_ascii=False) + "\n")


@mutation("M17 README paper count one lower", "J1")
def m17(root):
    p = root / "README.md"
    p.write_text(p.read_text().replace("for each of 15 papers", "for each of 14 papers"))


@mutation("M18 ledger entry invented outside the scan", "P5")
def m18(root):
    L = load(root)
    e = json.loads(json.dumps(L["entries"][0]))
    e["arxiv_id"] = "2699.99999v1"
    L["entries"].append(e)
    L["n_entries"] = len(L["entries"])
    save(root, L)


@mutation("M19 expected-role table disagrees with the ledger", "B8")
def m19(root):
    R = json.loads((root / "roles-expected.json").read_text())
    R["roles"]["2410.07073v2"]["claim_role"] = "benchmark_input_lengths"
    save_roles(root, R)


@mutation("M20 candidate dropped from the scan (population shrink)", "P4")
def m20(root):
    s = json.loads((root / "scan-candidates.json").read_text())
    s["candidates"] = s["candidates"][:-1]
    s["n_candidates"] = len(s["candidates"])
    (root / "scan-candidates.json").write_text(json.dumps(s, indent=1, ensure_ascii=False) + "\n")


def fresh_copy(dest):
    if dest.exists():
        shutil.rmtree(dest)
    shutil.copytree(ROOT, dest, ignore=shutil.ignore_patterns("__pycache__"))
    return dest


def main():
    keep = "--keep" in sys.argv
    scratch = os.environ.get("LCMM_SCRATCH") or tempfile.gettempdir()
    for a in sys.argv[1:]:
        if a.startswith("--scratch="):
            scratch = a.split("=", 1)[1]
    Path(scratch).mkdir(parents=True, exist_ok=True)
    tmp = Path(tempfile.mkdtemp(prefix="lcmm-mut-", dir=scratch))
    print(f"scratch: {tmp} (removed on exit unless --keep)")
    try:
        control = fresh_copy(tmp / "control")
        rc, verdict, failed = run_verify(control)
        print(f"  [{'PASS' if rc == 0 else 'FAIL'}] control (pristine copy must verify) -> rc={rc} {verdict}")
        if rc != 0:
            print("control FAILED; aborting battery")
            return 1

        caught = 0
        for name, fn, expect in MUTATIONS:
            d = fresh_copy(tmp / name.split(" ")[0])
            fn(d)
            rc, verdict, failed = run_verify(d)
            hit = [f for f in failed if expect and f.startswith(expect)]
            ok = rc != 0 and (not expect or bool(hit))
            caught += ok
            why = "" if ok else f"  <- wanted a {expect} failure, saw {failed[:4]}"
            print(f"  [{'PASS' if ok else 'FAIL'}] {name} -> rc={rc} {verdict}{why}")

        total = len(MUTATIONS)
        print(f"\n{caught}/{total} mutations caught")
        print("VERDICT:", "FALSIFIABLE" if caught == total else "NOT FALSIFIABLE")
        return 0 if caught == total else 1
    finally:
        if not keep:
            shutil.rmtree(tmp, ignore_errors=True)


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