stack/agents/darkwing/work/queue-169/mutants.py

#!/usr/bin/env python3
# Row 169 mutants (agents/darkwing/work/queue-169/BUILD.md). Each mutant is exact-string edits to
# scripts/stack-release.sh in the r169 worktree; the test it names must fail. CONTROL makes no edit
# and must pass. The touched file is backed up first, restored after each mutant and checked by
# sha256. Run with python3 -I, only on Sage's slot go, behind busy.sh. DRY=1 checks the anchors
# without running anything.
import hashlib, os, re, shutil, signal, subprocess, sys, time

HOME = os.environ["HOME"]
R = f"{HOME}/darkwing-scratch/r169"
WT = f"{R}/wt"
OUT = f"{R}/out/mutants-{time.strftime('%Y%m%dT%H%M%SZ', time.gmtime())}"
SR = "scripts/stack-release.sh"
T_SR = "cli/tests/stack-release.test.mjs"

M = []
def mut(mid, what, edits, tests):
    M.append((mid, what, edits, tests))

# Each test is "pkg/file::T-number"; the runner passes --test-name-pattern for the T-number.
T32, T33, T34, T35, T36 = (f"{T_SR}::T{n}" for n in range(32, 37))
mut("CONTROL", "no edit; T32 to T36 pass", [], [T32, T33, T34, T35, T36])
# Item 1
mut("1a", "pm-after-harness-target removed", [
    (SR, '    plugins has "$TO" harness 2>/dev/null || { check_refuse pm-after-harness-target "$TO has no harness plug-in kind, so there is nothing to wait for; use --pm"; return 0; }\n', ''),
], [T32])
mut("1b", "pm-after-harness-set removed", [
    (SR, '    [ "$out" = none ] || check_refuse pm-after-harness-set "the business file already sets harness $out; use --pm"\n', ''),
], [T32])
mut("1c", "--pm with --pm-after-harness no longer a usage error", [
    (SR, '    [ "$WANT_PM" = 0 ] || [ "$AFTER_HARNESS" = 0 ] || usage\n', ''),
], [T32])
mut("1d", "--pm-after-harness accepts a pointer without pm", [
    (SR, '        [ "$P_PM" = 1 ] || d+=$\'\\n\'"  pm: pointer $P_PM, --pm-after-harness needs a pointer with pm 1"\n', '        :\n'),
], [T32])
mut("1e", "phase 1 installs and gates --pm", [
    (SR, 'then RUN_PM=0 RUN_PENDING=1\n', 'then RUN_PM=1 RUN_PENDING=1\n'),
], [T32])
mut("1f", "phase 1 writes no pmPending", [
    (SR, '...(pending==="1"?{pmPending:true}:{})', '...({})'),
], [T32])
# Item 2
mut("2a", "read_pointer takes any pmPending", [
    (SR, '&&(p.pmPending===undefined||(p.pmPending===true&&p.pm===false))', ''),
], [T33])
mut("2b", "read_pointer takes pmPending beside pm true", [
    (SR, '(p.pmPending===true&&p.pm===false)', 'p.pmPending===true'),
], [T33])
mut("2c", "pm-pending removed", [
    (SR, '    [ "$P_PENDING" = 0 ] || check_refuse pm-pending "the pointer is pending (activate --pm-after-harness); run pm-on or rollback first"\n', ''),
], [T33])
mut("2d", "status prints no pending line", [
    (SR, '    [ "$P_PENDING" = 0 ] || echo "pm: pending (activate --pm-after-harness); next: Jason\'s plugins set of harness, then stack-release.sh pm-on $BUSINESS"\n', ''),
], [T33])
# Item 3
mut("3a", "not-pending removed", [
    (SR, '    [ "$P_PENDING" = 1 ] || refuse not-pending "the pointer is not pending; pm-on follows activate --pm-after-harness"\n', ''),
], [T34])
mut("3b", "harness-unset for a missing harness removed", [
    (SR, '    [ "$out" != none ] || refuse harness-unset "the business file sets no harness; Jason sets it with plugins set first"\n', ''),
], [T34])
mut("3c", "harness-unset for an empty list removed", [
    (SR, "process.exit(Array.isArray(h)&&h.length>0?0:1)", "process.exit(Array.isArray(h)?0:1)"),
], [T34])
mut("3d", "pm-on installs without --pm", [
    (SR, 'scripts/bus-service.sh install --copy "$P_COPY" --copies "$COPIES" --pm --claude', 'scripts/bus-service.sh install --copy "$P_COPY" --copies "$COPIES" --claude'),
], [T34])
mut("3e", "pm-on gates without pm_launched", [
    (SR, '    gate "$P_COPY" 1 || pm_on_fail "$GATE"\n', '    gate "$P_COPY" 0 || pm_on_fail "$GATE"\n'),
], [T34])
mut("3f", "a failed pm-on keeps the --pm unit", [
    (SR, '    elif ! put_back; then pcode=put-back\n', ''),
], [T34])
mut("3g", "a failed pm-on skips the phase-1 gate", [
    (SR, '    elif ! gate "$P_COPY" 0; then pcode="$GATE"\n', ''),
], [T34])
# Item 4
mut("4a", "rollback from pending installs without --pm", [
    (SR, '    elif [ "$P_PENDING" = 1 ]; then RUN_PM=1\n', ''),
], [T35])
mut("4b", "unknown-kind-for-target removed", [
    (SR, '    if [ -n "$out" ]; then\n        [ "$OP" != rollback ]', '    if false; then\n        [ "$OP" != rollback ]'),
], [T35])
mut("4c", "unknown-kind-for-target names only the first kind", [
    (SR, '.filter((k)=>!Object.hasOwn(KINDS,k)).join(" ")', '.filter((k)=>!Object.hasOwn(KINDS,k)).slice(0,1).join(" ")'),
], [T35])
# --plan (brief item 4, last point)
mut("5a", "--plan prints no pending pointer line", [
    (SR, '    [ "$P_PENDING" = 0 ] || echo "pm: the pointer is pending (activate --pm-after-harness); the bus runs without its PM"\n', ''),
], [T36])
mut("5b", "--plan lists no refusal", [
    (SR, '    for r in "${PLAN_REFUSE[@]}"; do echo "refuses $r"; done\n', ''),
], [T36])
mut("5c", "--plan skips the plug-in checks", [
    (SR, '    if [ "$PLAN" = 1 ]; then check_target; plugin_checks; plan; exit 0; fi\n    precheck\n    plugin_checks\n    new_activation\n    run_steps migrate', '    if [ "$PLAN" = 1 ]; then check_target; plan; exit 0; fi\n    precheck\n    plugin_checks\n    new_activation\n    run_steps migrate'),
], [T36])

def sha(p):
    return hashlib.sha256(open(p, "rb").read()).hexdigest()

def main():
    only = set(sys.argv[1:])
    dry = bool(os.environ.get("DRY"))
    files = [SR]
    pristine = {}
    for f in files:
        p = f"{R}/edit/pristine-{f.replace('/', '_')}"
        shutil.copy2(f"{WT}/{f}", p)
        if sha(p) != sha(f"{WT}/{f}"):
            sys.exit(f"ABORT: backup of {f} failed")
        pristine[f] = (p, sha(p))
    if not dry:
        os.makedirs(OUT, exist_ok=False)
    env = dict(os.environ, TMPDIR=f"{R}/tmp")
    results = []
    try:
        for mid, what, edits, tests in M:
            if only and mid not in only:
                continue
            texts = {}
            for f, old, new in edits:
                text = texts.get(f, open(pristine[f][0]).read())
                if text.count(old) != 1:
                    sys.exit(f"ABORT {mid}: expected 1 match of {old[:80]!r} in {f}, found {text.count(old)}")
                texts[f] = text.replace(old, new)
            if dry:
                print(f"{mid} ok {what}", flush=True); continue
            for f, text in texts.items():
                open(f"{WT}/{f}", "w").write(text)
            t0 = time.time(); per = []
            try:
                for t in tests:
                    path, tn = t.split("::")
                    pkg, rel = path.split("/", 1)
                    log = f"{OUT}/{mid}_{tn}.log"
                    with open(log, "w") as fh:
                        p = subprocess.Popen(["node", "--test", f"--test-name-pattern=^{tn} ", rel], cwd=f"{WT}/packages/{pkg}", env=env,
                                             stdout=fh, stderr=subprocess.STDOUT, start_new_session=True)
                        try:
                            rc = p.wait(timeout=600)
                        except subprocess.TimeoutExpired:
                            os.killpg(p.pid, signal.SIGKILL); p.wait(); rc = "timeout"
                    body = open(log).read()
                    fails = re.search(r"^ℹ fail (\d+)", body, re.M)
                    passes = re.search(r"^ℹ pass (\d+)", body, re.M)
                    per.append((tn, rc, int(fails.group(1)) if fails else -1, int(passes.group(1)) if passes else -1))
            finally:
                for f in texts:
                    shutil.copy2(pristine[f][0], f"{WT}/{f}")
                    if sha(f"{WT}/{f}") != pristine[f][1]:
                        sys.exit(f"ABORT {mid}: restore of {f} failed")
            if mid == "CONTROL":
                verdict = "PASS" if all(rc == 0 and n == 0 and k > 0 for _, rc, n, k in per) else "FAIL"
            else:
                verdict = "KILLED" if any(rc != 0 and n > 0 for _, rc, n, _ in per) else "SURVIVED"
            detail = " ".join(f"{t}:rc={rc},fail={n},pass={k}" for t, rc, n, k in per)
            print(f"{mid} {verdict} {time.time()-t0:.0f}s {what} | {detail}", flush=True)
            results.append((mid, verdict))
    finally:
        for f, (p, s) in pristine.items():
            shutil.copy2(p, f"{WT}/{f}")
            if sha(f"{WT}/{f}") != s:
                print(f"RESTORE FAILED {f}", flush=True)
    if dry:
        print(f"DRY {len(M)} mutants ok"); return
    surv = [m for m, v in results if v not in ("KILLED", "PASS")]
    print(f"DONE {len(results)} mutants, {len(results)-len(surv)} killed, survived: {surv} OUT={OUT}", flush=True)
    sys.exit(1 if surv else 0)

main()