stackchain/troupe-standup
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b9751c87fa3846b0a93390c0762bc0ff1484c8ce
troupe-standup/tools/verify.py
T
volt b9751c87fa tools (volt): verify.py -- a FAULT row may not carry a '0 fatal' line 
Found while demonstrating the NO_PROBE fix: the gate graded selftest-sweep FAULT and
printed 'sweep: 15 tools, 14 OK, 0 fatal' beside it, because pick_detail took the
first hint-matching line and 'fatal' matched a summary asserting zero of them.

- pick_detail: zero-count summaries (' 0 fatal' / ' 0 fault') are second-pass only,
  so a clean run keeps its sweep line and a red run names the red axis
- selftest 24 -> 26 cells: c13 a zero-fault summary is not a reason line,
  c13b a clean run still gets its sweep line; both advertised counts updated
2026-10-18 21:57:21 +00:00

449 lines
20 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""verify.py -- the HOME the index gate did not have: one command, three legs,
one exit code, and a row per leg so a partial run cannot look like a clean one.
WHY THIS EXISTS
On 2026-10-18 this lane told Alexander, three ticks running, to "wire
`queue_index --check` into the tick's check.sh". There is no `check.sh` in
this repo and no `scripts/` directory at all -- `git ls-files` shows `tools/`
and nothing else. The advice named a target that does not exist, which is why
it was never adopted. The gate did not need a decision about policy; it needed
a home. This is that home, and it is a task, not a policy call: nothing here
obliges any lane to run it, and nothing is enforced until a lane does.
The second reason is narrower and this lane's own: three separate counts of
"which commits left the index stale" came out wrong in one week, each from a
hand-assembled probe loop. A gate whose legs are remembered is a gate that
gets run selectively. Three legs, fixed order, one rc -- that is the whole
design.
WHAT IT RUNS -- one row per leg, cheapest first, never in parallel
index-current tools/queue_index.py --check docs/queue-index.md
the derived document, byte-compared against a fresh render
append-audit tools/queue_append_audit.py --guard --to HEAD
the newest queue-touching commit, graded in ITS OWN tree
selftest-sweep tools/selftest_sweep.py
every tool's --selftest plus its two read-refusal axes
(skip with --fast; the row is still printed and counted)
The legs are ordered by cost, and a leg is never skipped because an earlier
one failed: a stale index and a broken selftest are different defects and the
run should name both in one pass.
THE EXIT CONTRACT -- the house three codes
0 every graded leg passed
1 at least one leg ran and graded a fault
2 at least one leg could not grade -- a missing tool, a timeout, an
unexpected exit code, an unreadable repo -- or no leg was runnable at all
rc 2 OUTRANKS rc 1 deliberately. rc 1 means "I ran and I graded a fault";
a run with an ungradable leg has not made that claim about the whole gate,
and folding the two (as the 15:16Z correction caught `queue_append_audit`
doing with ungradable commits) is how a "3 stale" reading hides a "2 could
not be read". The summary always prints both counts, so precedence never
erases a fault -- it only refuses to certify the run that found it.
A skipped leg is not a pass and not a fault: it is counted as `skipped` and
printed as SKIP. `--fast` on a green tree exits 0 with one row SKIP, and the
summary says so rather than pretending the sweep ran.
TOOLS ARE RESOLVED FROM --repo, NOT FROM THIS FILE'S DIRECTORY
`verify.py --repo /path/to/worktree` runs THAT tree's `queue_index.py`,
`queue_append_audit.py` and `selftest_sweep.py`. That is the instrument
lesson the 15:16Z correction paid for: a gate run in a working tree at HEAD
says nothing about a cited revision, so grading a revision means grading its
own tools from its own checkout.
--queue and --index stay REPO-RELATIVE (defaults WORK-QUEUE.md and
docs/queue-index.md) because `queue_append_audit` reads them with
`git show <rev>:<path>`, which has no absolute-path form.
`--selftest` fixture corpus, 26 cells
"""
import argparse
import json
import os
import subprocess
import sys
import tempfile
import time
QUEUE_DEFAULT = "WORK-QUEUE.md"
INDEX_DEFAULT = "docs/queue-index.md"
PASS, FAULT, UNGRADABLE, SKIP = "PASS", "FAULT", "UNGRADABLE", "SKIP"
# Legs, in the fixed order they are printed and run. `tool` is resolved inside
# the repo being graded. `slow` marks the leg --fast drops.
LEGS = (
{"key": "index-current", "tool": "queue_index.py", "slow": False,
"what": "render the index and byte-compare it with the document"},
{"key": "append-audit", "tool": "queue_append_audit.py", "slow": False,
"what": "grade the newest queue-touching commit in its own tree"},
{"key": "selftest-sweep", "tool": "selftest_sweep.py", "slow": True,
"what": "every tool's --selftest and both read-refusal axes"},
)
# A detail line is a courtesy, not the verdict -- but a verdict with no line
# beside it is what the read-refusal axis calls SILENT, so it is graded too.
HINTS = ("STALE:", "FAIL", "FAILED", "fatal", "cannot", "TIMEOUT", "NO_",
"error", "Error", "refus", "not a ")
def pick_detail(text, limit=150):
"""First line that names a reason, else the last line that says anything.
A summary line that reports ZERO faults is not a reason: `selftest_sweep`
prints `sweep: 15 tools, 15 OK, 0 fatal` even on the run where its input
axes went red, and that line was being carried onto a FAULT row -- the row
said FAULT beside the words "0 fatal". Zero-count summaries are therefore
only reachable on the second pass (a clean run still gets its sweep line),
and any line that names a real reason outranks them.
"""
lines = [ln.strip() for ln in text.splitlines() if ln.strip()]
hinted = [ln for ln in lines if any(h in ln for h in HINTS)]
reasons = [ln for ln in hinted
if " 0 fatal" not in ln and " 0 fault" not in ln]
if reasons:
return reasons[0][:limit]
if hinted:
return hinted[0][:limit]
return lines[-1][:limit] if lines else ""
def leg_argv(key, repo, queue, index):
if key == "index-current":
return [sys.executable, os.path.join(repo, "tools", "queue_index.py"),
"--repo", repo, "--queue", queue, "--check", index]
if key == "append-audit":
return [sys.executable, os.path.join(repo, "tools", "queue_append_audit.py"),
"--repo", repo, "--queue", queue, "--index", index,
"--guard", "--to", "HEAD"]
if key == "selftest-sweep":
return [sys.executable, os.path.join(repo, "tools", "selftest_sweep.py"),
"--repo", repo, "--quiet"]
raise KeyError(key)
def run_leg(leg, repo, queue, index, timeout):
"""One leg -> one row. rc 0/1/2 are graded; anything else is ungradable."""
tool = os.path.join(repo, "tools", leg["tool"])
started = time.time()
row = {"key": leg["key"], "tool": leg["tool"], "verdict": None,
"rc": None, "detail": "", "what": leg["what"]}
if not os.path.exists(tool):
row.update(verdict=UNGRADABLE, detail="no such tool: %s" % tool)
row["ms"] = 0
return row
try:
proc = subprocess.run(leg_argv(leg["key"], repo, queue, index),
cwd=repo, capture_output=True, text=True,
timeout=timeout)
except subprocess.TimeoutExpired:
row.update(verdict=UNGRADABLE, detail="TIMEOUT after %ss" % timeout)
row["ms"] = int((time.time() - started) * 1000)
return row
except OSError as exc:
row.update(verdict=UNGRADABLE, detail="could not run: %s" % exc)
row["ms"] = int((time.time() - started) * 1000)
return row
out = (proc.stdout or "") + (proc.stderr or "")
row["rc"] = proc.returncode
row["detail"] = pick_detail(out)
if proc.returncode == 0:
row["verdict"] = PASS
elif proc.returncode == 1:
row["verdict"] = FAULT
elif proc.returncode == 2:
row["verdict"] = UNGRADABLE
else:
# rc 3 is retired in this repo (one meaning, two codes); any other code
# is a tool that left the contract, and the run cannot grade through it.
row["verdict"] = UNGRADABLE
row["detail"] = "unexpected rc %d: %s" % (proc.returncode, row["detail"])
row["ms"] = int((time.time() - started) * 1000)
return row
def counts(rows):
c = {PASS: 0, FAULT: 0, UNGRADABLE: 0, SKIP: 0}
for r in rows:
c[r["verdict"]] += 1
return c
def decide(rows):
"""The precedence, in one place: ungradable outranks fault outranks clean."""
c = counts(rows)
if c[UNGRADABLE]:
return 2
if c[FAULT]:
return 1
return 0
def report(rows, repo, quiet=False):
c = counts(rows)
if not quiet:
print("verify -- repo: %s" % repo)
for r in rows:
rc = "-" if r["rc"] is None else str(r["rc"])
print(" %-16s %-10s rc=%-2s %-6s %s"
% (r["key"], r["verdict"], rc, "%dms" % r["ms"], r["detail"]))
print()
print("verify: %d legs, %d pass, %d fault, %d ungradable, %d skipped"
% (len(rows), c[PASS], c[FAULT], c[UNGRADABLE], c[SKIP]))
faulted = [r["key"] for r in rows if r["verdict"] == FAULT]
ungraded = [r["key"] for r in rows if r["verdict"] == UNGRADABLE]
if faulted:
print("fault: %s" % " ".join(faulted))
if ungraded:
print("ungradable: %s" % " ".join(ungraded))
rc = decide(rows)
if rc == 0:
print("gate: OK -- %d of %d graded leg(s) passed%s"
% (c[PASS], c[PASS] + c[FAULT],
"" if not c[SKIP] else
" (%d skipped: not a pass, not a fault)" % c[SKIP]))
elif rc == 1:
print("gate: FAULT -- %d leg(s) ran and graded a fault: %s"
% (c[FAULT], " ".join(faulted)))
else:
print("gate: NOT CERTIFIED -- %d leg(s) could not grade: %s"
% (c[UNGRADABLE], " ".join(ungraded)))
return rc
def main(argv=None):
ap = argparse.ArgumentParser(
description="Run the repo's gates as one command: index currency, the "
"append audit, and the selftest sweep.")
ap.add_argument("--repo", default=".")
ap.add_argument("--queue", default=QUEUE_DEFAULT,
help="repo-relative queue path (default %s)" % QUEUE_DEFAULT)
ap.add_argument("--index", default=INDEX_DEFAULT,
help="repo-relative index document (default %s)" % INDEX_DEFAULT)
ap.add_argument("--fast", action="store_true",
help="skip the selftest-sweep leg; the row is printed as SKIP")
ap.add_argument("--timeout", type=int, default=300, metavar="S",
help="per-leg timeout in seconds (default 300)")
ap.add_argument("--json", metavar="PATH", help="write the rows here")
ap.add_argument("--quiet", action="store_true")
ap.add_argument("--selftest", action="store_true", help="fixture corpus, 26 cells")
a = ap.parse_args(argv)
if a.selftest:
return selftest()
repo = os.path.abspath(a.repo)
if not os.path.isdir(repo):
print("cannot read repo: %s is not a directory" % repo, file=sys.stderr)
return 2
if not os.path.isdir(os.path.join(repo, ".git")):
print("cannot read repo: %s is not a git repository" % repo, file=sys.stderr)
return 2
rows = []
for leg in LEGS:
if a.fast and leg["slow"]:
rows.append({"key": leg["key"], "tool": leg["tool"], "verdict": SKIP,
"rc": None, "detail": "--fast: leg not run",
"what": leg["what"], "ms": 0})
continue
rows.append(run_leg(leg, repo, a.queue, a.index, a.timeout))
rc = report(rows, repo, a.quiet)
if a.json:
c = counts(rows)
with open(a.json, "w") as fh:
json.dump({"repo": repo, "queue": a.queue, "index": a.index,
"fast": a.fast, "rows": rows, "counts": c, "rc": rc},
fh, indent=1)
return rc
# ------------------------------------------------------------------- selftest
def _stub(path, rc, out="", sleep=0.0, marker=None):
"""A stand-in leg: it prints what the fixture needs and exits with its rc."""
body = ["#!/usr/bin/env python3", "import sys"]
if sleep:
body.append("import time; time.sleep(%r)" % sleep)
if marker:
body.append("open(%r, 'a').write('hit\\n')" % marker)
if out:
body.append("sys.stdout.write(%r)" % out)
body.append("sys.exit(%d)" % rc)
with open(path, "w") as fh:
fh.write("\n".join(body) + "\n")
os.chmod(path, 0o755)
def _fixture(base, rcs, outs=None, sleep=0.0, markers=None):
"""A throwaway repo with three stub legs. rcs: {tool: rc}."""
repo = os.path.join(base, "repo")
os.makedirs(os.path.join(repo, "tools"))
os.makedirs(os.path.join(repo, ".git"))
with open(os.path.join(repo, QUEUE_DEFAULT), "w") as fh:
fh.write("## Q1 -- fixture\n")
os.makedirs(os.path.join(repo, "docs"))
with open(os.path.join(repo, INDEX_DEFAULT), "w") as fh:
fh.write("fixture\n")
for tool, rc in rcs.items():
if rc is None: # tool absent on purpose
continue
_stub(os.path.join(repo, "tools", tool), rc,
out=(outs or {}).get(tool, ""), sleep=sleep,
marker=(markers or {}).get(tool))
return repo
def _verify(*argv):
"""Run this file as the CLI, so the fixture grades the real entry point."""
proc = subprocess.run([sys.executable, os.path.abspath(__file__)] + list(argv),
capture_output=True, text=True)
return proc.returncode, (proc.stdout or "") + (proc.stderr or "")
THREE = {"queue_index.py": 0, "queue_append_audit.py": 0, "selftest_sweep.py": 0}
def selftest():
cells = 0
passed = 0
failures = []
def check(label, got, want):
nonlocal cells, passed
cells += 1
if got == want:
passed += 1
else:
failures.append("%s\n got: %r\n want: %r" % (label, got, want))
with tempfile.TemporaryDirectory(prefix="verify-selftest-") as td:
# 1. three green stubs -> rc 0
repo = _fixture(os.path.join(td, "c1"), THREE)
rc, out = _verify("--repo", repo)
check("1 all three legs green -> rc 0", rc, 0)
check("1b summary counts the three legs", "verify: 3 legs, 3 pass, 0 fault, "
"0 ungradable, 0 skipped" in out, True)
# 2. a stale index is a FAULT, and its own line reaches the row
repo = _fixture(os.path.join(td, "c2"),
dict(THREE, **{"queue_index.py": 1}),
outs={"queue_index.py":
"STALE: rendered document differs at line 7\n"})
rc, out = _verify("--repo", repo)
check("2 stale index -> rc 1", rc, 1)
check("2b the STALE line is carried into the row",
"index-current" in out and "differs at line 7" in out, True)
# 3. a refusal is UNGRADABLE, not a fault
repo = _fixture(os.path.join(td, "c3"),
dict(THREE, **{"queue_index.py": 2}),
outs={"queue_index.py": "cannot read queue: nope\n"})
rc, out = _verify("--repo", repo)
check("3 refusal -> rc 2", rc, 2)
check("3b refusal is named in the summary", "ungradable: index-current" in out, True)
# 4. precedence: ungradable outranks fault, and the fault is NOT erased
repo = _fixture(os.path.join(td, "c4"),
dict(THREE, **{"queue_index.py": 1,
"selftest_sweep.py": 2}),
outs={"queue_index.py": "STALE: line 3\n",
"selftest_sweep.py": "cannot read: no tools\n"})
rc, out = _verify("--repo", repo)
check("4 ungradable outranks fault -> rc 2", rc, 2)
check("4b both classes are still printed in one run",
"1 fault, 1 ungradable" in out and "fault: index-current" in out,
True)
# 5. a missing leg tool is ungradable, and says which path was missing
repo = _fixture(os.path.join(td, "c5"),
dict(THREE, **{"queue_append_audit.py": None}))
rc, out = _verify("--repo", repo)
check("5 missing tool -> rc 2", rc, 2)
check("5b the missing path is named", "no such tool:" in out, True)
# 6. --fast skips the sweep and does not run it
marker = os.path.join(td, "c6-sweep-ran")
repo = _fixture(os.path.join(td, "c6"), THREE, markers={"selftest_sweep.py": marker})
rc, out = _verify("--repo", repo, "--fast")
check("6 --fast on a green tree -> rc 0", rc, 0)
check("6b the skipped leg is printed as SKIP", "selftest-sweep" in out
and "SKIP" in out, True)
check("6c the skipped stub never ran", os.path.exists(marker), False)
# 7. --json is a real machine-readable record
repo = _fixture(os.path.join(td, "c7"), THREE)
jpath = os.path.join(td, "c7.json")
rc, out = _verify("--repo", repo, "--json", jpath)
obj = json.load(open(jpath)) if os.path.exists(jpath) else {}
check("7 --json writes the rows", [r["key"] for r in obj.get("rows", [])],
["index-current", "append-audit", "selftest-sweep"])
check("7b --json carries the same rc the process returned",
obj.get("rc"), rc)
# 8. a repo that is not there is refused before any leg runs
rc, out = _verify("--repo", os.path.join(td, "no-such-repo"))
check("8 missing repo -> rc 2", rc, 2)
check("8b the refusal names the path", "cannot read repo:" in out, True)
# 9. rc 3 is retired: it is ungradable, never a fault and never a pass
repo = _fixture(os.path.join(td, "c9"),
dict(THREE, **{"queue_index.py": 3}),
outs={"queue_index.py": "unknown state\n"})
rc, out = _verify("--repo", repo)
check("9 rc 3 -> rc 2 (retired code)", rc, 2)
check("9b the unexpected code is named", "unexpected rc 3" in out, True)
# 10. a hung leg is a timeout, not a pass and not a fault
repo = _fixture(os.path.join(td, "c10"),
dict(THREE, **{"selftest_sweep.py": 0}), sleep=5.0)
rc, out = _verify("--repo", repo, "--timeout", "1")
check("10 timeout -> rc 2", rc, 2)
check("10b the timeout is named", "TIMEOUT after 1s" in out, True)
# 11. order is fixed, cheapest first: the table is not re-sorted by outcome
repo = _fixture(os.path.join(td, "c11"),
dict(THREE, **{"queue_index.py": 1}),
outs={"queue_index.py": "STALE: line 1\n"})
rc, out = _verify("--repo", repo)
order = [out.find(k) for k in ("index-current", "append-audit", "selftest-sweep")]
check("11 legs print in fixed order", order == sorted(order)
and -1 not in order, True)
# 12. a skip is counted as a skip, never folded into pass
repo = _fixture(os.path.join(td, "c12"), THREE)
jpath = os.path.join(td, "c12.json")
rc, out = _verify("--repo", repo, "--fast", "--json", jpath)
obj = json.load(open(jpath)) if os.path.exists(jpath) else {}
check("12 skipped leg is its own count in --json",
obj.get("counts", {}).get("SKIP"), 1)
check("12b and the skipped leg is not counted as a pass",
obj.get("counts", {}).get(PASS), 2)
# 13. the detail line must not contradict the verdict: a row graded
# FAULT beside "0 fatal" is the SILENT failure this gate exists to catch.
check("13 a zero-fault summary is not a reason line",
pick_detail("sweep: 15 tools, 15 OK, 0 fatal\n"
"read-refusal: 20 probes, 19 ok, 1 fault (NO_PROBE)\n"
"malformed-input: 18 probes, 17 ok, 1 fault (NO_PROBE)\n"),
"read-refusal: 20 probes, 19 ok, 1 fault (NO_PROBE)")
check("13b a clean run still gets its sweep line",
pick_detail("sweep: 14 tools, 14 OK, 0 fatal\n"
"read-refusal: 19 probes, 19 ok, 0 fault\n"),
"sweep: 14 tools, 14 OK, 0 fatal")
print("selftest: %d cells, %d passed, %d failed" % (cells, passed, len(failures)))
for f in failures:
print(" FAIL %s" % f)
return 0 if passed == cells else 1
if __name__ == "__main__":
sys.exit(main())
Reference in new issue View Git Blame Copy Permalink