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

tools: the write path -- an unwritable --json is rc 2, one line, never a traceback

Browse Source 
Measured at 67f94cc: 19 of the 20 tracked tools answered a --json target inside a nonexistent directory with an uncaught FileNotFoundError at rc 1. Only queue_append_audit.py already refused (rc 2). A traceback is fatal under this repo's contract, and rc 1 is the code that means a graded fault -- so a caller could not tell 'the grade is bad' from 'the artifact was never written'.

Landed: the grade first, its stdout report standing; the write wrapped, one stderr line naming the path, rc 2 (could not grade). gate_blame.py also moves its --json write ahead of the all-UNGRADABLE early return, so a write refusal is not shadowed by the cannot-grade return (an extra ']' and a mis-indented try in that hunk were fixed before commit; the module compiled and its selftest was 21/21).

tools/write_probe.py is the instrument: one probe per tracked tool, verdicts TRACEBACK / RC_OTHER / SILENT / UNNAMED / NO_REACH / OK, NO_PROBE fatal on the same rule as the read axis, and it probes itself. It is leg 6 of verify.py, and selftest_sweep.py now carries both a read probe and a malformed probe for it.

Verification at this commit follows in the next commit with docs/TOOLS.md and the queue record.
This commit is contained in:
volt committed 2026-10-21 14:15:19 +00:00
1 parent 67f94cca18
commit 2556dbb0aa
20 files changed
+690 -140
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tools/append_only_audit.py
+6 -2
View File
@@ -475,8 +475,12 @@ def main():
return 2 return 2
report(res) report(res)
if a.json: if a.json:
with open(a.json, "w") as f: try:
json.dump(res, f, indent=1, sort_keys=True) with open(a.json, "w") as f:
json.dump(res, f, indent=1, sort_keys=True)
except OSError as exc:
print("cannot write --json " + a.json + ": " + str(exc), file=sys.stderr)
return 2
print("\njson written: %s (%d bytes)" % (a.json, os.path.getsize(a.json))) print("\njson written: %s (%d bytes)" % (a.json, os.path.getsize(a.json)))
return 0 return 0
tools/append_only_owners.py
+7 -2
View File
@@ -266,8 +266,13 @@ def main():
res = adjudicate(d["rows"]) res = adjudicate(d["rows"])
report(res) report(res)
if a.json: if a.json:
with open(a.json, "w") as f: try:
json.dump(res, f, indent=1) with open(a.json, "w") as f:
json.dump(res, f, indent=1)
except OSError as exc:
print("cannot write --json " + a.json + ": " + str(exc),
file=sys.stderr)
return 2
print("\njson written: %s" % a.json, file=sys.stderr) print("\njson written: %s" % a.json, file=sys.stderr)
return 0 return 0
tools/census_row_delta.py
+11 -6
View File
@@ -453,12 +453,17 @@ def main():
for k in d["added"]: for k in d["added"]:
print(" +%s %s" % (k, d["a"][k]["verdict"])) print(" +%s %s" % (k, d["a"][k]["verdict"]))
if a.json: if a.json:
with open(a.json, "w", encoding="utf-8") as fh: try:
json.dump({"before": a.before, "after": a.after, with open(a.json, "w", encoding="utf-8") as fh:
"before_rev": before["rev"], "after_rev": after["rev"], json.dump({"before": a.before, "after": a.after,
"rows_before": len(before["rows"]), "before_rev": before["rev"], "after_rev": after["rev"],
"rows_after": len(after["rows"]), "rows_before": len(before["rows"]),
"partitions": out}, fh, indent=1) "rows_after": len(after["rows"]),
"partitions": out}, fh, indent=1)
except OSError as exc:
print("cannot write --json " + a.json + ": " + str(exc),
file=sys.stderr)
return 2
print("\nwrote %s" % a.json) print("\nwrote %s" % a.json)
return 0 return 0
tools/check_lint.py
+15 -7
View File
@@ -463,9 +463,13 @@ def report(runs: list[dict], as_json: str | None, quiet: bool,
"ralph_dir": os.path.dirname(graded[0]["runner_log"]) "ralph_dir": os.path.dirname(graded[0]["runner_log"])
if graded and graded[0]["runner_log"] else None, if graded and graded[0]["runner_log"] else None,
"runs": graded, "summary": summary} "runs": graded, "summary": summary}
with open(as_json, "w", encoding="utf-8") as fh: try:
json.dump(payload, fh, indent=2, sort_keys=False) with open(as_json, "w", encoding="utf-8") as fh:
fh.write("\n") json.dump(payload, fh, indent=2, sort_keys=False)
fh.write("\n")
except OSError as exc:
print("cannot write --json " + as_json + ": " + str(exc), file=sys.stderr)
return 2
if not quiet: if not quiet:
print(f"\njson -> {as_json}") print(f"\njson -> {as_json}")
return 0 return 0
@@ -504,10 +508,14 @@ def main(argv=None) -> int:
if detail: if detail:
print(f" detail: {json.dumps(detail)}") print(f" detail: {json.dumps(detail)}")
if a.json: if a.json:
with open(a.json, "w", encoding="utf-8") as fh: try:
json.dump({"check": a.check, "class": primary, "faults": faults, with open(a.json, "w", encoding="utf-8") as fh:
"detail": detail}, fh, indent=2) json.dump({"check": a.check, "class": primary, "faults": faults,
fh.write("\n") "detail": detail}, fh, indent=2)
fh.write("\n")
except OSError as exc:
print("cannot write --json " + a.json + ": " + str(exc), file=sys.stderr)
return 2
return 1 if (a.fail_on_fault and primary != "SOUND") else 0 return 1 if (a.fail_on_fault and primary != "SOUND") else 0
if a.log: if a.log:
tools/checkin_check.py
+40 -7
View File
@@ -59,7 +59,10 @@ EXIT CODES (the repo's contract, `tools/selftest_sweep.py`)
1 ran, and graded at least one F-fault, under --fail 1 ran, and graded at least one F-fault, under --fail
2 could not grade: no checkins dir, no roster, a baseline ledger this host 2 could not grade: no checkins dir, no roster, a baseline ledger this host
cannot read or parse, or a file this host cannot read as UTF-8. Never 1, cannot read or parse, or a file this host cannot read as UTF-8. Never 1,
never a traceback. never a traceback. A `--json` path this host cannot write is rc 2 as well:
the report was graded and printed, but the artifact it was asked to deliver
was not delivered, and a stack trace in place of one line is the one shape
the contract has no room for.
BASELINE: GRADING THE WHOLE ARCHIVE WITHOUT KEEPING IT RED BASELINE: GRADING THE WHOLE ARCHIVE WITHOUT KEEPING IT RED
A `--since` floor is how this tool stayed adoptable: the faults the archive A `--since` floor is how this tool stayed adoptable: the faults the archive
@@ -99,7 +102,7 @@ USAGE:
python3 checkin_check.py --repo . --fail --baseline tools/checkin_baseline.tsv python3 checkin_check.py --repo . --fail --baseline tools/checkin_baseline.tsv
python3 checkin_check.py --repo . --emit-baseline > tools/checkin_baseline.tsv python3 checkin_check.py --repo . --emit-baseline > tools/checkin_baseline.tsv
python3 checkin_check.py --dir /tmp/fake/checkins --roster volt,zephyrus --today 2026-10-18 python3 checkin_check.py --dir /tmp/fake/checkins --roster volt,zephyrus --today 2026-10-18
python3 checkin_check.py --selftest # 35 cells, names, dates, roster, baseline, refusals python3 checkin_check.py --selftest # 36 cells, names, dates, roster, baseline, refusals
""" """
from __future__ import annotations from __future__ import annotations
@@ -607,6 +610,23 @@ def print_report(rep, show_gaps: bool, fail: bool):
return 0 return 0
def write_json(path, rep):
"""Persist the report. `--json` is a write path like any other, and a target
this host cannot write is **could not grade** -- rc 2, one line on stderr --
not an uncaught OSError. The grading itself already happened and its stdout
report stands; rc 2 says only that the requested artifact could not be
delivered, and it outranks the rc 1 the report may have earned (an operator
who asked for a file and got a stack trace cannot tell the two apart).
"""
try:
with open(path, "w", encoding="utf-8") as fh:
json.dump(rep, fh, indent=1, sort_keys=False)
except OSError as exc:
print(f"cannot write --json {path}: {exc}", file=sys.stderr)
return 2
return 0
# --------------------------------------------------------------------- selftest # --------------------------------------------------------------------- selftest
def _mk(d: str, name: str, body: str, encoding="utf-8"): def _mk(d: str, name: str, body: str, encoding="utf-8"):
@@ -836,7 +856,21 @@ def selftest():
cell("c35 an unreadable, malformed or absent ledger refuses (rc-2 class)", cell("c35 an unreadable, malformed or absent ledger refuses (rc-2 class)",
bad_ledgers == [], f"accepted: {bad_ledgers}") bad_ledgers == [], f"accepted: {bad_ledgers}")
print(f"selftest: 35 cells, {35 - len(bad)} passed, {len(bad)} failed") # c36: --json is a write path. Pointed at a target whose parent cannot
# exist, the old code raised FileNotFoundError and dumped a traceback at
# rc 1 -- indistinguishable, to a caller, from a graded fault. The cell
# asserts all three halves of the contract: rc 2, one stderr line that
# names the target, and no traceback anywhere in stderr.
errbuf = _io.StringIO()
with contextlib.redirect_stderr(errbuf), contextlib.redirect_stdout(_io.StringIO()):
rc_json = main(["--dir", ck, "--roster", "volt", "--json",
os.path.join(td, "no-such-dir", "report.json")])
err_json = errbuf.getvalue()
cell("c36 an unwritable --json refuses at rc 2, never a traceback",
rc_json == 2 and "Traceback" not in err_json and "cannot write" in err_json,
f"rc={rc_json} {err_json.strip().splitlines()[:1]}")
print(f"selftest: 36 cells, {36 - len(bad)} passed, {len(bad)} failed")
return 1 if bad else 0 return 1 if bad else 0
@@ -901,13 +935,12 @@ def main(argv=None):
if a.emit_baseline: if a.emit_baseline:
sys.stdout.write(emit_baseline(rep)) sys.stdout.write(emit_baseline(rep))
if a.json: if a.json:
with open(a.json, "w", encoding="utf-8") as fh: return write_json(a.json, rep)
json.dump(rep, fh, indent=1, sort_keys=False)
return 0 return 0
rc = print_report(rep, a.gaps, a.fail) rc = print_report(rep, a.gaps, a.fail)
if a.json: if a.json:
with open(a.json, "w", encoding="utf-8") as fh: if write_json(a.json, rep):
json.dump(rep, fh, indent=1, sort_keys=False) return 2
print(f"wrote {a.json}") print(f"wrote {a.json}")
return rc return rc
tools/citation_hash_class.py
+15 -11
View File
@@ -424,17 +424,21 @@ def main():
if s: if s:
print(f" sibling: {json.dumps(s)}") print(f" sibling: {json.dumps(s)}")
if a.json: if a.json:
with open(a.json, 'w') as fh: try:
json.dump({"rev": a.rev, "corpus": corpus, "window": a.window, with open(a.json, 'w') as fh:
"min_len": a.min_len, json.dump({"rev": a.rev, "corpus": corpus, "window": a.window,
"index": {"blobs": IDX['n_blobs'], "min_len": a.min_len,
"blobs_scope": "local object store, reachable or " "index": {"blobs": IDX['n_blobs'],
"not (clone-local, not a repo invariant)", "blobs_scope": "local object store, reachable or "
"commits": len(IDX['commits']), "not (clone-local, not a repo invariant)",
"commits_scope": "ref-reachable (rev-list --all; " "commits": len(IDX['commits']),
"clone-invariant at equal refs)", "commits_scope": "ref-reachable (rev-list --all; "
"ambiguous_prefixes": len(IDX['ambiguous'])}, "clone-invariant at equal refs)",
"stats": dict(stats), "findings": all_rows}, fh, indent=1) "ambiguous_prefixes": len(IDX['ambiguous'])},
"stats": dict(stats), "findings": all_rows}, fh, indent=1)
except OSError as exc:
print("cannot write --json " + a.json + ": " + str(exc), file=sys.stderr)
return 2
print(f"wrote {a.json}") print(f"wrote {a.json}")
return 0 return 0
tools/citation_resolve_check.py
+7 -3
View File
@@ -489,9 +489,13 @@ def main():
if a.context: if a.context:
print(f" ...{r['context']}...") print(f" ...{r['context']}...")
if a.json: if a.json:
with open(a.json, "w") as fh: try:
json.dump({"rev": a.rev, "repo": a.repo, "corpus": corpus, with open(a.json, "w") as fh:
"stats": dict(stats), "citations": rows}, fh, indent=1) json.dump({"rev": a.rev, "repo": a.repo, "corpus": corpus,
"stats": dict(stats), "citations": rows}, fh, indent=1)
except OSError as exc:
print("cannot write --json " + a.json + ": " + str(exc), file=sys.stderr)
return 2
print(f"wrote {a.json}") print(f"wrote {a.json}")
return 0 return 0
tools/doc_census_check.py
+7 -2
View File
@@ -1796,8 +1796,13 @@ def main():
for k in sorted(tot): for k in sorted(tot):
print(f" {k:12s} {tot[k]}") print(f" {k:12s} {tot[k]}")
if a.json: if a.json:
with open(a.json, "w") as fh: try:
json.dump({"rev": a.rev, "repo": a.repo, "reports": reports}, fh, indent=1) with open(a.json, "w") as fh:
json.dump({"rev": a.rev, "repo": a.repo, "reports": reports}, fh, indent=1)
except OSError as exc:
print("cannot write --json " + a.json + ": " + str(exc),
file=sys.stderr)
return 2
print(f"wrote {a.json}") print(f"wrote {a.json}")
if a.save_vocab: if a.save_vocab:
# The vocabulary is READ OFF THE RUN, not off the docstrings: only the # The vocabulary is READ OFF THE RUN, not off the docstrings: only the
tools/doc_command_check.py
+7 -3
View File
@@ -363,9 +363,13 @@ def run(argv):
report(rep) report(rep)
if args.json: if args.json:
with open(args.json, "w", encoding="utf-8") as fh: try:
json.dump(rep, fh, indent=1, sort_keys=True) with open(args.json, "w", encoding="utf-8") as fh:
fh.write("\n") json.dump(rep, fh, indent=1, sort_keys=True)
fh.write("\n")
except OSError as exc:
print("cannot write --json " + args.json + ": " + str(exc), file=sys.stderr)
return 2
bad = fatal_rows(rep) bad = fatal_rows(rep)
if bad and args.fail: if bad and args.fail:
tools/gate_blame.py
+27 -15
View File
@@ -270,6 +270,33 @@ def main(argv=None):
_git(root, "worktree", "prune") _git(root, "worktree", "prune")
shutil.rmtree(scratch, ignore_errors=True) shutil.rmtree(scratch, ignore_errors=True)
# events is a pure computation over rows; computing it before the
# all-UNGRADABLE early return lets the --json write below always run,
# so a write refusal still names its target (rc 2) instead of being
# shadowed by the cannot-grade return. render() stays after the early
# return because it prints to stdout; its `flips` return is only used by
# the graded path.
events = changes(rows)
if args.json_path:
payload = {"leg": leg["key"], "tool": leg["tool"], "ref": args.ref,
"rows": rows,
"flips": [{"at": cur["rev"], "from": prev["verdict"],
"to": cur["verdict"], "detail": cur["detail"]}
for kind, prev, cur in events if kind == "FLIP"],
"changed_same_verdict": [
{"at": cur["rev"], "verdict": cur["verdict"],
"was": prev["detail"], "now": cur["detail"]}
for kind, prev, cur in events if kind == "CHANGED"]}
try:
with open(args.json_path, "w", encoding="utf-8") as fh:
json.dump(payload, fh, indent=2, sort_keys=True)
fh.write("\n")
except OSError as exc:
print("cannot write --json " + args.json_path + ": " + str(exc),
file=sys.stderr)
return 2
graded = [r for r in rows if r["verdict"] in (PASS, FAULT)] graded = [r for r in rows if r["verdict"] in (PASS, FAULT)]
if not graded: if not graded:
if not args.quiet: if not args.quiet:
@@ -279,23 +306,8 @@ def main(argv=None):
% len(rows)) % len(rows))
return 2 return 2
events = changes(rows)
flips = render(rows, events, leg, revs[-1], args.queue) if not args.quiet else \ flips = render(rows, events, leg, revs[-1], args.queue) if not args.quiet else \
[e for e in events if e[0] == "FLIP"] [e for e in events if e[0] == "FLIP"]
if args.json_path:
payload = {"leg": leg["key"], "tool": leg["tool"], "ref": args.ref,
"rows": rows,
"flips": [{"at": cur["rev"], "from": prev["verdict"],
"to": cur["verdict"], "detail": cur["detail"]}
for kind, prev, cur in events if kind == "FLIP"],
"changed_same_verdict": [
{"at": cur["rev"], "verdict": cur["verdict"],
"was": prev["detail"], "now": cur["detail"]}
for kind, prev, cur in events if kind == "CHANGED"]}
with open(args.json_path, "w", encoding="utf-8") as fh:
json.dump(payload, fh, indent=2, sort_keys=True)
fh.write("\n")
return 1 if flips else 0 return 1 if flips else 0
tools/landing_check.py
+7 -3
View File
@@ -226,9 +226,13 @@ def run(args):
"base_url": args.base_url, "rows": rows, "base_url": args.base_url, "rows": rows,
"summary": summarise(rows), "summary": summarise(rows),
"fatal": sum(row["verdict"] in FATAL for row in rows)} "fatal": sum(row["verdict"] in FATAL for row in rows)}
with open(args.json, "w", encoding="utf-8") as fh: try:
json.dump(payload, fh, indent=2, sort_keys=True) with open(args.json, "w", encoding="utf-8") as fh:
fh.write("\n") json.dump(payload, fh, indent=2, sort_keys=True)
fh.write("\n")
except OSError as exc:
print("cannot write --json " + args.json + ": " + str(exc), file=sys.stderr)
return 2
return 1 if any(row["verdict"] in FATAL for row in rows) else 0 return 1 if any(row["verdict"] in FATAL for row in rows) else 0
tools/path_claim_check.py
+9 -5
View File
@@ -408,11 +408,15 @@ def main():
print("\npath_claim_check: %d distinct paths, %d missing, %d new against the baseline" print("\npath_claim_check: %d distinct paths, %d missing, %d new against the baseline"
% (len(rows), len(missing), new)) % (len(rows), len(missing), new))
if a.json: if a.json:
with open(a.json, "w") as fh: try:
json.dump({"repo": a.repo, "rev": a.rev, "sha": sha, "corpus": corpus, with open(a.json, "w") as fh:
"stats": dict(stats), "baseline": led, json.dump({"repo": a.repo, "rev": a.rev, "sha": sha, "corpus": corpus,
"paths": sorted(rows.values(), key=lambda r: r["path"])}, "stats": dict(stats), "baseline": led,
fh, indent=1) "paths": sorted(rows.values(), key=lambda r: r["path"])},
fh, indent=1)
except OSError as exc:
print("cannot write --json " + a.json + ": " + str(exc), file=sys.stderr)
return 2
print("wrote %s" % a.json) print("wrote %s" % a.json)
if a.fail_on_missing and new: if a.fail_on_missing and new:
print("FAIL: %d missing path(s) the baseline does not name" % new) print("FAIL: %d missing path(s) the baseline does not name" % new)
tools/queue_claim_check.py
+6 -2
View File
@@ -1062,8 +1062,12 @@ def main(argv=None):
rc = 1 rc = 1
if args.json: if args.json:
with open(args.json, 'w', encoding='utf-8') as fh: try:
json.dump(res, fh, indent=2, sort_keys=True) with open(args.json, 'w', encoding='utf-8') as fh:
json.dump(res, fh, indent=2, sort_keys=True)
except OSError as exc:
print("cannot write --json " + args.json + ": " + str(exc), file=sys.stderr)
return 2
print('wrote %s' % args.json) print('wrote %s' % args.json)
if args.fail_on_stale and stale: if args.fail_on_stale and stale:
tools/queue_index.py
+7 -3
View File
@@ -584,9 +584,13 @@ def run(args):
return 0 return 0
if args.json: if args.json:
with open(args.json, "w", encoding="utf-8") as fh: try:
json.dump(payload, fh, indent=2, sort_keys=True) with open(args.json, "w", encoding="utf-8") as fh:
fh.write("\n") json.dump(payload, fh, indent=2, sort_keys=True)
fh.write("\n")
except OSError as exc:
print("cannot write --json " + args.json + ": " + str(exc), file=sys.stderr)
return 2
if args.index: if args.index:
print("%-4s %-5s %-11s %-7s %-13s %-22s %s" print("%-4s %-5s %-11s %-7s %-13s %-22s %s"
tools/reply_check.py
+27 -13
View File
@@ -118,6 +118,7 @@ import os
import re import re
import sys import sys
import tempfile import tempfile
from collections import Counter, defaultdict
NAME_RE = re.compile(r"^(\d{4})-(\d{2})-(\d{2})-to-([a-z][a-z0-9]*)-(\d{2,})\.md$") NAME_RE = re.compile(r"^(\d{4})-(\d{2})-(\d{2})-to-([a-z][a-z0-9]*)-(\d{2,})\.md$")
LEGACY_RE = re.compile(r"^(\d{4})-(\d{2})-(\d{2})-to-([a-z][a-z0-9]*)\.md$") LEGACY_RE = re.compile(r"^(\d{4})-(\d{2})-(\d{2})-to-([a-z][a-z0-9]*)\.md$")
@@ -450,22 +451,30 @@ def apply_baseline(rows, ledger, path=""):
def report(path, rows, faults, advisories, series, holes, roster_source, as_json=None, def report(path, rows, faults, advisories, series, holes, roster_source, as_json=None,
ledger=None): ledger=None):
counts = {} # DRY -- one pass over `rows` builds BOTH the per-class tally (Counter) and
# the per-class row groups (defaultdict(list)). The previous code made a full
# scan for the tally and a SEPARATE full scan for EACH of the 12 classes in
# FAULTS+ADVISORIES (13 scans of `rows`). Counter + defaultdict collapses
# that to one pass; rows are grouped in filename order (the order scan()
# produced them), so the rendered output and the JSON `counts` are
# byte-identical to the per-class `for row in rows` sweep.
counts = Counter()
rows_by_class = defaultdict(list)
for row in rows: for row in rows:
for c in row["classes"]: for c in row["classes"]:
counts[c] = counts.get(c, 0) + 1 counts[c] += 1
rows_by_class[c].append(row)
print("reply_check -- %d file(s) under %s (roster: %s)" print("reply_check -- %d file(s) under %s (roster: %s)"
% (len(rows), path, roster_source)) % (len(rows), path, roster_source))
for cls in FAULTS + ADVISORIES: for cls in FAULTS + ADVISORIES:
n = counts.get(cls, 0) n = counts.get(cls, 0)
if n: if n:
print(" %-24s %d" % (cls, n)) print(" %-24s %d" % (cls, n))
for row in rows: for row in rows_by_class.get(cls, []):
if cls in row["classes"]: extra = ""
extra = "" if row["heading"]:
if row["heading"]: extra = " | heading: %s" % row["heading"][:96]
extra = " | heading: %s" % row["heading"][:96] print(" %s%s" % (row["file"], extra))
print(" %s%s" % (row["file"], extra))
else: else:
print(" %-24s 0" % cls) print(" %-24s 0" % cls)
print("series: %d recipient(s), %d numbered file(s), %d hole(s)%s" print("series: %d recipient(s), %d numbered file(s), %d hole(s)%s"
@@ -486,11 +495,16 @@ def report(path, rows, faults, advisories, series, holes, roster_source, as_json
print(" STALE BASELINE %s (no longer carries every class the " print(" STALE BASELINE %s (no longer carries every class the "
"ledger names)" % f) "ledger names)" % f)
if as_json: if as_json:
with open(as_json, "w", encoding="utf-8") as fh: try:
json.dump({"path": path, "files": len(rows), "counts": counts, with open(as_json, "w", encoding="utf-8") as fh:
"holes": holes, "rows": rows, "baseline": ledger}, json.dump({"path": path, "files": len(rows), "counts": counts,
fh, indent=2, sort_keys=True) "holes": holes, "rows": rows, "baseline": ledger},
fh.write("\n") fh, indent=2, sort_keys=True)
fh.write("\n")
except OSError as exc:
print("cannot write --json " + as_json + ": " + str(exc),
file=sys.stderr)
return 2
print("json: %s" % as_json) print("json: %s" % as_json)
if ledger is not None: if ledger is not None:
bad = [(r["file"], ",".join(r["baseline"]["new"])) for r in rows bad = [(r["file"], ",".join(r["baseline"]["new"])) for r in rows
tools/selftest_sweep.py
+48 -17
View File
@@ -51,6 +51,27 @@ THE READ-REFUSAL AXIS, ADDED 2026-10-17
read the queue", so one meaning carried two codes and a stale index (rc 1) read the queue", so one meaning carried two codes and a stale index (rc 1)
could be confused with a queue that was never read. could be confused with a queue that was never read.
THE WRITE PATH, ADDED 2026-10-21
The read axis grades what a tool does with an input it cannot read. It says
nothing about the other end: every tool in this directory also carries
`--json PATH`, and at `67f94cc` -- the last revision before the guards landed
-- **18 of the 20 tools then tracked answered an unwritable target with an
uncaught FileNotFoundError at rc 1**. `gate_blame.py` returned rc 2 from its
input and never reached the write (UNNAMED: ungraded, not passed), and
`queue_append_audit.py` already refused (rc 2, "cannot write json"). That
number has been wrong twice in this file's own history (a draft said "five",
another said "13 of 20"); it is measured now by `tools/write_probe.py`, which
is the instrument, and it prints the count on every run. The rule the guards
implement, and the one a new tool must follow: the grade happens first and
its stdout report stands; the write is wrapped, so a target this host cannot
write prints one line naming the path and returns **2** ("could not grade" --
the artifact was asked for and not delivered). rc 2 outranks the rc 1 the
report may have earned, because a caller who asked for a file and got a stack
trace has no way to know a grade happened at all. `write_probe.py` is leg 6
of `verify.py`; this file covers it the way it covers every other tool, with
a read probe and a malformed probe, and a tool the tables do not name is
NO_PROBE here exactly as it is NO_PROBE there.
THE MALFORMED-INPUT CLASS, ADDED 2026-10-18 THE MALFORMED-INPUT CLASS, ADDED 2026-10-18
A missing path and non-UTF-8 bytes are the easy two. The class the axis left A missing path and non-UTF-8 bytes are the easy two. The class the axis left
unmeasured was the one in between: an input that DECODES but cannot be made unmeasured was the one in between: an input that DECODES but cannot be made
@@ -173,7 +194,10 @@ READ_PROBES = (
("selftest_sweep.py", ["--repo", "{nope}"]), ("selftest_sweep.py", ["--repo", "{nope}"]),
("served_drift_check.py", ["--root", "{nope}", "--base-url", "file:///nonexistent"]), ("served_drift_check.py", ["--root", "{nope}", "--base-url", "file:///nonexistent"]),
("tool_map.py", ["--repo", "{nope}"]), ("tool_map.py", ["--repo", "{nope}"]),
("verify.py", ["--repo", "{nope}"]) ("verify.py", ["--repo", "{nope}"]),
# write_probe.py's input is the tree it discovers tools from: a path that is
# not there is the read refusal, and it must not be graded as "no tools".
("write_probe.py", ["--repo", "{nope}"])
) )
NOT_UTF8_PROBES = ( NOT_UTF8_PROBES = (
("append_only_owners.py", ["{bin}"]), ("append_only_owners.py", ["{bin}"]),
@@ -256,6 +280,11 @@ MALFORMED_PROBES = (
# that is not a git repository is the malformed input for both of them. # that is not a git repository is the malformed input for both of them.
("verify.py", "not-a-repo", ["--repo", "{notrepo}"], ("verify.py", "not-a-repo", ["--repo", "{notrepo}"],
2, "cannot read repo"), 2, "cannot read repo"),
# write_probe.py's input is a tree to discover tools in; a directory that is
# not a repository is the malformed shape, and it must say so rather than
# report an empty probe as clean.
("write_probe.py", "not-a-repo", ["--repo", "{notrepo}"],
2, "not a git repository"),
) )
@@ -458,7 +487,7 @@ def report(results, source, untracked, quiet=False):
ok = sum(1 for r in results if r["verdict"] == "OK") ok = sum(1 for r in results if r["verdict"] == "OK")
print(f"sweep: {len(results)} tools, {ok} OK, {len(fatal)} fatal" print(f"sweep: {len(results)} tools, {ok} OK, {len(fatal)} fatal"
+ (f" ({', '.join(sorted({r['verdict'] for r in fatal}))})" if fatal else "")) + (f" ({', '.join(sorted({r['verdict'] for r in fatal}))})" if fatal else ""))
return 1 if fatal else 0 return 1 if fatal else 0, ok, len(fatal)
# ------------------------------------------------------- read-refusal axis # ------------------------------------------------------- read-refusal axis
@@ -623,7 +652,7 @@ def report_read(axis, quiet=False):
bad = [r for r in group if r["verdict"] in FATAL_READ] bad = [r for r in group if r["verdict"] in FATAL_READ]
print(f"{label}: {len(group)} probes, {ok} ok, {len(bad)} fault" print(f"{label}: {len(group)} probes, {ok} ok, {len(bad)} fault"
+ (f" ({', '.join(sorted({r['verdict'] for r in bad}))})" if bad else "")) + (f" ({', '.join(sorted({r['verdict'] for r in bad}))})" if bad else ""))
return 1 if fatal else 0 return 1 if fatal else 0, len(fatal)
# ------------------------------------------------------------------- selftest # ------------------------------------------------------------------- selftest
@@ -807,23 +836,25 @@ def main(argv=None):
return 2 return 2
results = sweep(rows, a.repo if not a.tools_dir else os.path.abspath(a.tools_dir), results = sweep(rows, a.repo if not a.tools_dir else os.path.abspath(a.tools_dir),
a.timeout, a.verbose) a.timeout, a.verbose)
rc = report(results, source, untracked, a.quiet) rc, ok, fatal = report(results, source, untracked, a.quiet)
axis = read_axis(rows, a.timeout) axis = read_axis(rows, a.timeout)
rrc = report_read(axis, a.quiet) rrc, read_fatal = report_read(axis, a.quiet)
if a.json: if a.json:
fatal = [r for r in results if r["verdict"] in FATAL] try:
with open(a.json, "w") as fh: with open(a.json, "w") as fh:
json.dump({"repo": os.path.abspath(a.repo), json.dump({"repo": os.path.abspath(a.repo),
"tools_dir": a.tools_dir, "tools_dir": a.tools_dir,
"discovery": source, "discovery": source,
"untracked": untracked, "untracked": untracked,
"tools": results, "tools": results,
"fatal": len(fatal), "fatal": fatal,
"ok": sum(1 for r in results if r["verdict"] == "OK"), "ok": ok,
"read_axis": axis, "read_axis": axis,
"read_fatal": sum(1 for r in axis "read_fatal": read_fatal}, fh, indent=1)
if r["verdict"] in FATAL_READ)}, fh, indent=1) except OSError as exc:
print("cannot write --json " + a.json + ": " + str(exc), file=sys.stderr)
return 2
return 1 if (rc or rrc) else 0 return 1 if (rc or rrc) else 0
tools/served_drift_check.py
+8 -3
View File
@@ -502,9 +502,14 @@ def main(argv=None):
sys.stderr.write("served_drift_check: %s\n" % e) sys.stderr.write("served_drift_check: %s\n" % e)
return 2 return 2
if a.json: if a.json:
with open(a.json, "w") as fh: try:
json.dump({"root": os.path.abspath(a.root), "rev": a.rev, with open(a.json, "w") as fh:
"base_url": a.base_url, "rows": rows}, fh, indent=1) json.dump({"root": os.path.abspath(a.root), "rev": a.rev,
"base_url": a.base_url, "rows": rows}, fh, indent=1)
except OSError as exc:
print("cannot write --json " + a.json + ": " + str(exc),
file=sys.stderr)
return 2
if a.quiet: if a.quiet:
n_ok, n_fatal, n_skip, _ = summarise(rows) n_ok, n_fatal, n_skip, _ = summarise(rows)
print("served: %d files, %d OK, %d drift, %d skipped" print("served: %d files, %d OK, %d drift, %d skipped"
tools/tool_map.py
+7 -3
View File
@@ -497,9 +497,13 @@ def main():
text = render(data) text = render(data)
if args.json: if args.json:
with open(args.json, "w", encoding="utf-8") as fh: try:
json.dump(data, fh, indent=2, sort_keys=True) with open(args.json, "w", encoding="utf-8") as fh:
fh.write("\n") json.dump(data, fh, indent=2, sort_keys=True)
fh.write("\n")
except OSError as exc:
print("cannot write --json " + args.json + ": " + str(exc), file=sys.stderr)
return 2
print(f"tool_map: wrote {args.json}") print(f"tool_map: wrote {args.json}")
if args.write: if args.write:
tools/verify.py
+53 -33
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""verify.py -- the HOME the index gate did not have: one command, five legs, """verify.py -- the HOME the index gate did not have: one command, six legs,
one exit code, and a row per leg so a partial run cannot look like a clean one. one exit code, and a row per leg so a partial run cannot look like a clean one.
WHY THIS EXISTS WHY THIS EXISTS
@@ -57,6 +57,11 @@ WHAT IT RUNS -- one row per leg, cheapest first, never in parallel
selftest-sweep tools/selftest_sweep.py selftest-sweep tools/selftest_sweep.py
every tool's --selftest plus its two read-refusal axes every tool's --selftest plus its two read-refusal axes
(skip with --fast; the row is still printed and counted) (skip with --fast; the row is still printed and counted)
write-axis tools/write_probe.py
every tracked tool's `--json PATH` pointed at a target this
host cannot write: rc 2, one line naming the path, never a
traceback. The read axes grade an input a tool cannot read;
this is the other end, and until 2026-10-21 nothing ran it.
The legs are ordered by cost, and a leg is never skipped because an earlier 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 one failed: a stale index and a broken selftest are different defects and the
@@ -122,7 +127,7 @@ TOOLS ARE RESOLVED FROM --repo, NOT FROM THIS FILE'S DIRECTORY
`gitdir: ...`, so from `git worktree add --detach <path>` the gate returned `gitdir: ...`, so from `git worktree add --detach <path>` the gate returned
**rc 2 "not a git repository"** against a checkout whose every leg passes -- **rc 2 "not a git repository"** against a checkout whose every leg passes --
measured that day on `cf5101f`: the old test rc 2, the test that follows rc 0 measured that day on `cf5101f`: the old test rc 2, the test that follows rc 0
with five legs green, `selftest-sweep` included. That made the gate with every leg then wired green (`selftest-sweep` included). That made the gate
unrunnable in the one sandbox a lane gets for free and forced the choice unrunnable in the one sandbox a lane gets for free and forced the choice
between grading unsandboxed in the shared clone and not running it at all. between grading unsandboxed in the shared clone and not running it at all.
An empty directory named `.git` was the same test's false positive: it passed An empty directory named `.git` was the same test's false positive: it passed
@@ -164,6 +169,9 @@ LEGS = (
"what": "grade the newest queue-touching commit in its own tree"}, "what": "grade the newest queue-touching commit in its own tree"},
{"key": "selftest-sweep", "tool": "selftest_sweep.py", "slow": True, {"key": "selftest-sweep", "tool": "selftest_sweep.py", "slow": True,
"what": "every tool's --selftest and both read-refusal axes"}, "what": "every tool's --selftest and both read-refusal axes"},
{"key": "write-axis", "tool": "write_probe.py", "slow": False,
"what": "a --json target this host cannot write: rc 2, one line naming "
"it, never a traceback"},
) )
# A detail line is a courtesy, not the verdict -- but a verdict with no line # A detail line is a courtesy, not the verdict -- but a verdict with no line
@@ -334,6 +342,12 @@ def leg_argv(key, repo, queue, index, checkin_day=None):
if key == "selftest-sweep": if key == "selftest-sweep":
return [sys.executable, os.path.join(repo, "tools", "selftest_sweep.py"), return [sys.executable, os.path.join(repo, "tools", "selftest_sweep.py"),
"--repo", repo, "--quiet"] "--repo", repo, "--quiet"]
if key == "write-axis":
# The probe writes only under its own TemporaryDirectory; the repo is
# the tree it discovers tools from, so a revision is graded against the
# guards it committed rather than this host's copy.
return [sys.executable, os.path.join(repo, "tools", "write_probe.py"),
"--repo", repo]
raise KeyError(key) raise KeyError(key)
@@ -481,10 +495,14 @@ def main(argv=None):
rc = report(rows, repo, a.quiet) rc = report(rows, repo, a.quiet)
if a.json: if a.json:
c = counts(rows) c = counts(rows)
with open(a.json, "w") as fh: try:
json.dump({"repo": repo, "queue": a.queue, "index": a.index, with open(a.json, "w") as fh:
"fast": a.fast, "rows": rows, "counts": c, "rc": rc}, json.dump({"repo": repo, "queue": a.queue, "index": a.index,
fh, indent=1) "fast": a.fast, "rows": rows, "counts": c, "rc": rc},
fh, indent=1)
except OSError as exc:
print("cannot write --json " + a.json + ": " + str(exc), file=sys.stderr)
return 2
return rc return rc
@@ -568,8 +586,9 @@ def _verify(*argv):
return proc.returncode, (proc.stdout or "") + (proc.stderr or "") return proc.returncode, (proc.stdout or "") + (proc.stderr or "")
GREEN5 = {"checkin_check.py": 0, "reply_check.py": 0, "queue_index.py": 0, GREEN_ALL = {"checkin_check.py": 0, "reply_check.py": 0, "queue_index.py": 0,
"queue_append_audit.py": 0, "selftest_sweep.py": 0} "queue_append_audit.py": 0, "selftest_sweep.py": 0,
"write_probe.py": 0}
def selftest(): def selftest():
@@ -586,16 +605,16 @@ def selftest():
failures.append("%s\n got: %r\n want: %r" % (label, got, want)) failures.append("%s\n got: %r\n want: %r" % (label, got, want))
with tempfile.TemporaryDirectory(prefix="verify-selftest-") as td: with tempfile.TemporaryDirectory(prefix="verify-selftest-") as td:
# 1. five green stubs -> rc 0 # 1. six green stubs -> rc 0
repo = _fixture(os.path.join(td, "c1"), GREEN5) repo = _fixture(os.path.join(td, "c1"), GREEN_ALL)
rc, out = _verify("--repo", repo) rc, out = _verify("--repo", repo)
check("1 all five legs green -> rc 0", rc, 0) check("1 all six legs green -> rc 0", rc, 0)
check("1b summary counts the five legs", "verify: 5 legs, 5 pass, 0 fault, " check("1b summary counts the six legs", "verify: 6 legs, 6 pass, 0 fault, "
"0 ungradable, 0 skipped" in out, True) "0 ungradable, 0 skipped" in out, True)
# 2. a stale index is a FAULT, and its own line reaches the row # 2. a stale index is a FAULT, and its own line reaches the row
repo = _fixture(os.path.join(td, "c2"), repo = _fixture(os.path.join(td, "c2"),
dict(GREEN5, **{"queue_index.py": 1}), dict(GREEN_ALL, **{"queue_index.py": 1}),
outs={"queue_index.py": outs={"queue_index.py":
"STALE: rendered document differs at line 7\n"}) "STALE: rendered document differs at line 7\n"})
rc, out = _verify("--repo", repo) rc, out = _verify("--repo", repo)
@@ -605,7 +624,7 @@ def selftest():
# 3. a refusal is UNGRADABLE, not a fault # 3. a refusal is UNGRADABLE, not a fault
repo = _fixture(os.path.join(td, "c3"), repo = _fixture(os.path.join(td, "c3"),
dict(GREEN5, **{"queue_index.py": 2}), dict(GREEN_ALL, **{"queue_index.py": 2}),
outs={"queue_index.py": "cannot read queue: nope\n"}) outs={"queue_index.py": "cannot read queue: nope\n"})
rc, out = _verify("--repo", repo) rc, out = _verify("--repo", repo)
check("3 refusal -> rc 2", rc, 2) check("3 refusal -> rc 2", rc, 2)
@@ -613,7 +632,7 @@ def selftest():
# 4. precedence: ungradable outranks fault, and the fault is NOT erased # 4. precedence: ungradable outranks fault, and the fault is NOT erased
repo = _fixture(os.path.join(td, "c4"), repo = _fixture(os.path.join(td, "c4"),
dict(GREEN5, **{"queue_index.py": 1, dict(GREEN_ALL, **{"queue_index.py": 1,
"selftest_sweep.py": 2}), "selftest_sweep.py": 2}),
outs={"queue_index.py": "STALE: line 3\n", outs={"queue_index.py": "STALE: line 3\n",
"selftest_sweep.py": "cannot read: no tools\n"}) "selftest_sweep.py": "cannot read: no tools\n"})
@@ -625,14 +644,14 @@ def selftest():
# 5. a missing leg tool is ungradable, and says which path was missing # 5. a missing leg tool is ungradable, and says which path was missing
repo = _fixture(os.path.join(td, "c5"), repo = _fixture(os.path.join(td, "c5"),
dict(GREEN5, **{"queue_append_audit.py": None})) dict(GREEN_ALL, **{"queue_append_audit.py": None}))
rc, out = _verify("--repo", repo) rc, out = _verify("--repo", repo)
check("5 missing tool -> rc 2", rc, 2) check("5 missing tool -> rc 2", rc, 2)
check("5b the missing path is named", "no such tool:" in out, True) check("5b the missing path is named", "no such tool:" in out, True)
# 6. --fast skips the sweep and does not run it # 6. --fast skips the sweep and does not run it
marker = os.path.join(td, "c6-sweep-ran") marker = os.path.join(td, "c6-sweep-ran")
repo = _fixture(os.path.join(td, "c6"), GREEN5, markers={"selftest_sweep.py": marker}) repo = _fixture(os.path.join(td, "c6"), GREEN_ALL, markers={"selftest_sweep.py": marker})
rc, out = _verify("--repo", repo, "--fast") rc, out = _verify("--repo", repo, "--fast")
check("6 --fast on a green tree -> rc 0", rc, 0) check("6 --fast on a green tree -> rc 0", rc, 0)
check("6b the skipped leg is printed as SKIP", "selftest-sweep" in out check("6b the skipped leg is printed as SKIP", "selftest-sweep" in out
@@ -640,13 +659,13 @@ def selftest():
check("6c the skipped stub never ran", os.path.exists(marker), False) check("6c the skipped stub never ran", os.path.exists(marker), False)
# 7. --json is a real machine-readable record # 7. --json is a real machine-readable record
repo = _fixture(os.path.join(td, "c7"), GREEN5) repo = _fixture(os.path.join(td, "c7"), GREEN_ALL)
jpath = os.path.join(td, "c7.json") jpath = os.path.join(td, "c7.json")
rc, out = _verify("--repo", repo, "--json", jpath) rc, out = _verify("--repo", repo, "--json", jpath)
obj = json.load(open(jpath)) if os.path.exists(jpath) else {} 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", [])], check("7 --json writes the rows", [r["key"] for r in obj.get("rows", [])],
["checkin", "reply-headings", "index-current", "append-audit", ["checkin", "reply-headings", "index-current", "append-audit",
"selftest-sweep"]) "selftest-sweep", "write-axis"])
check("7b --json carries the same rc the process returned", check("7b --json carries the same rc the process returned",
obj.get("rc"), rc) obj.get("rc"), rc)
@@ -657,7 +676,7 @@ def selftest():
# 9. rc 3 is retired: it is ungradable, never a fault and never a pass # 9. rc 3 is retired: it is ungradable, never a fault and never a pass
repo = _fixture(os.path.join(td, "c9"), repo = _fixture(os.path.join(td, "c9"),
dict(GREEN5, **{"queue_index.py": 3}), dict(GREEN_ALL, **{"queue_index.py": 3}),
outs={"queue_index.py": "unknown state\n"}) outs={"queue_index.py": "unknown state\n"})
rc, out = _verify("--repo", repo) rc, out = _verify("--repo", repo)
check("9 rc 3 -> rc 2 (retired code)", rc, 2) check("9 rc 3 -> rc 2 (retired code)", rc, 2)
@@ -665,30 +684,31 @@ def selftest():
# 10. a hung leg is a timeout, not a pass and not a fault # 10. a hung leg is a timeout, not a pass and not a fault
repo = _fixture(os.path.join(td, "c10"), repo = _fixture(os.path.join(td, "c10"),
dict(GREEN5, **{"selftest_sweep.py": 0}), sleep=5.0) dict(GREEN_ALL, **{"selftest_sweep.py": 0}), sleep=5.0)
rc, out = _verify("--repo", repo, "--timeout", "1") rc, out = _verify("--repo", repo, "--timeout", "1")
check("10 timeout -> rc 2", rc, 2) check("10 timeout -> rc 2", rc, 2)
check("10b the timeout is named", "TIMEOUT after 1s" in out, True) 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 # 11. order is fixed, cheapest first: the table is not re-sorted by outcome
repo = _fixture(os.path.join(td, "c11"), repo = _fixture(os.path.join(td, "c11"),
dict(GREEN5, **{"queue_index.py": 1}), dict(GREEN_ALL, **{"queue_index.py": 1}),
outs={"queue_index.py": "STALE: line 1\n"}) outs={"queue_index.py": "STALE: line 1\n"})
rc, out = _verify("--repo", repo) rc, out = _verify("--repo", repo)
order = [out.find(k) for k in ("checkin", "reply-headings", "index-current", order = [out.find(k) for k in ("checkin", "reply-headings", "index-current",
"append-audit", "selftest-sweep")] "append-audit", "selftest-sweep",
"write-axis")]
check("11 legs print in fixed order", order == sorted(order) check("11 legs print in fixed order", order == sorted(order)
and -1 not in order, True) and -1 not in order, True)
# 12. a skip is counted as a skip, never folded into pass # 12. a skip is counted as a skip, never folded into pass
repo = _fixture(os.path.join(td, "c12"), GREEN5) repo = _fixture(os.path.join(td, "c12"), GREEN_ALL)
jpath = os.path.join(td, "c12.json") jpath = os.path.join(td, "c12.json")
rc, out = _verify("--repo", repo, "--fast", "--json", jpath) rc, out = _verify("--repo", repo, "--fast", "--json", jpath)
obj = json.load(open(jpath)) if os.path.exists(jpath) else {} obj = json.load(open(jpath)) if os.path.exists(jpath) else {}
check("12 skipped leg is its own count in --json", check("12 skipped leg is its own count in --json",
obj.get("counts", {}).get("SKIP"), 1) obj.get("counts", {}).get("SKIP"), 1)
check("12b and the skipped leg is not counted as a pass", check("12b and the skipped leg is not counted as a pass",
obj.get("counts", {}).get(PASS), 4) obj.get("counts", {}).get(PASS), 5)
# 13. the detail line must not contradict the verdict: a row graded # 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. # FAULT beside "0 fatal" is the SILENT failure this gate exists to catch.
@@ -705,7 +725,7 @@ def selftest():
# 14. THE HOLE THIS LEG CLOSES: a contract-breaking check-in is rc 1, # 14. THE HOLE THIS LEG CLOSES: a contract-breaking check-in is rc 1,
# and the gate is no longer green on a tree that carries one. # and the gate is no longer green on a tree that carries one.
repo = _fixture(os.path.join(td, "c14"), repo = _fixture(os.path.join(td, "c14"),
dict(GREEN5, **{"checkin_check.py": 1}), dict(GREEN_ALL, **{"checkin_check.py": 1}),
outs={"checkin_check.py": outs={"checkin_check.py":
"FAIL: 2 file(s) violate the check-in contract\n"}) "FAIL: 2 file(s) violate the check-in contract\n"})
rc, out = _verify("--repo", repo, "--fast") rc, out = _verify("--repo", repo, "--fast")
@@ -715,7 +735,7 @@ def selftest():
and "fault: checkin" in out, True) and "fault: checkin" in out, True)
# 15. no checkins/ directory is ungradable, not a pass and not a fault # 15. no checkins/ directory is ungradable, not a pass and not a fault
repo = _fixture(os.path.join(td, "c15"), GREEN5, checkins=None) repo = _fixture(os.path.join(td, "c15"), GREEN_ALL, checkins=None)
rc, out = _verify("--repo", repo, "--fast") rc, out = _verify("--repo", repo, "--fast")
check("15 no checkins dir -> rc 2", rc, 2) check("15 no checkins dir -> rc 2", rc, 2)
check("15b the ungradable row says which directory is missing", check("15b the ungradable row says which directory is missing",
@@ -724,7 +744,7 @@ def selftest():
# 16. the leg grades the WHOLE archive against its own ledger -- the # 16. the leg grades the WHOLE archive against its own ledger -- the
# fix for a newest-day floor that made a backdated fault invisible. # fix for a newest-day floor that made a backdated fault invisible.
av = os.path.join(td, "c16-argv") av = os.path.join(td, "c16-argv")
repo = _fixture(os.path.join(td, "c16"), GREEN5, repo = _fixture(os.path.join(td, "c16"), GREEN_ALL,
argv_markers={"checkin_check.py": av}, argv_markers={"checkin_check.py": av},
checkins=("2026-01-01-fixture.md", "2026-03-04-fixture.md")) checkins=("2026-01-01-fixture.md", "2026-03-04-fixture.md"))
rc, out = _verify("--repo", repo, "--fast") rc, out = _verify("--repo", repo, "--fast")
@@ -737,7 +757,7 @@ def selftest():
# 17. --checkin-day still narrows the window explicitly, in the open # 17. --checkin-day still narrows the window explicitly, in the open
av = os.path.join(td, "c17-argv") av = os.path.join(td, "c17-argv")
repo = _fixture(os.path.join(td, "c17"), GREEN5, repo = _fixture(os.path.join(td, "c17"), GREEN_ALL,
argv_markers={"checkin_check.py": av}, argv_markers={"checkin_check.py": av},
checkins=("2026-01-01-fixture.md", "2026-03-04-fixture.md")) checkins=("2026-01-01-fixture.md", "2026-03-04-fixture.md"))
rc, out = _verify("--repo", repo, "--fast", "--checkin-day", "2026-01-01") rc, out = _verify("--repo", repo, "--fast", "--checkin-day", "2026-01-01")
@@ -749,7 +769,7 @@ def selftest():
# the tool's own reason has to reach the row, because "no ledger" and # the tool's own reason has to reach the row, because "no ledger" and
# "clean archive" print nothing alike and must not grade alike. # "clean archive" print nothing alike and must not grade alike.
repo = _fixture(os.path.join(td, "c19"), repo = _fixture(os.path.join(td, "c19"),
dict(GREEN5, **{"checkin_check.py": 2}), dict(GREEN_ALL, **{"checkin_check.py": 2}),
outs={"checkin_check.py": outs={"checkin_check.py":
"cannot grade: baseline %s/tools/checkin_baseline.tsv: " "cannot grade: baseline %s/tools/checkin_baseline.tsv: "
"[Errno 2] No such file or directory\n" "[Errno 2] No such file or directory\n"
@@ -785,7 +805,7 @@ def selftest():
# FAULT on the gate and the row is named -- the class that was invisible # FAULT on the gate and the row is named -- the class that was invisible
# for as long as `replies/` had no reader. # for as long as `replies/` had no reader.
repo = _fixture(os.path.join(td, "c20"), repo = _fixture(os.path.join(td, "c20"),
dict(GREEN5, **{"reply_check.py": 1}), dict(GREEN_ALL, **{"reply_check.py": 1}),
outs={"reply_check.py": outs={"reply_check.py":
"reply_check: 192 file(s), 1 new fault(s), " "reply_check: 192 file(s), 1 new fault(s), "
"0 known-faulting in the baseline\n"}) "0 known-faulting in the baseline\n"})
@@ -796,7 +816,7 @@ def selftest():
# 21. the ledger is read from the tree being graded, not from the host # 21. the ledger is read from the tree being graded, not from the host
av = os.path.join(td, "c21-argv") av = os.path.join(td, "c21-argv")
repo = _fixture(os.path.join(td, "c21"), GREEN5, repo = _fixture(os.path.join(td, "c21"), GREEN_ALL,
argv_markers={"reply_check.py": av}) argv_markers={"reply_check.py": av})
rc, out = _verify("--repo", repo, "--fast") rc, out = _verify("--repo", repo, "--fast")
seen = open(av).read() if os.path.exists(av) else "" seen = open(av).read() if os.path.exists(av) else ""
@@ -823,7 +843,7 @@ def selftest():
# test this gate used until 2026-10-19 -- answered "not a git repository" # test this gate used until 2026-10-19 -- answered "not a git repository"
# (rc 2) on a checkout every leg would have graded. Measured RED against # (rc 2) on a checkout every leg would have graded. Measured RED against
# the old test on this exact fixture, GREEN here. # the old test on this exact fixture, GREEN here.
repo = _fixture(os.path.join(td, "c23"), GREEN5, commit=True) repo = _fixture(os.path.join(td, "c23"), GREEN_ALL, commit=True)
wt = os.path.join(td, "c23-wt") wt = os.path.join(td, "c23-wt")
subprocess.run(["git", "-C", repo, "worktree", "add", "--detach", wt, subprocess.run(["git", "-C", repo, "worktree", "add", "--detach", wt,
"HEAD"], check=True, capture_output=True, text=True) "HEAD"], check=True, capture_output=True, text=True)
tools/write_probe.py
+376
View File
@@ -0,0 +1,376 @@
#!/usr/bin/env python3
"""Measure the WRITE end of every tool in tools/: `--json PATH` at a target the
host cannot write.
WHY THIS EXISTS
`selftest_sweep.py`'s read axis grades what a tool does with an input it
cannot read. Nothing graded the other end. Measured at `67f94cc`, the last
revision before the guards landed, over the 20 tools then tracked: **18
answered a `--json` target inside a nonexistent directory with an uncaught
FileNotFoundError at rc 1**; `gate_blame.py` returned rc 2 from its own input
and never reached the write (the UNNAMED class below, ungraded rather than
passed); `queue_append_audit.py` already refused (rc 2, "cannot write json").
A traceback is what this repo's contract calls fatal, and rc 1 is the code
that means a graded fault, so those 18 made "the grade is bad"
indistinguishable from "the artifact was never written". Two earlier drafts
of this paragraph said "five" and "13 of 20"; neither is what the instrument
below prints, and the number that counts is the one a reader can reproduce:
`git worktree add --detach <dir> 67f94cc && cp tools/write_probe.py
<dir>/tools/ && python3 <dir>/tools/write_probe.py --repo <dir>`. The guards
landed that day; this instrument is what proves them and what a new tool has
to pass -- including this one, which probes itself.
THE CONTRACT IT GRADES
The grade runs first and its stdout report stands. The write is wrapped, so a
target this host cannot write prints one line naming the path on stderr and
returns 2 -- "could not grade": the artifact was asked for and not delivered.
rc 2 outranks the 0/1 the report may have earned, because a caller who asked
for a file and got a stack trace cannot know a grade happened at all.
VERDICTS, IN THIS ORDER (a traceback is a defect whatever the exit code)
TRACEBACK uncaught exception text in either stream
RC_OTHER clean stream, rc != 2 -- the write failed and the tool did not
say so in the one code that means "could not grade"
SILENT rc 2 with no message at all -- a refusal nobody can read
UNNAMED rc 2, a message, but the message never names the target
OK rc 2, one line naming the target, no traceback
NO_REACH the tool never got as far as the write on this probe: its own
rc 2 came from the input, so the write path is UNGRADED here
(counted, never passed -- an ungraded path is not a clean one)
NO_JSON no --json argument in the source: nothing to probe
FIXTURES
Every probe runs in a throwaway tree, so the grade each tool performs is over
a few bytes rather than over this repository: the write path is what is being
graded, not the tool's subject matter. `{target}` is the only path that does
not exist -- it must not.
READ-ONLY on the repo: the only files created are under a TemporaryDirectory.
USAGE
python3 tools/write_probe.py [--repo DIR] [--only TOOL] [--timeout S]
[--json PATH] [--verbose]
python3 tools/write_probe.py --selftest # 10 cells, judge, table, fixture
EXIT
0 every probed tool is OK; 1 at least one is not (including NO_REACH and
NO_JSON: an ungraded write path may not be reported clean); 2 could not
grade -- no tracked tools discovered, or --json itself unwritable.
"""
import argparse
import json
import os
import re
import subprocess
import sys
import tempfile
JSON_FLAG = re.compile(r"add_argument\(\s*[\"']--json[\"']")
TRACEBACK = "Traceback (most recent call last)"
# (tool, argv) -- `{...}` placeholders are filled by fixture_paths(). One entry
# per tracked tool: a tool the table does not name is NO_PROBE (fatal), on the
# same rule as the read axis -- an ungraded path may not pass.
WRITE_PROBES = (
("append_only_audit.py", ["--repo", "{tinyrepo}", "--json", "{target}"]),
("append_only_owners.py", ["{audit}", "--json", "{target}"]),
("census_row_delta.py", ["{census_a}", "{census_b}", "--json", "{target}"]),
("check_lint.py", ["--check", "echo ok", "--json", "{target}"]),
("checkin_check.py", ["--dir", "{checkins}", "--roster", "volt",
"--json", "{target}"]),
("citation_hash_class.py", ["--repo", "{tinyrepo}", "--json", "{target}"]),
("citation_resolve_check.py", ["--repo", "{tinyrepo}", "--json", "{target}"]),
("doc_census_check.py", ["--repo", "{tinyrepo}", "--no-run",
"--json", "{target}"]),
("doc_command_check.py", ["--repo", "{tinyrepo}", "--json", "{target}"]),
("gate_blame.py", ["--repo", "{tinyrepo}", "--leg", "selftest-sweep",
"--range", "1", "--json", "{target}"]),
("landing_check.py", ["--repo", "{tinyrepo}", "--json", "{target}"]),
("path_claim_check.py", ["--repo", "{tinyrepo}", "--json", "{target}"]),
("queue_append_audit.py", ["--repo", "{tinyrepo}", "--json", "{target}"]),
("queue_claim_check.py", ["--queue", "{queue}", "--json", "{target}"]),
("queue_index.py", ["--queue", "{queue}", "--json", "{target}"]),
("reply_check.py", ["--dir", "{replies}", "--json", "{target}"]),
("selftest_sweep.py", ["--repo", "{tinyrepo}", "--json", "{target}"]),
("served_drift_check.py", ["--root", "{tinyrepo}", "--base-url", "{baseurl}",
"--json", "{target}"]),
("tool_map.py", ["--repo", "{tinyrepo}", "--json", "{target}"]),
("verify.py", ["--repo", "{tinyrepo}", "--fast", "--json", "{target}"]),
# It grades itself: run over the one-tool fixture repo, this probe reports
# that fixture tool as NO_PROBE (rc 1), then turns to its own --json write,
# which is the path being graded. Committing the instrument without this
# row would make the table's own NO_PROBE rule fail on its author.
("write_probe.py", ["--repo", "{tinyrepo}", "--json", "{target}"]),
)
def _write(path, text):
d = os.path.dirname(path)
if d and not os.path.isdir(d):
os.makedirs(d, exist_ok=True)
with open(path, "w", encoding="utf-8") as fh:
fh.write(text)
def build_fixtures(td, real_repo):
"""A one-commit git repo, one fake instrument, one reply, one check-in, one
log, one queue and two census files: enough for every tool in the table to
earn a 0 or a 1 and then turn to its write. Sizes are deliberately tiny."""
p = {}
tiny = os.path.join(td, "tiny")
os.makedirs(os.path.join(tiny, "docs"), exist_ok=True)
os.makedirs(os.path.join(tiny, "checkins"), exist_ok=True)
os.makedirs(os.path.join(tiny, "replies"), exist_ok=True)
os.makedirs(os.path.join(tiny, "tools"), exist_ok=True)
_write(os.path.join(tiny, "README.md"), "# tiny\n\nA fixture repo.\n")
_write(os.path.join(tiny, "WORK-QUEUE.md"),
"# Work queue\n\n**Q1 -- OPEN (pipe, 2026-01-01). One item.**\n\n"
"Body.\n")
_write(os.path.join(tiny, "docs", "queue-index.md"),
"# Index\n\n| id | state |\n| --- | --- |\n| Q1 | OPEN |\n")
_write(os.path.join(tiny, "tools", "fake_tool.py"),
'#!/usr/bin/env python3\n"""fake_tool.py -- a fake instrument.\n\n'
'USAGE:\n python3 fake_tool.py --selftest # 1 cells\n"""\n'
'import sys\nif "--selftest" in sys.argv:\n'
' print("selftest: 1 cells, 1 passed, 0 failed")\n'
' sys.exit(0)\nprint("fake")\n')
_write(os.path.join(tiny, "checkins", "2026-01-01-volt.md"),
"# volt check-in 2026-01-01\n\n"
"- Built: the fixture.\n- Next: nothing.\n")
_write(os.path.join(tiny, "replies", "2026-01-01-to-merlin-01.md"),
"# reply 01 to merlin\n\n## 1. Subject\n\nBody text, one finding.\n\n"
"-- volt\n")
for cmd in (["init", "-q", "-b", "main"], ["add", "-A"],
["-c", "user.email=v@example.invalid",
"-c", "user.name=volt", "commit", "-q", "-m", "fixture"]):
subprocess.run(["git", "-C", tiny] + cmd, capture_output=True, text=True)
p["tinyrepo"] = tiny
p["queue"] = os.path.join(tiny, "WORK-QUEUE.md")
p["index"] = os.path.join(tiny, "docs", "queue-index.md")
p["replies"] = os.path.join(tiny, "replies")
p["checkins"] = os.path.join(tiny, "checkins")
p["toolsdir"] = os.path.join(tiny, "tools")
p["audit"] = os.path.join(td, "audit.json")
_write(p["audit"], json.dumps({"rows": []}) + "\n")
p["log"] = os.path.join(td, "log.txt")
_write(p["log"], "checked 2026-01-01 Q1 clean\n")
for key in ("census_a", "census_b"):
p[key] = os.path.join(td, key + ".json")
_write(p[key], json.dumps({"rev": "abc1234", "repo": tiny,
"reports": [], "rows": []}) + "\n")
p["baseurl"] = "file://" + tiny
p["target"] = os.path.join(td, "no-such-dir", "out.json")
# The probe is meaningless if the target's parent exists: an existing
# directory would let the write succeed and grade the wrong thing.
assert not os.path.isdir(os.path.dirname(p["target"]))
p["repo"] = real_repo
return p
def judge(rc, out, err, target):
text = (out or "") + (err or "")
if TRACEBACK in text:
tail = [l for l in text.splitlines() if l.strip()][-1:]
return {"verdict": "TRACEBACK", "rc": rc,
"detail": (tail[0] if tail else "")[:110]}
if rc != 2:
first = next((l.strip() for l in (err or out or "").splitlines()
if l.strip()), "")
return {"verdict": "RC_OTHER", "rc": rc,
"detail": (first or "rc %s, no message" % rc)[:110]}
msg = [l for l in (err or "").splitlines() if l.strip()] or \
[l for l in (out or "").splitlines() if l.strip()]
if not msg:
return {"verdict": "SILENT", "rc": rc, "detail": "rc 2 with no message"}
if target not in text:
return {"verdict": "UNNAMED", "rc": rc,
"detail": "rc 2 but the message never names the target: "
+ msg[-1][:64]}
if "cannot write" not in text and "cannot be written" not in text:
# rc 2 and the path is named, but not as a write refusal: the tool
# refused its INPUT. The write path is therefore ungraded here.
return {"verdict": "NO_REACH", "rc": rc,
"detail": "rc 2 on the input, never reached the write: "
+ msg[-1][:64]}
return {"verdict": "OK", "rc": rc, "detail": msg[-1][:110]}
def run_probe(path, argv, timeout, cwd):
try:
p = subprocess.run([sys.executable, path] + argv, cwd=cwd,
capture_output=True, text=True, errors="replace",
timeout=timeout)
return p.returncode, p.stdout or "", p.stderr or ""
except subprocess.TimeoutExpired:
return None, "", "TIMEOUT after %ss" % timeout
except OSError as exc:
return None, "", "<exec failed: %s>" % exc
def tracked_tools(repo):
p = subprocess.run(["git", "-C", repo, "ls-tree", "-r", "--name-only",
"HEAD", "tools/"], capture_output=True, text=True)
if p.returncode != 0:
return None, (p.stderr or "git ls-tree failed").strip()
return [os.path.basename(x) for x in p.stdout.splitlines()
if x.endswith(".py")], "git ls-tree HEAD tools/"
def selftest(repo):
"""Ten cells: six for the judge's ordered verdicts, four for the table and
the fixture. The judge cells matter because the order is the contract -- a
traceback outranks a right code, and a rc-2 input refusal (NO_REACH) must
never read as a passed write path.
"""
bad = []
def cell(name, ok, why=""):
print((" ok " if ok else " FAIL ") + name
+ ("" if ok else " [%s]" % why))
if not ok:
bad.append(name)
t = os.path.join("no-such-dir", "out.json")
tb = TRACEBACK + "\n File \"x.py\", line 1\nFileNotFoundError: [Errno 2]"
cell("c1 a traceback is TRACEBACK whatever the exit code",
judge(2, "", tb, t)["verdict"] == "TRACEBACK")
cell("c2 a clean rc 1 is RC_OTHER, not a pass",
judge(1, "", "cannot write --json " + t + "\n", t)["verdict"] == "RC_OTHER")
cell("c3 rc 2 with no message is SILENT",
judge(2, "", "", t)["verdict"] == "SILENT")
cell("c4 rc 2 that never names the target is UNNAMED",
judge(2, "", "gate_blame: cannot grade: all 1 commit(s) UNGRADABLE\n",
t)["verdict"] == "UNNAMED")
cell("c5 rc 2 naming the target but refusing INPUT is NO_REACH (ungraded)",
judge(2, "", "cannot read " + t + ": no such file\n",
t)["verdict"] == "NO_REACH")
cell("c6 rc 2 naming the target as a WRITE refusal is OK",
judge(2, "", "cannot write --json " + t + ": [Errno 2]\n",
t)["verdict"] == "OK")
names, source = tracked_tools(repo)
table = dict(WRITE_PROBES)
dupes = len([n for n, _ in WRITE_PROBES]) - len(table)
cell("c7 the table names every tracked tool once, and no tool it cannot see",
names is not None and not dupes and set(table) == set(names),
"%s: table-only=%s tool-only=%s dupes=%s"
% (source, sorted(set(table) - set(names or [])),
sorted(set(names or []) - set(table)), dupes))
with tempfile.TemporaryDirectory(prefix="write-selftest-") as td:
fx = build_fixtures(td, repo)
cell("c8 the probe target's parent does not exist (or the write would succeed)",
not os.path.isdir(os.path.dirname(fx["target"])), fx["target"])
if names and "check_lint.py" in names:
rc, out, err = run_probe(os.path.join(repo, "tools", "check_lint.py"),
["--check", "echo ok", "--json", fx["target"]],
120, fx["tinyrepo"])
v = judge(rc, out, err, fx["target"])
cell("c9 a real tool, probed: rc 2, one line naming the target, no traceback",
v["verdict"] == "OK", str(v))
else:
cell("c9 a real tool, probed: rc 2, one line naming the target", False,
"check_lint.py not discovered under %s" % repo)
rows = [{"tool": "x", "verdict": "OK"}, {"tool": "y", "verdict": "NO_PROBE"}]
cell("c10 NO_PROBE counts as a fault, so an ungraded path cannot pass",
len([r for r in rows if r["verdict"] != "OK"]) == 1)
print("selftest: 10 cells, %d passed, %d failed" % (10 - len(bad), len(bad)))
return 1 if bad else 0
def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--repo", default="/root/troupe-standup")
ap.add_argument("--only", default=None, metavar="TOOL")
ap.add_argument("--timeout", type=int, default=180, metavar="S")
ap.add_argument("--json", default=None, metavar="PATH")
ap.add_argument("--selftest", action="store_true",
help="prove the judge and the table, then exit")
ap.add_argument("--verbose", action="store_true")
a = ap.parse_args(argv)
repo = os.path.abspath(a.repo)
if a.selftest:
return selftest(repo)
names, source = tracked_tools(repo)
if names is None:
print("cannot read: " + source, file=sys.stderr)
return 2
if not names:
print("cannot read: no tracked tools under " + repo, file=sys.stderr)
return 2
table = dict(WRITE_PROBES)
rows = []
with tempfile.TemporaryDirectory(prefix="write-probe-") as td:
fx = build_fixtures(td, repo)
for name in names:
if a.only and name != a.only:
continue
path = os.path.join(repo, "tools", name)
try:
with open(path, encoding="utf-8") as fh:
src = fh.read()
except OSError as exc:
rows.append({"tool": name, "verdict": "UNREADABLE", "rc": None,
"detail": str(exc)[:110]})
continue
if name not in table:
rows.append({"tool": name, "verdict": "NO_PROBE", "rc": None,
"detail": "no write probe in the table -- ungraded"})
continue
if not JSON_FLAG.search(src):
rows.append({"tool": name, "verdict": "NO_JSON", "rc": None,
"detail": "no --json argument in source"})
continue
argv = [re.sub(r"\{(\w+)\}", lambda m: fx[m.group(1)], x)
for x in table[name]]
rc, out, err = run_probe(path, argv, a.timeout, fx["tinyrepo"])
rec = {"tool": name, "probe": table[name]}
rec.update(judge(rc, out, err, fx["target"]))
rows.append(rec)
if a.verbose:
print(" . %-24s %-9s rc=%s" % (name, rec["verdict"], rc))
if not rows:
print("cannot read: no tool matched --only " + str(a.only),
file=sys.stderr)
return 2
bad = [r for r in rows if r["verdict"] != "OK"]
width = max([len(r["tool"]) for r in rows] + [4])
print("write probe -- %d of %d tracked tool(s) probed under %s (%s)"
% (len(rows), len(names), os.path.join(repo, "tools"), source))
print()
print("%-*s %-9s %4s detail" % (width, "tool", "verdict", "rc"))
for r in rows:
print("%-*s %-9s %4s %s"
% (width, r["tool"], r["verdict"],
"-" if r.get("rc") is None else r["rc"], r["detail"]))
print()
print("write axis: %d probed, %d ok, %d fault(s)"
% (len(rows), len(rows) - len(bad), len(bad))
+ ("" if not bad else " (" + ", ".join(sorted({r["verdict"]
for r in bad})) + ")"))
if a.json:
payload = {"repo": repo, "discovery": source, "probed": len(rows),
"ok": len(rows) - len(bad), "faults": len(bad),
"tools": rows}
try:
with open(a.json, "w", encoding="utf-8") as fh:
json.dump(payload, fh, indent=1, sort_keys=True)
fh.write("\n")
except OSError as exc:
print("cannot write --json " + a.json + ": " + str(exc),
file=sys.stderr)
return 2
return 1 if bad else 0
if __name__ == "__main__":
sys.exit(main())