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

#!/usr/bin/env python3
"""Row 101 (#1595) mutants, r1. Written in phase A; NOT RUN in phase A.

Run only after Sage sends "slot clear" and the suites pass:
  python3 -I agents/researcher/work/queue-101/r1/mutants.py <worktree-root>

Each mutant is one exact text replacement that must match exactly once. The killer
command runs with the mutant in place. The file is restored in a finally block and
its bytes are checked against the original. Each run has its own process group, and
the group is killed on exit and on timeout. A timeout is reported as a gate failure,
never as a kill. Baseline: every killer must pass unmutated, or nothing runs.
"""
import hashlib
import os
import pathlib
import signal
import subprocess
import sys

GATE_OWNER = ['node', '--test', 'tests/gate-owner.test.mjs']
TIMEOUT = 600

MUTANTS = [
    dict(id='M1', file='packages/queue/src/store.mjs',
         old='if (verb === "add" && PRIVILEGED.has(actor) && cargs.gateOwner === null) {',
         new='if (false) {',
         cwd='packages/queue', argv=GATE_OWNER,
         why='the write-time refusal is removed; a privileged add with no gate owner is written'),
    dict(id='M2', file='packages/queue/src/store.mjs',
         old='if (verb === "add" && PRIVILEGED.has(actor) && cargs.gateOwner === null) {',
         new='if (verb === "add" && cargs.gateOwner === null) {',
         cwd='packages/queue', argv=GATE_OWNER,
         why='the refusal also hits non-privileged adds, which must get jason'),
    dict(id='M3', file='packages/queue/src/store.mjs',
         old='if (verb === "add" && PRIVILEGED.has(actor) && cargs.gateOwner === null) {',
         new='if (verb === "add" && PRIVILEGED.has(actor) && cargs.gateOwner !== null) {',
         cwd='packages/queue', argv=GATE_OWNER,
         why='the condition is inverted; a privileged add that names its gate owner is refused'),
    dict(id='M4', file='packages/queue/src/queue.mjs',
         old='gate: a.gate, gateOwner: a.gateOwner ?? "jason", brief,',
         new='gate: a.gate, gateOwner: a.gateOwner ?? by, brief,',
         cwd='packages/queue', argv=GATE_OWNER,
         why='a non-privileged add gets its own seat as gate owner, not jason'),
    dict(id='M5', file='packages/queue/src/queue.mjs',
         old='for (const k of ["owner", "gateOwner", "after", "reviewers", "required"]) {',
         new='for (const k of ["owner", "after", "reviewers", "required"]) {',
         cwd='packages/queue', argv=GATE_OWNER,
         why='a non-privileged add may name a gate owner'),
]


def sha(data):
    return hashlib.sha256(data).hexdigest()


def run(argv, cwd, timeout):
    proc = subprocess.Popen(argv, cwd=cwd, start_new_session=True,
                            stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    try:
        out, _ = proc.communicate(timeout=timeout)
        return proc.returncode, out.decode(errors='replace')
    except subprocess.TimeoutExpired:
        os.killpg(proc.pid, signal.SIGKILL)
        proc.communicate()
        return 'timeout', ''
    finally:
        try:
            os.killpg(proc.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass


def main():
    root = pathlib.Path(sys.argv[1]).resolve()
    only = sys.argv[3] if len(sys.argv) > 3 and sys.argv[2] == '--only' else None
    seen = set()
    for m in MUTANTS:
        key = (m['cwd'], tuple(m['argv']))
        if key in seen:
            continue
        seen.add(key)
        rc, _ = run(m['argv'], root / m['cwd'], TIMEOUT)
        if rc != 0:
            print(f"BASELINE FAILED {m['cwd']} {' '.join(m['argv'])} rc={rc}; no mutant was run")
            return 2
        print(f"baseline ok {m['cwd']} {' '.join(m['argv'])}")
    results = []
    for m in MUTANTS:
        if only and m['id'] != only:
            continue
        path = root / m['file']
        orig = path.read_bytes()
        text = orig.decode()
        count = text.count(m['old'])
        if count != 1:
            results.append((m['id'], 'ANCHOR', f"old text matches {count} times"))
            continue
        path.write_bytes(text.replace(m['old'], m['new'], 1).encode())
        try:
            rc, _ = run(m['argv'], root / m['cwd'], TIMEOUT)
        finally:
            path.write_bytes(orig)
        if sha(path.read_bytes()) != sha(orig):
            print(f"RESTORE FAILED {m['file']}; stop and report")
            return 3
        if rc == 'timeout':
            verdict = 'TIMEOUT (gate failure, not a kill)'
        elif rc == 0:
            verdict = 'SURVIVED'
        else:
            verdict = 'KILLED'
        results.append((m['id'], verdict, m['why']))
        print(f"{m['id']}: {verdict} ({m['why']})")
    print('---')
    for r in results:
        print(*r)
    return 0


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