stackchain/troupe-standup
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
troupe-standup/wizard-work/forge/verifier_gate.py
T
volt 17d0180cef volt: gate the forge lane's 39 verifiers (Q87) 
wizard-work/forge had 39 committed verify_*.py and no CHECK; tools/verify.py
names the lane zero times. verifier_gate.py runs them, keeps one row per file
bound to the git blob the rc was measured against, and grades that ledger in
under a second. First full run at 7b42050: PASS 21 (18 of them PASS* -- they
read an absolute host root), FAIL 18, all named with causes.

Five verifiers in this lane clean up with rmtree(os.path.dirname(tmp)), i.e.
they delete $TMPDIR itself; each file now runs under its own sandbox so the
harness cannot be measured instead of the lane.
2026-10-31 11:12:45 +00:00

803 lines
35 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""verifier_gate.py -- the runner `wizard-work/forge` never had (Q87, 2026-10-31).
WHY THIS EXISTS
At `431bca1` the pipe measured 34 `verify_*.py` in this lane and no `CHECK`;
the correction at `d56105a` moved the count to 36 and the runner count stayed
zero. At `7b42050` the lane holds 39 and `tools/verify.py` still names
`wizard-work/forge` 0 times. A verifier nothing runs is an unread claim: its
verdict is whatever its author last saw by hand, and this lane would not
notice if all 39 were broken at once. This gate executes them, keeps one row
per file bound to the blob the rc was measured against, and re-grades that
ledger on every run without re-executing it.
WHAT IT MEASURES -- one row per committed verifier, one verdict per row
PASS rc 0: the file ran and its checks held.
FAIL rc 1: the file ran and its checks did NOT hold. A finding, not a
bookkeeping error -- the file's claim is false at this tree.
REFUSED rc 2: the file declined to grade. Not a pass.
OTHER_RC rc above 2 that the house contract does not define.
TIMEOUT no finish inside its per-file window (see WINDOWS below).
ERROR the interpreter could not run the file at all (no rc to grade).
MISSING the row's path is not in the tree at --rev.
A non-PASS row must carry a reason in its note column. A red with no recorded
cause is a FAULT (`CAUSELESS`), because "it is red and nobody wrote down why"
is the state this lane has been in since 2026-10-27.
THE LEDGER IS BOUND TO CONTENT
Column 7 is the git blob id each rc was measured against. `--check` recomputes
it at its own `--rev`; a mismatch is DRIFT, so editing a verifier is a request
for `--emit` rather than a free pass on an old green. Column 8 is a static
host-pin probe: the committed blob is parsed as an AST and every non-docstring
string constant is tested for an absolute host root (`/root/troupe-standup`,
`/root/.hermes`). A PASS carrying one prints as `PASS*` and is reported
separately, because a verifier that reads this host's copy graded this host
and not the tree that invoked it (Q62/Q106's class).
WHAT IT DELIBERATELY DOES NOT DO
`--check` (the default, what `CHECK` runs) does NOT execute anything: it
grades the ledger against the tree in well under a second. Execution is
`--emit`, it takes minutes here (measured: 470 s for 39 files at a 75 s
default window, plus three files that need longer), and a gate that takes
minutes is a gate that gets skipped. `--strict` promotes any non-PASS row to
a fatal fault, which is the switch between "are the books consistent?" and
"are the lane's verifiers true?" -- today the honest answer to the second is
no, and the rows say which ones.
It does not decide whether a verifier is *good*, only whether it exits 0. A
verifier that asserts nothing exits 0. That is the same boundary q65_runner
draws for the repo-wide population and it is not moved here.
WINDOWS
A single `--timeout` for the whole lane is wrong: three files here need more
than 75 s and one of them is a recursive sweep. `VERIFIER_WINDOWS.tsv` names
one window per path with the reason; `--timeout` stays the fallback. A
TIMEOUT row's note names the window it actually got, so the ledger cannot
quietly disagree with the run.
EXIT CODES -- the house contract
0 clean / 1 graded a fault / 2 could not grade (no git, no python3, no lane).
Usage:
python3 verifier_gate.py --check # grade the ledger (fast, no runs)
python3 verifier_gate.py --check --strict # ...and any non-PASS row is fatal
python3 verifier_gate.py --emit # execute every verifier, rewrite
python3 verifier_gate.py --census # who else runs each file
python3 verifier_gate.py --selftest # the gate's own cells
"""
import argparse
import ast
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path
LANE = "wizard-work/forge"
HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
LEDGER = HERE / "VERIFIER_LEDGER.tsv"
WINDOWS = HERE / "VERIFIER_WINDOWS.tsv"
PASS, FAIL, REFUSED = "PASS", "FAIL", "REFUSED"
OTHER_RC, TIMEOUT, ERROR, MISSING = "OTHER_RC", "TIMEOUT", "ERROR", "MISSING"
VERDICTS = (PASS, FAIL, REFUSED, OTHER_RC, TIMEOUT, ERROR, MISSING)
HOST_ROOTS = ("/root/troupe-standup", "/root/.hermes")
VERDICT_OF_RC = {0: PASS, 1: FAIL, 2: REFUSED}
def git(repo, *args):
"""(rc, stdout, stderr) -- never raises, so a missing git is rc 2 upstream."""
try:
r = subprocess.run(["git", "-C", str(repo)] + list(args),
capture_output=True, text=True, errors="replace",
timeout=120)
return r.returncode, r.stdout, r.stderr
except (OSError, subprocess.SubprocessError) as exc:
return 127, "", str(exc)
def population(repo, rev):
"""Committed verify_*.py under the lane at `rev` -- from the tree, not a list.
The population is read from git rather than from a directory glob because
the question is "what did this commit ship", and an untracked scratch file
beside the lane is not a claim about the commit.
"""
rc, out, err = git(repo, "ls-tree", "-r", "--name-only", rev, "--", LANE)
if rc != 0:
raise SystemExit("git ls-tree %s failed: %s" % (rev, (err or "").strip()))
paths = [p.strip() for p in out.splitlines() if p.strip()]
return sorted(p for p in paths
if p.startswith(LANE + "/")
and os.path.basename(p).startswith("verify_")
and p.endswith(".py"))
def blob_of(repo, rev, path):
rc, out, _ = git(repo, "rev-parse", "%s:%s" % (rev, path))
return out.strip() if rc == 0 else ""
def host_pin(repo, rev, path):
"""The first absolute host root this blob reads, or ''.
Parsed rather than grepped: a docstring that *names* the host root is a
description, and the Q85 lesson is that a text scan turns description into
finding. Only string constants outside docstrings count.
"""
rc, blob, _ = git(repo, "show", "%s:%s" % (rev, path))
if rc != 0:
return ""
try:
tree = ast.parse(blob)
except SyntaxError:
return "unparseable"
docs = set()
for node in ast.walk(tree):
if isinstance(node, (ast.Module, ast.FunctionDef, ast.AsyncFunctionDef,
ast.ClassDef)):
body = getattr(node, "body", [])
if body and isinstance(body[0], ast.Expr) \
and isinstance(body[0].value, ast.Constant) \
and isinstance(body[0].value.value, str):
docs.add(id(body[0].value))
for node in ast.walk(tree):
if isinstance(node, ast.Constant) and isinstance(node.value, str):
if id(node) in docs:
continue
for root in HOST_ROOTS:
if root in node.value:
hits = [l for l in node.value.splitlines() if root in l]
return (hits[0] if hits else root).strip()[:120]
return ""
def read_windows(path):
"""{path: (seconds, why)} -- one window per file, the reason mandatory."""
win = {}
p = Path(path)
if not p.exists():
return win
for line in p.read_text(encoding="utf-8").splitlines():
if not line.strip() or line.lstrip().startswith("#"):
continue
parts = line.split("\t")
if len(parts) < 3:
raise SystemExit("windows row needs path<TAB>seconds<TAB>why: %r" % line)
try:
win[parts[0].strip()] = (float(parts[1]), parts[2].strip())
except ValueError:
raise SystemExit("window seconds is not a number: %r" % line)
return win
def run_one(repo, rel, timeout, windows):
"""Execute one verifier in its own directory, under its own TMPDIR.
THE SANDBOX IS NOT A LUXURY. Five files in this lane clean up with
`shutil.rmtree(os.path.dirname(<their temp path>))`, which is the PARENT of
the directory they made -- i.e. `$TMPDIR` itself:
verify_append_only_audit_dry_refactor.py:304, verify_append_only_owners_
dry_refactor.py:238, verify_gate_blame_dry_refactor.py:178, verify_served_
drift_summarise_dry.py:192, verify_write_probe_dry_refactor.py:133.
Running them with the caller's TMPDIR deleted the caller's scratch tree (this
item's own first measurement, at 09:27, died that way: the probe's TSV was
gone mid-run) and every later file failed with FileNotFoundError naming a
directory it had just created, because `mkdtemp` raises on the missing
PARENT. Each file now gets `<own base>/tb` as TMPDIR, so a verifier that
deletes its parent takes out a throwaway directory the gate removes itself.
"""
window, why = windows.get(rel, (timeout, ""))
full = Path(repo) / rel
if not full.exists():
return {"path": rel, "verdict": MISSING, "rc": None, "seconds": 0.0,
"note": "not in worktree"}
base = tempfile.mkdtemp(prefix="forge_gate_")
tb = os.path.join(base, "tb")
os.makedirs(tb, exist_ok=True)
env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1", PYTHONUNBUFFERED="1",
TMPDIR=tb, TEMP=tb, TMP=tb)
env.pop("PYTHONOPTIMIZE", None)
t0 = time.time()
try:
r = subprocess.run([sys.executable, "-B", full.name], cwd=str(full.parent),
env=env, capture_output=True, timeout=window,
text=True, errors="replace")
rc, so, se = r.returncode, r.stdout, r.stderr
except subprocess.TimeoutExpired:
return {"path": rel, "verdict": TIMEOUT, "rc": None,
"seconds": round(time.time() - t0, 2),
"note": "no finish in %ss%s"
% (window, (" (%s)" % why) if why else "")}
except OSError as exc:
return {"path": rel, "verdict": ERROR, "rc": None, "seconds": 0.0,
"note": str(exc)[:160]}
finally:
shutil.rmtree(base, ignore_errors=True)
secs = round(time.time() - t0, 2)
verdict = VERDICT_OF_RC.get(rc, OTHER_RC if rc > 2 else ERROR)
note = verdict_line(so, se, rc)
return {"path": rel, "verdict": verdict, "rc": rc, "seconds": secs,
"note": note}
def verdict_line(so, se, rc):
"""The file's own sentence, not whichever stream happened to end last.
The first draft took stderr's last line, and `verify_append_only_audit_dry_
refactor.py` wrote its cause to stdout while stderr ended on a SyntaxWarning
fragment -- the row read `(b"dash \\u2014 here\\nplain\\n", ...)` and lost the
`VERDICT: MISMATCH` it had printed. A verdict-looking line is preferred, from
stdout first; the last line of either stream is the fallback.
"""
keys = ("VERDICT", "FAIL", "checks", "PASS", "failed", "Error")
for text in (so, se):
lines = [l.strip() for l in (text or "").splitlines() if l.strip()]
for line in reversed(lines):
if line.startswith("[ok") or line.startswith("ok:"):
continue # a passing cell is not this row's cause
if any(k in line for k in keys):
return line[:160]
# Nothing verdict-shaped: prefer stderr's last line, because that is where a
# traceback puts the reason, and stdout's last line is often a passing cell
# the file printed before the thing that killed it.
for text in (se, so):
lines = [l.strip() for l in (text or "").splitlines() if l.strip()]
if lines:
return lines[-1][:160]
return "" if rc == 0 else "exited rc %d with no stdout/stderr" % rc
HEADER = """\
# Q87 verifier ledger -- one row per committed verify_*.py in wizard-work/forge,
# with the exit code it actually returned at the stated rev.
# row: <invocation>\t<path>\t<verdict>\t<rc>\t<seconds>\t<note>\t<blob>\t<host>
# verdicts: PASS rc0 / FAIL rc1 / REFUSED rc2 / OTHER_RC >2 / TIMEOUT / ERROR /
# MISSING. A non-PASS row must carry a note: the cause, in the file's words.
# invocation is what runs the file: `gate` when this gate is the only runner,
# `also-run` when another committed line invokes it (the census evidence is
# `--census`'s job, not a column here).
# blob is the git blob id the rc was measured against; --check recomputes it and
# a mismatch is DRIFT, so editing a verifier is a request for --emit.
# host is the absolute host root the blob reads (AST, non-docstring strings), or
# empty. A PASS carrying one prints as PASS* and is counted separately.
# seconds is wall clock under that file's window (VERIFIER_WINDOWS.tsv); a
# TIMEOUT note names the window it got.
"""
def read_ledger(path):
rows, meta = [], {}
p = Path(path)
if not p.exists():
return rows, meta, "no ledger at %s (run --emit)" % p
for line in p.read_text(encoding="utf-8").splitlines():
if not line.strip():
continue
if line.startswith("#"):
m = re.match(r"#\s*(rev|emitted_at):\s*(\S+)", line)
if m:
meta[m.group(1)] = m.group(2)
continue
parts = line.split("\t")
if len(parts) < 6:
return rows, meta, "ledger row has %d column(s): %r" % (len(parts), line)
while len(parts) < 8:
parts.append("")
rows.append({"invocation": parts[0], "path": parts[1], "verdict": parts[2],
"rc": parts[3], "seconds": parts[4], "note": parts[5],
"blob": parts[6], "host": parts[7]})
return rows, meta, ""
def write_ledger(path, rows, repo, rev):
rc, out, _ = git(repo, "rev-parse", "--short", rev)
sha = out.strip() or rev
body = [HEADER.rstrip("\n"),
"# rev: %s" % sha,
"# emitted_at: %sZ" % time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime())]
for r in sorted(rows, key=lambda x: x["path"]):
body.append("\t".join([r["invocation"], r["path"], r["verdict"],
"" if r["rc"] is None else str(r["rc"]),
"%.2f" % float(r["seconds"]), r["note"],
r["blob"], r["host"]]))
Path(path).write_text("\n".join(body) + "\n", encoding="utf-8")
def rc_consistent(row):
"""Wording a verdict has to match. Returns a reason string or ''."""
v, rc = row["verdict"], row["rc"]
if v not in VERDICTS:
return "verdict %r is not one of %s" % (v, "/".join(VERDICTS))
if v in (TIMEOUT, ERROR, MISSING):
return "" if rc == "" else "%s carries rc %s" % (v, rc)
if rc == "" or not re.fullmatch(r"-?\d+", rc):
return "%s carries rc %r" % (v, rc)
n = int(rc)
if v == PASS and n != 0:
return "PASS with rc %d" % n
if v == FAIL and n != 1:
return "FAIL with rc %d" % n
if v == REFUSED and n != 2:
return "REFUSED with rc %d" % n
if v == OTHER_RC and n <= 2:
return "OTHER_RC with rc %d" % n
return ""
# ---------------------------------------------------------------- invocation
DOC_SUFFIXES = (".md", ".txt", ".log", ".tsv", ".json", ".pyc")
RUN_TOKENS = ("python3", "python ", "sys.executable", "subprocess", "os.system",
"pytest", "bash ")
def invocation_census(repo, rev, paths, exclude_rel):
"""{path: (class, evidence)} -- who runs this file, from committed content.
A mention is not an invocation. A reference counts as a RUN only when it sits
in an executable file (.py/.sh/CHECK*, not a document) on a line that also
names an interpreter or a spawner; every other reference is prose. That is
the distinction reply 21 asked for, and it is a rule with a test: --selftest
holds one prose mention and one `python3 x.py` line and requires the
classifier to separate them.
"""
rc, out, _ = git(repo, "ls-tree", "-r", "--name-only", rev)
if rc != 0:
raise SystemExit("git ls-tree failed at %s" % rev)
files = [f.strip() for f in out.splitlines() if f.strip()]
texts = {}
for f in files:
base = os.path.basename(f)
if base.endswith(DOC_SUFFIXES):
continue
rc, blob, _ = git(repo, "show", "%s:%s" % (rev, f))
if rc == 0:
texts[f] = blob
res = {}
for path in paths:
name = os.path.basename(path)
runs = []
named = 0
for f, blob in texts.items():
if f == path or f == exclude_rel:
continue
if name not in blob:
continue
for i, line in enumerate(blob.splitlines(), 1):
if name not in line:
continue
if line.lstrip().startswith(("#", "//", "*", "--")):
named += 1
continue
if any(t in line for t in RUN_TOKENS):
runs.append("%s:%d" % (f, i))
else:
named += 1
if runs:
res[path] = ("also-run", ", ".join(runs[:2]))
elif named:
res[path] = ("gate", "%d prose ref(s), no invocation" % named)
else:
res[path] = ("gate", "named by nothing else at all")
return res
def check(repo, rev, ledger_path, strict, quiet=False):
try:
pop = population(repo, rev)
except SystemExit as exc:
print("verifier_gate: %s" % exc)
return 2
if not pop:
print("verifier_gate: population is empty at %s -- could not grade" % rev)
return 2
rows, meta, err = read_ledger(ledger_path)
if err:
print("verifier_gate: %s" % err)
return 2
faults, notes = [], []
seen = {}
for r in rows:
seen[r["path"]] = seen.get(r["path"], 0) + 1
why = rc_consistent(r)
if why:
faults.append("INCONSISTENT %s: %s" % (r["path"], why))
if r["verdict"] != PASS and not r["note"].strip():
faults.append("CAUSELESS %s: %s row with no reason in the note"
% (r["path"], r["verdict"]))
if not r["blob"]:
faults.append("UNBOUND %s: no blob column -- the row does not say "
"which content its rc describes (re-emit)" % r["path"])
for p, n in sorted(seen.items()):
if n > 1:
faults.append("DUPLICATE %s: %d rows" % (p, n))
for r in rows:
if r["path"] not in pop:
continue
now = blob_of(repo, rev, r["path"])
if r["blob"] and now and now != r["blob"]:
faults.append("DRIFT %s: rc measured against blob %s, tree at %s has %s"
% (r["path"], r["blob"][:10], rev, now[:10]))
# A dirty worktree is not a commit, so it cannot be DRIFT -- but a verifier
# whose working bytes are not the blob its rc describes is exactly the case
# where the ledger's verdict is about content you are not running. Named,
# not fatal: the fix is a commit (then DRIFT/UNGRADED decide) or --emit.
for r in rows:
if r["path"] not in pop or not r["blob"]:
continue
rc_w, out_w, _ = git(repo, "hash-object", "--", r["path"])
if rc_w == 0 and out_w.strip() and out_w.strip() != r["blob"]:
notes.append("UNCOMMITTED %s: worktree bytes are not the blob its rc "
"was measured against -- commit it, then re-emit or "
"expect DRIFT" % r["path"])
ungraded = sorted(set(pop) - set(seen))
for p in ungraded:
faults.append("UNGRADED %s: committed at %s, no ledger row" % (p, rev))
stale = sorted(p for p in set(seen) - set(pop)
if not (Path(repo) / p).exists())
for p in stale:
faults.append("STALE %s: ledger row, file not in the tree" % p)
outside = sorted(p for p in set(seen) - set(pop) if (Path(repo) / p).exists())
for p in outside:
notes.append("OUTSIDE %s: ledger row, not a committed verifier at %s "
"(untracked beside the lane or renamed)" % (p, rev))
counts = {}
for r in rows:
counts[r["verdict"]] = counts.get(r["verdict"], 0) + 1
pinned = [r for r in rows if r["verdict"] == PASS and r["host"]]
reds = [r for r in rows if r["verdict"] != PASS]
clean = len(rows) - len(reds) - len(pinned)
print("verifier_gate -- lane %s at %s, %d committed verifier(s), %d ledger row(s)"
% (LANE, rev, len(pop), len(rows)))
if meta.get("rev"):
print(" ledger rev : %s emitted %s"
% (meta["rev"], meta.get("emitted_at", "?")))
print(" rows : %s" % ", ".join("%s %d" % (v, counts[v])
for v in VERDICTS if counts.get(v)))
print(" graded-clean : %d of %d (PASS on a host-independent blob)"
% (clean, len(pop)))
if pinned:
print(" host-pinned : %d PASS row(s) read as PASS* -- they read an "
"absolute host root" % len(pinned))
for r in pinned:
print(" PASS* %s %s" % (r["path"], r["host"]))
if reds:
print(" known-red : %d of %d row(s) -- every one named, every one "
"with its cause" % (len(reds), len(rows)))
for r in sorted(reds, key=lambda x: x["path"]):
print(" %-8s rc %-3s %s %s" % (r["verdict"], r["rc"] or "-",
r["path"], r["note"][:100]))
for n in notes:
print(" note: " + n)
if strict:
for r in reds:
faults.append("%s %s rc %s %s" % (r["verdict"], r["path"], r["rc"] or "-",
r["note"][:80]))
for f in faults:
print(" " + f)
print("verifier_gate: %d fault(s)%s" % (len(faults),
" (strict: every non-PASS row counts)"
if strict else ""))
return 1 if faults else 0
def emit(repo, rev, ledger_path, timeout, windows, only=None, merge=False):
pop = population(repo, rev)
if only:
pop = [p for p in pop if only in p]
if not pop:
print("verifier_gate: nothing to emit")
return 2
keep = []
if merge and Path(ledger_path).exists():
have, _, err = read_ledger(ledger_path)
if err:
print("verifier_gate: %s" % err)
return 2
keep = [r for r in have if r["path"] not in set(pop)]
rc, before, _ = git(repo, "status", "--porcelain")
t0 = time.time()
rows = list(keep)
for p in pop:
r = run_one(repo, p, timeout, windows)
r["blob"] = blob_of(repo, rev, p)
r["host"] = host_pin(repo, rev, p)
r["invocation"] = "gate"
rows.append(r)
print(" %-8s %6.2fs %s" % (r["verdict"], r["seconds"], p), flush=True)
census = {}
try:
census = invocation_census(repo, rev, [r["path"] for r in rows],
os.path.relpath(ledger_path, repo))
except SystemExit as exc:
print("verifier_gate: census skipped (%s)" % exc)
for r in rows:
cls, ev = census.get(r["path"], ("gate", ""))
r["invocation"] = "also-run" if cls == "also-run" else "gate"
write_ledger(ledger_path, rows, repo, rev)
rc2, after, _ = git(repo, "status", "--porcelain")
led_rel = os.path.relpath(str(ledger_path), str(repo))
dirt = sorted(d for d in set(after.splitlines()) - set(before.splitlines())
if led_rel not in d)
counts = {}
for r in rows:
counts[r["verdict"]] = counts.get(r["verdict"], 0) + 1
print("verifier_gate: %d row(s) in %.1fs -> %s"
% (len(rows), time.time() - t0, ledger_path))
print(" " + ", ".join("%s %d" % (v, counts[v]) for v in VERDICTS
if counts.get(v)))
if dirt:
print(" WARNING: the run changed the worktree (these verifiers write "
"into it):")
for d in dirt:
print(" " + d)
return 0
def census_report(repo, rev, ledger_path):
pop = population(repo, rev)
res = invocation_census(repo, rev, pop, os.path.relpath(str(ledger_path), str(repo)))
run = sorted(p for p, (c, _) in res.items() if c == "also-run")
print("verifier_census -- %d committed verifier(s) in %s at %s"
% (len(pop), LANE, rev))
print(" invoked by another committed line : %d" % len(run))
for p in run:
print(" %s %s" % (p, res[p][1]))
print(" invoked by this gate only : %d" % (len(pop) - len(run)))
for p in pop:
if p in run:
continue
print(" %s %s" % (p, res[p][1]))
return 0
# ------------------------------------------------------------------- selftest
def selftest():
"""Cells for the rules that can silently pass. Every cell builds a throwaway
git repo, so the gate's logic is graded on a tree it did not ship with."""
cells = []
def cell(name, fn):
try:
ok, detail = fn()
except Exception as exc: # noqa: BLE001 - a cell that raises is a cell that failed
ok, detail = False, "%s: %s" % (type(exc).__name__, exc)
cells.append((name, ok, detail))
def mk(tmp, files, commit=True):
repo = Path(tmp)
(repo / LANE).mkdir(parents=True, exist_ok=True)
for rel, body in files.items():
p = repo / rel
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(body, encoding="utf-8")
git(repo, "init", "-q")
git(repo, "config", "user.email", "t@example.invalid")
git(repo, "config", "user.name", "t")
if commit:
git(repo, "add", "-A")
git(repo, "commit", "-qm", "cell")
return repo
V = LANE + "/verify_alpha.py"
def c1():
bad = {"path": V, "verdict": PASS, "rc": "1", "note": "", "blob": "x",
"host": "", "seconds": "0", "invocation": "gate"}
good = dict(bad, verdict=FAIL)
a, b = rc_consistent(bad), rc_consistent(good)
return (bool(a) and not b, "PASS with rc 1")
def c2():
with tempfile.TemporaryDirectory() as tmp:
repo = mk(tmp, {V: "print(0)\n"})
n1 = len(population(repo, "HEAD"))
(repo / LANE / "verify_beta.py").write_text("print(0)\n")
git(repo, "add", "-A")
git(repo, "commit", "-qm", "two")
n2 = len(population(repo, "HEAD"))
return (n1 == 1 and n2 == 2, "population follows the rev")
def c3():
with tempfile.TemporaryDirectory() as tmp:
repo = mk(tmp, {V: "print(0)\n"})
led = repo / LANE / "L.tsv"
write_ledger(led, [{"path": V, "verdict": PASS, "rc": 0, "seconds": 0.0,
"note": "", "blob": blob_of(repo, "HEAD", V),
"host": "", "invocation": "gate"}], repo, "HEAD")
r0 = check(repo, "HEAD", led, False)
(repo / V).write_text("print(1)\n")
git(repo, "add", "-A")
git(repo, "commit", "-qm", "edit")
r1 = check(repo, "HEAD", led, False)
return (r0 == 0 and r1 == 1, "a drifted blob is a fault")
def c4():
with tempfile.TemporaryDirectory() as tmp:
repo = mk(tmp, {V: "print(0)\n"})
led = repo / LANE / "L.tsv"
write_ledger(led, [], repo, "HEAD")
return (check(repo, "HEAD", led, False) == 1, "a committed verifier with no row faults")
def c5():
with tempfile.TemporaryDirectory() as tmp:
repo = mk(tmp, {V: "print(0)\n"})
led = repo / LANE / "L.tsv"
write_ledger(led, [{"path": LANE + "/verify_gone.py", "verdict": PASS,
"rc": 0, "seconds": 0.0, "note": "", "blob": "abc",
"host": "", "invocation": "gate"}], repo, "HEAD")
return (check(repo, "HEAD", led, False) == 1, "a row for a deleted file faults")
def c6():
with tempfile.TemporaryDirectory() as tmp:
body = ('"""Reads /root/troupe-standup in a docstring."""\n'
'P = "/root/troupe-standup/inbox"\n')
repo = mk(tmp, {V: body})
hit = host_pin(repo, "HEAD", V)
repo2 = mk(tmp + "/b", {V: '"""Reads /root/troupe-standup."""\n'})
miss = host_pin(repo2, "HEAD", V)
return (hit != "" and miss == "", "host pin is AST, not prose")
def c7():
with tempfile.TemporaryDirectory() as tmp:
repo = mk(tmp, {V: "print(0)\n",
"CHECK": "python3 %s\n" % V,
"notes/x.md": "see %s\n" % V})
res = invocation_census(repo, "HEAD", [V], "")
return (res[V][0] == "also-run", "prose is not an invocation")
def c8():
with tempfile.TemporaryDirectory() as tmp:
repo = mk(tmp, {V: "import sys\nsys.exit(1)\n",
LANE + "/verify_ok.py": "print(0)\n"})
led = repo / LANE / "L.tsv"
rc = emit(repo, "HEAD", led, 30.0, {}, only=None)
rows, _, err = read_ledger(led)
by = {r["path"]: r for r in rows}
ok = (rc == 0 and not err and by[V]["verdict"] == FAIL
and by[V]["note"] != "" and by[LANE + "/verify_ok.py"]["verdict"] == PASS)
return (ok, "emit records rc and a cause")
def c9():
with tempfile.TemporaryDirectory() as tmp:
repo = mk(tmp, {V: "print(0)\n"})
led = repo / LANE / "L.tsv"
write_ledger(led, [{"path": V, "verdict": FAIL, "rc": 1, "seconds": 0.0,
"note": "", "blob": blob_of(repo, "HEAD", V),
"host": "", "invocation": "gate"}], repo, "HEAD")
return (check(repo, "HEAD", led, False) == 1, "a red with no cause is a fault")
def c10():
with tempfile.TemporaryDirectory() as tmp:
repo = mk(tmp, {V: "print(0)\n"})
led = repo / LANE / "L.tsv"
blob = blob_of(repo, "HEAD", V)
write_ledger(led, [{"path": V, "verdict": FAIL, "rc": 1, "seconds": 0.0,
"note": "known red", "blob": blob, "host": "",
"invocation": "gate"}], repo, "HEAD")
loose = check(repo, "HEAD", led, False)
strict = check(repo, "HEAD", led, True)
return (loose == 0 and strict == 1, "strict promotes a red row")
def c11():
import contextlib
import io
with tempfile.TemporaryDirectory() as tmp:
repo = mk(tmp, {V: "print(0)\n"})
led = repo / LANE / "L.tsv"
write_ledger(led, [{"path": V, "verdict": PASS, "rc": 0, "seconds": 0.0,
"note": "", "blob": blob_of(repo, "HEAD", V),
"host": "", "invocation": "gate"}], repo, "HEAD")
(repo / V).write_text("print(0) # uncommitted edit\n")
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
rc = check(repo, "HEAD", led, False)
out = buf.getvalue()
return (rc == 0 and "UNCOMMITTED" in out,
"a dirty worktree is named, not fatal")
def c12():
body = ("import sys\n"
"sys.stderr.write('warning: bad escape \\\\u\\n')\n"
"print('VERDICT: MISMATCH -- behavior NOT preserved')\n"
"sys.exit(1)\n")
with tempfile.TemporaryDirectory() as tmp:
repo = mk(tmp, {V: body})
led = repo / LANE / "L.tsv"
emit(repo, "HEAD", led, 30.0, {}, only=None)
rows, _, err = read_ledger(led)
by = {r["path"]: r for r in rows}
note = by[V]["note"] if V in by else ""
return (not err and "VERDICT: MISMATCH" in note,
"the note is the file's verdict line: %r" % note[:60])
def c13():
body = ("import os, shutil, tempfile\n"
"sandbox = tempfile.mkdtemp(prefix='x_', dir=os.environ['TMPDIR'])\n"
"shutil.rmtree(os.path.dirname(sandbox), ignore_errors=True)\n"
"print('cleaned up my parent')\n")
with tempfile.TemporaryDirectory() as tmp:
repo = mk(tmp, {V: body})
led = repo / LANE / "L.tsv"
survive = tempfile.mkdtemp(prefix="survive_")
emit(repo, "HEAD", led, 30.0, {}, only=None)
rows, _, err = read_ledger(led)
by = {r["path"]: r for r in rows}
alive = os.path.isdir(survive)
shutil.rmtree(survive, ignore_errors=True)
return (not err and by[V]["verdict"] == PASS and alive,
"a verifier that deletes its TMPDIR leaves the caller's alone")
# the cell fixture needs a caller-level temp dir to watch
for name, fn in (("c1 verdict/rc pairing", c1),
("c2 population from the rev", c2),
("c3 blob drift", c3),
("c4 ungraded file", c4),
("c5 stale row", c5),
("c6 host-pin AST", c6),
("c7 named-vs-run", c7),
("c8 emit records", c8),
("c9 causeless red", c9),
("c10 strict", c10),
("c11 uncommitted worktree", c11),
("c12 note is the verdict line", c12),
("c13 sandboxed TMPDIR", c13)):
cell(name, fn)
bad = [c for c in cells if not c[1]]
for name, ok, detail in cells:
print(" [%s] %s -- %s" % ("ok " if ok else "FAIL", name, detail))
print("verifier_gate selftest: %d cells, %d passed, %d failed"
% (len(cells), len(cells) - len(bad), len(bad)))
return 0 if not bad else 1
def main():
ap = argparse.ArgumentParser(add_help=True)
ap.add_argument("--repo", default=str(ROOT))
ap.add_argument("--rev", default="HEAD")
ap.add_argument("--ledger", default=str(LEDGER))
ap.add_argument("--windows", default=str(WINDOWS))
ap.add_argument("--timeout", type=float, default=90.0)
ap.add_argument("--check", action="store_true")
ap.add_argument("--emit", action="store_true")
ap.add_argument("--census", action="store_true")
ap.add_argument("--strict", action="store_true")
ap.add_argument("--only")
ap.add_argument("--merge", action="store_true",
help="with --emit: keep the rows this run did not re-measure")
ap.add_argument("--selftest", action="store_true")
args = ap.parse_args()
if args.selftest:
return selftest()
repo = Path(args.repo).resolve()
if not (repo / ".git").exists():
print("verifier_gate: %s is not a git worktree -- could not grade" % repo)
return 2
ledger = Path(args.ledger)
if not ledger.is_absolute():
ledger = repo / ledger
if args.emit:
try:
win = read_windows(args.windows)
except SystemExit as exc:
print("verifier_gate: %s" % exc)
return 2
return emit(repo, args.rev, ledger, args.timeout, win, only=args.only,
merge=args.merge)
if args.census:
return census_report(repo, args.rev, ledger)
return check(repo, args.rev, ledger, args.strict)
if __name__ == "__main__":
sys.exit(main())
Reference in new issue View Git Blame Copy Permalink