ledger
2b35aa6523
The gate was red on every clone older than two minutes: its recency leg was
abs(now - extracted_at_utc) < 120 against the frozen 2026-10-28T18:31:34Z, and
the README's documented `bash CHECK` could not run at all (the file is Python;
rc 2, syntax error at line 27).
- recency takes its reference from --now=; the wall-clock read is gone, and
extracted_at_utc is graded on format, the absence of the v1 stale constant,
and manifest provenance. No reference -> one SKIP line, gate still decides.
- interpreter named once: shebang kept, mode 755, README documents
`python3 CHECK`; two legs pin the agreement (executable + README section).
- --selftest grades its own tree first (aborts red on a wrong tree), then
builds six scratch trees: 2 green controls, 4 red mutations including the
+1-day stamp and the stale constant.
Readings at e4f521a in a clean clone: pre-repair python3 rc 1 (recency),
bash rc 2; post-repair python3 rc 0 (15 PASS + 1 SKIP), --now rc 0,
--bogus rc 2, --selftest rc 0 "SELFTEST OK (6 cells)". Independent
reproduction: wizard-outputs/volt/verify_q122.py, 15 PASS, rc 0.
328 lines
13 KiB
Python
Executable File
328 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Independent verification for the workqueue-census-v2 artifact.
|
|
|
|
RE-APPROACH: this checker does NOT import the extractor. It re-derives the
|
|
expected rows from WORK-QUEUE.md with plain string operations (split on
|
|
whitespace, startswith); the committed TSV is the artifact of record.
|
|
The verdict is the script's exit code: 0 = correct, 1 = a check failed,
|
|
2 = could not grade (bad invocation / missing artifact).
|
|
|
|
Run it with the interpreter this file names:
|
|
|
|
python3 CHECK # portable: every leg that can be graded
|
|
# from the artifact alone is graded
|
|
python3 CHECK --now=2026-10-28T18:31:34Z
|
|
# adds the recency leg: the manifest's
|
|
# extraction time must be within 120s of
|
|
# the reference time SUPPLIED HERE
|
|
python3 CHECK --selftest # grade the tree, then prove
|
|
# every leg can fail
|
|
|
|
Checks:
|
|
1. every '## Q<n>' heading in the source has exactly one committed row, in
|
|
source order, with a verbatim title;
|
|
2. row counts, distinct-q count and reopened-q list in extract_manifest.json
|
|
agree with an independent recomputation;
|
|
3. extracted_at_utc provenance: it parses as an UTC 'Z' stamp and is not the
|
|
known stale constant. Its RECENCY is graded only against a reference time
|
|
passed by the caller (--now), never against the wall clock, so the gate
|
|
does not decay with time: a clone made a year from now reads the same
|
|
verdict a clone made a minute from now reads;
|
|
4. README names the interpreter and the file carries the python3 shebang and
|
|
the execute bit (the defect that made the documented invocation unrunnable);
|
|
5. no committed row has an empty q or title, and no status/done value
|
|
contains a literal tab (delimiter corruption).
|
|
"""
|
|
|
|
import csv
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
REPO_ROOT = os.path.abspath(os.path.join(HERE, "..", "..", ".."))
|
|
SRC_NAME = "WORK-QUEUE.md"
|
|
TSV_NAME = "workqueue-census.tsv"
|
|
MD_NAME = "workqueue-census.md"
|
|
MANIFEST_NAME = "extract_manifest.json"
|
|
README_NAME = "README.md"
|
|
STALE_CONSTANT = "2026-10-28T00:00:00Z"
|
|
TS_FMT = "%Y-%m-%dT%H:%M:%SZ"
|
|
RECENCY_WINDOW_S = 120
|
|
SHEBANG = "#!/usr/bin/env python3"
|
|
|
|
|
|
def parse_args(argv):
|
|
now_ref, selftest = None, False
|
|
for arg in argv:
|
|
if arg == "--selftest":
|
|
selftest = True
|
|
elif arg.startswith("--now="):
|
|
now_ref = arg.split("=", 1)[1]
|
|
elif arg in ("-h", "--help"):
|
|
print(__doc__)
|
|
sys.exit(0)
|
|
else:
|
|
print("usage: python3 CHECK [--now=YYYY-MM-DDTHH:MM:SSZ] [--selftest]")
|
|
sys.exit(2)
|
|
return now_ref, selftest
|
|
|
|
|
|
def rederive(src_path):
|
|
"""Re-derive (qid, title) in source order with plain string operations."""
|
|
golden = []
|
|
qid, title = None, ""
|
|
with open(src_path, encoding="utf-8") as fh:
|
|
src_lines = fh.read().splitlines()
|
|
|
|
for line in src_lines:
|
|
if line.startswith("## "):
|
|
parts = line[3:].split(None, 1)
|
|
if parts and len(parts[0]) > 1 and parts[0][0] == "Q" and parts[0][1:].isdigit():
|
|
if qid is not None:
|
|
golden.append((qid, title))
|
|
qid = parts[0]
|
|
title = parts[1].strip() if len(parts) > 1 else ""
|
|
continue
|
|
|
|
if qid is not None:
|
|
golden.append((qid, title))
|
|
return golden
|
|
|
|
|
|
def run(here, repo_root, now_ref):
|
|
"""Grade one tree. Returns (fails, stats)."""
|
|
fails = []
|
|
|
|
def check(name, cond, detail=""):
|
|
if cond:
|
|
print("PASS " + name)
|
|
else:
|
|
print("FAIL " + name + " " + repr(detail))
|
|
fails.append(name)
|
|
|
|
src = os.path.join(repo_root, SRC_NAME)
|
|
tsv = os.path.join(here, TSV_NAME)
|
|
manifest_path = os.path.join(here, MANIFEST_NAME)
|
|
readme_path = os.path.join(here, README_NAME)
|
|
self_path = os.path.join(here, "CHECK")
|
|
|
|
# --- 1. Independent re-derivation from the source (string ops, no regex) ---
|
|
golden = rederive(src)
|
|
|
|
# --- 2. Committed TSV ------------------------------------------------------
|
|
with open(tsv, encoding="utf-8", newline="") as fh:
|
|
rows = list(csv.reader(fh, delimiter="\t"))
|
|
header, committed = rows[0], rows[1:]
|
|
|
|
check("tsv header", header == ["q", "title", "status", "done_implies"], repr(header))
|
|
check("row count",
|
|
len(committed) == len(golden),
|
|
"committed=%d golden=%d" % (len(committed), len(golden)))
|
|
|
|
mism = [k for k, (c, g) in enumerate(zip(committed, golden))
|
|
if c[0] != g[0] or c[1] != g[1]]
|
|
check("qid+title identical to independent re-derive", not mism,
|
|
"first mismatch row %r" % (mism[0] if mism else None))
|
|
|
|
# --- 3. Manifest self-consistency (independent recomputation) --------------
|
|
with open(manifest_path, encoding="utf-8") as fh:
|
|
man = json.load(fh)
|
|
|
|
counts = {}
|
|
for q, _ in golden:
|
|
counts[q] = counts.get(q, 0) + 1
|
|
distinct = sorted(counts)
|
|
reopened = sorted(q for q, c in counts.items() if c > 1)
|
|
|
|
check("manifest.rows", man.get("rows") == len(golden),
|
|
"manifest=%r actual=%d" % (man.get("rows"), len(golden)))
|
|
check("manifest.distinct_q_labels", man.get("distinct_q_labels") == len(distinct),
|
|
"manifest=%r actual=%d" % (man.get("distinct_q_labels"), len(distinct)))
|
|
check("manifest.reopened_q_labels", man.get("reopened_q_labels") == reopened,
|
|
"manifest=%r actual=%r" % (man.get("reopened_q_labels"), reopened))
|
|
|
|
# --- 4. extracted_at_utc: provenance graded, recency by argument -----------
|
|
auto = man.get("extracted_at_utc", "")
|
|
check("extracted_at_utc key", isinstance(auto, str) and bool(auto), repr(auto))
|
|
dt = None
|
|
if isinstance(auto, str) and auto:
|
|
try:
|
|
dt = datetime.strptime(auto, TS_FMT).replace(tzinfo=timezone.utc)
|
|
check("extracted_at_utc format", True)
|
|
except ValueError:
|
|
check("extracted_at_utc format", False, repr(auto))
|
|
check("extracted_at_utc not the stale constant", auto != STALE_CONSTANT, repr(auto))
|
|
check("manifest provenance names the source",
|
|
isinstance(man.get("schema"), str) and man.get("schema") == "workqueue-census-v2"
|
|
and isinstance(man.get("generated_by"), str) and man.get("generated_by"),
|
|
"schema=%r generated_by=%r" % (man.get("schema"), man.get("generated_by")))
|
|
|
|
if now_ref is None:
|
|
print("SKIP extracted_at_utc recency "
|
|
"(no reference time; pass --now=YYYY-MM-DDTHH:MM:SSZ to grade it)")
|
|
else:
|
|
try:
|
|
ref = datetime.strptime(now_ref, TS_FMT).replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
check("--now format", False, repr(now_ref))
|
|
else:
|
|
if dt is None:
|
|
check("extracted_at_utc recency", False, "unparseable stamp %r" % (auto,))
|
|
else:
|
|
delta = abs((ref - dt).total_seconds())
|
|
check("extracted_at_utc recency", delta < RECENCY_WINDOW_S,
|
|
"auto=%s ref=%s delta=%ss" % (auto, now_ref, delta))
|
|
|
|
# --- 5. Interpreter contract (README vs this file) -------------------------
|
|
with open(self_path, encoding="utf-8") as fh:
|
|
first = fh.readline().rstrip("\n")
|
|
check("CHECK shebang is python3", first == SHEBANG, repr(first))
|
|
check("CHECK is executable", os.access(self_path, os.X_OK),
|
|
oct(os.stat(self_path).st_mode & 0o777))
|
|
with open(readme_path, encoding="utf-8") as fh:
|
|
readme = fh.read()
|
|
# The documented invocation lives in the Verification section; the prose may
|
|
# still quote the old broken command when it explains the repair.
|
|
vsec = readme.split("## Verification", 1)
|
|
vsec = vsec[1].split("\n## ", 1)[0] if len(vsec) > 1 else ""
|
|
check("README documents the python3 invocation",
|
|
"python3 CHECK" in vsec and "bash CHECK" not in vsec,
|
|
"python3 CHECK present=%r bash CHECK present=%r"
|
|
% ("python3 CHECK" in vsec, "bash CHECK" in vsec))
|
|
|
|
# --- 6. Corruption checks --------------------------------------------------
|
|
empty = [r for r in committed if not r[0] or not r[1]]
|
|
check("no empty q/title", not empty, repr(empty[:1]))
|
|
bad_tab = [r for r in committed if any("\t" in c for c in r[2:])]
|
|
check("status/done no tabs", not bad_tab, repr(bad_tab[:1]))
|
|
|
|
return fails, {"rows": len(golden), "distinct": len(distinct), "reopened": len(reopened)}
|
|
|
|
|
|
# --- selftest: one cell per leg, each built by mutating a scratch copy ---------
|
|
|
|
def build_tree(dst, manifest_mutate=None, tsv_mutate=None):
|
|
"""Copy the artifact + its source into dst as a self-contained repo root."""
|
|
here = os.path.join(dst, "wizard-work", "ledger", "workqueue-census-v2")
|
|
os.makedirs(here)
|
|
shutil.copy2(os.path.join(REPO_ROOT, SRC_NAME), os.path.join(dst, SRC_NAME))
|
|
|
|
for name in ("CHECK", TSV_NAME, MD_NAME, README_NAME):
|
|
shutil.copy2(os.path.join(HERE, name), os.path.join(here, name))
|
|
|
|
with open(os.path.join(HERE, MANIFEST_NAME), encoding="utf-8") as fh:
|
|
man = json.load(fh)
|
|
if manifest_mutate:
|
|
manifest_mutate(man)
|
|
with open(os.path.join(here, MANIFEST_NAME), "w", encoding="utf-8") as fh:
|
|
json.dump(man, fh, indent=2)
|
|
|
|
if tsv_mutate:
|
|
path = os.path.join(here, TSV_NAME)
|
|
with open(path, encoding="utf-8", newline="") as fh:
|
|
rows = list(csv.reader(fh, delimiter="\t"))
|
|
tsv_mutate(rows)
|
|
with open(path, "w", encoding="utf-8", newline="") as fh:
|
|
csv.writer(fh, delimiter="\t").writerows(rows)
|
|
|
|
return here
|
|
|
|
|
|
def shift_one_day(man):
|
|
dt = datetime.strptime(man["extracted_at_utc"], TS_FMT).replace(tzinfo=timezone.utc)
|
|
man["extracted_at_utc"] = (dt + timedelta(days=1)).strftime(TS_FMT)
|
|
|
|
|
|
def set_stale(man):
|
|
man["extracted_at_utc"] = STALE_CONSTANT
|
|
|
|
|
|
def set_garbage_stamp(man):
|
|
man["extracted_at_utc"] = "2026-10-28 18:31:34"
|
|
|
|
|
|
def corrupt_title(rows):
|
|
rows[1][1] = rows[1][1] + " (corrupted)"
|
|
|
|
|
|
def selftest():
|
|
with open(os.path.join(HERE, MANIFEST_NAME), encoding="utf-8") as fh:
|
|
true_stamp = json.load(fh)["extracted_at_utc"]
|
|
|
|
cells = []
|
|
|
|
def cell(name, mutate_kwargs, argv, want_rc, want_substr):
|
|
with tempfile.TemporaryDirectory(prefix="census2-selftest-") as tmp:
|
|
here = build_tree(tmp, **mutate_kwargs)
|
|
proc = subprocess.run([sys.executable, os.path.join(here, "CHECK")] + argv,
|
|
capture_output=True, text=True)
|
|
out = proc.stdout + proc.stderr
|
|
ok = proc.returncode == want_rc and want_substr in out
|
|
cells.append((name, ok))
|
|
print("%s %-34s rc=%d (want %d) %r%s"
|
|
% ("PASS" if ok else "FAIL", name, proc.returncode, want_rc,
|
|
want_substr, "" if ok else "\n" + out))
|
|
|
|
# control: artifact as committed, reference supplied -> green
|
|
cell("control-with-reference", {}, ["--now=" + true_stamp], 0,
|
|
"PASS extracted_at_utc recency")
|
|
# control: no reference -> recency SKIPs instead of decaying to red
|
|
cell("control-no-reference-skips", {}, [], 0,
|
|
"SKIP extracted_at_utc recency")
|
|
# the cell the item asks for: +1 day stamp, reference supplied -> red
|
|
cell("timestamp-plus-one-day", {"manifest_mutate": shift_one_day},
|
|
["--now=" + true_stamp], 1, "FAIL extracted_at_utc recency")
|
|
# the Q122 opening defect: the stale constant, no reference needed
|
|
cell("stale-constant", {"manifest_mutate": set_stale}, [], 1,
|
|
"FAIL extracted_at_utc not the stale constant")
|
|
# malformed stamp
|
|
cell("malformed-stamp", {"manifest_mutate": set_garbage_stamp}, [], 1,
|
|
"FAIL extracted_at_utc format")
|
|
# re-derivation leg still bites
|
|
cell("tsv-title-corrupted", {"tsv_mutate": corrupt_title}, [], 1,
|
|
"FAIL qid+title identical to independent re-derive")
|
|
|
|
bad = [n for n, ok in cells if not ok]
|
|
print("")
|
|
if bad:
|
|
print("SELFTEST FAILED (%d of %d cells): %s" % (len(bad), len(cells), ", ".join(bad)))
|
|
return 1
|
|
print("SELFTEST OK (%d cells: 2 green controls, 4 red mutations)" % len(cells))
|
|
return 0
|
|
|
|
|
|
def main():
|
|
now_ref, do_selftest = parse_args(sys.argv[1:])
|
|
if do_selftest:
|
|
fails, _ = run(HERE, REPO_ROOT, None)
|
|
if fails:
|
|
print("")
|
|
print("SELFTEST ABORTED: the in-tree artifact is already red (%d check(s)); "
|
|
"fix the tree before grading the gate" % len(fails))
|
|
return 1
|
|
print("")
|
|
print("in-tree gate green; grading the legs")
|
|
print("")
|
|
return selftest()
|
|
|
|
fails, stats = run(HERE, REPO_ROOT, now_ref)
|
|
if fails:
|
|
print("")
|
|
print("%d check(s) FAILED" % len(fails))
|
|
return 1
|
|
|
|
print("")
|
|
print("ALL CHECKS PASSED (%d rows, %d distinct labels, %d reopened)"
|
|
% (stats["rows"], stats["distinct"], stats["reopened"]))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|