#!/usr/bin/env python3 """Falsifiable gate for the WIZARDS-TABLE roster extract (`wizards.csv`). WHAT IT GRADES 1. PIN the source file's bytes are the bytes the extractor pinned: sha256 and size against `wizards.manifest.json`, plus the git blob id when this tree can answer for it. A roster table that moved without a re-extract is a FAIL, not a silent re-read. 2. SHAPE the CSV's header is exactly the manifest's column list, its row count matches, and its bytes hash to `csv_sha256`. 3. CONTENT every cell is re-derived from the pinned source by this file's own parser and compared one cell at a time, so a renamed Wizard or a corrupted Lane cell names the row and the column instead of returning a green light (the defect Q113 filed). 4. CORRECTION the table appends a control-plane correction that withdraws one row's escalate-line. A transcription that copies the cell and drops the correction re-issues a withdrawn escalation, so each correction in the manifest must (a) carry its marker in the CSV, (b) name text that is still in the source cell, and (c) cite an authority span that still contains its phrase. A marker on a row with nothing withdrawn is a fault too, so the column cannot be sprayed green. The parse here is an independent re-implementation of the extractor's rule, not an oracle: the two agree by construction on the stated rule (5 cells, `lane signer` iff the Model cell says it has no model of its own), and their agreement is a transcription check. Paths are derived from this file's own location; `--root DIR` grades a tree this file did not ship from (how the cells build their mutants). Exit codes (repo tool contract): 0 every leg passed 1 at least one leg faulted 2 could not grade (missing/unparseable artifact, source or manifest) """ import argparse import csv import hashlib import io import json import os import subprocess import sys HERE = os.path.dirname(os.path.realpath(__file__)) ROOT = os.path.dirname(os.path.dirname(HERE)) CSV_NAME = "wizards.csv" MANIFEST_NAME = "wizards.manifest.json" TABLE_HEADER = ["Wizard", "Model", "Lane", "Latest check-in", "Current unit"] LANE_SIGNER_MARK = "no model of its own" FAULTS = [] CANNOT = [] def pass_(text): print("PASS " + text) def fail(text): FAULTS.append(text) print("FAIL " + text) def info(text): print("INFO " + text) def cannot(text): CANNOT.append(text) print("CANNOT GRADE " + text) def sha256(data): return hashlib.sha256(data).hexdigest() def cells_of(line): parts = [c.strip() for c in line.strip().split("|")] if parts and parts[0] == "": parts = parts[1:] if parts and parts[-1] == "": parts = parts[:-1] return parts def parse_source(text): """Independent parse: -> (list of 5-cell rows, source lines).""" lines = text.splitlines() start = None for i, line in enumerate(lines): if cells_of(line) == TABLE_HEADER: start = i break if start is None: raise ValueError("roster header not found") sep = cells_of(lines[start + 1]) if start + 1 < len(lines) else [] if not sep or not all(set(c) <= set("-: ") for c in sep): raise ValueError("no header separator after the roster header") rows = [] for line in lines[start + 2:]: s = line.strip() if not s or s.startswith("#") or not s.startswith("|"): break row = cells_of(s) if len(row) != 5: raise ValueError("row is not 5 cells: %r" % s[:70]) rows.append(row) return rows, lines def git_blob(root, relpath): try: proc = subprocess.run(["git", "-C", root, "hash-object", relpath], capture_output=True, text=True, timeout=20) except (OSError, subprocess.SubprocessError): return None, "git unavailable" if proc.returncode != 0: return None, (proc.stderr.strip().splitlines() or ["git failed"])[0] return proc.stdout.strip(), "" def main(argv=None): ap = argparse.ArgumentParser( description="Grade the WIZARDS-TABLE roster extract against its pin.") ap.add_argument("--root", default=ROOT, help="tree to grade (default: the tree this file sits in)") args = ap.parse_args(argv) root = os.path.realpath(args.root) ledger = os.path.join(root, "wizard-work", "ledger") # --- inputs ----------------------------------------------------------- man_path = os.path.join(ledger, MANIFEST_NAME) csv_path = os.path.join(ledger, CSV_NAME) try: with open(man_path, "rb") as fh: manifest = json.loads(fh.read().decode("utf-8")) except (OSError, ValueError) as exc: cannot("%s not readable/parseable: %s" % (man_path, exc)) return 2 try: with open(csv_path, "rb") as fh: csv_bytes = fh.read() except OSError as exc: cannot("%s not readable: %s" % (csv_path, exc)) return 2 source_rel = manifest.get("source") if not isinstance(source_rel, str) or not source_rel: cannot("%s carries no 'source' path" % MANIFEST_NAME) return 2 source_path = os.path.join(root, source_rel) try: with open(source_path, "rb") as fh: source_bytes = fh.read() except OSError as exc: cannot("%s not readable: %s" % (source_path, exc)) return 2 # --- 1. pin ----------------------------------------------------------- want_sha = manifest.get("source_sha256") got_sha = sha256(source_bytes) if got_sha == want_sha: pass_("pin: %s sha256=%s bytes=%d" % (source_rel, got_sha[:12], len(source_bytes))) else: fail("pin: %s sha256=%s != pinned %s -- source moved without a " "re-extract; run: python3 wizard-work/ledger/extract_wizards.py" % (source_rel, got_sha[:12], str(want_sha)[:12])) if manifest.get("source_bytes") not in (None, len(source_bytes)): fail("pin: %s is %d bytes, manifest pins %s" % (source_rel, len(source_bytes), manifest.get("source_bytes"))) blob, why = git_blob(root, source_rel) want_blob = str(manifest.get("source_blob") or "") if blob is None: info("pin: git blob not graded (%s)" % why) elif len(want_blob) != 40: fail("pin: manifest source_blob=%r is not a 40-hex blob id" % want_blob) elif blob != want_blob: fail("pin: %s blob=%s != pinned %s" % (source_rel, blob[:12], want_blob[:12])) else: pass_("pin: git blob %s matches the manifest" % blob[:12]) commit = str(manifest.get("source_commit") or "") if len(commit) == 40 and all(c in "0123456789abcdef" for c in commit): info("pin: blob read at commit %s, source states roster as of %s" % (commit[:12], manifest.get("source_stated_date"))) else: info("pin: source_commit=%r is not a 40-hex commit id" % commit) # --- 2. shape --------------------------------------------------------- columns = manifest.get("columns") or [] try: parsed = list(csv.reader(io.StringIO(csv_bytes.decode("utf-8")))) except (UnicodeDecodeError, csv.Error) as exc: cannot("%s is not parseable CSV: %s" % (csv_path, exc)) return 2 if not parsed: cannot("%s is empty" % csv_path) return 2 header, data = parsed[0], parsed[1:] if header != columns: fail("shape: header %r != manifest columns %r" % (header, columns)) else: pass_("shape: header %d columns" % len(header)) if len(data) != manifest.get("row_count"): fail("shape: %d data row(s) != manifest row_count %r" % (len(data), manifest.get("row_count"))) else: pass_("shape: %d data row(s)" % len(data)) if sha256(csv_bytes) != manifest.get("csv_sha256"): fail("shape: csv sha256=%s != pinned %s -- the CSV was edited after " "extraction" % (sha256(csv_bytes)[:12], str(manifest.get("csv_sha256"))[:12])) else: pass_("shape: csv sha256=%s matches the pin" % sha256(csv_bytes)[:12]) # --- 3. content, cell by cell ---------------------------------------- try: source_rows, source_lines = parse_source(source_bytes.decode("utf-8")) except (UnicodeDecodeError, ValueError) as exc: cannot("%s: %s" % (source_rel, exc)) return 2 corrections = manifest.get("corrections") or [] markers = {c["row"]: c["marker"] for c in corrections if "row" in c} expected = [] for lead, model, lane, latest, unit in source_rows: expected.append([ "lane signer" if LANE_SIGNER_MARK in model else "wizard", lead, model, lane, latest, unit, markers.get(lead, ""), ]) faults_before = len(FAULTS) compared = 0 # Name the rows that are absent or extra before the positional diff, so a # deleted roster row reads as a deleted row rather than only as a cascade # of shifted cells. data_leads = [r[1] for r in data if len(r) > 1] for want_row in expected: if want_row[1] not in data_leads: fail("content: row %r missing from the CSV" % want_row[1]) for got_row in data: if len(got_row) > 1 and got_row[1] not in [e[1] for e in expected]: fail("content: CSV row %r is not in the source" % got_row[1]) for idx in range(max(len(expected), len(data))): if idx >= len(data): fail("content: row %d (%s) missing from the CSV" % (idx + 1, expected[idx][1])) continue if idx >= len(expected): fail("content: CSV row %d (%r) is not in the source" % (idx + 1, data[idx][0])) continue if len(data[idx]) != len(columns): fail("content: row %d (%s) has %d column(s), the header has %d" % (idx + 1, expected[idx][1], len(data[idx]), len(columns))) for col, (want, got) in enumerate(zip(expected[idx], data[idx])): compared += 1 if want != got: name = columns[col] if col < len(columns) else "col%d" % col fail("content: row %d (%s) column %r: %r != source %r" % (idx + 1, expected[idx][1], name, got, want)) if len(FAULTS) == faults_before: pass_("content: %d rows x %d column(s) = %d cell(s) re-derived" % (len(expected), len(columns), compared)) # --- 4. corrections --------------------------------------------------- cell_index = {c: i for i, c in enumerate(columns)} rows_by_lead = {r[1]: r for r in data if len(r) > 1} for corr in corrections: lead = corr.get("row") col = corr.get("column") marker = corr.get("marker") row = rows_by_lead.get(lead) if row is None: fail("correction: row %r is not in the CSV" % lead) continue idx = cell_index.get(col) if idx is None or idx >= len(row): fail("correction: row %r names column %r, not in the CSV" % (lead, col)) continue if row[-1] != marker: fail("correction: row %r Correction cell does not carry the " "withdrawn marker" % lead) else: pass_("correction: row %r carries its marker" % lead) withdrawn = corr.get("withdrawn_text") or "" if withdrawn and withdrawn not in row[idx]: fail("correction: row %r no longer contains the withdrawn text " "in %r -- the marker outlives its subject" % (lead, col)) start = corr.get("authority_line_start") end = corr.get("authority_line_end") phrase = corr.get("authority_phrase") or "" if isinstance(start, int) and isinstance(end, int): span = "\n".join(source_lines[start - 1:end]) if phrase and phrase in span: pass_("correction: authority %s:%d-%d still states %r" % (source_rel, start, end, phrase[:40])) else: fail("correction: authority %s:%d-%d no longer states %r" % (source_rel, start, end, phrase[:40])) else: fail("correction: row %r has no integer authority span" % lead) for row in data: if len(row) < len(columns): continue lead, marker = row[1], row[-1] if marker and lead not in markers: fail("correction: row %r carries a marker with no withdrawal " "recorded in the manifest" % lead) # Generic direction: any source cell holding a withdrawn phrase must be # marked, whether or not its row is the one the manifest was written for. for corr in corrections: withdrawn = corr.get("withdrawn_text") or "" col = cell_index.get(corr.get("column")) if not withdrawn or col is None: continue for row in data: if col < len(row) and withdrawn in row[col] and not row[-1]: fail("correction: row %r holds a withdrawn phrase and no marker" % row[1]) # --- summary ---------------------------------------------------------- checked = len(data) print("verify_wizards: %d row(s), %d column(s), %d cell(s) compared, " "%d fault(s)" % (checked, len(columns), compared, len(FAULTS))) if CANNOT: return 2 if FAULTS: print("VERDICT FALSE: %d fault(s) -- see the FAIL lines above" % len(FAULTS)) return 1 print("VERDICT TRUE: %s replays from %s at its pinned bytes" % (CSV_NAME, source_rel)) return 0 if __name__ == "__main__": sys.exit(main())