stack/agents/dewey/work/queue-76/mutants.py

#!/usr/bin/env python3
# Dewey row 76 mutants: single-site edits in a mutant worktree (never the build tree), each run
# against the whole webui suite. The candidate is uncommitted there, so each file is restored
# from the copy read before the edit, not from git.
# Usage: mutants.py [mutant ...]. A site that doesn't match exactly once is BAD-SITE.
import os, subprocess, sys

R = os.path.expanduser('~/dewey-scratch/r76')
T = R + '/mut'
OUT = R + '/out/mut'
SERVE = 'packages/webui/src/serve.mjs'
BUS = 'packages/webui/src/bus.mjs'
PAGE = 'packages/webui/src/public/bus.js'
TESTS = ['node', '--test', 'packages/webui/tests/*.test.mjs']

MUTANTS = [
    # name, file, edits [(old, new)]
    ('send-header-unchecked', SERVE, [("  if (one(req, preview.header) !== preview.user) return json(res, 401, { error: 'sign-in required' });\n", "")]),
    ('send-origin-unchecked', SERVE, [("  if (one(req, 'origin') !== `https://${preview.host}`) return refuse(", "  if (false) return refuse(")]),
    ('send-sfs-unchecked', SERVE, [("  if (one(req, 'sec-fetch-site') !== 'same-origin') return refuse(", "  if (false) return refuse(")]),
    ('send-sfs-same-site', SERVE, [("  if (one(req, 'sec-fetch-site') !== 'same-origin') return", "  if (!['same-origin', 'same-site'].includes(one(req, 'sec-fetch-site'))) return")]),
    ('send-limit-64k', SERVE, [("await body(req, 16384)", "await body(req, 65536)")]),
    ('send-extra-keys', SERVE, [(" || Object.keys(value).some(k => !SEND_KEYS.includes(k))", "")]),
    ('send-drops-reply-to', SERVE, [("  if ('in_reply_to' in value) args.in_reply_to = value.in_reply_to;\n", "")]),
    ('send-no-id-check', SERVE, [("  if (!subjectOk(result?.id)) return refuse(", "  if (false) return refuse(")]),
    ('send-log-unbounded', SERVE, [("JSON.stringify(to.slice(0, 64))", "to")]),
    ('config-send-loopback', SERVE, [("send: Boolean(send) && named", "send: Boolean(send)")]),
    ('route-open-without-send', SERVE, [("    if (send && path === '/api/bus/message') {", "    if (path === '/api/bus/message') {")]),
    ('send-opens-writes', SERVE, [("    if (preview !== null && req.method !== 'GET' && req.method !== 'HEAD') return json(res, 403, { error: 'read-only'", "    if (preview !== null && !send && req.method !== 'GET' && req.method !== 'HEAD') return json(res, 403, { error: 'read-only'")]),
    ('outcome-all-definite', BUS, [("const OPEN = ['outcome-unknown', 'invalid-response', 'response-too-large'];", "const OPEN = ['outcome-unknown'];")]),
    ('status-unknown-role-502', BUS, [("'unknown-role': 400, ", "")]),
    ('page-no-pending-guard', PAGE, [("    if (sending) return;\n", "")]),
    ('page-clears-edited-text', PAGE, [("      if (draft.body === text) draft.body = '';", "      draft.body = '';")]),
    ('page-5xx-refusal-unknown', PAGE, [(" || body?.sent === false)", ")")]),
    ('page-retries-unknown', PAGE, [("    } else receipt = { kind: 'unknown', to };", "    } else { receipt = { kind: 'unknown', to }; await fetch('/api/bus/message', { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ to, body: text }), cache: 'no-store', redirect: 'manual' }).catch(() => {}); }")]),
    ('page-redirect-unknown', PAGE, [("  if (res?.type === 'opaqueredirect') {", "  if (false) {")]),
    ('page-follows-redirect', PAGE, [(", cache: 'no-store', redirect: 'manual' });", ", cache: 'no-store' });")]),
    ('page-trail-any-id', PAGE, [("      if (!of(tr, 'messages').some(m => m.id === ref)) return notFound(", "      if (false) return notFound(")]),
]

def run(cmd):
    try:
        p = subprocess.run(cmd, cwd=T, capture_output=True, text=True, env={**os.environ, 'TMPDIR': R + '/tmp'}, timeout=600)
    except subprocess.TimeoutExpired:
        return 'TIMEOUT', 'timed out after 600 s'
    return p.returncode, p.stdout + p.stderr

os.makedirs(OUT, exist_ok=True)
os.makedirs(R + '/tmp', exist_ok=True)
before = run(['git', 'diff'])[1]
rc, out = run(TESTS)
open(f'{OUT}/baseline.txt', 'w').write(out)
print('baseline', 'pass' if rc == 0 else f'FAIL rc={rc}', flush=True)
want = sys.argv[1:]
results = []
for name, f, edits in MUTANTS:
    if want and name not in want:
        continue
    path = os.path.join(T, f)
    original = open(path).read()
    bad = [old for old, _ in edits if original.count(old) != 1]
    if bad:
        results.append(f'{name} BAD-SITE ({[original.count(o) for o in bad]} matches)')
        print(results[-1], flush=True)
        continue
    src = original
    for old, new in edits:
        src = src.replace(old, new)
    open(path, 'w').write(src)
    try:
        rc, out = run(TESTS)
    finally:
        open(path, 'w').write(original)
    open(f'{OUT}/mut-{name}.txt', 'w').write(out)
    failed = [l[2:].split(' (')[0] for l in out.splitlines() if l.startswith('✖ ') and 'failing tests' not in l]
    verdict = 'TIMEOUT' if rc == 'TIMEOUT' else 'killed' if rc else 'SURVIVED'
    results.append(f"{name} {verdict}" + (f": {sorted(set(failed))}" if failed else ''))
    print(results[-1], flush=True)
print('worktree unchanged after:', run(['git', 'diff'])[1] == before)
ran = [r for r in results if 'BAD-SITE' not in r]
print(f'summary {sum(1 for r in ran if " killed" in r)} of {len(ran)} killed')