302 lines
12 KiB
Python
302 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Q65 cycle-7 controls -- the pin, the inert branch, and the four omissions.
|
|
|
|
One re-runnable file; every cell is a fresh clone of ``--rev`` in this repo.
|
|
It grades the row written for ``wizard-work/forge/verify_probe_forge_refactor.py``
|
|
and answers four questions the ledger cannot answer about itself:
|
|
|
|
PART 1 Does the PASS cover the committed tree?
|
|
The verifier runs the two probe subjects; both subjects import
|
|
``chunker`` from ``/root/.hermes/wizard-outputs/forge``. So sabotage
|
|
the CLONE's ``wizard-work/forge/chunker.py`` and see whether the row
|
|
notices. It does not (B = rc 0): the row is host-dependent in
|
|
substance while the runner's host-pin scan reads the verifier's own
|
|
source and finds nothing -- a both-levels pin.
|
|
C and D are the matched controls: rewrite the subjects' one sys.path
|
|
line into the clone, and the same sabotage turns rc 1 (named FAIL),
|
|
with a green run when the chunker is restored.
|
|
|
|
PART 2 Is the ``for called in (...)`` block in check_helpers_extracted
|
|
able to fail? Static: the AST of its body holds no Raise/Return/
|
|
Assign/exit. Dynamic: delete the block and both subject states give
|
|
the same rc and the same message (green 0/0, helper-call removed
|
|
1/1). It is a branch no input can fail -- the enforcement is the
|
|
name_calls block below it.
|
|
|
|
PART 3 The four omissions: census row dropped, ledger row dropped, verifier
|
|
edited and committed after rowing (DRIFT / STALE), and the gate on an
|
|
untouched clone (rc 0). The census-row cell is the one that changed
|
|
this instrument: dropped alone it is not a fault (the runner row
|
|
satisfies "not unwired"), but the scan used to print that file as
|
|
UNCLASSED -- false, because the runner had graded it. The cell now
|
|
requires the corrected wording: coverage drops to 109 of 110 and the
|
|
printer names the file CENSUS_ROW_MISSING, not UNCLASSED.
|
|
|
|
Usage:
|
|
python3 wizard-work/volt/q65_c7_controls.py [--rev REV] [--keep]
|
|
|
|
Exit 0 when every cell behaves as its ``expect`` line prints; 1 otherwise.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ast
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
VER = "wizard-work/forge/verify_probe_forge_refactor.py"
|
|
PROBE_R = "wizard-work/forge/_probe_forge_refactored.py"
|
|
PROBE_O = "wizard-work/forge/_probe_forge.py"
|
|
CHUNKER = "wizard-work/forge/chunker.py"
|
|
CENSUS = "wizard-work/volt/q63_unwired_baseline.tsv"
|
|
LEDGER = "wizard-work/volt/q65_runner_baseline.tsv"
|
|
TARGET = "verify_probe_forge_refactor"
|
|
|
|
SABOTAGE = '''
|
|
|
|
# --- control: make reconstruct lossy without touching chunk_text -----------
|
|
_reconstruct_virgin = reconstruct
|
|
|
|
|
|
def reconstruct(chunks, overlap_tokens=32): # noqa: F811
|
|
return _reconstruct_virgin(chunks, overlap_tokens) + "SABOTAGE"
|
|
'''
|
|
|
|
WHICH = (
|
|
"import runpy, sys\n"
|
|
"try:\n"
|
|
" runpy.run_path(%r, run_name='__main__')\n"
|
|
"except SystemExit:\n"
|
|
" pass\n"
|
|
"print('CHUNKER_FILE:', sys.modules['chunker'].__file__, file=sys.stderr)\n"
|
|
)
|
|
|
|
RESULTS = []
|
|
|
|
|
|
def sh(cmd, cwd):
|
|
return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=900)
|
|
|
|
|
|
def clone(work, name, rev):
|
|
d = os.path.join(work, name)
|
|
if os.path.isdir(d):
|
|
shutil.rmtree(d)
|
|
r = subprocess.run(["git", "clone", "-q", REPO, d], capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
sys.exit("clone failed: " + r.stderr)
|
|
r = subprocess.run(["git", "checkout", "-q", rev], cwd=d,
|
|
capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
sys.exit("checkout %s failed: %s" % (rev, r.stderr))
|
|
return d
|
|
|
|
|
|
def rewrite_paths(d):
|
|
"""Point both probe subjects' chunker import at the clone's own forge dir."""
|
|
n = 0
|
|
for p in (PROBE_R, PROBE_O):
|
|
full = os.path.join(d, p)
|
|
src = open(full, encoding="utf-8").read()
|
|
old = 'sys.path.insert(0, "/root/.hermes/wizard-outputs/forge")'
|
|
new = 'sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))'
|
|
if old in src:
|
|
if "\nimport os\n" not in src:
|
|
src = src.replace("import sys\n", "import os\nimport sys\n", 1)
|
|
open(full, "w", encoding="utf-8").write(src.replace(old, new))
|
|
n += 1
|
|
return n
|
|
|
|
|
|
def sabotage_chunker(d):
|
|
with open(os.path.join(d, CHUNKER), "a", encoding="utf-8") as fh:
|
|
fh.write(SABOTAGE)
|
|
|
|
|
|
def run_verifier(d):
|
|
r = sh([sys.executable, "-B", VER], d)
|
|
lines = (r.stdout + r.stderr).splitlines()
|
|
fails = [ln for ln in lines if ln.startswith("FAIL")]
|
|
oks = [ln for ln in lines if ln.startswith("OK:")]
|
|
summary = [ln for ln in lines if ln.startswith("=== Verification")]
|
|
return r.returncode, len(oks), (fails[0] if fails else ""), \
|
|
(summary[0].strip("= ") if summary else "-")
|
|
|
|
|
|
def cell(label, got, expect, detail):
|
|
ok = got == expect
|
|
RESULTS.append(ok)
|
|
print("%-4s %-46s rc=%-2d expect=%d %s"
|
|
% ("ok" if ok else "FAIL", label, got, expect, detail))
|
|
return ok
|
|
|
|
|
|
def part1(work, rev):
|
|
print("PART 1 -- does the PASS cover the committed tree?")
|
|
d_b = clone(work, "cB", rev)
|
|
sabotage_chunker(d_b)
|
|
which = os.path.join(work, "which_chunker.py")
|
|
open(which, "w", encoding="utf-8").write(WHICH % PROBE_R)
|
|
r = sh([sys.executable, "-B", which], d_b)
|
|
line = [ln for ln in r.stderr.splitlines() if ln.startswith("CHUNKER_FILE:")]
|
|
path = line[0].split(": ", 1)[1] if line else "(none)"
|
|
inside = path.startswith(d_b + os.sep)
|
|
print(" import resolution (cwd=clone, chunker sabotaged): %s" % path)
|
|
print(" resolves inside the clone: %s" % inside)
|
|
|
|
rc_a, ok_a, _, _ = run_verifier(clone(work, "cA", rev))
|
|
rc_b, ok_b, _, _ = run_verifier(d_b)
|
|
d_c = clone(work, "cC", rev)
|
|
rw = rewrite_paths(d_c)
|
|
sabotage_chunker(d_c)
|
|
rc_c, ok_c, fail_c, _ = run_verifier(d_c)
|
|
d_d = clone(work, "cD", rev)
|
|
rewrite_paths(d_d)
|
|
rc_d, ok_d, _, _ = run_verifier(d_d)
|
|
print()
|
|
cell("A clone, subject untouched", rc_a, 0, "%d/9 checks" % ok_a)
|
|
cell("B clone's chunker.py sabotaged (pinned)", rc_b, 0,
|
|
"%d/9 checks -- the sabotage is invisible" % ok_b)
|
|
cell("C sabotage + import un-pinned", rc_c, 1, "%d/9, %s" % (ok_c, fail_c))
|
|
cell("D import un-pinned, chunker restored", rc_d, 0, "%d/9 checks" % ok_d)
|
|
print(" (C rewrote %d subject import line(s))" % rw)
|
|
print()
|
|
|
|
|
|
def part2(work, rev):
|
|
print("PART 2 -- can the `for called in (...)` block fail?")
|
|
d = clone(work, "cE", rev)
|
|
lines = open(os.path.join(d, VER), encoding="utf-8").read().splitlines()
|
|
start = next(i for i, ln in enumerate(lines)
|
|
if 'for called in ("_find_first_diff"' in ln)
|
|
end = next(i for i in range(start, len(lines)) if lines[i].strip() == "pass")
|
|
tree = ast.parse("\n".join(lines))
|
|
node = next(n for n in ast.walk(tree)
|
|
if isinstance(n, ast.For) and getattr(n, "lineno", 0) == start + 1)
|
|
kinds = sorted({type(n).__name__ for st in node.body for n in ast.walk(st)})
|
|
effectful = [k for k in kinds if k in
|
|
{"Raise", "Return", "Assign", "AugAssign", "Exit", "Global"}]
|
|
print(" block L%d..L%d body node types: %s"
|
|
% (start + 1, end + 1, ", ".join(kinds)))
|
|
print(" effectful of those: %s" % (", ".join(effectful) or "NONE"))
|
|
RESULTS.append(not effectful)
|
|
|
|
intact = "\n".join(lines) + "\n"
|
|
noblock = "\n".join(lines[:start] + lines[end + 1:]) + "\n"
|
|
for label, mutate in (("green subject", False),
|
|
("main() no longer calls _find_first_diff", True)):
|
|
got = []
|
|
for variant, text in (("intact", intact), ("loop-deleted", noblock)):
|
|
dd = clone(work, "cF-" + variant, rev)
|
|
open(os.path.join(dd, VER), "w", encoding="utf-8").write(text)
|
|
if mutate:
|
|
p = os.path.join(dd, PROBE_R)
|
|
s = open(p, encoding="utf-8").read()
|
|
open(p, "w", encoding="utf-8").write(
|
|
s.replace(" diff = _find_first_diff(actual, expected)",
|
|
" diff = None", 1))
|
|
got.append(run_verifier(dd))
|
|
same = got[0][0] == got[1][0] and got[0][2] == got[1][2]
|
|
RESULTS.append(same)
|
|
print(" %-38s intact rc=%d | loop-deleted rc=%d | identical=%s"
|
|
% (label, got[0][0], got[1][0], same))
|
|
if got[0][2] or got[1][2]:
|
|
print(" message: %s" % (got[0][2] or got[1][2]))
|
|
print()
|
|
|
|
|
|
def part3(work, rev):
|
|
print("PART 3 -- the omissions and the drift")
|
|
d = clone(work, "cO", rev)
|
|
|
|
def runner():
|
|
return sh([sys.executable, "wizard-work/volt/q65_runner.py", "--check",
|
|
"--class", "gate-run,undeclared"], d)
|
|
|
|
def rc(proc):
|
|
return proc.returncode
|
|
|
|
def named(proc, needle=TARGET):
|
|
return any(needle in ln for ln in (proc.stdout + proc.stderr).splitlines())
|
|
|
|
p = runner()
|
|
cell("gate on an untouched clone (runner --check)", rc(p), 0, "0 fault(s)")
|
|
|
|
full = os.path.join(d, CENSUS)
|
|
keep = [ln for ln in open(full, encoding="utf-8").readlines() if TARGET not in ln]
|
|
dropped = os.path.join(work, "census_dropped.tsv")
|
|
open(dropped, "w", encoding="utf-8").writelines(keep)
|
|
p = sh([sys.executable, "wizard-work/volt/q63_verify_scan.py", "--repo", ".",
|
|
"--baseline", dropped], d)
|
|
cov = [ln.strip() for ln in p.stdout.splitlines() if "census coverage" in ln]
|
|
miss = [ln.strip() for ln in p.stdout.splitlines() if "CENSUS_ROW_MISSING" in ln]
|
|
ok = (rc(p) == 0 and not named(p, "NEW_UNWIRED")
|
|
and bool(cov) and "carry a census row" in cov[0] and bool(miss))
|
|
cell("census row dropped alone -> scan", 0 if ok else 1, 0,
|
|
"CENSUS_ROW_MISSING named" if ok else "NOT NAMED (rc %d)" % rc(p))
|
|
print(" %s" % (cov[0] if cov else "-"))
|
|
print(" %s" % (miss[0] if miss else "-"))
|
|
print(" (not a fault: the runner row satisfies 'not unwired'; the coverage "
|
|
"NUMBER drops and the printer names the file CENSUS_ROW_MISSING, not UNCLASSED)")
|
|
|
|
fresh = clone(work, "cO2", rev)
|
|
d = fresh
|
|
full = os.path.join(d, LEDGER)
|
|
keep = [ln for ln in open(full, encoding="utf-8").readlines() if TARGET not in ln]
|
|
open(full, "w", encoding="utf-8").writelines(keep)
|
|
p = runner()
|
|
cell("ledger row dropped -> runner", rc(p), 1, "UNGRADED, named"
|
|
if named(p, "UNGRADED") else "NOT NAMED")
|
|
RESULTS[-1] = RESULTS[-1] and named(p, "UNGRADED")
|
|
|
|
d = clone(work, "cO3", rev)
|
|
pth = os.path.join(d, VER)
|
|
src = open(pth, encoding="utf-8").read()
|
|
open(pth, "w", encoding="utf-8").write(src.replace(
|
|
'print("OK: py_compile passes")',
|
|
'print("OK: py_compile passes (edited)")', 1))
|
|
subprocess.run(["git", "-c", "user.name=ctl", "-c", "user.email=ctl@x.invalid",
|
|
"commit", "-q", "-am", "control: edit the verifier after its row"],
|
|
cwd=d, capture_output=True, text=True)
|
|
p = runner()
|
|
cell("verifier edited+committed after rowing -> runner", rc(p), 1,
|
|
"DRIFT, named" if named(p, "DRIFT") else "NOT NAMED")
|
|
RESULTS[-1] = RESULTS[-1] and named(p, "DRIFT")
|
|
p = sh([sys.executable, "wizard-work/volt/q65_rc_probe.py", "--repo", ".",
|
|
"--check"], d)
|
|
cell("same edit -> rc-probe ledger", rc(p), 1,
|
|
"STALE, named" if named(p, "STALE") else "NOT NAMED")
|
|
RESULTS[-1] = RESULTS[-1] and named(p, "STALE")
|
|
|
|
d = clone(work, "cO4", rev)
|
|
p = sh(["bash", "wizard-work/volt/CHECK"], d)
|
|
cell("bash wizard-work/volt/CHECK on a clean clone", rc(p), 0,
|
|
p.stdout.strip().splitlines()[-1][:60] if p.stdout.strip() else "-")
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--rev", default="HEAD")
|
|
ap.add_argument("--keep", action="store_true",
|
|
help="keep the clones (default: workspace removed on exit)")
|
|
args = ap.parse_args()
|
|
rev = sh(["git", "rev-parse", "--short", args.rev], REPO).stdout.strip()
|
|
work = tempfile.mkdtemp(prefix="q65-c7-")
|
|
print("q65 c7 controls -- repo %s, rev %s, workspace %s\n" % (REPO, rev, work))
|
|
part1(work, rev)
|
|
part2(work, rev)
|
|
part3(work, rev)
|
|
print()
|
|
print("%d/%d cells as expected" % (sum(RESULTS), len(RESULTS)))
|
|
if not args.keep:
|
|
shutil.rmtree(work)
|
|
return 0 if all(RESULTS) else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|