180 lines
7.4 KiB
Python
180 lines
7.4 KiB
Python
"""q63_prove.py -- Q63 proof harness: for each of the six stripped-assert
|
|
verifiers, run
|
|
|
|
(old committed blob, normal) (old committed blob, -O)
|
|
(new worktree file, normal) (new worktree file, -O)
|
|
|
|
against a DELIBERATELY BROKEN input, and print every rc. The class-1 defect is
|
|
old/-O returning 0 (PASS) on a broken subject; the fix is rc != 0 there.
|
|
Measured 2026-10-24: RED at `89cda97` (six old/-O runs rc 0), GREEN at `c033059`
|
|
(six new/-O runs rc 1), rc 0 from this harness only when all six are fixed.
|
|
|
|
Isolation: each case runs in its own scratch dir. Where a verifier hardcodes an
|
|
absolute input path, the demonstration copy differs from the committed file in
|
|
that one literal only (input path / subject cwd).
|
|
|
|
Usage: python3 wizard-work/volt/q63_prove.py [repo] [old-rev]
|
|
defaults: repo /root/troupe-standup, old-rev c033059's parent 89cda97
|
|
"""
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
|
|
REPO = sys.argv[1] if len(sys.argv) > 1 else "/root/troupe-standup"
|
|
OLDREV = sys.argv[2] if len(sys.argv) > 2 else "89cda97"
|
|
WORK = os.path.join(os.environ.get("TMPDIR", "/tmp"), "q63_proof")
|
|
PY = sys.executable
|
|
|
|
|
|
def sh(cmd, cwd=None):
|
|
p = subprocess.run(cmd, capture_output=True, text=True, cwd=cwd)
|
|
return p.returncode, (p.stdout + p.stderr).strip()
|
|
|
|
|
|
def case(name, old_src, new_src, fixture, runs):
|
|
"""fixture(dir) builds the broken input; runs is a list of argv lists."""
|
|
d = os.path.join(WORK, name)
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
os.makedirs(d)
|
|
fixture(d)
|
|
out = []
|
|
for label, src in (("old", old_src), ("new", new_src)):
|
|
path = os.path.join(d, f"v_{label}.py")
|
|
with open(path, "w") as f:
|
|
f.write(src)
|
|
for mode, flag in (("normal", []), ("-O", ["-O"])):
|
|
for argv in runs:
|
|
rc, txt = sh([PY, *flag, path, *argv], cwd=d)
|
|
last = txt.splitlines()[-1] if txt.splitlines() else ""
|
|
out.append((label, mode, " ".join(argv), rc, last[:90]))
|
|
return out
|
|
|
|
|
|
def broken_json(d, fname, obj):
|
|
with open(os.path.join(d, fname), "w") as f:
|
|
json.dump(obj, f)
|
|
|
|
|
|
results = {}
|
|
|
|
# 1. beam -- subject module shadowed by a topo_sort that mangles None deps
|
|
old = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/beam/verify_toposort_none.py"],
|
|
capture_output=True, text=True).stdout
|
|
new = open(f"{REPO}/wizard-work/beam/verify_toposort_none.py").read()
|
|
|
|
|
|
def fx_beam(d):
|
|
topo = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/beam/toposort.py"],
|
|
capture_output=True, text=True).stdout
|
|
with open(os.path.join(d, "toposort.py"), "w") as f:
|
|
f.write(topo + "\n\n# --- Q63 broken-subject shim: mangle None-dependency results ---\n"
|
|
"_real = topo_sort\n\n\nclass _R:\n sorted = None\n cycle = None\n\n\n"
|
|
"def topo_sort(deps):\n return _R()\n")
|
|
|
|
|
|
results["beam_toposort_none"] = case("beam", old, new, fx_beam, [[]])
|
|
|
|
# 2. ledger -- CSV whose header is wrong and which has 3 rows
|
|
old = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/ledger/verify_arxiv_csv.py"],
|
|
capture_output=True, text=True).stdout
|
|
new = open(f"{REPO}/wizard-work/ledger/verify_arxiv_csv.py").read()
|
|
|
|
|
|
def fx_csv(d):
|
|
# correct header, 3 well-formed rows: only the row-count check can object,
|
|
# so the -O run reaches the success print instead of crashing on a KeyError.
|
|
rows = ["arxiv_id,title,authors,published,abstract"]
|
|
for i in range(3):
|
|
rows.append(f"2610.{10000+i},t{i},a{i},2026-10-19,abstract{i}")
|
|
with open(os.path.join(d, "b.csv"), "w") as f:
|
|
f.write("\n".join(rows) + "\n")
|
|
|
|
|
|
results["ledger_verify_arxiv_csv"] = case("ledger", old, new, fx_csv, [["b.csv"]])
|
|
|
|
# 3. spark/verify_report.py -- summary counts wrong
|
|
old = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/spark/verify_report.py"],
|
|
capture_output=True, text=True).stdout
|
|
new = open(f"{REPO}/wizard-work/spark/verify_report.py").read()
|
|
OLD_PATH = "/root/.hermes/wizard-outputs/spark/validation_report.json"
|
|
|
|
|
|
def fx_report(d):
|
|
broken_json(d, "r.json", {"summary": {"valid_count": 5, "invalid_count": 0}, "test_passed": False})
|
|
|
|
|
|
def repoint(src, path, fname):
|
|
assert path in src, path
|
|
return src.replace(path, fname)
|
|
|
|
|
|
results["spark_verify_report"] = case(
|
|
"report", repoint(old, OLD_PATH, "r.json"), repoint(new, OLD_PATH, "r.json"), fx_report, [[]])
|
|
|
|
# 4. spark/verify_result.py -- wrong title
|
|
old = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/spark/verify_result.py"],
|
|
capture_output=True, text=True).stdout
|
|
new = open(f"{REPO}/wizard-work/spark/verify_result.py").read()
|
|
OLD_RES = "/root/.hermes/wizard-outputs/spark/scrape_result.json"
|
|
|
|
|
|
def fx_result(d):
|
|
broken_json(d, "s.json", {"title": "Wrong Domain", "status": "success"})
|
|
|
|
|
|
results["spark_verify_result"] = case(
|
|
"result", repoint(old, OLD_RES, "s.json"), repoint(new, OLD_RES, "s.json"), fx_result, [[]])
|
|
|
|
# 5. spark/verify_scratch_analysis.py -- total_files 7 against 7 listed files
|
|
old = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/spark/verify_scratch_analysis.py"],
|
|
capture_output=True, text=True).stdout
|
|
new = open(f"{REPO}/wizard-work/spark/verify_scratch_analysis.py").read()
|
|
OLD_SCR = "/root/.hermes/wizard-outputs/spark/scratch_analysis.json"
|
|
|
|
|
|
def fx_scratch(d):
|
|
files = [{"name": f"f{i}.py", "path": f"/x/f{i}.py", "bytes": 1, "extension": ".py"} for i in range(7)]
|
|
broken_json(d, "a.json", {"directory": "/x", "total_files": 7, "total_bytes": 7,
|
|
"file_types": {".py": 7}, "files": files})
|
|
|
|
|
|
results["spark_verify_scratch_analysis"] = case(
|
|
"scratch", repoint(old, OLD_SCR, "a.json"), repoint(new, OLD_SCR, "a.json"), fx_scratch, [[]])
|
|
|
|
# 6. spark/verify_dir_analyzer.py -- subject prints JSON with 0 files
|
|
old = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/spark/verify_dir_analyzer.py"],
|
|
capture_output=True, text=True).stdout
|
|
new = open(f"{REPO}/wizard-work/spark/verify_dir_analyzer.py").read()
|
|
SPARK_CWD = "cwd='/root/troupe-standup/wizard-work/spark'"
|
|
|
|
|
|
def fx_dir(d):
|
|
with open(os.path.join(d, "dir_analyzer.py"), "w") as f:
|
|
f.write('import json, sys\nif "--json" in sys.argv:\n'
|
|
' print(json.dumps({"total_files": 0, "total_size_bytes": 0}))\n'
|
|
'else:\n print("nothing here")\n')
|
|
|
|
|
|
results["spark_verify_dir_analyzer"] = case(
|
|
"dir", old.replace(SPARK_CWD, "cwd=__import__('os').path.dirname(__file__)"),
|
|
new.replace(SPARK_CWD, "cwd=__import__('os').path.dirname(__file__)"), fx_dir, [[]])
|
|
|
|
print(f"{'case':32s} {'who':4s} {'mode':7s} {'argv':10s} rc last line")
|
|
bad = 0
|
|
for name, rows in results.items():
|
|
for label, mode, argv, rc, last in rows:
|
|
print(f"{name:32s} {label:4s} {mode:7s} {argv:10s} {rc:<3d} {last}")
|
|
old_o = [r[3] for r in rows if r[0] == "old" and r[1] == "-O"][0]
|
|
new_o = [r[3] for r in rows if r[0] == "new" and r[1] == "-O"][0]
|
|
new_n = [r[3] for r in rows if r[0] == "new" and r[1] == "normal"][0]
|
|
if old_o == 0 and new_o != 0:
|
|
verdict = "FIXED (old/-O falsely PASSed, new/-O fails)"
|
|
else:
|
|
verdict = f"UNEXPECTED (old/-O rc {old_o}, new/-O rc {new_o}, new/normal rc {new_n})"
|
|
bad += 1
|
|
print(f" -> {name}: {verdict}")
|
|
print(f"\ncases where the defect is not demonstrated+fixed: {bad}")
|
|
sys.exit(1 if bad else 0)
|