#!/usr/bin/env python3 """Q89: grade the import-time-read detector on every spelling it names, and on one spelling no list can name. What it grades, in legs: 1. clean the three refactor harnesses are green on the tree as committed and print their own totals (14/14, 14/14, 13/13 today). 2. evasion the four Q89 spellings -- ``Path(...).read_text()``, ``os.open(...)``, ``io.open(...)``, ``Path(...).open()`` -- are appended at module scope to each harness's module in turn and the harness must go red (rc 1) naming the site. Before the repair all four read rc 0 with "0 module-scope io calls; 0 runtime open calls" printed. 3. control a module-scope ``len(DEFAULT_PATH)`` is not a read: it must stay green, so the detector is not a constant-red. 4. blinding the two instruments are disabled in turn. With the static walk blinded the audit hook must be the one that reds check 12; with the audit hook blinded the static walk must. With both blinded check 12 prints OK and the mutant battery (check 13) must still red the harness -- a detector cannot be emptied quietly. 5. no-list ``_F = open`` / ``_F(DEFAULT_PATH)`` resolves to a name no set contains; the audit hook must catch the real open event anyway. This is the difference between a longer list and a closed class. 6. fourth the fourth copy of this detector (``verify_dump_seed_refactor.py``, forge's own later hardening) still passes 16/16 and the shared walk agrees with its 12-spelling corpus, so this module is a superset rather than a third convention. Everything runs in a temporary copy of ``wizard-work/forge`` beside a copy of ``inbox/seed-2026-09-26.json``; the tree the instrument ships in is never mutated, which is what lets it grade in one checkout and be run in another. rc 0 = every leg as expected; 1 = a graded fault; 2 = could not grade. """ from __future__ import annotations import importlib.util import json import os import shutil import subprocess import sys import tempfile ROOT = os.path.dirname(os.path.dirname(os.path.dirname( os.path.abspath(__file__)))) SEED = os.path.join("inbox", "seed-2026-09-26.json") SEED_SHA_PIN = None # the seed is copied, not pinned: its bytes are not graded HARNESSES = ( ("probe", "verify_triage_probe_refactor.py", "_triage_probe_refactored.py"), ("dump", "verify_triage_dump_refactor.py", "_triage_dump_refactored.py"), ("pending", "verify_triage_dump_pending_refactor.py", "_triage_dump_pending_refactored.py"), ) FOURTH = ("verify_dump_seed_refactor.py", "_dump_seed_refactored.py") EVASIONS = { "pathlib.read_text": "\nfrom pathlib import Path as _P\n_evade = _P(DEFAULT_PATH).read_text()\n", "os.open": "\nimport os as _o\n_fd = _o.open(DEFAULT_PATH, _o.O_RDONLY)\n_o.close(_fd)\n", "io.open": "\nimport io as _io\n_h = _io.open(DEFAULT_PATH)\n_h.close()\n", "Path.open": "\nfrom pathlib import Path as _P\nwith _P(DEFAULT_PATH).open() as _fh:\n" " _evade = _fh.read()\n", } CONTROL = "\n_neg = len(DEFAULT_PATH)\n" NO_LIST = "_F = open\n_h = _F(DEFAULT_PATH)\n_h.close()\n" BLIND_STATIC = "\nMODELLED_NAMES = frozenset()\nMODELLED_ATTRS = frozenset()\n" BLIND_AUDIT = ('\n\ndef audit_reads(module_path, timeout=120):\n' ' return {"rc": 0, "events": [], "import_error": None,\n' ' "error": None}\n') class Report: def __init__(self) -> None: self.faults = 0 self.legs = 0 def ok(self, leg: str, note: str) -> None: self.legs += 1 print("ok %-22s %s" % (leg, note)) def fail(self, leg: str, note: str) -> None: self.legs += 1 self.faults += 1 print("FAIL %-20s %s" % (leg, note)) def run(forge: str, harness: str): proc = subprocess.run(["python3", harness], cwd=forge, capture_output=True, text=True, timeout=300) out = proc.stdout or "" total = next((l.strip() for l in out.splitlines() if l.startswith("=== Verification")), "") first_fail = next((l.strip() for l in out.splitlines() if l.strip().startswith("FAIL")), "") # The io detector's own verdict, wherever it lands: a harness whose earlier # checks red first (the dump lane's dead-import check shadows it) must still # be graded on whether the io detector fired, not on check order. io_fail = next((l.strip() for l in out.splitlines() if l.strip().startswith("FAIL") and ("static read-spelling" in l or "open event(s)" in l)), "") return proc.returncode, total, first_fail, io_fail, out def load(name: str, path: str): spec = importlib.util.spec_from_file_location(name, path) if spec is None or spec.loader is None: raise SystemExit("cannot load %s" % path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def build_copy(base: str) -> str: """Copy the forge lane and the seed the modules read; return the copy's forge.""" work = os.path.join(base, "tree") forge = os.path.join(work, "wizard-work", "forge") os.makedirs(os.path.dirname(forge)) shutil.copytree(os.path.join(ROOT, "wizard-work", "forge"), forge, ignore=shutil.ignore_patterns("__pycache__", ".pytest_cache")) os.makedirs(os.path.join(work, "inbox")) shutil.copy2(os.path.join(ROOT, SEED), os.path.join(work, SEED)) return forge def with_suffix(forge: str, name: str, suffix: str, fn): """Run ``fn`` with ``suffix`` appended to one module, then restore the bytes.""" path = os.path.join(forge, name) with open(path, encoding="utf-8") as handle: original = handle.read() try: with open(path, "w", encoding="utf-8") as handle: handle.write(original + suffix) return fn() finally: with open(path, "w", encoding="utf-8") as handle: handle.write(original) with open(path, encoding="utf-8") as handle: if handle.read() != original: raise SystemExit(2) def main() -> int: if not os.path.isdir(os.path.join(ROOT, "wizard-work", "forge")): print("FAIL tree: %s is not this instrument's repo" % ROOT) return 2 report = Report() base = tempfile.mkdtemp(prefix="q89-") try: forge = build_copy(base) # (1) clean totals = {} for label, harness, _module in HARNESSES: rc, total, fail, _io_fail, _out = run(forge, harness) totals[label] = total if rc == 0 and total: report.ok("clean " + label, total.replace("=== Verification: ", "")) else: report.fail("clean " + label, "rc=%d %s %s" % (rc, total, fail)) # (2) evasion: every spelling must red every harness for label, harness, module in HARNESSES: for spelling, body in EVASIONS.items(): rc, _total, fail, io_fail, _out = with_suffix( forge, module, body, lambda: run(forge, harness)) if rc == 1 and io_fail: report.ok("%s %s" % (label, spelling), "rc=1, %s" % io_fail[:60]) else: report.fail("%s %s" % (label, spelling), "rc=%d, io detector said %r (first FAIL %r)" % (rc, io_fail[:50], fail[:50])) # (3) the negative control is not a read for label, harness, module in HARNESSES: rc, _total, fail, _io_fail, _out = with_suffix( forge, module, CONTROL, lambda: run(forge, harness)) if rc == 0: report.ok("control " + label, "len() at module scope stays green") else: report.fail("control " + label, "rc=%d %r" % (rc, fail[:80])) # (4) each instrument is load-bearing, and the battery is the tripwire probe_module = HARNESSES[0][2] probe_harness = HARNESSES[0][1] def _blinded(static_patch: str, audit_patch: str): detector = os.path.join(forge, "import_io_detector.py") with open(detector, encoding="utf-8") as handle: saved = handle.read() def go(): with open(detector, "w", encoding="utf-8") as handle: handle.write(saved + static_patch + audit_patch) return run(forge, probe_harness) try: return with_suffix(forge, probe_module, EVASIONS["pathlib.read_text"], go) finally: with open(detector, "w", encoding="utf-8") as handle: handle.write(saved) rc, _t, fail, io_fail, out = _blinded(BLIND_STATIC, "") if rc == 1 and io_fail and "open event(s) during import" in io_fail: report.ok("blinded static", "audit hook alone reds check 12") else: report.fail("blinded static", "rc=%d %r" % (rc, io_fail[:90])) rc, _t, fail, io_fail, out = _blinded("", BLIND_AUDIT) if rc == 1 and io_fail and "static read-spelling" in io_fail: report.ok("blinded audit", "static walk alone reds check 12") else: report.fail("blinded audit", "rc=%d %r" % (rc, io_fail[:90])) rc, _t, fail, io_fail, out = _blinded(BLIND_STATIC, BLIND_AUDIT) check12_clean = ("OK: no import-time read" in out and "modelled spellings" in out) battery_red = "spelling mutant(s) survived" in out if rc == 1 and check12_clean and battery_red: report.ok("blinded both", "check 12 green, check 13 battery reds it") else: report.fail("blinded both", "rc=%d check12_clean=%s battery_red=%s" % (rc, check12_clean, battery_red)) # (5) a spelling no list names rc, _t, fail, io_fail, out = with_suffix( forge, probe_module, NO_LIST, lambda: run(forge, probe_harness)) if rc == 1 and io_fail and "open event(s) during import" in io_fail: report.ok("no-list spelling", "_F = open caught by the audit hook") else: report.fail("no-list spelling", "rc=%d %r" % (rc, io_fail[:100])) # (6) the fourth copy still passes, and the shared walk is a superset rc, total, fail, _io_fail, _out = run(forge, FOURTH[0]) if rc == 0 and "16/16" in total: report.ok("fourth copy", total.replace("=== Verification: ", "")) else: report.fail("fourth copy", "rc=%d %s %s" % (rc, total, fail)) detector = load("q89_detector", os.path.join(forge, "import_io_detector.py")) fourth = load("q89_fourth", os.path.join(forge, FOURTH[0])) with open(os.path.join(forge, fourth.REFACTORED), encoding="utf-8") as handle: source = handle.read() missed = [label for label, snippet in fourth._IO_SPELLINGS if not detector.static_reads(source + "\n" + snippet + "\n")] if not missed: report.ok("corpus agreement", "%d/%d of the fourth's spellings flagged" % (len(fourth._IO_SPELLINGS), len(fourth._IO_SPELLINGS))) else: report.fail("corpus agreement", "shared walk missed %s" % missed) except (OSError, subprocess.SubprocessError, SystemExit) as exc: print("FAIL instrument: %s: %s" % (type(exc).__name__, exc)) return 2 finally: shutil.rmtree(base, ignore_errors=True) if report.faults: print("q89: FAULT -- %d leg line(s) red of %d graded" % (report.faults, report.legs)) return 1 print("q89: %d leg line(s), 0 fault -- OK" % report.legs) return 0 if __name__ == "__main__": sys.exit(main())