volt
10ec491129
Two test cells opened with sys.path.insert(0, /root/.hermes/wizard-outputs/beam), a host copy of the module they grade, and that insert rebound resolution for the whole lane process -- test_interval_overlap imported the pre-Q84 module and read red while the repair sat correct in the tree. Each cell now binds Path(__file__).resolve().parent and asserts the import landed there. New instrument: q133_verify.py behind CHECK-q133 (5 legs) and q133_oracle.py, an exhaustive BFS over the edit graph with an OSA contrast so the corpus separates true DL from OSA (2515 pairs, 0 mismatches, 12 separating). verify.py: 10 legs, 10 pass, 0 fault. beam/CHECK: 595 tests, 0 failures.
377 lines
15 KiB
Python
377 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Q133 verifier -- the beam lane's test cells must import the module beside them.
|
|
|
|
The fault (measured at HEAD `e1182ce`, 2026-10-31)
|
|
---------------------------------------------------
|
|
Two `test_*.py` files in `wizard-work/beam/` opened with
|
|
|
|
sys.path.insert(0, "/root/.hermes/wizard-outputs/beam")
|
|
|
|
an absolute path to a *host copy* of the module they grade. The copies were
|
|
byte-identical to the tree's modules when each file landed, which is what made
|
|
the fault quiet: the cells were green, and they were grading the wrong file.
|
|
It was not only their own module either -- the insert put that directory at
|
|
`sys.path[0]` for the whole lane process, and `beam/CHECK` runs every test file
|
|
in one process, so `test_interval_overlap.py` (written for Q84's O(n log n)
|
|
early exit) imported the host's PRE-Q84 `interval_overlap` and read red, with
|
|
the traceback naming `/root/.hermes/wizard-outputs/beam/interval_overlap.py`.
|
|
Measured on 3000 fully-overlapping intervals: 26.381 s through the host copy,
|
|
0.0 s through the one in this tree. A repair in the tree was invisible while a
|
|
green gate said everything was fine.
|
|
|
|
What this instrument grades
|
|
---------------------------
|
|
bindings `tools/import_resolve_check.py` over the whole tree: rc 0, 0 faults
|
|
scan a second, independent scan of every test_*.py for a `sys.path`
|
|
call whose argument is an absolute literal outside the root
|
|
sibling the two repaired cells, copied OUT of the tree with their modules
|
|
and run there: rc 0, and the module they imported is the copy's
|
|
mutant a copy whose binding is rewritten to the host path must fail, and
|
|
must name that path -- the binding assertion is not decorative
|
|
oracle `q133_oracle.py`: an exhaustive BFS over the edit graph against the
|
|
module, with an OSA contrast so the corpus is shown to separate
|
|
true DL from OSA, i.e. the leg cannot be vacuous
|
|
|
|
Exit codes, per the repo's tools/ contract:
|
|
0 every leg as expected
|
|
1 a graded fault
|
|
2 could not grade (missing file, no python3, a fixture that could not be built)
|
|
|
|
`--root` grades another tree; paths otherwise resolve from __file__, so a clone
|
|
grades its own checkout.
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
DEFAULT_ROOT = os.path.dirname(os.path.dirname(HERE))
|
|
|
|
# The two files this item repaired, and the module names they import.
|
|
CELLS = [
|
|
("test_damerau_levenshtein.py", "damerau_levenshtein"),
|
|
("test_expr.py", "expr_eval"),
|
|
]
|
|
LANE = os.path.join("wizard-work", "beam")
|
|
|
|
# The host directory the pre-repair cells pointed at. Used by the mutant leg
|
|
# only, and named once so the diagnosis and the fixture cannot drift apart.
|
|
HOST_COPY = "/root/.hermes/wizard-outputs/beam"
|
|
|
|
SYS_PATH_CALL = re.compile(
|
|
r"sys\.path\s*\.\s*(insert|append|extend)\s*\(\s*[^,]+,\s*(['\"])"
|
|
r"(/[^'\"]*)\2"
|
|
)
|
|
|
|
|
|
class Report:
|
|
def __init__(self):
|
|
self.passed = []
|
|
self.failed = []
|
|
self.info = []
|
|
|
|
def ok(self, leg, detail):
|
|
self.passed.append(leg)
|
|
print("PASS %s: %s" % (leg, detail))
|
|
|
|
def bad(self, leg, detail):
|
|
# `FAIL:` -- with the colon -- is the token tools/verify.py treats as a
|
|
# STRONG hint, so a gate wrapping this reads THIS line (Q130).
|
|
self.failed.append(leg)
|
|
print("FAIL: %s -- %s" % (leg, detail))
|
|
|
|
def note(self, leg, detail):
|
|
self.info.append(leg)
|
|
print("INFO %s: %s" % (leg, detail))
|
|
|
|
|
|
def run(cmd, cwd=None, timeout=300, env=None):
|
|
return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True,
|
|
timeout=timeout, env=env)
|
|
|
|
|
|
def test_files(root):
|
|
out = []
|
|
for base in ("wizard-work", "tools", "replies"):
|
|
top = os.path.join(root, base)
|
|
if not os.path.isdir(top):
|
|
continue
|
|
for dirpath, dirnames, filenames in os.walk(top):
|
|
dirnames[:] = [d for d in dirnames if d not in
|
|
("__pycache__", ".git", ".pytest_cache")]
|
|
for fn in filenames:
|
|
if fn.startswith("test_") and fn.endswith(".py"):
|
|
out.append(os.path.join(dirpath, fn))
|
|
return sorted(out)
|
|
|
|
|
|
def leg_bindings(root, rep):
|
|
tool = os.path.join(root, "tools", "import_resolve_check.py")
|
|
if not os.path.exists(tool):
|
|
return "cannot"
|
|
r = run([sys.executable, tool, "--repo", root])
|
|
lines = [ln for ln in r.stdout.splitlines() if ln.strip()]
|
|
detail = lines[-1] if lines else (r.stderr or "").strip() or "(no output)"
|
|
if r.returncode == 2:
|
|
# The tool's own contract: rc 2 = it could not grade this root (e.g. a
|
|
# tree with no git, where `git rev-parse` is how it names the rev). A
|
|
# leg that read that as a fault would accuse a tree of the thing the
|
|
# tool declined to measure -- the verdict class Q133's record names.
|
|
rep.note("bindings", "could not grade: %s" % detail)
|
|
return "cannot"
|
|
if r.returncode != 0 or "0 fault(s)" not in r.stdout:
|
|
faults = [ln for ln in r.stdout.splitlines() if ln.startswith("FAULT")]
|
|
rep.bad("bindings", "%s | %s" % (detail, "; ".join(faults[:4])))
|
|
return "fault"
|
|
rep.ok("bindings", detail)
|
|
return "pass"
|
|
|
|
|
|
def _foreign_literals(path, root_real):
|
|
"""Absolute string literals handed to sys.path.* in real code, not prose.
|
|
|
|
Parsed with `ast` rather than matched as text: three of the lane's repaired
|
|
cells still *quote* the old absolute line in a comment that records the
|
|
repair (test_lfu_cache.py, test_robin_hood_hash.py, and Q133's own two), and
|
|
a text scan reads those quotations as live bindings.
|
|
"""
|
|
import ast
|
|
|
|
with open(path, encoding="utf-8", errors="replace") as fh:
|
|
tree = ast.parse(fh.read(), filename=path)
|
|
found = []
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.Call):
|
|
continue
|
|
fn = node.func
|
|
if not (isinstance(fn, ast.Attribute) and fn.attr in
|
|
("insert", "append", "extend")):
|
|
continue
|
|
base = fn.value
|
|
if not (isinstance(base, ast.Attribute) and base.attr == "path" and
|
|
isinstance(base.value, ast.Name) and base.value.id == "sys"):
|
|
continue
|
|
for arg in node.args:
|
|
if isinstance(arg, ast.Constant) and isinstance(arg.value, str) \
|
|
and arg.value.startswith("/"):
|
|
target = os.path.realpath(arg.value)
|
|
if target != root_real and not target.startswith(root_real + os.sep):
|
|
found.append((node.lineno, arg.value))
|
|
return found
|
|
|
|
|
|
def leg_scan(root, rep):
|
|
"""A second implementation of the same rule, so the two must agree."""
|
|
root_real = os.path.realpath(root)
|
|
hits = []
|
|
graded = 0
|
|
for path in test_files(root):
|
|
graded += 1
|
|
try:
|
|
found = _foreign_literals(path, root_real)
|
|
except SyntaxError as exc:
|
|
rep.bad("scan", "%s does not parse (%s)" % (path, exc))
|
|
return "fault"
|
|
except OSError as exc:
|
|
rep.bad("scan", "unreadable %s (%s)" % (path, exc))
|
|
return "fault"
|
|
for lineno, lit in found:
|
|
hits.append("%s:%d: %s" % (os.path.relpath(path, root), lineno, lit))
|
|
if hits:
|
|
rep.bad("scan", "%d absolute binding(s) outside the tree: %s"
|
|
% (len(hits), "; ".join(hits[:4])))
|
|
return "fault"
|
|
rep.ok("scan", "%d test file(s) parsed, 0 foreign sys.path binding(s)"
|
|
% graded)
|
|
return "pass"
|
|
|
|
|
|
def _cell_runner(dest, cell, module):
|
|
"""A loader that prints where the cell's import actually resolved."""
|
|
return (
|
|
"import json, runpy, sys\n"
|
|
"dest = %r\n"
|
|
"sys.path.insert(0, dest)\n"
|
|
"runpy.run_path(dest + '/' + %r, run_name='__cell__')\n"
|
|
"mod = sys.modules.get(%r)\n"
|
|
"print(json.dumps({'file': getattr(mod, '__file__', None)}))\n"
|
|
% (dest, cell, module)
|
|
)
|
|
|
|
|
|
def leg_sibling(root, rep):
|
|
"""Copy the cells OUT of the tree and run them: the binding must travel."""
|
|
lane = os.path.join(root, LANE)
|
|
missing = [c for c, _ in CELLS if not os.path.exists(os.path.join(lane, c))]
|
|
if missing:
|
|
rep.bad("sibling", "missing cell file(s): %s" % ", ".join(missing))
|
|
return "cannot"
|
|
tmp = tempfile.mkdtemp(prefix="q133-sibling-")
|
|
try:
|
|
dest = os.path.join(tmp, "lane")
|
|
os.makedirs(dest)
|
|
copied = []
|
|
for cell, module in CELLS:
|
|
for name in (cell, module + ".py"):
|
|
src = os.path.join(lane, name)
|
|
if os.path.exists(src):
|
|
shutil.copy2(src, os.path.join(dest, name))
|
|
copied.append(name)
|
|
runner = os.path.join(tmp, "runner.py")
|
|
readings = []
|
|
env = dict(os.environ)
|
|
env.pop("PYTHONPATH", None)
|
|
for cell, module in CELLS:
|
|
with open(runner, "w") as fh:
|
|
fh.write(_cell_runner(dest, cell, module))
|
|
r = run([sys.executable, runner], cwd=dest, env=env)
|
|
if r.returncode != 0:
|
|
blob = (r.stderr or "") + (r.stdout or "")
|
|
rep.bad("sibling", "%s copied to %s reads rc %d: %s"
|
|
% (cell, dest, r.returncode,
|
|
" | ".join(blob.strip().splitlines()[-2:])))
|
|
return "fault"
|
|
try:
|
|
got = json.loads(r.stdout.strip().splitlines()[-1])
|
|
except (ValueError, IndexError):
|
|
rep.bad("sibling", "%s printed no resolvable path: %r"
|
|
% (cell, r.stdout[-200:]))
|
|
return "fault"
|
|
where = got.get("file") or ""
|
|
if os.path.dirname(os.path.realpath(where)) != os.path.realpath(dest):
|
|
rep.bad("sibling", "%s imported %s -- not the module beside it"
|
|
% (cell, where))
|
|
return "fault"
|
|
readings.append("%s -> %s" % (cell, module + ".py"))
|
|
rep.ok("sibling", "%d file(s) copied to a temp lane and run from there: %s"
|
|
% (len(copied), "; ".join(readings)))
|
|
return "pass"
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
|
|
def leg_mutant(root, rep):
|
|
"""Rewrite the binding to the host path: the cell must fail and name it."""
|
|
lane = os.path.join(root, LANE)
|
|
cell, module = CELLS[0]
|
|
src = os.path.join(lane, cell)
|
|
if not os.path.exists(src):
|
|
rep.bad("mutant", "missing %s" % src)
|
|
return "cannot"
|
|
with open(src, encoding="utf-8") as fh:
|
|
text = fh.read()
|
|
mutant = text.replace(
|
|
"sys.path.insert(0, str(HERE))",
|
|
"sys.path.insert(0, %r)" % HOST_COPY, 1)
|
|
if mutant == text:
|
|
# The precondition of this leg is the repaired binding line itself. A
|
|
# tree without it is not a tree this leg cannot grade -- it is a tree
|
|
# that regressed, and it must read as a FAULT (rc 1), not as rc 2.
|
|
# Measured on the reverted tree: with rc 2 here, `q133_verify.py` on the
|
|
# pre-repair bytes read "could not grade" and hid the three real faults
|
|
# the other legs had just printed.
|
|
rep.bad("mutant", "the repaired binding line is not present in %s -- the "
|
|
"tree has regressed (pre-repair rc 2 here hid three faults)" % cell)
|
|
return "fault"
|
|
tmp = tempfile.mkdtemp(prefix="q133-mutant-")
|
|
try:
|
|
dest = os.path.join(tmp, "lane")
|
|
os.makedirs(dest)
|
|
# the module goes in the copy too, so only the binding can be at fault
|
|
shutil.copy2(os.path.join(lane, module + ".py"),
|
|
os.path.join(dest, module + ".py"))
|
|
with open(os.path.join(dest, cell), "w") as fh:
|
|
fh.write(mutant)
|
|
r = run([sys.executable, os.path.join(dest, cell)], cwd=dest)
|
|
blob = (r.stderr or "") + (r.stdout or "")
|
|
if r.returncode == 0:
|
|
rep.bad("mutant", "cell bound to %s read rc 0 -- the binding "
|
|
"assertion does not bite" % HOST_COPY)
|
|
return "fault"
|
|
if HOST_COPY not in blob and "AssertionError" not in blob:
|
|
rep.bad("mutant", "cell failed (rc %d) without naming the binding: %s"
|
|
% (r.returncode, " | ".join(blob.strip().splitlines()[-2:])))
|
|
return "fault"
|
|
rep.ok("mutant", "host-bound copy reads rc %d and names %s"
|
|
% (r.returncode, HOST_COPY))
|
|
return "pass"
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
|
|
def leg_oracle(root, rep):
|
|
"""Grade the module against an exhaustive BFS (q133_oracle.py)."""
|
|
oracle = os.path.join(HERE, "q133_oracle.py")
|
|
if not os.path.exists(oracle):
|
|
rep.bad("oracle", "missing %s" % oracle)
|
|
return "cannot"
|
|
r = run([sys.executable, oracle, os.path.join(root, LANE)], timeout=180)
|
|
if r.returncode != 0:
|
|
rep.bad("oracle", "probe rc %d: %s"
|
|
% (r.returncode, ((r.stderr or "") + (r.stdout or "")).strip()[-300:]))
|
|
return "fault"
|
|
try:
|
|
got = json.loads(r.stdout.strip().splitlines()[-1])
|
|
except (ValueError, IndexError):
|
|
rep.bad("oracle", "probe printed no reading: %r" % r.stdout[-300:])
|
|
return "fault"
|
|
if got["mismatches"]:
|
|
rep.bad("oracle", "%d mismatch(es) vs BFS, e.g. %s"
|
|
% (got["mismatches"], got["examples"][:3]))
|
|
return "fault"
|
|
if got["bfs_vs_osa"] == 0:
|
|
rep.bad("oracle", "the corpus cannot separate true DL from OSA over %d "
|
|
"pairs -- the leg would be vacuous" % got["decided"])
|
|
return "fault"
|
|
rep.ok("oracle", "%d pair(s) decided by BFS, 0 mismatches, %d pair(s) where "
|
|
"BFS and OSA disagree (so the corpus separates them)"
|
|
% (got["decided"], got["bfs_vs_osa"]))
|
|
return "pass"
|
|
|
|
|
|
LEGS = (("bindings", leg_bindings), ("scan", leg_scan),
|
|
("sibling", leg_sibling), ("mutant", leg_mutant),
|
|
("oracle", leg_oracle))
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(description="Q133 verifier")
|
|
ap.add_argument("--root", default=DEFAULT_ROOT)
|
|
ap.add_argument("--json", action="store_true")
|
|
args = ap.parse_args(argv)
|
|
root = os.path.abspath(args.root)
|
|
if not os.path.isdir(os.path.join(root, LANE)):
|
|
print("q133: could not grade -- no %s under %s" % (LANE, root))
|
|
return 2
|
|
|
|
rep = Report()
|
|
outcomes = {}
|
|
for name, fn in LEGS:
|
|
outcomes[name] = fn(root, rep)
|
|
|
|
cannot = [k for k, v in outcomes.items() if v == "cannot"]
|
|
faults = [k for k, v in outcomes.items() if v == "fault"]
|
|
if args.json:
|
|
print(json.dumps({"root": root, "outcomes": outcomes,
|
|
"passed": rep.passed, "failed": rep.failed},
|
|
sort_keys=True))
|
|
if cannot:
|
|
print("q133: could not grade -- %s" % ", ".join(cannot))
|
|
return 2
|
|
if faults:
|
|
print("q133: %d leg(s), %d passed, %d fault(s) -- %s"
|
|
% (len(outcomes), len(rep.passed), len(faults), ", ".join(faults)))
|
|
return 1
|
|
print("q133: %d leg(s), %d passed, 0 fault(s) -- beam cells import the module "
|
|
"beside them" % (len(outcomes), len(rep.passed)))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|