197 lines
7.5 KiB
Python
197 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent, pin-bound verification of the 2026-11-02 workqueue-census.
|
|
|
|
RE-APPROACH: this grader does NOT import the extractor. It reads the queue bytes
|
|
that the manifest pins -- `git cat-file blob <source_blob_sha>` -- and re-derives
|
|
the expected (Q, Title) rows with plain string operations, then compares them
|
|
cell for cell against the committed TSV. Grading the pinned blob rather than the
|
|
worktree is the difference between a gate on an artifact and a gate on today's
|
|
file: the queue is append-only, so every later item would otherwise redden a
|
|
census whose rows were never wrong.
|
|
|
|
WHAT IT GRADES
|
|
1. pin-readable the pinned object exists and is a blob (else rc 2: cannot
|
|
grade -- neither a pass nor a fault)
|
|
2. pin-digest sha256(blob) == manifest.source_sha256
|
|
3. pin-matches-rev `git rev-parse <source_rev>:WORK-QUEUE.md` == source_blob_sha
|
|
4. row-count re-derived rows from the blob == committed TSV rows
|
|
5. q-title every (Q, Title) pair identical, in order
|
|
6. headings `## Q<n> -- ` headings in the blob == committed rows
|
|
7. manifest rows / columns / artifact / generated_by / stamp shape
|
|
8. cells no empty Q or Title; no literal tab inside Status or Target
|
|
9. drift ADVISORY only: the worktree queue differs from the pin
|
|
(reported, never graded -- a snapshot must not red on the
|
|
queue's growth)
|
|
|
|
Run: bash CHECK (the tree's invocation; wrapper over this file)
|
|
python3 grade_census.py
|
|
Exit: 0 green, 1 a fault, 2 could not grade.
|
|
"""
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(HERE)))
|
|
TSV_NAME = "workqueue-census.tsv"
|
|
MANIFEST_NAME = "extract_manifest.json"
|
|
SRC_REL = "WORK-QUEUE.md"
|
|
TS_FMT = "%Y-%m-%dT%H:%M:%SZ"
|
|
SEP = "\u2014" # em dash
|
|
HEADER = ["Q", "Title", "Status", "Target"]
|
|
|
|
|
|
def git(*args, stdin_bytes=None):
|
|
return subprocess.run(
|
|
["git", "-C", REPO_ROOT] + list(args),
|
|
input=stdin_bytes, capture_output=True, check=False)
|
|
|
|
|
|
def rederive(text):
|
|
"""Independent derivation: emit (qid, title) for every '## Q<n> -- <title>'."""
|
|
golden = []
|
|
qid, title = None, ""
|
|
for line in text.splitlines():
|
|
if not line.startswith("## "):
|
|
continue
|
|
rest = line[3:].strip()
|
|
if " " + SEP + " " not in rest:
|
|
continue
|
|
pid, _, t = rest.partition(" " + SEP + " ")
|
|
if not (pid.startswith("Q") and pid[1:].isdigit()):
|
|
continue
|
|
t = re.sub(r"\s*\*.*$", "", t).strip()
|
|
if qid is not None:
|
|
golden.append((qid, title))
|
|
qid, title = pid[1:], t
|
|
if qid is not None:
|
|
golden.append((qid, title))
|
|
return golden
|
|
|
|
|
|
def heading_count(text):
|
|
n = 0
|
|
for line in text.splitlines():
|
|
rest = line[3:].strip() if line.startswith("## ") else ""
|
|
if not rest or (" " + SEP + " ") not in rest:
|
|
continue
|
|
pid = rest.split(" " + SEP + " ", 1)[0]
|
|
if pid.startswith("Q") and pid[1:].isdigit():
|
|
n += 1
|
|
return n
|
|
|
|
|
|
def main():
|
|
fails = []
|
|
|
|
def check(name, cond, detail=""):
|
|
print(("PASS " if cond else "FAIL ") + name
|
|
+ ("" if cond else " " + repr(detail)))
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
try:
|
|
man = json.load(open(os.path.join(HERE, MANIFEST_NAME), encoding="utf-8"))
|
|
except (OSError, ValueError) as exc:
|
|
print("CANNOT GRADE: manifest unreadable: %r" % (exc,))
|
|
return 2
|
|
|
|
blob_sha = man.get("source_blob_sha")
|
|
rev = man.get("source_rev")
|
|
if not (isinstance(blob_sha, str) and re.fullmatch(r"[0-9a-f]{40}", blob_sha or "")
|
|
and isinstance(rev, str) and rev):
|
|
print("CANNOT GRADE: manifest pins no (source_rev, source_blob_sha): %r"
|
|
% ({"source_rev": rev, "source_blob_sha": blob_sha},))
|
|
return 2
|
|
|
|
kind = git("cat-file", "-t", blob_sha)
|
|
if kind.returncode != 0 or kind.stdout.strip() != b"blob":
|
|
print("CANNOT GRADE: pinned object %s is not a readable blob" % blob_sha)
|
|
return 2
|
|
blob = git("cat-file", "blob", blob_sha).stdout
|
|
text = blob.decode("utf-8")
|
|
check("pin-readable", True, "")
|
|
|
|
digest = hashlib.sha256(blob).hexdigest()
|
|
check("pin-digest", digest == man.get("source_sha256"),
|
|
"blob=%s manifest=%r" % (digest, man.get("source_sha256")))
|
|
|
|
at_rev = git("rev-parse", "%s:%s" % (rev, SRC_REL)).stdout.decode().strip()
|
|
check("pin-matches-rev", at_rev == blob_sha,
|
|
"rev %s holds %r" % (rev, at_rev))
|
|
|
|
check("manifest.source is repo-relative", man.get("source") == SRC_REL,
|
|
repr(man.get("source")))
|
|
|
|
try:
|
|
with open(os.path.join(HERE, TSV_NAME), encoding="utf-8", newline="") as fh:
|
|
rows = list(csv.reader(fh, delimiter="\t"))
|
|
except OSError as exc:
|
|
print("CANNOT GRADE: committed TSV unreadable: %r" % (exc,))
|
|
return 2
|
|
header, committed = rows[0], rows[1:]
|
|
check("tsv header", header == HEADER, repr(header))
|
|
|
|
golden = rederive(text)
|
|
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 len(c) < 2 or c[0] != g[0] or c[1] != g[1]]
|
|
check("q-title identical to independent re-derive", not mism,
|
|
"first mismatch row %r" % (mism[0] if mism else None))
|
|
check("headings == committed rows",
|
|
heading_count(text) == len(committed),
|
|
"headings=%d rows=%d" % (heading_count(text), len(committed)))
|
|
|
|
check("manifest.rows == re-derived rows", man.get("rows") == len(golden),
|
|
"manifest=%r golden=%d" % (man.get("rows"), len(golden)))
|
|
check("manifest.columns", man.get("columns") == HEADER, repr(man.get("columns")))
|
|
check("manifest.artifact", man.get("artifact") == TSV_NAME,
|
|
repr(man.get("artifact")))
|
|
check("manifest.generated_by present",
|
|
isinstance(man.get("generated_by"), str) and bool(man["generated_by"]),
|
|
repr(man.get("generated_by")))
|
|
stamp = man.get("extracted_at_utc", "")
|
|
ok_stamp = False
|
|
if isinstance(stamp, str) and stamp:
|
|
try:
|
|
datetime.strptime(stamp, TS_FMT)
|
|
ok_stamp = True
|
|
except ValueError:
|
|
pass
|
|
check("manifest.extracted_at_utc parseable UTC", ok_stamp, repr(stamp))
|
|
|
|
empty = [r for r in committed if len(r) < 2 or not r[0] or not r[1]]
|
|
check("no empty q/title", not empty, repr(empty[:1]))
|
|
check("status carries no literal tab",
|
|
not [r for r in committed if len(r) >= 3 and "\t" in r[2]], "")
|
|
check("target carries no literal tab",
|
|
not [r for r in committed if len(r) >= 4 and "\t" in r[3]], "")
|
|
|
|
live = os.path.join(REPO_ROOT, SRC_REL)
|
|
try:
|
|
with open(live, "rb") as fh:
|
|
live_digest = hashlib.sha256(fh.read()).hexdigest()
|
|
except OSError:
|
|
live_digest = None
|
|
if live_digest != digest:
|
|
print("ADVISORY source-moved worktree=%s pinned=%s"
|
|
% ((live_digest or "absent")[:12], digest[:12]))
|
|
|
|
print()
|
|
if fails:
|
|
print("%d check(s) FAILED" % len(fails))
|
|
return 1
|
|
print("ALL CHECKS PASSED (%d rows, %d distinct q-numbers, pin %s@%s)"
|
|
% (len(committed), len({g[0] for g in golden}), blob_sha[:7], rev[:7]))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|