Files
troupe-standup/replies/q60_truncated_census.py
T

472 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""Q60 — census of truncated identifiers in the archive, and the grading of each one.
A truncated identifier is a token written as a prefix followed by an ellipsis: `beae0b1f-…`.
This tool does three things, and prints its own matcher for each.
census every truncated token in the tracked tree at a rev, by file:line, with the counts in both
units the archive keeps confusing: occurrences and distinct tokens.
table the same measurement rendered as the markdown table the queue item asks for.
check re-measure everything and grade a written table against it. A row naming a token the
census does not contain, a row that grades a token differently from the measurement, or a
census token with no row is a fault. This is the artifact's CHECK line.
rule the item's enforcement clause, measured instead of stated: every line that carries a
truncated token *and* a probe/deadness verdict (`→ 200`, `404`, `URL DEAD`, `dead link`,
`no longer resolves`) is graded by whether the address it attributes that verdict to is
written. CARRIES when the full value is on the line; RECOVERABLE when the value survives
elsewhere in the tree or in history (the row can be repaired now, so it is a fault);
EXHAUSTED when it survives nowhere (the row is Testimony only, graded at write time and
not repaired later by inference). `--baseline` is the ledger: a fault row a reader has
already reviewed, printed on every run with its owner, never dropped in silence.
Search ladder for one token, in order, all of it printed per token:
S1 tree full UUID-shaped values in every tracked file at the rev
S2 extend a longer prefix on record -- the same leading group followed by more hex and an
ellipsis: a site that narrowed the truncation without completing it
S3 history `git log --all -S<prefix>` -> every commit whose patch mentions the prefix, then the
full UUID shape re-scanned in each such commit's tree
Scope is the archive: tracked bytes at the rev. An untracked file on this host is not a record, so it
cannot make a row checkable; the worktree was swept separately, by hand, in the artifact's report.
One pass over the tree: blobs are read through `git ls-tree` + `git cat-file --batch` rather than one
`git grep` per token, because the per-token extension pattern walked this repository for 180 s
without finishing (unbounded `(-[0-9a-f]{4,})*` backtracks on a line holding a long hex run and no
ellipsis). The quantifiers here are bounded.
Exit codes follow the tools/ contract: 0 clean, 1 ran and graded a fault, 2 could not grade.
"""
from __future__ import annotations
import argparse
import os
import re
import subprocess
import sys
ELL = "\u2026"
DOGS = "..."
CANON_RE = r"[0-9a-f]{6,12}(?:-[0-9a-f]{1,12}){0,3}-" + ELL
ANY_RE = r"[0-9a-f]{6,12}(?:-[0-9a-f]{1,12}){0,3}-(?:" + ELL + r"|\.\.\.)"
CANON = re.compile(CANON_RE)
ANY = re.compile(ANY_RE)
UUID_RE = r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
UUID = re.compile(UUID_RE)
# The verdict vocabulary for `rule`, printed by the tool so a reader can audit what fired. A row
# that attributes a read to an address is only as checkable as the address it writes; these are the
# shapes this archive uses to attribute a read. The status-code pattern excludes a digit or a dash
# on either side so `$150–200/hr` is not a status, while `(200 re-check`, `→ 200` and `200, 37,412 B`
# are. Every cue that fired on a row is printed with the row.
VERDICT_CUES = (
("status", re.compile(r"(?<![\$0-9\u2013-])\b(?:200|201|204|301|302|304|400|403|404|410|429|500|502|503)\b(?![0-9])")),
("URL-DEAD", re.compile(r"\bURL[\s-]*DEAD\b", re.I)),
("dead-link", re.compile(r"\bdead[\s-]link\b", re.I)),
("no-longer-resolves", re.compile(r"no longer (?:resolves|resolving)", re.I)),
("does-not-resolve", re.compile(r"(?:does not|doesn't|did not) (?:resolve|resolves)", re.I)),
)
# A truncated token written *inside* a URL-ish token: `jobs.micro1.ai/post/3d79ba82-…`. Advisory
# only -- `rule` grades it, `census`/`table` do not depend on it.
URL_CTX = re.compile(r"(?:[A-Za-z0-9_.:-]+/)+[A-Za-z0-9_./:%-]*$")
RULE_CLASSES = ("CARRIES", "RECOVERABLE", "RECOVERABLE-HIST", "EXHAUSTED")
def git(repo, *args):
p = subprocess.run(["git", *args], cwd=repo, capture_output=True, text=True)
if p.returncode != 0:
raise SystemExit(f"cannot grade: git {' '.join(args)} rc={p.returncode}: "
f"{p.stderr.strip()[:300]}")
return p.stdout
def iter_tree(repo, rev):
"""Yield (path, lineno, text) for every text line of every tracked file at `rev`, one pass."""
listing = git(repo, "ls-tree", "-r", "-z", "--full-tree", rev)
entries = []
for chunk in listing.split("\0"):
if not chunk:
continue
meta, path = chunk.split("\t", 1)
mode, otype, sha = meta.split()
if otype == "blob" and mode in ("100644", "100755"):
entries.append((sha, path))
if not entries:
raise SystemExit(f"cannot grade: no blobs at {rev}")
proc = subprocess.Popen(["git", "cat-file", "--batch"], cwd=repo,
stdin=subprocess.PIPE, stdout=subprocess.PIPE)
stdin = b"".join(sha.encode() + b"\n" for sha, _ in entries)
out, _ = proc.communicate(stdin)
pos = 0
for sha, path in entries:
nl = out.index(b"\n", pos)
header = out[pos:nl].split()
size = int(header[2])
blob = out[nl + 1: nl + 1 + size]
pos = nl + 1 + size + 1
if b"\x00" in blob:
continue
for i, line in enumerate(blob.decode("utf-8", "replace").splitlines(), 1):
yield path, i, line
def measure(repo, rev):
trunc = [] # [(path, lineno, token)] -- canonical: prefix-… as the item defines it
uuid_index = {} # first 8 hex -> [(path, lineno, uuid)]
ext_index = {} # first 8 hex -> [(path, lineno, written)] -- same id, more hex, still cut
text_of = {} # (path, lineno) -> line text, only for lines carrying a canonical token
for path, lineno, line in iter_tree(repo, rev):
canon_spans = []
for m in CANON.finditer(line):
tok = m.group(0)
trunc.append((path, lineno, tok))
canon_spans.append((m.start(), m.end()))
if canon_spans:
text_of[(path, lineno)] = line
for m in UUID.finditer(line):
uuid_index.setdefault(m.group(0)[:8], []).append((path, lineno, m.group(0)))
for m in ANY.finditer(line):
written = m.group(0)
if written.count("-") < 2:
continue # leading group + ellipsis only: already a canonical token above
ext_index.setdefault(written.split("-")[0], []).append((path, lineno, written))
tokens = sorted({t for _, _, t in trunc})
sites = {}
for path, lineno, token in trunc:
sites.setdefault(token, []).append((path, lineno))
files = sorted({p for p, _, _ in trunc})
return trunc, tokens, sites, uuid_index, ext_index, files, text_of
def prefix_of(token):
return token.split("-")[0]
def placeholder(uuid):
"""True for a UUID-shaped string that is not a value: all-zero groups, or a repeated character.
`beae0b1f-0000-0000-0000-000000000000` is on record at
`replies/2026-10-24-to-zephyrus-86.md:12` as the near-miss probe whose 404 proves nothing. It
has the prefix, it is a syntactically valid UUID, and it is not a completion; promoting it to
"full value present" is the exact failure the queue item calls out, so the classifier refuses it
by shape rather than by the judgement of whoever wrote the row.
"""
body = uuid.replace("-", "")
return len(set(body)) <= 2 or all(set(g) == {"0"} for g in uuid.split("-")[1:])
def history(repo, prefix):
out = git(repo, "log", "--all", "-S" + prefix, "--format=%H %s")
return [ln for ln in out.splitlines() if ln.strip()]
def history_full(repo, prefix, commits):
pat = prefix + r"-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
real, fake = [], []
for line in commits:
sha = line.split()[0]
p = subprocess.run(["git", "grep", "-nE", "-a", pat, sha, "--", "."],
cwd=repo, capture_output=True, text=True)
if p.returncode == 0:
for row in p.stdout.splitlines():
parts = row.split(":", 3)
if len(parts) != 4:
continue
found = UUID.search(parts[3])
if not found:
continue
rec = (sha[:12], parts[1], int(parts[2]), found.group(0))
(fake if placeholder(found.group(0)) else real).append(rec)
return real, fake
def resolve(repo, tokens, uuid_index, ext_index):
out = {}
for token in tokens:
p = prefix_of(token)
candidates = [r for r in uuid_index.get(p, []) if r[2].startswith(p)]
full = [r for r in candidates if not placeholder(r[2])]
fake = [r for r in candidates if placeholder(r[2])]
ext = [r for r in ext_index.get(p, []) if r[2] != token]
commits = [] if full else history(repo, p)
hist, hist_fake = history_full(repo, p, commits) if commits else ([], [])
if full:
grade = "FULL"
elif hist:
grade = "HIST"
else:
grade = "UNRECOVERED"
out[token] = {"sites": [], "full": full, "fake": fake, "ext": ext,
"commits": commits, "hist": hist, "hist_fake": hist_fake, "grade": grade}
return out
def fmt_sites(sites):
return " · ".join(f"`{p}:{n}`" for p, n in sites)
def fmt_grade(token, info):
if info["grade"] == "FULL":
where = " · ".join(f"`{p}:{n}`" for p, n, _ in info["full"])
return f"full value present at {where} (`{info['full'][0][2]}`)"
if info["grade"] == "HIST":
where = " · ".join(f"`{c}:{p}:{n}`" for c, p, n, _ in info["hist"])
return f"full value present at {where} (historic tree, not at HEAD)"
bits = []
if info["ext"]:
bits.append("prefix extended, still truncated: " + " · ".join(
f"`{p}:{n}` → `{t}`" for p, n, t in info["ext"]))
if info["fake"]:
bits.append("S1 tree: only placeholder " + " · ".join(
f"`{p}:{n}` = `{u}` (not a completion)" for p, n, u in info["fake"]))
else:
bits.append("S1 tree: no full UUID with this prefix at any path at this rev")
hist_fake = info.get("hist_fake") or []
bits.append(f"S3 history: `git log --all -S{prefix_of(token)}` → "
f"{len(info['commits'])} commit(s) whose patch mentions the prefix; "
+ ("only placeholders in their trees"
if hist_fake else "no full value in any of their trees"))
return "**Unrecovered** — " + "; ".join(bits)
def cmd_census(args, repo):
trunc, tokens, sites, uuid_index, ext_index, files, _text = measure(repo, args.rev)
print(f"matcher: git grep -nE '{CANON_RE}' {args.rev} -- . (canonical: ellipsis only)")
print(f"rev {args.rev}: {len(trunc)} occurrence(s), {len(tokens)} distinct token(s), "
f"{len(files)} file(s)")
info = resolve(repo, tokens, uuid_index, ext_index)
for token in tokens:
i = info[token]
print(f"\n{token} [{i['grade']}] {len(sites[token])} site(s)")
for path, lineno in sites[token]:
print(f" {path}:{lineno}")
for path, lineno, uuid in i["full"]:
print(f" FULL {path}:{lineno} {uuid}")
for path, lineno, uuid in i["fake"]:
print(f" FAKE {path}:{lineno} {uuid} (placeholder, not a completion)")
for path, lineno, written in i["ext"]:
print(f" EXT {path}:{lineno} {written}")
print(f" hist {len(i['commits'])} commit(s)")
return 0
def cmd_table(args, repo):
trunc, tokens, sites, uuid_index, ext_index, files, _text = measure(repo, args.rev)
info = resolve(repo, tokens, uuid_index, ext_index)
lines = [f"<!-- generated: wizard-work/volt/q60_truncated_census.py table "
f"--rev {args.rev} -->",
"",
f"Census at `{args.rev}`: **{len(trunc)} occurrences / {len(tokens)} distinct "
f"tokens / {len(files)} files**.",
"",
"| token (as written) | file:line sites | grading |",
"|---|---|---|"]
for token in tokens:
info[token]["sites"] = sites[token]
lines.append(f"| `{token}` | {fmt_sites(sites[token])} | {fmt_grade(token, info[token])} |")
text = "\n".join(lines) + "\n"
if args.out:
with open(args.out, "w", encoding="utf-8") as fh:
fh.write(text)
print(f"wrote {args.out}: {len(tokens)} row(s)")
else:
sys.stdout.write(text)
return 0
ROW_RE = re.compile(r"^\| `([0-9a-f]{6,12}(?:-[0-9a-f]{1,12})*-\u2026)` \| (.*?) \| (.*) \|$")
def cmd_check(args, repo):
try:
with open(args.table, "r", encoding="utf-8") as fh:
text = fh.read()
except OSError as exc:
print(f"cannot grade: {args.table}: {exc}", file=sys.stderr)
return 2
trunc, tokens, sites, uuid_index, ext_index, files, _text = measure(repo, args.rev)
info = resolve(repo, tokens, uuid_index, ext_index)
rows = {}
for line in text.splitlines():
m = ROW_RE.match(line.strip())
if not m:
continue
if m.group(1) in rows:
print(f"FAULT duplicate row for {m.group(1)}", file=sys.stderr)
return 1
rows[m.group(1)] = (m.group(2), m.group(3))
problems = []
for token in tokens:
if token not in rows:
problems.append(f"missing row: {token}")
continue
sites_cell, grade_cell = rows[token]
for path, lineno in sites[token]:
if f"`{path}:{lineno}`" not in sites_cell:
problems.append(f"{token}: site not named in row: {path}:{lineno}")
g = info[token]["grade"]
if g == "FULL":
if "Unrecovered" in grade_cell:
problems.append(f"{token}: measured FULL, row says Unrecovered")
for path, lineno, _ in info[token]["full"]:
if f"`{path}:{lineno}`" not in grade_cell:
problems.append(f"{token}: full value at {path}:{lineno} not graded present")
else:
if "Unrecovered" not in grade_cell:
problems.append(f"{token}: measured {g}, row does not say Unrecovered")
if "full value present" in grade_cell:
problems.append(f"{token}: measured {g}, row claims a full value")
for path, lineno, _ in info[token]["fake"]:
if f"`{path}:{lineno}`" not in grade_cell:
problems.append(f"{token}: placeholder at {path}:{lineno} not named as one")
for token in rows:
if token not in tokens:
problems.append(f"row names a token the census does not contain: {token}")
if problems:
print(f"q60 truncated census: {len(problems)} fault(s) against {args.table} "
f"({len(tokens)} token(s) measured at {args.rev})", file=sys.stderr)
for p in problems:
print(f" {p}", file=sys.stderr)
return 1
full = sum(1 for t in tokens if info[t]["grade"] == "FULL")
print(f"q60 truncated census: {len(tokens)} token(s) at {args.rev}, {len(rows)} row(s), "
f"{full} full / {len(tokens) - full} Unrecovered — every row's site list and grading "
f"agrees with a fresh measurement")
return 0
def rule_rows(repo, rev):
"""One row per (line, token) where the line attributes a read and writes the address truncated."""
trunc, tokens, sites, uuid_index, ext_index, files, text_of = measure(repo, rev)
info = resolve(repo, tokens, uuid_index, ext_index)
out = []
for (path, lineno), line in sorted(text_of.items()):
cues = sorted({name for name, rx in VERDICT_CUES if rx.search(line)})
if not cues:
continue
full_on_line = {u[:8] for u in
(m.group(0) for m in UUID.finditer(line)) if not placeholder(u)}
for m in CANON.finditer(line):
tok = m.group(0)
p = prefix_of(tok)
i = info[tok]
if p in full_on_line:
cls = "CARRIES"
where = [r for r in i["full"] if r[0] == path and r[1] == lineno] or i["full"]
elif i["full"]:
cls = "RECOVERABLE"
where = i["full"]
elif i.get("hist"):
cls = "RECOVERABLE-HIST"
where = [("{}:{}:{}".format(c.split()[0][:12], pp, nn), pp, nn, uu)
for c, pp, nn, uu in i["hist"]]
else:
cls = "EXHAUSTED"
where = []
in_url = bool(URL_CTX.search(line[:m.start()]))
out.append({"path": path, "line": lineno, "token": tok, "cues": cues,
"class": cls, "where": where, "in_url": in_url, "text": line.strip()})
return out, tokens, info
def load_rule_baseline(path):
"""`<path>:<line>\\t<token>\\t<CLASS>\\t<owner> -- <reason>`; blank lines and # comments ignored."""
rows = {}
with open(path, "r", encoding="utf-8") as fh:
for raw in fh:
line = raw.rstrip("\n")
if not line.strip() or line.lstrip().startswith("#"):
continue
parts = line.split("\t")
if len(parts) < 4:
raise SystemExit(f"cannot grade: baseline row needs 4 fields, got {len(parts)}: {line!r}")
key, tok, cls, note = parts[0], parts[1], parts[2], parts[3]
if cls not in RULE_CLASSES:
raise SystemExit(f"cannot grade: baseline class {cls!r} not in {RULE_CLASSES}")
rows[(key, tok)] = (cls, note)
return rows
def cmd_rule(args, repo):
rows, tokens, info = rule_rows(repo, args.rev)
print("matcher: cue vocabulary = " + ", ".join(f"{n}({rx.pattern})" for n, rx in VERDICT_CUES))
print(f"rev {args.rev}: {len(rows)} verdict occurrence(s) on "
f"{len({(r['path'], r['line']) for r in rows})} line(s), "
f"{len({r['token'] for r in rows})} distinct token(s), "
f"{len({r['path'] for r in rows})} file(s)\n")
baseline = load_rule_baseline(args.baseline) if args.baseline else {}
faults, tolerated, cleared = [], [], []
for r in rows:
key = (f"{r['path']}:{r['line']}", r["token"])
where = ""
if r["where"]:
sites = " · ".join(f"`{p}:{n}`" for p, n, *_ in r["where"][:6])
val = ""
if r["where"] and len(r["where"][0]) == 4:
val = f" (`{r['where'][0][3]}`)"
where = f" — value at {sites}{val}"
flag = " url-context" if r["in_url"] else ""
body = f"{r['path']}:{r['line']} `{r['token']}` cues={{{','.join(r['cues'])}}}{flag}"
head = f"{r['class']:<16} {body}"
if r["class"] in ("RECOVERABLE", "RECOVERABLE-HIST", "EXHAUSTED"):
if key in baseline:
cls, note = baseline[key]
mark = "KNOWN" if cls == r["class"] else f"DRIFT({cls})"
tolerated.append((mark, body, where, note))
else:
faults.append((head, where))
else:
print(head + where)
for mark, body, where, note in tolerated:
print(f"{mark:<16} {body}{where}\n baseline: {note}")
print()
for key, (cls, note) in sorted(baseline.items()):
if not any((f"{r['path']}:{r['line']}", r["token"]) == key
and r["class"] in ("RECOVERABLE", "RECOVERABLE-HIST", "EXHAUSTED") for r in rows):
cleared.append((key, cls, note))
for (key, tok), cls, note in cleared:
print(f"STALE baseline row names no measured fault row: {key} `{tok}` ({cls})")
tally = {}
for r in rows:
tally[r["class"]] = tally.get(r["class"], 0) + 1
counts = " / ".join(f"{c} {tally.get(c, 0)}" for c in RULE_CLASSES)
print(f"rule: {counts}; {len(tolerated)} tolerated in the baseline, "
f"{len(faults)} new fault(s), {len(cleared)} stale baseline row(s)")
if faults:
print("\nNEW FAULTS — a verdict row whose address is not written and whose value survives:", file=sys.stderr)
for head, where in faults:
print(f" {head}{where}", file=sys.stderr)
return 1
if cleared:
print("FAULT: a baseline row outlives the row it tolerates", file=sys.stderr)
return 1
return 0
def main(argv=None):
ap = argparse.ArgumentParser(
description="Q60 census of truncated identifiers; see the module docstring.")
ap.add_argument("--repo", default=".")
ap.add_argument("--rev", default="HEAD")
sub = ap.add_subparsers(dest="cmd", required=True)
sub.add_parser("census")
t = sub.add_parser("table")
t.add_argument("--out", default=None)
c = sub.add_parser("check")
c.add_argument("table")
r = sub.add_parser("rule")
r.add_argument("--baseline", default=None)
args = ap.parse_args(argv)
repo = os.path.abspath(args.repo)
if args.cmd == "census":
return cmd_census(args, repo)
if args.cmd == "table":
return cmd_table(args, repo)
if args.cmd == "rule":
return cmd_rule(args, repo)
return cmd_check(args, repo)
if __name__ == "__main__":
sys.exit(main())