#!/usr/bin/env python3 """Q136 verifier -- the spark lane's word-frequency cell must grade the copy beside it. The fault (measured at HEAD `4d329fa`, 2026-10-31) -------------------------------------------------- `wizard-work/spark/test_wordfreq_counter.py` opened with sys.path.insert(0, '/root/.hermes/wizard-outputs/spark') and read its fixture from the same absolute path, so the cell graded a host copy of `wordfreq_counter.py` and a host copy of `test_input.txt` on a host that happened to hold them. The copies were byte-identical when the file landed (`0696858a...` module, `3127d0f8...` fixture), which is what made the fault quiet: green readings, wrong file, and no way to run the cell at all on a host without `/root/.hermes/wizard-outputs/spark`. `tools/import_resolve_check.py` named it at HEAD as the repo gate's only red leg: FAULT OUTSIDE_REPO wizard-work/spark/test_wordfreq_counter.py:8 sys.path.insert -> /root/.hermes/wizard-outputs/spark import_resolve_check: 157 file(s) graded, 1 fault(s) (OUTSIDE_REPO 1) The tree already tracked `wizard-work/spark/test_input.txt` with the same bytes, so the repair is a binding change and nothing else: `HERE = Path(__file__).resolve().parent`, the module imported from `HERE` and asserted to have landed there, the fixture `HERE / "test_input.txt"`, and the report's own `file` field asserted equal to that fixture. What this instrument grades --------------------------- scan `tools/import_resolve_check.py` over the whole tree: rc 0, 0 faults run the repaired cell, run in place from `cwd=/`: rc 0, three PASS lines sandbox the cell + module + fixture copied to a temp directory outside the tree, run under CPython's `open` audit hook: rc 0, the module and fixture it opened are the sandbox copies (compared by real path, so a symlinked TMPDIR cannot fake a miss), the cell's own printed module/fixture paths are the sandbox copies, and nothing under the tree or under the old host directory was opened counts the tracked fixture's 24 / 18 / `the`:4 recomputed by a tokenizer that does not call the module, so the constants are not read back from the thing they grade shadow the load-bearing leg. A tree stand-in holds a SABOTAGED module (+1 on `total_words`) and a separate "foreign" directory holds an intact copy. The PRE-REPAIR bytes, bound to the foreign directory, read **rc 0 / All tests passed** beside the sabotage -- the verdict came from outside the tree. The repaired cell on the same sabotage reads **rc 1 / Expected 24 words, got 25** mutant a throwaway git repository holding the same three paths: repaired bytes -> rc 0; the pre-repair bytes restored -> rc 1, and the row names `OUTSIDE_REPO` at `:8` (the gate leg is not decorative) gate `tools/verify.py --repo . --fast` over the graded tree: rc 0, 0 fault Exit codes, per the repo's tools/ contract: 0 every leg as expected 1 a graded fault 2 could not grade (a fixture could not be built, git or python3 missing) `--root` grades another tree; every path resolves from __file__ otherwise. The graded tree is only ever read: the shadow and mutant legs work on copies. """ from __future__ import annotations import argparse import hashlib import json import os import re import shutil import subprocess import sys import tempfile from pathlib import Path HERE = Path(__file__).resolve().parent REPO = HERE.parent.parent TEST_REL = "wizard-work/spark/test_wordfreq_counter.py" MODULE_REL = "wizard-work/spark/wordfreq_counter.py" FIXTURE_REL = "wizard-work/spark/test_input.txt" HOST_DIR = "/root/.hermes/wizard-outputs/spark" PRE_REPAIR_REV = "4d329fa" PRE_REPAIR_BLOB = "7855bdfc9170" PRE_REPAIR_SHA256 = "55cbacdab11c7fb7f852d8b3bc9cab1da542e4f8bae5db42422f409690a4b55f" FIXTURE_SHA256 = "3127d0f89b3e87760130ee60e68eb9d9bf213ee83019062119361f1529cfd777" AUDIT_DRIVER = r''' import json, os, runpy, sys from pathlib import Path root = Path(sys.argv[1]).resolve() opened = [] def hook(event, args): if event == "open": p = args[0] if isinstance(p, (str, bytes, os.PathLike)): try: opened.append(str(Path(os.fsdecode(p)).resolve())) except Exception: pass sys.addaudithook(hook) os.chdir(str(root)) sys.argv = ["test_wordfreq_counter.py"] try: runpy.run_path(str(root / "test_wordfreq_counter.py"), run_name="__main__") except SystemExit: pass print("AUDIT " + json.dumps(sorted(set(opened)))) ''' class GradeError(Exception): """The instrument could not build what it grades (rc 2, never a fault).""" def run(cmd, cwd=None, timeout=300, env=None): e = dict(os.environ) if env: e.update(env) try: p = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout, env=e) except FileNotFoundError as exc: raise GradeError("cannot run %r: %s" % (cmd[0], exc)) except subprocess.TimeoutExpired: raise GradeError("%r did not finish inside %ss" % (cmd, timeout)) return p.returncode, p.stdout, p.stderr def git(repo, *args, timeout=120): return run(["git", "-C", str(repo)] + list(args), timeout=timeout) def sha256(path): h = hashlib.sha256() with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(65536), b""): h.update(chunk) return h.hexdigest() def require(path, what): p = Path(path) if not p.exists(): raise GradeError("%s absent: %s" % (what, p)) return p # --------------------------------------------------------------------------- legs def leg_scan(repo): rc, out, err = run([sys.executable, "tools/import_resolve_check.py", "--repo", str(repo)], cwd=str(repo)) text = out + err ok = rc == 0 and re.search(r"(\d+) file\(s\) graded, 0 fault\(s\)", text) is not None m = re.search(r"(\d+) file\(s\) graded, (\d+) fault\(s\)", text) detail = (m.group(0) if m else text.strip().splitlines()[-1:] or ["no summary"]) if isinstance(detail, list): detail = detail[-1] return ok, "rc=%d %s" % (rc, detail) def leg_run(repo): test = require(repo / TEST_REL, "test") rc, out, err = run([sys.executable, str(test)], cwd="/") passes = out.count(": PASS") ok = rc == 0 and passes == 3 and "All tests passed!" in out return ok, "rc=%d %d PASS line(s), cwd=/" % (rc, passes) def leg_sandbox(repo): test = require(repo / TEST_REL, "test") mod = require(repo / MODULE_REL, "module") fixture = require(repo / FIXTURE_REL, "fixture") base = tempfile.mkdtemp(prefix="q136-sandbox-") try: box = os.path.join(base, "cell") os.makedirs(box) for src in (test, mod, fixture): shutil.copy2(src, os.path.join(box, src.name)) driver = os.path.join(box, "_audit_driver.py") with open(driver, "w", encoding="utf-8") as fh: fh.write(AUDIT_DRIVER) rc, out, err = run([sys.executable, driver, box], cwd="/") if rc != 0: return False, "sandbox run rc=%d: %s" % (rc, (err or out).strip().splitlines()[-1:]) line = [ln for ln in out.splitlines() if ln.startswith("AUDIT ")] if not line: raise GradeError("audit driver printed no AUDIT line") opened = json.loads(line[-1][len("AUDIT "):]) # Temp directories can live behind a symlink (a shifted TMPDIR, for one), # so both sides of the comparison are taken by real path: the hook # records resolved names and a raw join would read a false miss. box_real = os.path.realpath(box) opened_real = {os.path.realpath(p) for p in opened} need = {os.path.join(box_real, name) for name in ("wordfreq_counter.py", "test_input.txt")} missing = sorted(need - opened_real) repo_real = os.path.realpath(str(repo)) host_real = os.path.realpath(HOST_DIR) outside = sorted(p for p in opened_real if p == repo_real or p.startswith(repo_real + os.sep) or p == host_real or p.startswith(host_real + os.sep)) # The cell prints what it graded; those paths must be the sandbox copies # too, so the leg does not rest on the audit hook alone. printed = {} for ln in out.splitlines(): if ln.startswith("module: "): printed["module"] = os.path.realpath(ln[len("module: "):]) elif ln.startswith("fixture: "): printed["fixture"] = os.path.realpath(ln[len("fixture: "):]) expected = {"module": os.path.join(box_real, "wordfreq_counter.py"), "fixture": os.path.join(box_real, "test_input.txt")} wrong_print = sorted(k for k, v in printed.items() if v != expected[k]) ok = not missing and not outside and not wrong_print \ and set(printed) == {"module", "fixture"} detail = "%d path(s) opened, %d outside the sandbox" % (len(opened_real), len(outside)) if missing: detail += ", never opened: %s" % ", ".join(os.path.basename(p) for p in missing) if wrong_print: detail += ", printed %s not the sandbox copy" % ", ".join(wrong_print) if outside: detail += ", outside: %s" % ", ".join(outside) return ok, detail finally: shutil.rmtree(base, ignore_errors=True) def leg_counts(repo): fixture = require(repo / FIXTURE_REL, "fixture") digest = sha256(fixture) text = fixture.read_text(encoding="utf-8") toks = re.findall(r"\S+", text.lower()) counts = {} for t in toks: counts[t] = counts.get(t, 0) + 1 ok = (digest == FIXTURE_SHA256 and len(toks) == 24 and len(counts) == 18 and counts.get("the") == 4) return ok, ("sha256=%s total=%d distinct=%d the=%d" % (digest[:16], len(toks), len(counts), counts.get("the", 0))) def leg_shadow(repo): repo = Path(repo) test = require(repo / TEST_REL, "test") mod = require(repo / MODULE_REL, "module") fixture = require(repo / FIXTURE_REL, "fixture") rc0, old, err0 = git(repo, "show", "%s:%s" % (PRE_REPAIR_REV, TEST_REL)) if rc0 != 0: raise GradeError("cannot read %s:%s -- %s" % (PRE_REPAIR_REV, TEST_REL, err0.strip())) old_sha = hashlib.sha256(old.encode("utf-8")).hexdigest() if old_sha != PRE_REPAIR_SHA256: raise GradeError("pre-repair blob reads sha256=%s, expected %s" % (old_sha[:16], PRE_REPAIR_SHA256[:16])) if old.count(HOST_DIR) != 3: raise GradeError("pre-repair bytes carry %d host references, expected 3 " "(the sys.path insert and the two fixture paths)" % old.count(HOST_DIR)) base = tempfile.mkdtemp(prefix="q136-shadow-") try: tree = os.path.join(base, "tree") foreign = os.path.join(base, "foreign") os.makedirs(tree) os.makedirs(foreign) # the tree stand-in: sabotaged module, but the cell beside it is the repair sabotaged = mod.read_text(encoding="utf-8") needle = '"total_words": sum(counts.values()),' if sabotaged.count(needle) != 1: raise GradeError("sabotage target not unique in %s" % MODULE_REL) sabotaged = sabotaged.replace(needle, '"total_words": sum(counts.values()) + 1,') with open(os.path.join(tree, "wordfreq_counter.py"), "w", encoding="utf-8") as fh: fh.write(sabotaged) shutil.copy2(fixture, os.path.join(tree, "test_input.txt")) shutil.copy2(test, os.path.join(tree, "test_wordfreq_counter.py")) # the foreign directory: an intact copy of the same two files shutil.copy2(mod, os.path.join(foreign, "wordfreq_counter.py")) shutil.copy2(fixture, os.path.join(foreign, "test_input.txt")) # the pre-repair cell, bound to the foreign directory old_path = os.path.join(base, "old_test.py") with open(old_path, "w", encoding="utf-8") as fh: fh.write(old.replace(HOST_DIR, foreign)) rc_old, out_old, err_old = run([sys.executable, old_path], cwd=tree) rc_new, out_new, err_new = run([sys.executable, os.path.join(tree, "test_wordfreq_counter.py")], cwd=tree) old_green = rc_old == 0 and "All tests passed!" in out_old new_red = rc_new != 0 and "Expected 24 words, got 25" in (out_new + err_new) ok = old_green and new_red return ok, ("old rc=%d green-while-sibling-sabotaged=%s; repaired rc=%d names the sabotage=%s" % (rc_old, old_green, rc_new, new_red)) finally: shutil.rmtree(base, ignore_errors=True) def leg_mutant(repo): repo = Path(repo) test = require(repo / TEST_REL, "test") mod = require(repo / MODULE_REL, "module") fixture = require(repo / FIXTURE_REL, "fixture") rc0, old, err0 = git(repo, "show", "%s:%s" % (PRE_REPAIR_REV, TEST_REL)) if rc0 != 0: raise GradeError("cannot read the pre-repair bytes: %s" % err0.strip()) tool = require(repo / "tools/import_resolve_check.py", "import_resolve_check.py") base = tempfile.mkdtemp(prefix="q136-mutant-") try: mini = os.path.join(base, "mini") spark = os.path.join(mini, "wizard-work", "spark") os.makedirs(spark) for src in (test, mod, fixture): shutil.copy2(src, os.path.join(spark, src.name)) shutil.copytree(os.path.join(str(repo), "tools"), os.path.join(mini, "tools")) ident = ["-c", "user.name=q136", "-c", "user.email=q136@example.invalid"] rc, out, err = run(["git", "init", "-q", mini]) if rc != 0: raise GradeError("git init failed: %s" % err.strip()) run(["git", "-C", mini] + ident + ["add", "-A"]) rc, out, err = run(["git", "-C", mini] + ident + ["commit", "-qm", "repaired"]) if rc != 0: raise GradeError("git commit failed: %s" % err.strip()) rc_rep, out_rep, err_rep = run([sys.executable, tool, "--repo", mini], cwd=mini) with open(os.path.join(spark, os.path.basename(TEST_REL)), "w", encoding="utf-8") as fh: fh.write(old) run(["git", "-C", mini] + ident + ["add", "-A"]) rc_old, out_old, err_old = run([sys.executable, tool, "--repo", mini], cwd=mini) text = out_old + err_old named = ("OUTSIDE_REPO" in text and ":8" in text and "test_wordfreq_counter.py" in text) ok = rc_rep == 0 and rc_old == 1 and named return ok, ("repaired rc=%d, pre-repair rc=%d, row names OUTSIDE_REPO@:8=%s" % (rc_rep, rc_old, named)) finally: shutil.rmtree(base, ignore_errors=True) def leg_gate(repo): """The repo gate over the COMMITTED tree, in a detached worktree. Grading the worktree was the first draft and it read red for a reason outside this item: the spark lane's files were being edited in place while the run happened, so `path-claims` named a file that existed on disk and was not yet committed. A commit is what this item lands, so a commit is what gets graded -- HEAD, checked out beside the worktree, left dirty by nobody. """ repo = Path(repo) if not (repo / "tools/verify.py").exists(): raise GradeError("tools/verify.py absent in the graded tree") rc, head, err = git(repo, "rev-parse", "HEAD") if rc != 0: raise GradeError("cannot read HEAD: %s" % err.strip()) head = head.strip() base = tempfile.mkdtemp(prefix="q136-head-") wt = os.path.join(base, "head") try: rc, out, err = git(repo, "worktree", "add", "--detach", wt, "HEAD") if rc != 0: raise GradeError("cannot check out HEAD: %s" % err.strip()) rc, out, err = run([sys.executable, "tools/verify.py", "--repo", ".", "--fast"], cwd=wt, timeout=600) text = out + err m = re.search(r"(\d+) legs?, (\d+) pass, (\d+) fault", text) ok = rc == 0 and m is not None and m.group(3) == "0" return ok, ("at HEAD %.7s: %s" % (head, m.group(0) if m else text.strip().splitlines()[-1:])) finally: git(repo, "worktree", "remove", "--force", wt) shutil.rmtree(base, ignore_errors=True) LEGS = [("scan", leg_scan), ("run", leg_run), ("sandbox", leg_sandbox), ("counts", leg_counts), ("shadow", leg_shadow), ("mutant", leg_mutant), ("gate", leg_gate)] def main(argv=None): ap = argparse.ArgumentParser(description="Q136: the spark word-frequency cell grades the copy beside it") ap.add_argument("--root", default=str(REPO), help="the tree to grade (default: the checkout holding this file)") ap.add_argument("--leg", action="append", help="run only these legs") args = ap.parse_args(argv) repo = Path(args.root).resolve() wanted = [n for n, _ in LEGS if not args.leg or n in args.leg] faults = 0 for name, fn in LEGS: if name not in wanted: continue try: ok, detail = fn(repo) except GradeError as exc: print("q136: could not grade -- %s (%s)" % (name, exc)) return 2 print("q136: %-7s %-8s %s" % (name, "ok" if ok else "FAULT", detail)) if not ok: faults += 1 print("q136: %d leg(s), %d passed, %d fault(s)" % (len(wanted), len(wanted) - faults, faults)) return 0 if faults == 0 else 1 if __name__ == "__main__": sys.exit(main())