#!/usr/bin/env python3 """selftest_sweep.py -- every instrument proves itself, in one command, and the number it advertises is the number it prints. WHY THIS EXISTS `tools/append_only_audit.py` carried, on its own usage line, the words "synthetic repo, N cells" beside a selftest that printed a different N. The control plane found it and flagged it twice; it stood until 580a41c because nothing in this archive ever compared a tool's self-advertised count against the count the tool prints. `doc_census_check.py` now checks that class -- but its subject is the census blocks (AMBIGUOUS / MATCHES / NOT_PRINTED / ...) and it drives every tool against a revision, so a full run is minutes, not seconds. The selftest cell count is a cheaper and more basic claim: it is the number a reader uses to decide whether a tool is worth trusting, and the number a check-in quotes as proof. Eight tools carry one. One was wrong for days. This is the gate for that sentence, and it runs in seconds because a selftest is a unit test, not a census. WHAT IT MEASURES -- one row per tracked tool, one verdict per row NO_SELFTEST_FLAG the source never mentions "--selftest"; the tool is not run (a tool without the flag may do real work on `--selftest`). TIMEOUT the selftest did not finish inside --timeout seconds. RUN_FAILED non-zero exit and no count line: the run died, or failed before it could report. NO_COUNT_LINE exit 0 but nothing matching `selftest: cells|fixtures, passed, failed`. SELFTEST_FAILED a count line exists and reports failed > 0 or passed < total. SPLIT two different values of one unit on the tool's own `--selftest` usage line(s) -- the four-hundred-and-seventy-one over-count defect, at selftest scale. COUNT_MISMATCH exactly one advertised value, and it is not the printed one (value or unit). This is the append_only_audit defect. NO_CLAIM prints a count, advertises none. ADVISORY, not a fault: an absent claim cannot be false. OK printed a count, every advertised value agrees with it. FATAL = every class above except NO_CLAIM. Exit 1 if any row is fatal, else 0. THE READ-REFUSAL AXIS, ADDED 2026-10-17 The same command asks every tool the second question this archive kept getting wrong: what exit code does it give an input it cannot read? The contract is three codes and no others -- rc 0 ran clean, rc 1 ran and graded a fault, rc 2 could not grade. The axis runs each tool against a path that does not exist (and, for the four file readers, against bytes that are not UTF-8) and grades the answer OK / RC_NOT_2 / TRACEBACK / SILENT; the last three are fatal. So is NO_PROBE (below): a tool the tables never name is a tool this axis never graded, and an ungraded tool may not be reported clean. rc 3 is retired: the two queue readers used it for "cannot 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. 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 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 sense of -- a `.json` that is not JSON, a valid `{}` where an audit file should be, a directory that is not a git repository, a queue file with no records. Those are the inputs where a tool is most tempted to answer "0" and return 0, which reads exactly like a clean pass (`doc_census_check`'s `--tools` bug, `76043b7`, was this class). The axis runs one such probe per tool and grades each against ITS OWN stated answer, because "must exit 2" is not a universal truth: most tools must refuse at rc 2, while `check_lint` grades a junk log MISSING (rc 0 in report mode, rc 1 under --fail-on-fault), `checkin_check` grades a heading-less check-in as an advisory (rc 0, stated in its docstring) and `queue_claim_check` reports 0 opening records (rc 0, its --fail-on-stale is the gate). A silent 0 cannot pass as a refusal: every probe must print something. WHAT A REFUSAL IS GRADED ON, FIXED 2026-10-27 (Q93) Until this date each probe carried a phrase its tool was required to print ("not a git", "cannot read repo"), so the axis graded WORDING. Two tools that refused correctly -- rc 2, one line, no traceback -- were held red because their message was the boundary guard ("... at HEAD: Stopping at filesystem boundary") instead of the phrase the table pinned, and the axis would have gone green if either tool's prose had been degraded to match. A refusal is now graded on the CONTRACT: the documented rc, no traceback, and a non-empty message. Where a tool GRADES a malformed input instead (rc 0 or 1 with a verdict word) the word is still required, because there the word IS the verdict -- MISSING, EMPTY, "0 opening records". Which STREAM carries a refusal is not part of the contract either, and that is measured rather than assumed: `tool_map.py` prints its refusal on stdout (100 B, rc 2) while the two citation tools print on stderr (139 B, rc 2). The trade is stated in the table's own comment: the contract keeps "refused, legibly, at the documented code" and stops keeping "refused for the reason the table remembers". WHAT A REFUSAL MUST NOT BE, ADDED 2026-10-29 (Q97) The contract above has one blind spot and it is not theoretical: Python's `argparse` writes `usage: ...` and `prog: error: unrecognized arguments: ...` to stderr and exits 2, which satisfies "rc 2, no traceback, a message" exactly. Every probe invokes its tool with flags, so a tool whose flag is renamed -- or removed -- never reads the input at all, and an axis that exists to grade malformed-input handling reads clean. Measured at 9878f12: renaming `--repo` to `--repo-dir` in `citation_hash_class.py` alone left `malformed-input: 26 probes, 26 ok` and `verdict: rc 0 -- clean`. A refusal is now ALSO required to be about the input: a message that leads with `usage:` or carries an argparse failure shape (`unrecognized arguments`, `error: argument`, `the following arguments are required`, `invalid choice:`, `ambiguous option:`, `no such option`) is ECHOES_USAGE -- fatal on both input axes, the read axis included, because that axis has had the same hole since it was written (it pinned no prose at all). The check is negative on purpose: requiring the message to name the probe's path would redden tools that refuse correctly with a boundary line, which is the Q93 defect coming back the other way. One measurement the filed record does not carry: its mutant renames `--repo` to `--repo-dir` with the same dest, and that rename is INVISIBLE to argparse -- a long option may be abbreviated to any unambiguous prefix, so `--repo` still parses as `--repo-dir` and the tool reads its input normally (measured: same refusal line, rc 2, flag and renamed flag alike). The filed mutant therefore stays green for a reason other than the one the item names. Cell c48 and the verification below use a rename that is not a prefix (`--tree`), which is the shape that actually reaches the parser. ONE VERDICT LINE, ADDED 2026-10-27 (Q93) The leg printed `sweep: 22 tools, 22 OK, 0 fatal` and returned **rc 1**: the sweep line is about the TOOLS, the exit code is about the RUN, and nothing printed the run's own disposition -- so the summary read clean while the gate was red, and the two faults were named only under --verbose. Every run now ends with one line from `run_verdict()`, the same function whose return value is the exit code, so the line and the code cannot disagree: `verdict: rc 0 -- clean: tools 0 fatal, read-refusal 0 fault, ...` `verdict: rc 1 -- red: malformed-input 2 fault (tools 0 fatal, ...)` and every fault in an axis summary is named on that summary's own line as `VERDICT tool/probe`, so UNEXPLAINED never appears without the probe and tool that earned it. THE COUNT PIN, ADDED 2026-10-22 -- the docstring was not a witness Every verdict above compares a tool's printed count with a number in the SAME file. That catches a stale number; it does not catch a shrinking corpus, because a commit that deletes three fixtures and decrements the docstring agrees with itself, and every leg stays green. Measured on a clone of this repository: `doc_command_check.py` cut from 15 cells to 12 at both the print and the usage line reads `OK 12 cells` and `sweep: 21 tools, 21 OK, 0 fatal, rc 0` -- a third of its fixtures gone, reported clean. `tools/selftest_baseline.tsv` is the second witness: one row per tool, the count its selftest prints, committed and read from disk. The sweep compares every row against it and prints four classes -- PIN_OK / PIN_DRIFT (printed differs from pinned) / PIN_UNPINNED (a tool with no pinned row) / PIN_ORPHAN (a pinned row whose tool is gone). DRIFT, UNPINNED and ORPHAN are FATAL: a corpus may not shrink silently, and a new tool may not arrive unmeasured. PIN_NA (a tool with no --selftest) and PIN_NO_COUNT (its run already failed above) are printed and not counted twice. A missing baseline in repo mode is rc 2 -- could not grade -- never a clean run; `--no-baseline` is the explicit way to sweep without one, and `--write-baseline PATH` re-pins after a deliberate change. WHAT COUNTS AS A CLAIM, PRECISELY Only the module docstring (read through `ast.get_docstring`, so a fixture string in the BODY is not a claim -- `doc_census_check.py` contains the literal "selftest: N cells, N passed, N failed" inside a test, and reading the file as text would call that a claim), and only on a line that also contains the token `--selftest`. Prose elsewhere in the docstring is not a claim: the same doc_census_check docstring quotes a thirteen-cell claim as the defect the tool itself catches, and a scanner that flagged its own bug report would be wrong. DISCOVERY, AND WHY UNTRACKED FILES ARE PRINTED Tracked paths only, from `git ls-files` in --repo -- the lesson doc_census_check paid for: an untracked tool is invisible to a revision-scoped census, and this gate inherits that blindness rather than pretending to fix it. Every `tools/*.py` that IS present but untracked is therefore listed as SKIPPED_UNTRACKED with its name, never silently dropped. With --tools-dir, or with no git, discovery is a plain glob of that directory and the row says so. READ-ONLY: the only write path is `--json PATH`. It does not edit a tool, and it does not decide that a claim SHOULD exist -- NO_CLAIM is advisory by design. NO_PROBE is not: an ungraded tool cannot be reported clean (see the read axis). USAGE: python3 selftest_sweep.py --repo /root/troupe-standup python3 selftest_sweep.py --repo /root/troupe-standup --write-baseline tools/selftest_baseline.tsv python3 selftest_sweep.py --tools-dir /tmp/fake --json /tmp/sweep.json python3 selftest_sweep.py --selftest # selftest: 49 cells, parser, sweep, pin and both input axes """ from __future__ import annotations import argparse import ast import contextlib import glob import io import json import os import re import subprocess import sys import tempfile from collections import Counter # The default target is the checkout this file lives in -- the same path in # place, a different path in a copy. An absolute literal here would make a bare # `--repo`-less run grade /root/troupe-standup from anywhere on the host, which # is the wrong-tree defect `tool_map.py` documents at length. DEFAULT_REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) UNITS = ("cells", "fixtures") # `selftest: 18 cells, 18 passed, 0 failed` # `self-test: 19 fixtures, 19 passed, 0 failed` # `selftest: 14 fixtures (10 claim-check, 4 unit), 14 passed, 0 failed` PRINTED_RE = re.compile( r"(?:selftest|self-test)\s*:\s*([\d,]+)\s+(cells|fixtures)\b" r"[^\n]*?\b([\d,]+)\s+passed\b[^\n]*?\b([\d,]+)\s+failed\b" ) CLAIM_RE = re.compile(r"(? int: return int(text.replace(",", "")) def parse_printed(text: str): """(count, unit, passed, failed) from the first count line, or None.""" m = PRINTED_RE.search(text or "") if not m: return None return _int(m.group(1)), m.group(2), _int(m.group(3)), _int(m.group(4)) def usage_claims(doc: str): """[(value, unit, line)] -- claims on lines that name --selftest.""" out = [] for line in (doc or "").splitlines(): if "--selftest" not in line: continue for m in CLAIM_RE.finditer(line): out.append((_int(m.group(1)), m.group(2), line.strip())) return out def module_docstring(path: str): """(docstring, source). The docstring is AST-read so body fixtures are not claims.""" try: with open(path, "r", encoding="utf-8", errors="replace") as fh: src = fh.read() except OSError as exc: return None, f"" try: tree = ast.parse(src) except SyntaxError: return None, src return (ast.get_docstring(tree) or ""), src # ---------------------------------------------------------------- discovery def discover(repo: str, tools_dir: str | None): """(rows, source, untracked). rows: [(display_name, path)] -- path is absolute, display_name is tools/x.py. """ if tools_dir: d = os.path.abspath(tools_dir) paths = sorted(p for p in glob.glob(os.path.join(d, "*.py")) if os.path.isfile(p)) return ([(os.path.basename(p), p) for p in paths], f"glob of --tools-dir {d} (no git consulted)", []) d = os.path.join(os.path.abspath(repo), "tools") have_git = subprocess.run(["git", "-C", os.path.abspath(repo), "rev-parse", "--git-dir"], capture_output=True).returncode == 0 if not have_git: paths = sorted(p for p in glob.glob(os.path.join(d, "*.py")) if os.path.isfile(p)) return ([(os.path.basename(p), p) for p in paths], f"glob of {d} (not a git tree)", []) tracked = subprocess.run( ["git", "-C", os.path.abspath(repo), "ls-files", "--", "tools"], capture_output=True, text=True).stdout.splitlines() named = [t for t in tracked if t.startswith("tools/") and t.endswith(".py") and "/" not in t[6:]] rows = [(os.path.basename(t), os.path.join(os.path.abspath(repo), t)) for t in sorted(named)] untracked = subprocess.run( ["git", "-C", os.path.abspath(repo), "ls-files", "--others", "--exclude-standard", "--", "tools"], capture_output=True, text=True).stdout.splitlines() stray = [t for t in untracked if t.startswith("tools/") and t.endswith(".py") and "/" not in t[6:]] rev = subprocess.run(["git", "-C", os.path.abspath(repo), "rev-parse", "--short", "HEAD"], capture_output=True, text=True).stdout.strip() return rows, f"git ls-files at {rev or 'HEAD'} ({len(rows)} tracked in tools/)", sorted(stray) # ------------------------------------------------------------------- judging def judge(doc, out: str, rc: int, timed_out: bool): """One verdict dict from a completed run. Pure: no I/O, no clock.""" printed = parse_printed(out) claims = usage_claims(doc or "") rec = { "printed": ({"count": printed[0], "unit": printed[1], "passed": printed[2], "failed": printed[3]} if printed else None), "claims": [{"value": v, "unit": u, "line": ln} for v, u, ln in claims], "verdict": None, "detail": "", } if timed_out: rec["verdict"] = "TIMEOUT" rec["detail"] = "selftest did not finish" return rec if printed is None: if rc != 0: rec["verdict"] = "RUN_FAILED" tail = [l for l in (out or "").splitlines() if l.strip()][-1:] or [""] rec["detail"] = f"rc={rc}; last line: {tail[0][:160]}" else: rec["verdict"] = "NO_COUNT_LINE" rec["detail"] = "rc=0 but no ' cells|fixtures, passed, failed' line" return rec count, unit, passed, failed = printed if failed > 0 or passed < count: rec["verdict"] = "SELFTEST_FAILED" rec["detail"] = f"{passed}/{count} passed, {failed} failed" return rec same_unit = [(v, ln) for v, u, ln in claims if u == unit] values = sorted({v for v, _ in same_unit}) if len(values) > 1: rec["verdict"] = "SPLIT" rec["detail"] = ("advertises " + " and ".join(f"{v} {unit}" for v in values) + f"; prints {count}") return rec if not claims: rec["verdict"] = "NO_CLAIM" rec["detail"] = f"prints {count} {unit}; advertises none (advisory)" return rec if values and values[0] == count: rec["verdict"] = "OK" rec["detail"] = f"{count} {unit}" return rec rec["verdict"] = "COUNT_MISMATCH" adv = ", ".join(f"{v} {u}" for v, u, _ in claims) rec["detail"] = f"advertises {adv}; prints {count} {unit}" return rec def sweep(rows, repo: str, timeout: int, verbose: bool): results = [] for name, path in rows: doc, src = module_docstring(path) rec = {"tool": name, "path": path} if doc is None and not SELFTEST_FLAG_RE.search(src or ""): rec.update({"verdict": "UNREADABLE", "detail": "source unreadable", "printed": None, "claims": []}) results.append(rec) continue if not SELFTEST_FLAG_RE.search(src or ""): rec.update({"verdict": "NO_SELFTEST_FLAG", "printed": None, "claims": [{"value": v, "unit": u, "line": ln} for v, u, ln in usage_claims(doc or "")], "detail": "no \"--selftest\" in source; not run"}) results.append(rec) continue timed_out = False rc, out = 0, "" try: proc = subprocess.run([sys.executable, path, "--selftest"], cwd=repo, capture_output=True, text=True, errors="replace", timeout=timeout) rc, out = proc.returncode, (proc.stdout or "") + (proc.stderr or "") except subprocess.TimeoutExpired: timed_out = True except OSError as exc: rc, out = 127, f"" verdict = judge(doc, out, rc, timed_out) rec.update(verdict) rec["rc"] = rc results.append(rec) if verbose: print(f" . {name}: {rec['verdict']} rc={rc}") return results # --------------------------------------------------------------------- report def report(results, source, untracked, quiet=False): counts = Counter(r["verdict"] for r in results) # one census pass (was: fatal list-comp + ok sum = two full scans of results) if not quiet: print(f"discovery: {source}") for name in untracked: print(f" SKIPPED_UNTRACKED {name} (present in tools/, not in the tree)") width = max([len(r["tool"]) for r in results] + [4]) print() print(f"{'tool':<{width}} {'verdict':<16} {'claim':>12} {'printed':>12} detail") for r in results: claims = r.get("claims") or [] claim_s = (", ".join(f"{c['value']} {c['unit']}" for c in claims) if claims else "-") p = r.get("printed") printed_s = f"{p['count']} {p['unit']}" if p else "-" print(f"{r['tool']:<{width}} {r['verdict']:<16} {claim_s:>12} " f"{printed_s:>12} {r['detail']}") print() n_fatal = sum(counts[v] for v in FATAL) fatal_verdicts = sorted(v for v in counts if v in FATAL) print(f"sweep: {len(results)} tools, {counts['OK']} OK, {n_fatal} fatal" + (f" ({', '.join(fatal_verdicts)})" if n_fatal else "")) return 1 if n_fatal else 0, counts["OK"], n_fatal # --------------------------------------------------------------- count pin # WHY THIS IS A FILE. Every verdict in `report()` compares the tool's printed # count with a number inside the tool. That is a real check on a stale number # and no check at all on a shrinking corpus: delete three fixtures, decrement # the docstring, and the file agrees with itself. Measured on a clone of this # repository at 2026-10-22 -- `doc_command_check.py` cut 15 cells -> 12 on both # the print and the usage line still reads `OK 12 cells`, `sweep: 21 tools, 21 # OK, 0 fatal`, rc 0. The ledger is the witness the tool cannot edit in the # same commit without the change being visible in the diff. def default_baseline(repo: str) -> str: return os.path.join(os.path.abspath(repo), "tools", BASELINE_NAME) def load_baseline(path: str): """({tool: (count, unit)}, None) or (None, reason). Blank and # lines skipped.""" rows = {} try: with open(path, "r", encoding="utf-8") as fh: lines = fh.read().splitlines() except OSError as exc: return None, f"cannot read baseline {path}: {exc}" for n, line in enumerate(lines, 1): if not line.strip() or line.lstrip().startswith("#"): continue parts = line.split("\t") if len(parts) != 3: return None, (f"{path}:{n}: want 3 tab-separated columns " f"(tool, count, unit), got {len(parts)}") name, count_s, unit = (p.strip() for p in parts) if unit not in UNITS: return None, f"{path}:{n}: unit {unit!r} is not one of {UNITS}" try: count = _int(count_s) except ValueError: return None, f"{path}:{n}: count {count_s!r} is not an integer" if name in rows: return None, f"{path}:{n}: duplicate row for {name}" rows[name] = (count, unit) if not rows: return None, f"{path}: no pinned rows" return rows, None def render_baseline(results) -> str: """The ledger text for these results. Rows the selftest never printed are skipped here and named by grade_pin() instead -- a pin cannot invent a number the tool did not print.""" head = [ "# selftest_baseline.tsv -- the cell count each tool's selftest prints, pinned.", "# Written by: python3 tools/selftest_sweep.py --repo . --write-baseline tools/selftest_baseline.tsv", "#", "# WHY THIS FILE EXISTS: the count a tool advertises lives in its own docstring,", "# so a commit that deletes fixtures and decrements that number agrees with", "# itself and every leg stays green. This ledger is the outside witness.", "# `selftest_sweep.py` reads it on every run: a printed count that differs is", "# PIN_DRIFT and fatal; a tool with no row is PIN_UNPINNED and fatal; a row", "# whose tool is gone is PIN_ORPHAN and fatal. Re-pin on purpose, in the diff.", "#", "# toolcountunit", ] for r in sorted(results, key=lambda r: r["tool"]): p = r.get("printed") if not p: continue head.append(f"{r['tool']}\t{p['count']}\t{p['unit']}") return "\n".join(head) + "\n" def pin_faults(rows): return [r for r in rows if r["verdict"] in PIN_FAULT] def grade_pin(results, baseline): """One row per tool plus one per orphaned pin. Pure: no I/O, no clock.""" rows = [] for r in sorted(results, key=lambda r: r["tool"]): name, printed = r["tool"], r.get("printed") pin = baseline.get(name) pinned = {"count": pin[0], "unit": pin[1]} if pin else None seen = ({"count": printed["count"], "unit": printed["unit"]} if printed else None) row = {"tool": name, "pinned": pinned, "printed": seen} if r.get("verdict") == "NO_SELFTEST_FLAG": row.update(verdict="PIN_NA", detail="no --selftest to pin; not counted") elif pin is None: row.update(verdict="PIN_UNPINNED", detail=(f"prints {seen['count']} {seen['unit']}; no pinned row" if seen else "no pinned row, and no count to pin")) elif seen is None: row.update(verdict="PIN_NO_COUNT", detail=(f"pinned {pin[0]} {pin[1]}; the run printed no count " "(graded above)")) elif (seen["count"], seen["unit"]) == (pin[0], pin[1]): row.update(verdict="PIN_OK", detail=f"{pin[0]} {pin[1]}") else: row.update(verdict="PIN_DRIFT", detail=(f"pinned {pin[0]} {pin[1]}; prints " f"{seen['count']} {seen['unit']}")) rows.append(row) seen_tools = {r["tool"] for r in results} for name in sorted(set(baseline) - seen_tools): rows.append({"tool": name, "pinned": {"count": baseline[name][0], "unit": baseline[name][1]}, "printed": None, "verdict": "PIN_ORPHAN", "detail": (f"pinned {baseline[name][0]} {baseline[name][1]}; " "no tool of that name at this rev")}) return rows def report_pin(rows, path, quiet=False): bad = pin_faults(rows) if not quiet: print() print(f"count pin: {path} -- the number each selftest prints, pinned " "outside the tool it grades") width = max([len(r["tool"]) for r in rows] + [4]) for r in rows: pin, seen = r["pinned"], r["printed"] pin_s = f"{pin['count']} {pin['unit']}" if pin else "-" seen_s = f"{seen['count']} {seen['unit']}" if seen else "-" print(f"{r['tool']:<{width}} {r['verdict']:<13} " f"pinned {pin_s:>12} printed {seen_s:>12} {r['detail']}") agree = sum(1 for r in rows if r["verdict"] == "PIN_OK") pinned = sum(1 for r in rows if r["pinned"]) print(f"pin: {pinned} pinned, {agree} agree, {len(bad)} drift" + (f" ({', '.join(sorted({r['verdict'] for r in bad}))})" if bad else "")) return (1 if bad else 0), len(bad) # ------------------------------------------------------- read-refusal axis def _first_line(text: str) -> str: for line in (text or "").splitlines(): if line.strip(): return line.strip()[:96] return "" def echoes_invocation(text): """True when a rc-2 message is argparse failing to PARSE THE INVOCATION rather than the tool refusing the input (Q97, 2026-10-29). Two shapes, both measured from real argparse output rather than invented: the message's first non-empty line leads with `usage:`, or it carries one of ARGV_ECHO_MARKERS. The test is negative -- a refusal is red for what it IS, not for what a table remembers it saying (that was Q93's defect) -- and it is deliberately not "must name the probe's path", because two tools in this directory refuse correctly with a boundary line that names no path.""" low = (text or "").lower() if any(m in low for m in ARGV_ECHO_MARKERS): return True for line in (text or "").splitlines(): if line.strip(): return line.strip().lower().startswith("usage:") return False def judge_read(rc, text): """One probe's verdict. Three classes, in this order on purpose: a traceback is a defect whatever its exit code, and a silent rc 2 is a refusal an operator cannot read.""" if "Traceback (most recent call last)" in (text or ""): return {"verdict": "TRACEBACK", "rc": rc, "detail": _first_line(text)} if rc != READ_AXIS_UNIT: return {"verdict": "RC_NOT_2", "rc": rc, "detail": _first_line(text) or f"rc {rc}, no output"} if not (text or "").strip(): return {"verdict": "SILENT", "rc": rc, "detail": "rc 2 with no message"} if echoes_invocation(text): return {"verdict": "ECHOES_USAGE", "rc": rc, "detail": "argparse failure, not an input refusal: " + _first_line(text)} return {"verdict": "OK", "rc": rc, "detail": _first_line(text)} def run_read_probe(path: str, argv, timeout: int): """(rc, combined output) for one probe, run in a throwaway cwd.""" with tempfile.TemporaryDirectory() as td: binpath = os.path.join(td, "not-utf8.bin") with open(binpath, "wb") as fh: fh.write(b"\xff\xfe\x00binary\x00\n") bindir = os.path.join(td, "not-utf8-dir") os.makedirs(bindir, exist_ok=True) with open(os.path.join(bindir, "2026-01-01-volt.md"), "wb") as fh: fh.write(b"\xff\xfe\x00binary\x00\n") nope = os.path.join(td, "there-is-no-such-path") # the malformed class: each of these DECODES, none of it parses notrepo = os.path.join(td, "not-a-repo") os.makedirs(notrepo, exist_ok=True) with open(os.path.join(notrepo, "README.txt"), "w") as fh: fh.write("files, but no .git here\n") paths = {"badjson": os.path.join(td, "bad.json"), "emptyjson": os.path.join(td, "empty.json"), "junklog": os.path.join(td, "log.txt"), "junkqueue": os.path.join(td, "q.md"), "junkindex": os.path.join(td, "i.md")} bodies = {"badjson": "not json at all\n", "emptyjson": "{}\n", "junklog": "garbage line\nnot a lint entry\n", "junkqueue": "this is not a queue\nno headings here at all\nQ\n", "junkindex": "not an index either\n"} for key, body in bodies.items(): with open(paths[key], "w", encoding="utf-8") as fh: fh.write(body) badcheckins = os.path.join(td, "checkins") os.makedirs(badcheckins, exist_ok=True) with open(os.path.join(badcheckins, "2026-01-01-volt.md"), "w", encoding="utf-8") as fh: fh.write("no heading, no template\n") # {junkmanifest} = an artifact ROOT holding one artifact that presents a # files.sha256 with no parsable entry. Decodes as UTF-8 and is the # wrong SHAPE, which is what puts it on this axis rather than the # read-refusal one: the tool must reach EMPTY, not rc 2. junkmanifest = os.path.join(td, "junk-root") os.makedirs(os.path.join(junkmanifest, "art"), exist_ok=True) with open(os.path.join(junkmanifest, "art", "files.sha256"), "w", encoding="utf-8") as fh: fh.write("not a hash, not an entry\n") repl = {"{nope}": nope, "{bin}": binpath, "{bindir}": bindir, "{notrepo}": notrepo, "{badcheckins}": badcheckins, "{junkmanifest}": junkmanifest} repl.update({"{" + k + "}": v for k, v in paths.items()}) real = [] for a in argv: for key, val in repl.items(): a = a.replace(key, val) real.append(a) try: p = subprocess.run([sys.executable, path] + real, cwd=td, capture_output=True, text=True, errors="replace", timeout=timeout) except subprocess.TimeoutExpired: return None, "", "" except OSError as exc: return 127, "", f"" return p.returncode, (p.stdout or ""), (p.stderr or "") def judge_malformed(rc, out, err, expect_rc, need): """One malformed probe's verdict, against the answer its tool states. Graded in this order: a traceback is a defect whatever the code; a code that is not the documented one is the defect this class exists to catch (the silent rc 0); a refusal with no message cannot be acted on; and a right code with the wrong verdict word is not the documented answer. A REFUSAL (expect_rc == 2) is graded on the contract, not on prose (Q93): the documented code, no traceback, one non-empty message on either stream. Since Q97 (2026-10-29) the message must also NOT be argparse failing to parse the invocation -- that is ECHOES_USAGE, fatal, because the input was never read. Which stream is not part of the contract -- measured 2026-10-27, `tool_map.py` writes its refusal to stdout and `citation_hash_class.py` to stderr, both at rc 2, both correct. `need` is only consulted where the tool GRADED the malformed input (rc 0/1): there the word is its verdict.""" text = (out or "") + (err or "") if "Traceback (most recent call last)" in text: return {"verdict": "TRACEBACK", "rc": rc, "detail": _first_line(text)} if rc != expect_rc: return {"verdict": "RC_WRONG", "rc": rc, "detail": "rc %s, contract says %s: %s" % (rc, expect_rc, _first_line(text) or "no output")} if not text.strip(): return {"verdict": "SILENT", "rc": rc, "detail": "rc %s with no message" % rc} if expect_rc == REFUSAL_RC: if echoes_invocation(text): # Q97: rc 2 from the argument parser is not a refusal of the input. return {"verdict": "ECHOES_USAGE", "rc": rc, "contract": "refusal", "detail": "argparse failure, not an input refusal: " + _first_line(text)} # a refusal: the code and a readable line are the whole contract return {"verdict": "OK", "rc": rc, "contract": "refusal", "detail": "refusal rc 2: " + _first_line(text)} if need and need not in text: return {"verdict": "UNEXPLAINED", "rc": rc, "contract": "grade", "detail": "wanted %r, got: %s" % (need, _first_line(text))} return {"verdict": "OK", "rc": rc, "contract": "grade", "detail": _first_line(text)} def axis_faults(axis): """The rows that make the axis red. NO_PROBE is one of them: a tool with no probe entry has not been graded, and 'not graded' is not 'clean'.""" return [r for r in axis if r["verdict"] in FATAL_READ] def read_axis(rows, timeout: int): """One row per (tool, probe). Tools the sweep discovered are probed by path; a probe naming a tool that was not discovered is NO_TOOL (fatal -- the table and the probes have drifted apart), and a discovered tool the tables never name is NO_PROBE on each axis (also fatal: the tool is uncovered).""" paths = dict(rows) out = [] for probe, specs in (("missing", READ_PROBES), ("not-utf8", NOT_UTF8_PROBES)): for name, argv in specs: rec: dict = {"tool": name, "probe": probe} if name not in paths: rec.update({"verdict": "NO_TOOL", "rc": None, "detail": "probe names a tool the sweep did not discover"}) else: rc, sout, serr = run_read_probe(paths[name], argv, timeout) rec.update(judge_read(rc, (sout or "") + (serr or ""))) out.append(rec) for name, label, argv, expect_rc, need in MALFORMED_PROBES: rec = {"tool": name, "probe": label, "expect_rc": expect_rc} if name not in paths: rec.update({"verdict": "NO_TOOL", "rc": None, "detail": "probe names a tool the sweep did not discover"}) else: rc, sout, serr = run_read_probe(paths[name], argv, timeout) rec.update(judge_malformed(rc, sout, serr, expect_rc, need)) out.append(rec) probed = {n for n, _ in READ_PROBES} malformed = {t for t, *_ in MALFORMED_PROBES} for name, _p in rows: if name not in probed: out.append({"tool": name, "probe": "missing", "verdict": NO_PROBE, "rc": None, "detail": "no read probe defined -- ungraded, fatal"}) if name not in malformed: out.append({"tool": name, "probe": "malformed", "verdict": NO_PROBE, "rc": None, "detail": "no malformed probe defined -- ungraded, fatal"}) return out def report_read(axis, quiet=False): """Print each probe row, then one summary per axis, then the run's faults by name. A faulting axis names every row that earned it on the SAME line (Q93): `2 fault (UNEXPLAINED citation_hash_class.py/not-a-repo, ...)` -- the pre-fix summary printed the verdict word alone, so a reader of the gate output could not tell which probe, or which tool, to look at.""" fatal = axis_faults(axis) hard = [r for r in axis if r["probe"] in ("missing", "not-utf8")] soft = [r for r in axis if r["probe"] not in ("missing", "not-utf8")] if not quiet: print() print(f"read-refusal axis: an unreadable input exits {READ_AXIS_UNIT}, " "never 1 (a graded fault)") width = max([len(r["tool"]) for r in axis] + [4]) for r in axis: rc = "-" if r["rc"] is None else str(r["rc"]) print(f"{r['tool']:<{width}} {r['probe']:<15} rc={rc:<4} " f"{r['verdict']:<11} {r['detail']}") bad_by_axis = [] for label, group in (("read-refusal", hard), ("malformed-input", soft)): ok = sum(1 for r in group if r["verdict"] == "OK") bad = [r for r in group if r["verdict"] in FATAL_READ] bad_by_axis.append(len(bad)) named = ", ".join(f"{r['verdict']} {r['tool']}/{r['probe']}" for r in bad) print(f"{label}: {len(group)} probes, {ok} ok, {len(bad)} fault" + (f" ({named})" if bad else "")) return (1 if fatal else 0), bad_by_axis[0], bad_by_axis[1] def run_verdict(tool_fatal, read_fatal, malformed_fatal, pin_rc, pin_fatal): """The run's one-line verdict AND its exit code, from one function (Q93). Before this existed the sweep printed `sweep: 22 tools, 22 OK, 0 fatal` -- true of the TOOLS -- and returned 1 because an input axis was red. Nothing printed the run's own disposition, so the summary read clean while the gate was red. With the line and the code computed here, they cannot disagree: `rc 1` in the line is the code the process exits with, and every red axis is named with its count on that line.""" axes = [("tools", tool_fatal, "fatal"), ("read-refusal", read_fatal, "fault"), ("malformed-input", malformed_fatal, "fault"), ("pin", pin_fatal, "drift")] census = ", ".join(f"{name} {n} {unit}" for name, n, unit in axes) red = [(name, n, unit) for name, n, unit in axes if n] if pin_rc == 2: return f"verdict: rc 2 -- could not grade: pin ungradable ({census})", 2 if red: return ("verdict: rc 1 -- red: " + ", ".join(f"{name} {n} {unit}" for name, n, unit in red) + f" ({census})"), 1 return f"verdict: rc 0 -- clean: {census}", 0 # ------------------------------------------------------------------- selftest def selftest(): bad = [] ran = [] def cell(name, ok, detail=""): ran.append(name) print(f"{'PASS' if ok else 'FAIL'} {name}" + (f" [{detail}]" if (detail and not ok) else "")) if not ok: bad.append(name) cell("c1 parse plain cells line", parse_printed("selftest: 18 cells, 18 passed, 0 failed") == (18, "cells", 18, 0)) cell("c2 parse fixtures with parenthetical", parse_printed("selftest: 14 fixtures (10 claim-check, 4 unit), 14 passed, 0 failed") == (14, "fixtures", 14, 0)) cell("c3 parse hyphenated self-test", parse_printed("self-test: 19 fixtures, 19 passed, 0 failed") == (19, "fixtures", 19, 0)) cell("c4 parse thousands separator", parse_printed("selftest: 1,204 cells, 1,204 passed, 0 failed") == (1204, "cells", 1204, 0)) cell("c5 no count line -> None", parse_printed("all good, everything fine") is None) doc = ("USAGE:\n python3 t.py --selftest # synthetic repo, 18 cells\n") cell("c6 claim on the --selftest usage line", usage_claims(doc) and usage_claims(doc)[0][:2] == (18, "cells")) prose = (" returned a docstring claim of `13 cells` against a selftest that\n" " prints `18`\n") cell("c7 prose beside the word selftest is not a claim", usage_claims(prose) == []) cell("c8 claim with a unit the tool never prints is still a claim", usage_claims(" --selftest # 7 fixtures\n")[0][:2] == (7, "fixtures")) j = judge(doc, "selftest: 13 cells, 13 passed, 0 failed", 0, False) cell("c9 advertised 18 vs printed 13 -> COUNT_MISMATCH", j["verdict"] == "COUNT_MISMATCH", j["verdict"]) j = judge(doc, "selftest: 18 cells, 18 passed, 0 failed", 0, False) cell("c10 advertised 18 vs printed 18 -> OK", j["verdict"] == "OK", j["verdict"]) j = judge(" --selftest # 7 fixtures\n", "selftest: 7 cells, 7 passed, 0 failed", 0, False) cell("c11 unit mismatch is a mismatch, not a pass", j["verdict"] == "COUNT_MISMATCH", j["verdict"]) j = judge(" --selftest # 18 cells\n --selftest # 13 cells\n", "selftest: 18 cells, 18 passed, 0 failed", 0, False) cell("c12 two advertised values -> SPLIT", j["verdict"] == "SPLIT", j["verdict"]) j = judge("", "selftest: 8 cells, 7 passed, 1 failed", 0, False) cell("c13 failed>0 is fatal even at rc=0", j["verdict"] == "SELFTEST_FAILED", j["verdict"]) j = judge("", "Traceback (most recent call last):", 1, False) cell("c14 rc!=0 with no count line -> RUN_FAILED", j["verdict"] == "RUN_FAILED", j["verdict"]) j = judge("", "", 0, False) cell("c15 rc=0 with no count line -> NO_COUNT_LINE", j["verdict"] == "NO_COUNT_LINE", j["verdict"]) # c16/c17: end to end through main(), on a directory of fake instruments. good = ('#!/usr/bin/env python3\n"""fake_good.py -- a fake.\n\nUSAGE:\n' ' python3 fake_good.py --selftest # 7 cells\n"""\n' 'import sys\nif "--selftest" in sys.argv:\n' ' print("selftest: 7 cells, 7 passed, 0 failed")\n sys.exit(0)\n') bad_tool = ('#!/usr/bin/env python3\n"""fake_bad.py -- a fake.\n\nUSAGE:\n' ' python3 fake_bad.py --selftest # 13 cells\n"""\n' 'import sys\nif "--selftest" in sys.argv:\n' ' print("selftest: 18 cells, 18 passed, 0 failed")\n sys.exit(0)\n') flagless = ('#!/usr/bin/env python3\n"""fake_flagless.py -- no selftest."""\n' 'print("did some work")\n') with tempfile.TemporaryDirectory() as td: for nm, body in (("fake_good.py", good), ("fake_bad.py", bad_tool), ("fake_flagless.py", flagless)): with open(os.path.join(td, nm), "w") as fh: fh.write(body) rc_bad = main(["--tools-dir", td, "--quiet"]) rows, source, untracked = discover(td, td) res = sweep(rows, td, 30, False) verdicts = {r["tool"]: r["verdict"] for r in res} cell("c16 end to end: a wrong claim is fatal", rc_bad == 1 and verdicts.get("fake_bad.py") == "COUNT_MISMATCH", f"rc={rc_bad} {verdicts}") cell("c17 end to end: correct claim OK, flagless caught, exit 1", verdicts.get("fake_good.py") == "OK" and verdicts.get("fake_flagless.py") == "NO_SELFTEST_FLAG", str(verdicts)) jd = os.path.join(td, "sweep.json") main(["--tools-dir", td, "--quiet", "--json", jd]) try: with open(jd) as fh: blob = json.load(fh) ok_json = (isinstance(blob, dict) and len(blob.get("tools", [])) == 3 and blob.get("fatal") == 2) except (OSError, ValueError): ok_json = False cell("c18 --json carries every row and the fatal count", ok_json) # c19-c22: the read-refusal axis's judgement, one cell per class. The axis # is the gate on "an input I cannot read must not exit 1"; these cells are # the gate on the gate. jr = judge_read(2, "cannot read census /nope/a.json: [Errno 2]") cell("c19 a refusal at rc 2 with a message -> OK", jr["verdict"] == "OK", jr["verdict"]) jr = judge_read(1, "MISSING: the check has a fault") cell("c20 a graded fault's rc 1 on an unreadable input -> RC_NOT_2", jr["verdict"] == "RC_NOT_2", jr["verdict"]) jr = judge_read(2, "Traceback (most recent call last):\n File x") cell("c21 a traceback is fatal even when it exits 2", jr["verdict"] == "TRACEBACK", jr["verdict"]) jr = judge_read(2, "") cell("c22 rc 2 with no message -> SILENT", jr["verdict"] == "SILENT", jr["verdict"]) # c23-c28: the malformed class's judgement, and its coverage. A decodable # input has no single right answer -- each tool's contract decides -- so # these cells pin the judgement itself: the documented rc passes, a wrong # code fails, a refusal must print something, and a tool that GRADED the # input must print the verdict word it claims. jm = judge_malformed(0, "0 opening records; 0 closed; 0 open.", "", 0, "0 opening records") cell("c23 a documented rc 0 grade on malformed input -> OK", jm["verdict"] == "OK", jm["verdict"]) jm = judge_malformed(0, "no tools found at HEAD", "", 2, None) cell("c24 the silent-zero: rc 0 where the contract says 2 -> RC_WRONG", jm["verdict"] == "RC_WRONG", jm["verdict"]) jm = judge_malformed(2, " ", "", 2, None) cell("c25 a refusal with no message -> SILENT", jm["verdict"] == "SILENT", jm["verdict"]) # c26 is Q93's defect, at the judge: before this, the refusal below was # UNEXPLAINED because its line is not the phrase the table pinned -- a # correct refusal held red for its wording. jm = judge_malformed(2, "cannot read /tmp/x at HEAD: Stopping at filesystem boundary", "", 2, None) cell("c26 a refusal at rc 2 whose prose is not the pinned phrase -> OK (contract)", jm["verdict"] == "OK" and jm.get("contract") == "refusal", jm["verdict"]) jm = judge_malformed(2, "Traceback (most recent call last):", "", 2, None) cell("c27 a traceback is fatal even at the documented code", jm["verdict"] == "TRACEBACK", jm["verdict"]) missing_probed = {t for t, _ in READ_PROBES} malformed_probed = {t for t, *_ in MALFORMED_PROBES} cell("c28 every tool with a missing-probe also has a malformed probe", missing_probed == malformed_probed, str(sorted(missing_probed ^ malformed_probed))) # c29/c30: the NO_PROBE hole, measured from the other side -- a tool the # tables never name must come out uncovered on BOTH axes and must make the # axis red. Before this cell existed, read_axis emitted two advisory rows and # a 15-tool directory graded "14 OK, 0 fatal, rc 0". alone = read_axis([("zzz_unprobed.py", "/nonexistent/zzz_unprobed.py")], 5) mine = [r for r in alone if r["tool"] == "zzz_unprobed.py"] cell("c29 an unprobed tool is NO_PROBE on both axes and is a fault", sorted(r["probe"] for r in mine) == ["malformed", "missing"] and len(axis_faults(mine)) == 2, str([(r["probe"], r["verdict"]) for r in mine])) cell("c30 every FATAL_READ verdict is red and OK is not", all(axis_faults([{"verdict": v}]) for v in FATAL_READ) and not axis_faults([{"verdict": "OK"}]) and NO_PROBE in FATAL_READ, str(FATAL_READ)) # c31-c37: the count pin. The defect these grade is the one the classes # above cannot see: a corpus that shrinks while every number in the tool # agrees with every other number in the tool. pin_one = {"fake_good.py": (7, "cells")} pr = grade_pin([{"tool": "fake_good.py", "verdict": "OK", "printed": {"count": 7, "unit": "cells"}}], pin_one) cell("c31 a printed count that matches its pinned row -> PIN_OK", pr[0]["verdict"] == "PIN_OK" and not pin_faults(pr), pr[0]["verdict"]) pr = grade_pin([{"tool": "fake_good.py", "verdict": "OK", "printed": {"count": 4, "unit": "cells"}}], pin_one) cell("c32 four cells against a pin of seven -> PIN_DRIFT, fatal", pr[0]["verdict"] == "PIN_DRIFT" and len(pin_faults(pr)) == 1, f"{pr[0]['verdict']} {pr[0]['detail']}") pr = grade_pin([{"tool": "new_tool.py", "verdict": "OK", "printed": {"count": 3, "unit": "cells"}}], pin_one) mine = [r for r in pr if r["tool"] == "new_tool.py"] cell("c33 a tool with no pinned row -> PIN_UNPINNED, fatal", mine and mine[0]["verdict"] == "PIN_UNPINNED" and mine[0] in pin_faults(pr), mine[0]["verdict"] if mine else "no row") pr = grade_pin([], pin_one) cell("c34 a pinned row whose tool is gone -> PIN_ORPHAN, fatal", pr[0]["verdict"] == "PIN_ORPHAN" and len(pin_faults(pr)) == 1, pr[0]["verdict"]) pr = grade_pin([{"tool": "flagless.py", "verdict": "NO_SELFTEST_FLAG", "printed": None}], pin_one) mine = [r for r in pr if r["tool"] == "flagless.py"] cell("c35 a tool with no --selftest is PIN_NA, not a fault", mine and mine[0]["verdict"] == "PIN_NA" and mine[0] not in pin_faults(pr), mine[0]["verdict"] if mine else "no row") with tempfile.TemporaryDirectory() as td: bp = os.path.join(td, "pin.tsv") blob = render_baseline([{"tool": "b.py", "printed": {"count": 9, "unit": "fixtures"}}, {"tool": "a.py", "printed": {"count": 2, "unit": "cells"}}]) with open(bp, "w", encoding="utf-8") as fh: fh.write(blob) round_trip, why = load_baseline(bp) cell("c36 render/load round-trip keeps both rows and their units", round_trip == {"b.py": (9, "fixtures"), "a.py": (2, "cells")}, f"{round_trip} {why}") with open(bp, "w", encoding="utf-8") as fh: fh.write("a.py\tnine\tcells\n") bad_pin, why = load_baseline(bp) cell("c37 a non-integer count is refused, naming the line", bad_pin is None and ":1:" in (why or "") and "not an integer" in (why or ""), str(why)) with tempfile.TemporaryDirectory() as td: for nm, body in (("fake_good.py", good), ("fake_flagless.py", flagless)): with open(os.path.join(td, nm), "w") as fh: fh.write(body) bp = os.path.join(td, "pin.tsv") buf = io.StringIO() with contextlib.redirect_stdout(buf): main(["--tools-dir", td, "--write-baseline", bp, "--quiet"]) written, why = load_baseline(bp) cell("c38 --write-baseline pins exactly what the run printed", written == {"fake_good.py": (7, "cells")}, f"{written} {why}") with open(bp, "w", encoding="utf-8") as fh: fh.write("fake_good.py\t9\tcells\n") buf = io.StringIO() with contextlib.redirect_stdout(buf): rc_pin = main(["--tools-dir", td, "--baseline", bp, "--quiet"]) out = buf.getvalue() cell("c39 end to end: a drifted pin names PIN_DRIFT and turns the run red", "PIN_DRIFT" in out and "1 drift (PIN_DRIFT)" in out and rc_pin == 1, (out.strip().splitlines() or ["no output"])[-1]) # c40-c44: Q93 -- the axis must fail for the RIGHT reason, and the run's # verdict must be readable from its own line. The two shapes below are the # negative cells Q93 asks for, run through the real subprocess path rather # than the judge alone: a tool that refuses at rc 2 with prose nobody # pinned (OK -- the contract holds), and a tool that answers 0 on a # malformed input (RC_WRONG, fatal). jm = judge_malformed(1, "artifact root: /x\nEMPTY", "", 1, "EMPTY") jm2 = judge_malformed(1, "artifact root: /x\nSOMETHING", "", 1, "EMPTY") cell("c40 a GRADED row still needs its own verdict word -> UNEXPLAINED", jm["verdict"] == "OK" and jm2["verdict"] == "UNEXPLAINED", f"{jm['verdict']}/{jm2['verdict']}") with tempfile.TemporaryDirectory() as td: refuses = os.path.join(td, "refuses_oddly.py") with open(refuses, "w", encoding="utf-8") as fh: fh.write("import sys\n" "sys.stderr.write('cannot read at HEAD: Stopping at filesystem " "boundary\\n')\n" "sys.exit(2)\n") rc_e, out_e, err_e = run_read_probe(refuses, ["{junkqueue}"], 30) jm = judge_malformed(rc_e, out_e, err_e, 2, None) cell("c41 end to end: a refusal at rc 2 with unpinned prose -> OK (contract)", jm["verdict"] == "OK" and rc_e == 2, f"{jm['verdict']} rc={rc_e}") answers = os.path.join(td, "answers_zero.py") with open(answers, "w", encoding="utf-8") as fh: fh.write("print('0 opening records; all fine')\n" "raise SystemExit(0)\n") rc_e, out_e, err_e = run_read_probe(answers, ["{junkqueue}"], 30) jm = judge_malformed(rc_e, out_e, err_e, 2, None) cell("c42 end to end: a tool that exits 0 on a malformed input -> RC_WRONG, fatal", jm["verdict"] == "RC_WRONG" and axis_faults([jm]) == [jm], f"{jm['verdict']} rc={rc_e}") axis_fake = [{"tool": "t.py", "probe": "not-a-repo", "verdict": "UNEXPLAINED", "rc": 2, "expect_rc": 2, "detail": "wanted x, got: y"}, {"tool": "u.py", "probe": "junk-queue", "verdict": "OK", "rc": 0, "expect_rc": 0, "detail": "0 opening records"}] buf = io.StringIO() with contextlib.redirect_stdout(buf): rrc_, hard_, soft_ = report_read(axis_fake, quiet=True) line = buf.getvalue() cell("c43 a faulting axis names its probe and tool on the summary line", "malformed-input: 2 probes, 1 ok, 1 fault (UNEXPLAINED t.py/not-a-repo)" in line and "read-refusal: 0 probes, 0 ok, 0 fault" in line and (rrc_, hard_, soft_) == (1, 0, 1), line.strip().splitlines()[-1] if line.strip() else "no output") ok_line, ok_rc = run_verdict(0, 0, 0, 0, 0) red_line, red_rc = run_verdict(0, 0, 2, 0, 0) bad_line, bad_rc = run_verdict(0, 0, 0, 2, 0) def _stated(ln): return int(ln.split("rc ", 1)[1].split()[0]) cell("c44 the verdict line states the rc the same call returns", (ok_rc, red_rc, bad_rc) == (0, 1, 2) and [_stated(ok_line), _stated(red_line), _stated(bad_line)] == [0, 1, 2] and "malformed-input 2 fault" in red_line and "pin ungradable" in bad_line, f"{ok_line} | {red_line} | {bad_line}") # c45-c49: Q97 -- a refusal that is argparse failing to parse the # INVOCATION is not a refusal of the input. c45/c46 are the judge on both # axes; c47 is the control (a refusal that names its input stays OK); c48 # is end to end through run_read_probe against a stub whose REAL argparse # renames the flag the probe passes -- the filed mutant, shrunk to one # cell; c49 is the summary line naming the fault. jm = judge_malformed( 2, "", "usage: t.py [-h] --repo-dir REPO_DIT\nt.py: error: " "unrecognized arguments: --repo /tmp/x\n", 2, None) cell("c45 argparse usage at rc 2 on the malformed axis -> ECHOES_USAGE, fatal", jm["verdict"] == "ECHOES_USAGE" and axis_faults([jm]) == [jm], jm["verdict"]) jr = judge_read(2, "usage: t.py [-h] --queue QUEUE\nt.py: error: the " "following arguments are required: --queue\n") cell("c46 argparse usage at rc 2 on the read axis -> ECHOES_USAGE, fatal", jr["verdict"] == "ECHOES_USAGE" and axis_faults([jr]) == [jr], jr["verdict"]) jm = judge_malformed(2, "", "cannot read /tmp/nope: [Errno 2] No such file " "or directory\n", 2, None) jr = judge_read(2, "cannot read /tmp/nope: [Errno 2] No such file or directory") cell("c47 a refusal that names its input stays OK on both axes", jm["verdict"] == "OK" and jr["verdict"] == "OK", f"{jm['verdict']}/{jr['verdict']}") with tempfile.TemporaryDirectory() as td: stub = os.path.join(td, "renamed_flag.py") with open(stub, "w", encoding="utf-8") as fh: fh.write("import argparse\n" "ap = argparse.ArgumentParser()\n" # deliberately NOT a prefix of --repo: argparse accepts an # unambiguous abbreviation, so --repo -> --repo-dir still # parses and reads the input (measured, Q97). "ap.add_argument('--tree', dest='repo')\n" "a = ap.parse_args()\n" "print('cannot read ' + a.repo)\n" "raise SystemExit(2)\n") rc_m, out_m, err_m = run_read_probe(stub, ["--repo", "{nope}"], 30) jm_bad = judge_malformed(rc_m, out_m, err_m, 2, None) rc_c, out_c, err_c = run_read_probe(stub, ["--tree", "{nope}"], 30) jm_ok = judge_malformed(rc_c, out_c, err_c, 2, None) cell("c48 end to end: a renamed flag reads ECHOES_USAGE, its real flag OK", rc_m == 2 and jm_bad["verdict"] == "ECHOES_USAGE" and rc_c == 2 and jm_ok["verdict"] == "OK", f"{rc_m}:{jm_bad['verdict']} / {rc_c}:{jm_ok['verdict']}") axis_echo = [{"tool": "t.py", "probe": "not-a-repo", "verdict": "ECHOES_USAGE", "rc": 2, "expect_rc": 2, "detail": "argparse failure"}] buf = io.StringIO() with contextlib.redirect_stdout(buf): erc, ehard, esoft = report_read(axis_echo, quiet=True) line_e = buf.getvalue() cell("c49 the summary names ECHOES_USAGE with its probe, and the run is red", erc == 1 and (ehard, esoft) == (0, 1) and "malformed-input: 1 probes, 0 ok, 1 fault " "(ECHOES_USAGE t.py/not-a-repo)" in line_e, line_e.strip().splitlines()[-1] if line_e.strip() else "no output") print(f"selftest: {len(ran)} cells, {len(ran) - len(bad)} passed, {len(bad)} failed") return 1 if bad else 0 # ----------------------------------------------------------------------- main def main(argv=None): ap = argparse.ArgumentParser( description="Run every tool's --selftest and check its advertised cell count.") ap.add_argument("--repo", default=DEFAULT_REPO) ap.add_argument("--tools-dir", default=None, help="sweep this directory instead of the repo's tracked tools/") ap.add_argument("--timeout", type=int, default=120, metavar="S") ap.add_argument("--json", default=None, metavar="PATH") ap.add_argument("--baseline", default=None, metavar="PATH", help="the pinned counts to grade against " f"(default: /tools/{BASELINE_NAME} in repo mode)") ap.add_argument("--no-baseline", action="store_true", help="do not grade the count pin (the run says so out loud)") ap.add_argument("--write-baseline", default=None, metavar="PATH", help="write the pinned counts from this run to PATH") ap.add_argument("--quiet", action="store_true") ap.add_argument("--verbose", action="store_true") ap.add_argument("--selftest", action="store_true") a = ap.parse_args(argv) if a.selftest: return selftest() rows, source, untracked = discover(a.repo, a.tools_dir) if not rows: # A sweep that discovered nothing has verified nothing. It is not a # clean run (rc 0) and not a fatal verdict (rc 1): it is rc 2, the # repo-wide code for "could not grade" -- the same contract the # read-refusal axis below measures on every other tool. print("cannot read: no tools discovered in %s (%s)" % (a.tools_dir or os.path.join(a.repo, "tools"), source), file=sys.stderr) return 2 results = sweep(rows, a.repo if not a.tools_dir else os.path.abspath(a.tools_dir), a.timeout, a.verbose) rc, ok, fatal = report(results, source, untracked, a.quiet) axis = read_axis(rows, a.timeout) rrc, read_fatal, malformed_fatal = report_read(axis, a.quiet) # ---- the count pin: the witness outside the tool's own docstring if a.no_baseline: pin_path, pin_note = None, "pin: not graded (--no-baseline)" elif a.baseline: pin_path, pin_note = a.baseline, None elif a.write_baseline: pin_path, pin_note = a.write_baseline, None elif a.tools_dir: pin_path, pin_note = None, ("pin: not graded (--tools-dir: a synthetic " "directory carries no pinned counts)") else: pin_path, pin_note = default_baseline(a.repo), None pin_rc, pin_fatal, pin_rows = 0, 0, None if pin_path is None: if not a.quiet and pin_note: print() print(pin_note) else: if a.write_baseline: text = render_baseline(results) try: with open(a.write_baseline, "w", encoding="utf-8") as fh: fh.write(text) except OSError as exc: print("cannot write --write-baseline " + a.write_baseline + ": " + str(exc), file=sys.stderr) return 2 first = [l for l in text.splitlines() if l and not l.startswith("#")] print(f"pin: wrote {a.write_baseline} ({len(first)} row(s))") pinned, why = load_baseline(pin_path) if pinned is None: # A missing or malformed ledger is not a clean run: the pin graded # nothing. rc 2 (could not grade), never rc 1 and never 0. print("pin: cannot grade -- " + str(why) + " (run --write-baseline to create one, or --no-baseline to sweep without)", file=sys.stderr) pin_rc = 2 else: pin_rows = grade_pin(results, pinned) pin_rc, pin_fatal = report_pin(pin_rows, pin_path, a.quiet) line, run_rc = run_verdict(fatal, read_fatal, malformed_fatal, pin_rc, pin_fatal) if a.json: try: with open(a.json, "w") as fh: json.dump({"repo": os.path.abspath(a.repo), "tools_dir": a.tools_dir, "discovery": source, "untracked": untracked, "tools": results, "fatal": fatal, "ok": ok, "read_axis": axis, "read_fatal": read_fatal, "malformed_fatal": malformed_fatal, "verdict": {"line": line, "rc": run_rc}, "pin": {"path": pin_path, "rows": pin_rows, "fatal": pin_fatal}}, fh, indent=1) except OSError as exc: print("cannot write --json " + a.json + ": " + str(exc), file=sys.stderr) return 2 print() print(line) return run_rc if __name__ == "__main__": sys.exit(main())