311 lines
12 KiB
Python
311 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""audit.py -- Q140: population audit of a directory census against a pinned rev.
|
|
|
|
The ledger lane's `2026-10-31-ledger-dir-census.csv` records
|
|
(rel_path, size_bytes, sha256) for every file it finds under
|
|
wizard-work/ledger, and `verify_ledger_dir_census.py` re-derives that list by
|
|
walking the *live worktree* and comparing row for row. That population is a
|
|
function of the host: it contains ignored `__pycache__/*.pyc` files (which the
|
|
tree's own .gitignore calls "never content a reader can review, diff or
|
|
re-run") and it changes whenever any Python run recompiles, or any lane writes
|
|
the CSV again.
|
|
|
|
This tool answers the question the census never asked, deterministically:
|
|
|
|
of the rows in the census at rev R, which ones name a blob that actually
|
|
exists in the repository at R, and which name a file that exists only on
|
|
the machine that generated it?
|
|
|
|
For every census row it reads the CSV at `<rev>:<csv>` (never the worktree) and
|
|
classifies it:
|
|
|
|
TRACKED_MATCH path is a blob at rev R, same size and same sha256
|
|
TRACKED_DRIFT path is a blob at rev R, size or sha256 differs
|
|
UNTRACKED_IGNORED path is not at R and is covered by the tree's ignore rule
|
|
UNTRACKED_OTHER path is not at R and is not ignored (an orphan row)
|
|
|
|
and it counts the reverse direction:
|
|
|
|
MISSING a blob at rev R under the censused directory with no row
|
|
|
|
`--check` compares every count against PIN (below) and exits 1 on any move, so
|
|
a stale pin and a drifted census are both visible. Exit codes: 0 ok, 1 pin
|
|
fault, 2 could not grade.
|
|
|
|
Usage:
|
|
python3 audit.py --repo . --rev <sha> --check
|
|
python3 audit.py --repo . --rev <sha> # report only
|
|
python3 audit.py --repo . --rev <sha> --json
|
|
python3 audit.py --repo . --rev <sha> --csv-file /tmp/mutant.csv # controls
|
|
"""
|
|
import argparse
|
|
import csv
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
DEFAULT_REPO = os.path.abspath(os.path.join(HERE, "..", "..", ".."))
|
|
DEFAULT_CSV = "wizard-work/ledger/2026-10-31-ledger-dir-census.csv"
|
|
CENSUS_DIR = "wizard-work/ledger"
|
|
|
|
# The pin: every count this tool prints for the rev it was measured at.
|
|
# Moving it is a deliberate edit made with the commit that moved the census.
|
|
PIN = {
|
|
"rev": "78f1139dd681f1b41c69463103a6c05a13490282",
|
|
"csv_blob": "a27e0193f7a1b8a888eeef66f56677503934cf9a",
|
|
"csv_sha256": "e52e907b14a8f0d617a810ff04d6830aa1ec5aa9ab592d9d159e24ea2ddf7aa5",
|
|
"rows": 594,
|
|
"tracked_in_population": 575,
|
|
"tracked_match": 574,
|
|
"tracked_drift": 0,
|
|
"untracked_ignored": 20,
|
|
"untracked_other": 0,
|
|
"missing": 0, # the census CSV itself is excluded on both sides
|
|
}
|
|
|
|
|
|
class Ungradable(Exception):
|
|
"""Raised when the rev or the CSV cannot be read: exit 2, never exit 1."""
|
|
|
|
|
|
def git(repo, *args, stdin=None, check=True):
|
|
proc = subprocess.run(["git", "-C", repo, *args], input=stdin,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
if check and proc.returncode != 0:
|
|
raise Ungradable("git %s -> rc %d: %s" % (
|
|
" ".join(args), proc.returncode,
|
|
proc.stderr.decode("utf-8", "replace").strip()[:400]))
|
|
return proc
|
|
|
|
|
|
def resolve_rev(repo, rev):
|
|
proc = git(repo, "rev-parse", "--verify", rev + "^{commit}", check=False)
|
|
if proc.returncode != 0:
|
|
raise Ungradable("rev %r is not a commit in %s" % (rev, repo))
|
|
return proc.stdout.decode().strip()
|
|
|
|
|
|
def read_csv_at_rev(repo, rev, rel):
|
|
proc = git(repo, "rev-parse", "--verify", "%s:%s" % (rev, rel), check=False)
|
|
if proc.returncode != 0:
|
|
raise Ungradable("%s has no %s" % (rev, rel))
|
|
blob = proc.stdout.decode().strip()
|
|
text = git(repo, "cat-file", "blob", blob).stdout.decode("utf-8")
|
|
return blob, text
|
|
|
|
|
|
def read_csv_file(path):
|
|
with open(path, encoding="utf-8", newline="") as f:
|
|
return f.read()
|
|
|
|
|
|
def parse_rows(text):
|
|
r = csv.reader(io.StringIO(text, newline=""))
|
|
try:
|
|
header = next(r)
|
|
except StopIteration:
|
|
raise Ungradable("census csv is empty")
|
|
if header != ["rel_path", "size_bytes", "sha256"]:
|
|
raise Ungradable("census header is %r, expected rel_path,size_bytes,sha256"
|
|
% (header,))
|
|
rows = []
|
|
for lineno, row in enumerate(r, start=2):
|
|
if not row:
|
|
continue
|
|
if len(row) != 3:
|
|
raise Ungradable("census line %d has %d fields" % (lineno, len(row)))
|
|
rows.append(tuple(row))
|
|
return rows
|
|
|
|
|
|
def tracked_population(repo, rev, directory):
|
|
"""path -> (size, blob_oid, sha256) for every blob at rev under directory."""
|
|
out = git(repo, "ls-tree", "-r", "-l", "--full-tree", rev, "--", directory).stdout
|
|
entries = []
|
|
for line in out.decode("utf-8").splitlines():
|
|
if not line.strip():
|
|
continue
|
|
meta, path = line.split("\t", 1)
|
|
parts = meta.split()
|
|
if len(parts) != 4:
|
|
raise Ungradable("unexpected ls-tree row: %r" % (line,))
|
|
mode, typ, oid, size = parts
|
|
if typ != "blob":
|
|
continue
|
|
entries.append((path, int(size), oid))
|
|
|
|
# one cat-file --batch for the whole population, not one subprocess per blob
|
|
stdin = "".join(oid + "\n" for _, _, oid in entries).encode()
|
|
proc = subprocess.run(["git", "-C", repo, "cat-file", "--batch"],
|
|
input=stdin, stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE)
|
|
if proc.returncode != 0:
|
|
raise Ungradable("git cat-file --batch -> rc %d: %s" % (
|
|
proc.returncode, proc.stderr.decode("utf-8", "replace")[:400]))
|
|
buf = proc.stdout
|
|
pos = 0
|
|
digests = []
|
|
for _ in entries:
|
|
nl = buf.index(b"\n", pos)
|
|
header = buf[pos:nl].decode("utf-8", "replace").split()
|
|
pos = nl + 1
|
|
if len(header) < 3 or header[1] != "blob":
|
|
raise Ungradable("unexpected batch header: %r" % (header,))
|
|
size = int(header[2])
|
|
content = buf[pos:pos + size]
|
|
pos += size + 1 # trailing newline
|
|
digests.append(hashlib.sha256(content).hexdigest())
|
|
pop = {}
|
|
for (path, size, _oid), digest in zip(entries, digests):
|
|
pop[path] = (size, digest)
|
|
return pop
|
|
|
|
|
|
def ignored_set(repo, paths):
|
|
"""Measured with the tree's own rule; pattern-classified if git cannot read it."""
|
|
if not paths:
|
|
return set()
|
|
proc = git(repo, "check-ignore", "--stdin", check=False, stdin=(
|
|
"".join(p + "\n" for p in paths)).encode())
|
|
if proc.returncode not in (0, 1):
|
|
return None
|
|
return set(l for l in proc.stdout.decode("utf-8", "replace").splitlines() if l)
|
|
|
|
|
|
def classify(rows, pop, csv_rel):
|
|
counts = {"rows": len(rows), "tracked_in_population": len(pop),
|
|
"tracked_match": 0, "tracked_drift": 0,
|
|
"untracked_ignored": 0, "untracked_other": 0, "missing": 0}
|
|
drift, untracked_ignored, untracked_other = [], [], []
|
|
seen = set()
|
|
for path, size, sha in rows:
|
|
seen.add(path)
|
|
if path in pop:
|
|
tsize, tdigest = pop[path]
|
|
if int(size) == tsize and sha.strip().lower() == tdigest:
|
|
counts["tracked_match"] += 1
|
|
else:
|
|
counts["tracked_drift"] += 1
|
|
drift.append({"path": path, "census": [int(size), sha],
|
|
"rev": [tsize, tdigest]})
|
|
elif "__pycache__/" in path or path.endswith((".pyc", ".pyo")):
|
|
counts["untracked_ignored"] += 1
|
|
untracked_ignored.append(path)
|
|
else:
|
|
counts["untracked_other"] += 1
|
|
untracked_other.append(path)
|
|
missing = sorted(p for p in pop if p not in seen and p != csv_rel)
|
|
counts["missing"] = len(missing)
|
|
return counts, {"drift": drift, "untracked_ignored": untracked_ignored,
|
|
"untracked_other": untracked_other, "missing": missing}
|
|
|
|
|
|
def faults(counts, expected):
|
|
out = []
|
|
for key in ("rows", "tracked_in_population", "tracked_match",
|
|
"tracked_drift", "untracked_ignored", "untracked_other",
|
|
"missing"):
|
|
if counts[key] != expected[key]:
|
|
out.append("count %s moved %d -> %d" % (key, expected[key], counts[key]))
|
|
return out
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(add_help=True,
|
|
description="population audit of a directory census")
|
|
ap.add_argument("--repo", default=DEFAULT_REPO)
|
|
ap.add_argument("--rev", default="HEAD")
|
|
ap.add_argument("--csv", default=DEFAULT_CSV, help="census path relative to repo")
|
|
ap.add_argument("--csv-file", default=None,
|
|
help="read rows from this file instead of <rev>:<csv> (controls)")
|
|
ap.add_argument("--dir", default=CENSUS_DIR, help="population directory at rev")
|
|
ap.add_argument("--check", action="store_true", help="compare counts against PIN")
|
|
ap.add_argument("--json", action="store_true", help="emit machine-readable result")
|
|
args = ap.parse_args(argv)
|
|
|
|
repo = os.path.abspath(args.repo)
|
|
try:
|
|
rev = resolve_rev(repo, args.rev)
|
|
if args.csv_file:
|
|
blob, text = "(worktree)", read_csv_file(args.csv_file)
|
|
else:
|
|
blob, text = read_csv_at_rev(repo, rev, args.csv)
|
|
rows = parse_rows(text)
|
|
pop = tracked_population(repo, rev, args.dir)
|
|
counts, detail = classify(rows, pop, args.csv)
|
|
except Ungradable as exc:
|
|
print("UNGRADABLE: %s" % exc)
|
|
return 2
|
|
|
|
ignored = ignored_set(repo, detail["untracked_ignored"] + detail["untracked_other"])
|
|
rules_agree = None
|
|
if ignored is not None:
|
|
by_pattern = set(detail["untracked_ignored"])
|
|
rules_agree = by_pattern == ignored
|
|
|
|
result = {"repo": repo, "rev": rev, "csv": args.csv, "csv_blob": blob,
|
|
"csv_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
|
|
"counts": counts, "detail": detail,
|
|
"ignore_rule_agrees": rules_agree,
|
|
"population_is_tracked_only": counts["rows"] == counts["tracked_in_population"] - 1
|
|
and counts["untracked_ignored"] == 0 and counts["untracked_other"] == 0}
|
|
|
|
if args.json:
|
|
print(json.dumps(result, indent=2, sort_keys=True))
|
|
else:
|
|
print("rev %s csv %s" % (rev[:12], args.csv))
|
|
print(" population tracked at rev: %d" % counts["tracked_in_population"])
|
|
print(" rows in census: %d" % counts["rows"])
|
|
print(" tracked match: %d" % counts["tracked_match"])
|
|
print(" tracked drift: %d" % counts["tracked_drift"])
|
|
print(" untracked (ignored): %d" % counts["untracked_ignored"])
|
|
print(" untracked (not ignored): %d" % counts["untracked_other"])
|
|
print(" tracked blobs with no row: %d" % counts["missing"])
|
|
if rules_agree is not None:
|
|
print(" tree ignore rule agrees: %s" % ("yes" if rules_agree else "NO"))
|
|
if detail["untracked_other"]:
|
|
for p in detail["untracked_other"][:10]:
|
|
print(" orphan row: %s" % p)
|
|
if detail["untracked_ignored"]:
|
|
print(" ignored rows (first 3):")
|
|
for p in detail["untracked_ignored"][:3]:
|
|
print(" %s" % p)
|
|
if detail["missing"]:
|
|
print(" unrowed blobs (first 3):")
|
|
for p in detail["missing"][:3]:
|
|
print(" %s" % p)
|
|
print(" population is tracked-only: %s" % (
|
|
"yes" if result["population_is_tracked_only"] else "NO"))
|
|
|
|
if not args.check:
|
|
return 0
|
|
|
|
expected = dict(PIN)
|
|
expected_rev = expected.pop("rev")
|
|
faults_found = faults(counts, expected)
|
|
if not args.csv_file:
|
|
if rev != expected_rev:
|
|
faults_found.insert(0, "rev moved %s -> %s" % (expected_rev[:12], rev[:12]))
|
|
if blob != expected.pop("csv_blob"):
|
|
faults_found.insert(1, "csv blob moved %s -> %s" % (
|
|
PIN["csv_blob"][:12], blob[:12]))
|
|
if faults_found:
|
|
print("FAULT: %d" % len(faults_found))
|
|
for f in faults_found:
|
|
print(" " + f)
|
|
return 1
|
|
print("CHECK OK: %d rows = %d tracked + %d ignored (%d orphan), %d unrowed, "
|
|
"counts match the pin" % (
|
|
counts["rows"], counts["tracked_match"],
|
|
counts["untracked_ignored"], counts["untracked_other"],
|
|
counts["missing"]))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|