Files
stack/agents/huey/work/queue-272/fallback-mutants-r2.py

63 lines
3.4 KiB
Python

#!/usr/bin/env python3
"""Run only on Sage's slot. Mutate disposable sources, never the candidate tree."""
import argparse
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
GUARD = "if (err.code === 'access-refused' && !match[3]) return null;"
CONTROLS = " if (current.legacyDefault) return '<p role=\"status\">Sessions are not available for this account.</p>';"
MUTANTS = {
'F1-any-code': (GUARD, "if (!match[3]) return null;", 'human-required catalog failure'),
'F2-explicit-link': (GUARD, "if (err.code === 'access-refused') return null;", 'granted explicit-session route'),
'F3-controls': (CONTROLS, ' // mutant: show session controls on a legacy fallback', 'granted legacy route'),
'F4-widened-key': (GUARD, "if (['access-refused', 'unreachable', 'no-bus-host', 'invalid-response'].includes(err.code) && !match[3]) return null;", 'timeout catalog failure'),
}
def main():
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument('--out', type=Path, required=True)
ap.add_argument('--mutant', choices=list(MUTANTS), action='append')
args = ap.parse_args()
source = Path(__file__).resolve().parents[4]
args.out.mkdir(parents=True, exist_ok=True)
for name in args.mutant or MUTANTS:
# Explicit mktemp, no underscores: same TMPDIR rule as the real gate.
scratch = Path(subprocess.check_output(['mktemp', '-d', '/tmp/huey272-fallback-mutant-XXXXXXXX'], text=True).strip())
try:
for directory in ['packages', 'roles']:
shutil.copytree(source / directory, scratch / directory)
if (source / 'node_modules').exists():
(scratch / 'node_modules').symlink_to((source / 'node_modules').resolve(), target_is_directory=True)
for filename in ['package.json', 'RELEASE']:
if (source / filename).exists():
shutil.copy2(source / filename, scratch / filename)
path = scratch / 'packages/webui/src/public/chat.js'
old, new, target = MUTANTS[name]
text = path.read_text()
if text.count(old) != 1:
raise RuntimeError(f'{name}: mutation anchor is not unique')
path.write_text(text.replace(old, new, 1))
env = dict(os.environ, TMPDIR=str(scratch))
result = subprocess.run(['timeout', '-k', '10', '300', 'node', '--test', '--test-reporter=tap',
'packages/webui/tests/chat-sessions-access.test.mjs'],
cwd=scratch, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
(args.out / f'{name}.tap').write_text(result.stdout)
killed = result.returncode == 1 and 'ERR_ASSERTION' in result.stdout and bool(
re.search(r'not ok [^\n]*' + re.escape(target), result.stdout))
print(json.dumps({'mutant': name, 'returncode': result.returncode, 'assertion_kill': killed}), flush=True)
if not killed:
raise RuntimeError(f'{name}: not an assertion kill; stop, diagnose and report before any fix')
finally:
# These tests never use compose. Remove the disposable copy after browser teardown;
# retained TAP lives only in --out. No live network/volume/service cleanup is performed.
shutil.rmtree(scratch)
if __name__ == '__main__':
main()