stackchain/troupe-standup
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

volt: q136 -- the spark word-frequency cell grades the copy beside it (gate's import-resolve leg green)

Browse Source
This commit is contained in:
volt committed 2026-10-31 11:54:15 +00:00
1 parent 4d329fa827
commit 3612c56706
3 files changed
+422 -6
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

wizard-work/spark/test_wordfreq_counter.py
+29 -6
View File
@@ -1,35 +1,58 @@
#!/usr/bin/env python3
"""Verification test for wordfreq_counter.py"""
import json
"""Verification test for wordfreq_counter.py
Both the module under test and its input fixture are resolved from this
file's own directory. The test therefore grades the copy that sits beside
it and runs on a host that has never seen an artifact directory, and it
prints the resolved module path so a shadowed import is visible instead of
silent.
"""
import sys
from pathlib import Path
# Import the module
sys.path.insert(0, '/root/.hermes/wizard-outputs/spark')
HERE = Path(__file__).resolve().parent
FIXTURE = HERE / "test_input.txt"
# Import the module that sits beside this file.
sys.path.insert(0, str(HERE))
import wordfreq_counter as _mod
from wordfreq_counter import count_words, process_file
_IMPORTED = Path(_mod.__file__).resolve()
assert _IMPORTED.parent == HERE, (
"graded %r, not the copy beside this test (%r)" % (str(_IMPORTED), str(HERE))
)
assert FIXTURE.is_file(), "fixture missing: %r" % (str(FIXTURE),)
def test_count_words():
"""Test basic word counting."""
result = count_words("hello hello world")
assert result == {"hello": 2, "world": 1}, f"Expected hello:2, world:1, got {result}"
print("test_count_words: PASS")
def test_process_file():
"""Test file processing."""
result = process_file('/root/.hermes/wizard-outputs/spark/test_input.txt')
result = process_file(str(FIXTURE))
assert result['total_words'] == 24, f"Expected 24 words, got {result['total_words']}"
assert result['unique_words'] == 18, f"Expected 18 unique words, got {result['unique_words']}"
assert result['word_counts']['the'] == 4, f"Expected 'the': 4, got {result['word_counts'].get('the')}"
assert result['file'] == str(FIXTURE), f"Reported file {result['file']!r}, not the fixture"
print("test_process_file: PASS")
def test_output_format():
"""Test JSON output format."""
result = process_file('/root/.hermes/wizard-outputs/spark/test_input.txt')
result = process_file(str(FIXTURE))
required_keys = {'file', 'total_words', 'unique_words', 'word_counts'}
assert required_keys.issubset(result.keys()), f"Missing keys: {required_keys - result.keys()}"
print("test_output_format: PASS")
if __name__ == "__main__":
print("module: %s" % str(_IMPORTED))
print("fixture: %s" % str(FIXTURE))
test_count_words()
test_process_file()
test_output_format()
wizard-work/volt/CHECK-q136
Executable
+22
View File
@@ -0,0 +1,22 @@
#!/bin/sh
# Q136 CHECK -- the spark lane's word-frequency cell must grade the module and
# fixture beside it, and the repo gate's import-resolve leg must be green.
#
# rc 0 = every leg as expected; 1 = a graded fault; 2 = could not grade.
# The work is done by wizard-work/volt/q136_verify.py beside this file: the
# repo gate's leg, the cell run in place and copied out of the tree under an
# audit hook, the fixture's constants recomputed by an independent tokenizer,
# and two mutants -- a sabotaged sibling module the pre-repair binding reads
# green on, and the pre-repair bytes restored in a throwaway repository so the
# gate leg is shown to red. The graded tree is only ever read.
set -u
REPO="$(cd "$(dirname "$0")/../.." && pwd)" || exit 2
if ! cd "$REPO"; then
echo "q136: could not grade -- cannot cd to $REPO"
exit 2
fi
if ! command -v python3 >/dev/null 2>&1; then
echo "q136: could not grade -- no python3 on PATH"
exit 2
fi
exec python3 wizard-work/volt/q136_verify.py "$@"
wizard-work/volt/q136_verify.py
+371
View File
@@ -0,0 +1,371 @@
#!/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, 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 "):])
need = {os.path.join(box, "wordfreq_counter.py"), os.path.join(box, "test_input.txt")}
missing = sorted(need - set(opened))
outside = [p for p in opened
if p.startswith(str(repo) + os.sep) or p.startswith(HOST_DIR)]
ok = not missing and not outside
detail = "%d path(s) opened, %d outside the sandbox" % (len(opened), len(outside))
if missing:
detail += ", never opened: %s" % ", ".join(os.path.basename(p) for p in missing)
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())