stack/agents/researcher/work/queue-99/r1/mutants.py

#!/usr/bin/env python3
"""Row 99 mutants. PHASE B ONLY: not run in phase A.

Usage: python3 -I mutants.py <worktree-root> <out-dir>

Runs packages/cli/tests/cli.test.mjs on the unmutated source first (baseline),
then once per mutant of packages/cli/src/cli.mjs. Each mutant is one exact text
replacement that must match exactly once. A mutant counts as killed only when
its run exits nonzero, did not time out, and the named test reports "not ok".
A timeout, a missing match, or a run that passes counts as NOT killed.

Each run starts in its own process group. The group is killed after the run,
and nothing outside that group is signalled. The source is restored in a
finally block and checked against the candidate hash.
"""
import hashlib
import json
import os
import pathlib
import signal
import subprocess
import sys

ROOT = pathlib.Path(sys.argv[1]).resolve()
OUT = pathlib.Path(sys.argv[2]).resolve()
SRC = ROOT / "packages/cli/src/cli.mjs"
TEST = "packages/cli/tests/cli.test.mjs"
TIMEOUT_S = 120
CANDIDATE_CLI_SHA256 = "afcb9d9276c96e543cdc4dac1856b2dd6e973dfa2f1c68d179ddb5fc0e25b1ad"
ENV = dict(os.environ, DOCKER_HOST="unix:///nonexistent.sock")

T_A = "text-mode inbox on a bus without the messages verb prints the decisions and the unavailable line, exit 0"
T_B = "text-mode inbox still fails on any other messages error, and prints nothing"
T_C = "text-mode inbox prints the unread count and every unread message across pages"
T_D = "a later page answered unknown-verb drops the partial list and prints only the unavailable line"

FALLBACK = r'return io.stdout.write(`${formatInbox(decisions)}\nmessages: not available until the bus restarts\n`);'

# (name, old text, new text, the named test that must die)
MUTANTS = [
    (
        "M1-remove-fallback",
        r'if (e.code !== "unknown-verb") throw e;',
        "throw e;",
        T_A,
    ),
    (
        "M2-widen-to-every-error",
        r'if (e.code !== "unknown-verb") throw e;',
        "if (false) throw e;",
        T_B,
    ),
    (
        "M3-changed-line-text",
        "messages: not available until the bus restarts",
        "messages: unavailable",
        T_A,
    ),
    (
        "M4-drop-decisions-in-fallback",
        r"${formatInbox(decisions)}\nmessages: not available",
        r"\nmessages: not available",
        T_A,
    ),
    (
        "M5-fallback-exits-nonzero",
        FALLBACK,
        r'{ io.stdout.write(`${formatInbox(decisions)}\nmessages: not available until the bus restarts\n`); '
        r'throw new CliError("messages not available", 1); }',
        T_A,
    ),
    (
        "M7-read-messages-counted-as-unread",
        "messages.push(...page.messages.filter(m => !m.read));",
        "messages.push(...page.messages);",
        T_C,
    ),
    (
        "M6-partial-list-printed-as-count",
        FALLBACK,
        r"return print({ decisions, messages }, value => formatInbox(value.decisions) + "
        r"`\nmessages: ${value.messages.length} unread\n` + value.messages.map(formatMessage).join(''));",
        T_D,
    ),
]


def group_alive(pgid):
    try:
        os.killpg(pgid, 0)
        return True
    except ProcessLookupError:
        return False


def run_suite(label):
    proc = subprocess.Popen(
        ["node", "--test", "--test-reporter=tap", TEST],
        cwd=ROOT,
        env=ENV,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        start_new_session=True,
    )
    timed_out = False
    try:
        out, _ = proc.communicate(timeout=TIMEOUT_S)
    except subprocess.TimeoutExpired:
        timed_out = True
        os.killpg(proc.pid, signal.SIGKILL)
        out, _ = proc.communicate()
    finally:
        try:
            os.killpg(proc.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
    text = out.decode("utf8", "replace")
    (OUT / f"{label}.log").write_text(text, encoding="utf8")
    return {
        "label": label,
        "exit": proc.returncode,
        "timed_out": timed_out,
        "not_ok": [line.strip() for line in text.splitlines() if "not ok" in line],
        "group_clear": not group_alive(proc.pid),
    }


def killed(res, test_name):
    return (not res["timed_out"]) and res["exit"] != 0 and any(test_name in line for line in res["not_ok"])


def main():
    OUT.mkdir(parents=True, exist_ok=True)
    original = SRC.read_bytes()
    if hashlib.sha256(original).hexdigest() != CANDIDATE_CLI_SHA256:
        print("refusing: cli.mjs is not the candidate hash", file=sys.stderr)
        return 2
    rows = []
    try:
        base = run_suite("baseline")
        rows.append({**base, "status": "baseline"})
        if base["exit"] != 0 or base["timed_out"] or base["not_ok"]:
            print("baseline is not green; no mutants run", file=sys.stderr)
            return 2
        text = original.decode("utf8")
        for name, old, new, test_name in MUTANTS:
            if text.count(old) != 1:
                rows.append({"label": name, "status": "no-match (counted as not killed)"})
                continue
            SRC.write_bytes(text.replace(old, new, 1).encode("utf8"))
            res = run_suite(name)
            if res["timed_out"]:
                status = "TIMEOUT (counted as failure, not killed)"
            else:
                status = "killed" if killed(res, test_name) else "NOT killed"
            rows.append({**res, "status": status, "named_test": test_name})
    finally:
        SRC.write_bytes(original)
        restored = hashlib.sha256(SRC.read_bytes()).hexdigest() == CANDIDATE_CLI_SHA256
        rows.append({"label": "restore", "status": "restored to candidate" if restored else "RESTORE FAILED"})
    (OUT / "results.json").write_text(json.dumps(rows, indent=2) + "\n", encoding="utf8")
    for row in rows:
        print(f"{row['label']}: {row['status']}")
    all_killed = all(r.get("status") == "killed" for r in rows if r["label"].startswith("M"))
    return 0 if all_killed and rows[-1]["status"] == "restored to candidate" else 1


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