#!/usr/bin/env python3 """Independent verifier for draft-compliance-report-2026-10-31.{tsv,md}. WHAT IT GRADES, AND AGAINST WHAT The committed TSV is a claim about the .md drafts beside it. This verifier re-derives every row from the files with its OWN code -- its own line counter, its own `VERDICT:`/`CHECK:` detection and its own implementation of the bare- command rule -- and compares in both directions: every TSV row against the file it names, and every file against the row it should have. It also grades the report's Evidence table against the counts it just measured, so a stale numeral in the prose is a fault rather than a reader's problem. WHICH TREE -- three modes, because a dated census describes a revision (default) the lane as it stands at the pin in `draft-compliance-report-2026-10-31.pin`, extracted with `git archive`; the committed TSV and report are graded against it. Drift since the pin is printed as a note, never a fault. --worktree the live worktree; this is the "is the census current?" reading and it faults on any difference, additions included. --rev REV the lane at REV, for probing whether a given tree's census is true and for grading an older landing. Why: the 2026-10-31 table was byte-true at 2f8ef83 and read red at HEAD for eight drafts that landed after it -- a snapshot cannot be graded against a tree it never described. And the fix must not be "grade nothing": the pin is what makes the claim falsifiable (Q154). Paths come from THIS FILE's location. The 2026-10-31 version hardcoded an absolute home-directory path in both the scanner and the verifier, so a clone's CHECK graded the host's tree and a mutated copy of the committed TSV still read PASS; `--selftest` cell `no-host-path` greps the sources for that class, because this file's own fix must not reintroduce it. EXIT CODES -- the archive's convention 0 the TSV and the report re-derive from the tree they describe 1 a fault: a row, a file-set entry or a reported count disagrees, or the committed TSV is not the bytes the pin names 2 could not grade: the report, the TSV or the pin is missing/unreadable, the pin names no rev, the rev cannot be extracted, or the repository root cannot be found. Never a pass. --selftest runs the cells in a temp tree: one green control, eight mutants and four pin-contract cells, so a green reading is never the only thing this checker can print. NOTE ON THE SECOND IMPLEMENTATION OF `is_bare` It reads operators on the raw string and only checks quoting for parity, so an operator inside a quoted argument reads as not bare where the shlex-based rule would read it as bare. That is deliberate and conservative (it over-reports "not bare", never over-reports "bare"), and `rule-agreement` fails loudly if the two implementations ever disagree on a command this corpus actually holds. """ import argparse import hashlib import os import re import shutil import subprocess import sys import tempfile HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import draft_compliance_rule as rule # noqa: E402 # The path inside the repository that holds this lane. Pinned grading archives # exactly this directory at the pinned rev; nothing outside it is read. PATHSPEC = "wizard-work/quill" # The report's Evidence table, label -> the metric it claims. The labels are # matched verbatim; a missing label is a fault (the report must say what it # measured). REPORT_LABELS = ( ("Total `.md` drafts scanned", "total"), ("Drafts with `VERDICT:`", "verdict"), ("Drafts with `CHECK:`", "check"), ("Drafts with a bare `CHECK:` command", "bare"), ("Drafts failing at least one column", "failing"), ("`VERDICT:` and `CHECK:` but a non-bare command", "nonbare_both"), ) NULL_RE = re.compile(r"(?:^|[\s;&|(])(true|false)(?=$|[\s;&|)])") def own_is_bare(cmd): """The same contract as `rule.is_bare`, implemented differently. Quoted regions are removed before the operator scan, so an operator inside a quoted argument is an argument, not an operator; parity of quotes is checked first, because a command that does not close its quotes does not tokenize and is therefore not bare under the contract. """ if not cmd: return False if cmd.count("'") % 2 or cmd.count('"') % 2: return False stripped = re.sub(r"'[^']*'|\"[^\"]*\"", "''", cmd) if ";" in stripped or "||" in stripped: return False rest = stripped.replace("&&", "") if "&" in rest or "|" in rest: return False return NULL_RE.search(stripped) is None def own_line_count(text): """Line count without splitlines(): newlines plus a final unterminated line.""" return text.count("\n") + (1 if text and not text.endswith("\n") else 0) def own_rows(lane, exclude=rule.ARTIFACT_MD): """Re-derive the table from the files, without calling the generator.""" rows = [] names = sorted(n for n in os.listdir(lane) if n.endswith(".md") and n != exclude) for name in names: path = os.path.join(lane, name) if not os.path.isfile(path): continue with open(path, "r", encoding="utf-8", errors="replace") as f: text = f.read() has_v = has_c = False cmd = "" for ln in text.split("\n"): if ln.startswith("VERDICT:"): has_v = True elif ln.startswith("CHECK:"): has_c = True rest = ln[len("CHECK:"):].strip() if rest and not cmd: cmd = rest rows.append((name, str(own_line_count(text)), str(has_v), str(has_c), str(own_is_bare(cmd)), cmd)) return rows def read_tsv(path): with open(path, "r", encoding="utf-8") as f: raw = f.read().split("\n") comment = "" if raw and raw[0].startswith("#"): comment = raw[0] raw = raw[1:] body = [ln for ln in raw if ln.strip()] header = body[0].split("\t") if body else [] rows = [tuple(ln.split("\t")) for ln in body[1:]] return comment, header, rows def read_report_counts(path): counts = {} with open(path, "r", encoding="utf-8") as f: for ln in f: m = re.match(r"^\|\s*(.+?)\s*\|\s*(\d+)\s*\|\s*$", ln.rstrip("\n")) if m: counts[m.group(1)] = int(m.group(2)) return counts def metrics(rows): def ok(r): return r[2] == "True" and r[3] == "True" and r[4] == "True" return { "total": len(rows), "verdict": sum(1 for r in rows if r[2] == "True"), "check": sum(1 for r in rows if r[3] == "True"), "bare": sum(1 for r in rows if r[4] == "True"), "failing": sum(1 for r in rows if not ok(r)), "nonbare_both": sum(1 for r in rows if r[2] == "True" and r[3] == "True" and r[4] != "True"), } def check(lane=None, report=None, tsv=None): """Return (rc, faults, notes). Never raises for bad input.""" lane = lane or HERE report = report or os.path.join(lane, rule.ARTIFACT_MD) tsv = tsv or os.path.join(lane, rule.ARTIFACT_TSV) faults, notes = [], [] if not os.path.isdir(lane): return 2, ["lane directory missing: %s" % lane], notes for path, what in ((report, "report"), (tsv, "TSV")): if not os.path.isfile(path): return 2, ["%s missing: %s" % (what, path)], notes try: comment, header, claimed = read_tsv(tsv) reported = read_report_counts(report) except (OSError, UnicodeDecodeError) as exc: return 2, ["cannot read the artifact: %s" % exc], notes if tuple(header) != rule.COLUMNS: faults.append("TSV header is %r, contract is %r" % (tuple(header), rule.COLUMNS)) derived = own_rows(lane) der_map = {r[0]: r for r in derived} claim_map = {r[0]: r for r in claimed} for name in sorted(set(claim_map) - set(der_map)): faults.append("TSV row names a file the lane does not hold: %s" % name) for name in sorted(set(der_map) - set(claim_map)): faults.append("lane file has no TSV row: %s" % name) for name in sorted(set(der_map) & set(claim_map)): if der_map[name] != claim_map[name]: faults.append("row mismatch for %s: claimed=%r derived=%r" % (name, claim_map[name], der_map[name])) expected = "# measured %d markdown files" % len(derived) if comment != expected: faults.append("TSV comment is %r, measured %r" % (comment, expected)) m = metrics(derived) for label, key in REPORT_LABELS: if label not in reported: faults.append("report table row missing: %r" % label) elif reported[label] != m[key]: faults.append("report says %r = %d, measured %d" % (label, reported[label], m[key])) # The contract has two implementations; they must agree on every command the # corpus carries, or one of them is grading a different rule. for r in derived: if rule.is_bare(r[5]) != own_is_bare(r[5]): faults.append("rule divergence on %s: rule=%s verifier=%s cmd=%r" % (r[0], rule.is_bare(r[5]), own_is_bare(r[5]), r[5])) notes.append("population %d, verdict %d, check %d, bare %d, failing %d" % (m["total"], m["verdict"], m["check"], m["bare"], m["failing"])) return (1 if faults else 0), faults, notes # --- the pin: which revision this census describes --------------------------- def repo_root(start): """Nearest ancestor that holds a `.git` entry, or None.""" path = os.path.abspath(start) while True: if os.path.exists(os.path.join(path, ".git")): return path parent = os.path.dirname(path) if parent == path: return None path = parent def materialize(rev, root, dest): """Extract PATHSPEC at `rev` under `dest`; return (lane_dir, error).""" try: src = subprocess.run(["git", "-C", root, "archive", rev, "--", PATHSPEC], stdout=subprocess.PIPE, stderr=subprocess.PIPE) except OSError as exc: return None, "cannot run git: %s" % exc if src.returncode != 0: return None, "cannot archive %s: %s" % ( rev, src.stderr.decode("utf-8", "replace").strip()[:200]) lane = os.path.join(dest, "lane") os.makedirs(lane) try: untar = subprocess.run(["tar", "-x", "-C", lane], input=src.stdout, stdout=subprocess.PIPE, stderr=subprocess.PIPE) except OSError as exc: return None, "cannot run tar: %s" % exc if untar.returncode != 0: return None, "cannot extract %s: %s" % ( rev, untar.stderr.decode("utf-8", "replace").strip()[:200]) return os.path.join(lane, PATHSPEC), "" def read_pin(path): """The pin as a dict; raises OSError when it cannot be read.""" values = {} with open(path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue parts = line.split(None, 1) if len(parts) == 2: values[parts[0]] = parts[1].strip() return values def pin_verdict(pin, tsv): """(rc, faults) for the pin contract: 2 ungradable, 1 not the bytes, 0 ok. The digest is what makes "the committed table" a fixed claim: without it a silent edit to the TSV would be graded as the new truth. """ try: values = read_pin(pin) except OSError as exc: return 2, ["pin missing or unreadable: %s" % exc] if not values.get("rev"): return 2, ["pin names no rev: %s" % pin] try: with open(tsv, "rb") as f: digest = hashlib.sha256(f.read()).hexdigest() except OSError as exc: return 2, ["TSV unreadable: %s" % exc] want = values.get("tsv") if want and want != digest: return 1, ["committed TSV is not the pinned bytes: sha256 %s, pin %s" % (digest, want)] return 0, [] def grade_at_rev(rev, claim_dir, lane=None): """Grade the artifact in `claim_dir` against the drafts at `rev`.""" root = repo_root(claim_dir) if root is None: return 2, ["cannot find the repository root above %s" % claim_dir], [] tmp = tempfile.mkdtemp(prefix="q154-rev-") try: tree, error = materialize(rev, root, tmp) if tree is None: return 2, [error], [] return check(tree, os.path.join(claim_dir, rule.ARTIFACT_MD), os.path.join(claim_dir, rule.ARTIFACT_TSV)) finally: shutil.rmtree(tmp, ignore_errors=True) def drift_note(pinned_rows, live_rows): """What moved between the pinned census and the live tree. Advisory text.""" pin_map = {r[0]: r for r in pinned_rows} live_map = {r[0]: r for r in live_rows} added = sorted(set(live_map) - set(pin_map)) gone = sorted(set(pin_map) - set(live_map)) changed = sorted(n for n in set(pin_map) & set(live_map) if pin_map[n] != live_map[n]) text = "drift since the pin: %d added, %d changed, %d removed" % ( len(added), len(changed), len(gone)) if added: text += "; added: " + ", ".join(added[:6]) + (" ..." if len(added) > 6 else "") return text FIXTURE_FILES = { "good.md": "# t\n\nVERDICT: ok\nCHECK: python3 x.py\n", "plain.md": "# nothing\n", "piped.md": "# t\n\nVERDICT: ok\nCHECK: python3 x.py | grep -q ok\n", } def _fixture_report(claimed): lines = ["# fixture", "", "| Metric | Value |", "|---|---|"] for label, _key in REPORT_LABELS: lines.append("| %s | %d |" % (label, claimed[label])) return "\n".join(lines) + "\n" def _fixture_tree(root, mutate=None): lane = os.path.join(root, "lane") os.makedirs(lane) for name, text in FIXTURE_FILES.items(): with open(os.path.join(lane, name), "w", encoding="utf-8") as f: f.write(text) rows = own_rows(lane) m = metrics(rows) claimed = {label: m[key] for label, key in REPORT_LABELS} report = os.path.join(lane, rule.ARTIFACT_MD) with open(report, "w", encoding="utf-8") as f: f.write(_fixture_report(claimed)) tsv = os.path.join(lane, rule.ARTIFACT_TSV) with open(tsv, "w", encoding="utf-8") as f: f.write("# measured %d markdown files\n" % len(rows)) f.write("\t".join(rule.COLUMNS) + "\n") for r in rows: f.write("\t".join(r) + "\n") if mutate: mutate(lane, report, tsv) return lane, report, tsv def selftest(verbose=True): cells, failed = [], [] def cell(name, want, mutate=None): tmp = tempfile.mkdtemp(prefix="q142-") try: lane, report, tsv = _fixture_tree(tmp, mutate) got, faults, _notes = check(lane, report, tsv) finally: shutil.rmtree(tmp, ignore_errors=True) cells.append(name) if got != want: failed.append((name, want, got, faults[:3])) if verbose: print("%-22s rc=%d want=%d %s" % (name, got, want, "ok" if got == want else "FAIL")) def bump_line(lane, report, tsv): with open(os.path.join(lane, "good.md"), "a", encoding="utf-8") as f: f.write("one more line\n") def drop_verdict(lane, report, tsv): path = os.path.join(lane, "good.md") with open(path, encoding="utf-8") as f: text = f.read() with open(path, "w", encoding="utf-8") as f: f.write(text.replace("VERDICT: ok\n", "")) def drop_row(lane, report, tsv): with open(tsv, encoding="utf-8") as f: lines = f.read().split("\n") with open(tsv, "w", encoding="utf-8") as f: f.write("\n".join(ln for ln in lines if not ln.startswith("piped.md"))) def extra_file(lane, report, tsv): with open(os.path.join(lane, "late.md"), "w", encoding="utf-8") as f: f.write("# late\n") def no_report(lane, report, tsv): os.remove(report) def no_tsv(lane, report, tsv): os.remove(tsv) def stale_count(lane, report, tsv): with open(report, encoding="utf-8") as f: text = f.read() text = re.sub(r"(\| Total `\.md` drafts scanned \| )3( \|)", r"\g<1>2\g<2>", text) with open(report, "w", encoding="utf-8") as f: f.write(text) def check_line_edited(lane, report, tsv): with open(os.path.join(lane, "good.md"), "w", encoding="utf-8") as f: f.write("# t\n\nVERDICT: ok\nCHECK: python3 y.py\n") cell("green-control", 0) cell("line-count-moved", 1, bump_line) cell("verdict-removed", 1, drop_verdict) cell("row-deleted", 1, drop_row) cell("file-without-row", 1, extra_file) cell("report-missing", 2, no_report) cell("tsv-missing", 2, no_tsv) cell("report-count-stale", 1, stale_count) cell("check-line-edited", 1, check_line_edited) # The pin contract: a snapshot is graded against the bytes and the rev it # names, and an unreadable pin is ungradable rather than a pass. good_tsv = "x\n" def pin_cell(name, want, pin_text): tmp = tempfile.mkdtemp(prefix="q154-pin-") try: tsv = os.path.join(tmp, "t.tsv") with open(tsv, "w", encoding="utf-8") as f: f.write(good_tsv) pin = os.path.join(tmp, "p.pin") if pin_text is not None: with open(pin, "w", encoding="utf-8") as f: f.write(pin_text) got, faults = pin_verdict(pin, tsv) finally: shutil.rmtree(tmp, ignore_errors=True) cells.append(name) if got != want: failed.append((name, want, got, faults[:2])) if verbose: print("%-22s rc=%d want=%d %s" % (name, got, want, "ok" if got == want else "FAIL")) digest = hashlib.sha256(good_tsv.encode("utf-8")).hexdigest() pin_cell("pin-ok", 0, "rev 0123456789abcdef\ntsv %s\n" % digest) pin_cell("pin-tsv-digest-moved", 1, "rev 0123456789abcdef\ntsv %s\n" % ("0" * 64)) pin_cell("pin-no-rev", 2, "tsv %s\n" % digest) pin_cell("pin-absent", 2, None) # The two implementations of the rule, over the fixture cases and this # directory's own corpus. A divergence here is what the 2026-10-31 copies had. divergence = [cmd for cmd, want, _why in rule.SELFTEST_CASES if own_is_bare(cmd) != want] for r in own_rows(HERE): if rule.is_bare(r[5]) != own_is_bare(r[5]): divergence.append(r[0]) cells.append("rule-agreement") if divergence: failed.append(("rule-agreement", 0, 1, divergence[:3])) if verbose: print("%-22s %s" % ("rule-agreement", "ok" if not divergence else "FAIL %r" % divergence[:3])) # No host path in the sources: the 2026-10-31 defect, graded at the source. offenders = [] needle = os.sep + "root" + os.sep for name in (os.path.basename(__file__), "draft_compliance_report.py", "draft_compliance_rule.py"): with open(os.path.join(HERE, name), encoding="utf-8") as f: if needle in f.read(): offenders.append(name) cells.append("no-host-path") if offenders: failed.append(("no-host-path", 0, 1, offenders)) if verbose: print("%-22s %s" % ("no-host-path", "ok" if not offenders else "FAIL %r" % offenders)) print("\nselftest: %d cells, %d passed, %d failed" % (len(cells), len(cells) - len(failed), len(failed))) for name, want, got, detail in failed: print(" FAIL %s: rc %d wanted %d -- %r" % (name, got, want, detail)) return 1 if failed else 0 def _emit(rc, faults, notes, mode): for note in notes: print(note) if rc == 0: print("VERDICT: PASS -- TSV and report re-derive from %s (%s)" % (mode, notes[0] if notes else "no counts")) elif rc == 2: print("VERDICT: UNGRADABLE") for f in faults: print(" -", f) else: print("VERDICT: FAIL") for f in faults: print(" -", f) return rc def main(argv=None): ap = argparse.ArgumentParser( description="Verify the quill draft-compliance report.") ap.add_argument("--lane", default=None, help="grade this directory directly (fixtures; default: " "the pinned revision)") ap.add_argument("--report", default=None, help="report .md (default: in --lane)") ap.add_argument("--tsv", default=None, help="report .tsv (default: in --lane)") ap.add_argument("--pin", default=None, help="pin file (default: %s beside the artifact)" % rule.ARTIFACT_PIN) ap.add_argument("--rev", default=None, help="grade the committed artifact against the lane at REV") ap.add_argument("--worktree", action="store_true", help="grade against the live worktree instead of the pin") ap.add_argument("--selftest", action="store_true", help="run the cells") a = ap.parse_args(argv) if a.selftest: return selftest() claim = a.lane or HERE report = a.report or os.path.join(claim, rule.ARTIFACT_MD) tsv = a.tsv or os.path.join(claim, rule.ARTIFACT_TSV) pin = a.pin or os.path.join(claim, rule.ARTIFACT_PIN) if a.worktree or a.lane: rc, faults, notes = check(claim, report, tsv) mode = "the live worktree" if a.worktree else "the tree at %s" % claim elif a.rev: rc, faults, notes = grade_at_rev(a.rev, claim) mode = "the lane at %s" % a.rev else: rc_pin, faults_pin = pin_verdict(pin, tsv) if rc_pin: return _emit(rc_pin, faults_pin, [], "the pin") rev = read_pin(pin)["rev"] rc, faults, notes = grade_at_rev(rev, claim) mode = "the pinned revision %s" % rev[:12] # A snapshot is graded against the revision it names; what has landed since # is drift to report, not a fault of the table. `--worktree` grades it. if mode != "the live worktree": try: notes = list(notes) + [drift_note(read_tsv(tsv)[2], own_rows(HERE))] except OSError: pass return _emit(rc, faults, notes, mode) if __name__ == "__main__": raise SystemExit(main())