volt
926f9a1c59
The census was red at HEAD (96 rows vs 97 headings), named its source by absolute host path, and decided its verdict from the worktree at read time. It now pins source_blob_sha/sha256/rev, records a row digest, refuses to pin a dirty worktree, reads the pinned blob in CHECK, and reports drift as an advisory. 9 selftest cells; the non-decay claim is measured on real commits.
278 lines
10 KiB
Python
278 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Corrected, complete census of every numbered work-queue (Q) item.
|
|
|
|
Extract lane: data extraction and tables. One row per "## Q<n> ..." heading in
|
|
WORK-QUEUE.md, built from the source with a real parser and verified by an
|
|
INDEPENDENT re-derivation in CHECK (the old census's CHECK re-imported its own
|
|
extractor, so it could never fail; that is the fault this version removes).
|
|
|
|
Output (all in this directory):
|
|
workqueue-census.tsv one row per Q heading, tab-separated
|
|
workqueue-census.md markdown table of the same rows
|
|
extract_manifest.json schema + counts + provenance (extracted_at_utc = NOW)
|
|
README.md documentation for this artifact
|
|
CHECK real verification (independent re-derivation)
|
|
|
|
Provenance (Q135, 2026-10-31). The manifest used to name its source by
|
|
absolute host path (`/root/troupe-standup/WORK-QUEUE.md`), which is the
|
|
Q69/Q106/Q109 host-pin class, and it carried no pin to the bytes it read, so
|
|
the artifact's own rows could not be re-derived by anyone who reads it after
|
|
the queue has moved on. It now records:
|
|
|
|
* source repo-relative path (`WORK-QUEUE.md`), never a host path
|
|
* source_rev `git rev-parse HEAD` at generation
|
|
* source_blob_sha `git rev-parse HEAD:WORK-QUEUE.md` -- the pinned bytes
|
|
* source_sha256 sha256 of the bytes actually read
|
|
* source_lines line count of those bytes
|
|
* row_digest sha256 of the emitted TSV data rows
|
|
|
|
The pin is honest or the extractor refuses. If the worktree's WORK-QUEUE.md
|
|
differs from the HEAD blob, this exits 1 and writes nothing, because a pin to
|
|
a blob that does not contain the rows in the TSV is a false provenance claim;
|
|
`--allow-dirty` writes the snapshot anyway and records `source_blob_sha: null`
|
|
with `source_matches_head_blob: false`, so CHECK grades the worktree bytes and
|
|
says so.
|
|
|
|
Conventions
|
|
-----------
|
|
* A row is created for every line matching '## <Qnum> ...' where the '<Qnum>' is
|
|
'Q' followed by digits (continuation labels such as 'closure', 'section 9
|
|
follow-on', 'verified', 'claimed' are kept verbatim in the title).
|
|
* q : the label 'Q<num>' (dupes kept as separate rows).
|
|
* title : verbatim heading text after '## <Qnum>' (trimmed).
|
|
* status : the first '**Status:' line in the block, whitespace collapsed,
|
|
truncated at the first sentence end, single trailing period;
|
|
empty when no such marker precedes the next heading.
|
|
* done_implies : the first '**Done means:' line in the block, trimmed; empty
|
|
when no such marker precedes the next heading.
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
REPO_ROOT = os.path.abspath(os.path.join(HERE, "..", "..", ".."))
|
|
SRC = os.path.join(REPO_ROOT, "WORK-QUEUE.md")
|
|
OUT = HERE
|
|
|
|
# '## Q' + digits, then anything (heading title begins). Captures Q1..Q98
|
|
# headings, including continuation/closure labels that follow the number.
|
|
HEADING_RE = re.compile(r"^##\s+Q(\d+)(.*)$")
|
|
|
|
STATUS_RE = re.compile(r"^\*\*Status:\s*(.*)$")
|
|
DONE_RE = re.compile(r"^\*\*Done means:\s*(.*)$")
|
|
|
|
|
|
def _sentence(s):
|
|
s = re.sub(r"\s+", " ", s).strip()
|
|
m = re.search(r"\.\s", s)
|
|
if m:
|
|
s = s[: m.start()] + "."
|
|
if s.endswith("."):
|
|
s = s[: -1]
|
|
return s + "."
|
|
|
|
|
|
def split_blocks(lines):
|
|
"""Return list of dicts: idx (heading line number), qid, title, body.
|
|
|
|
A 'block' is the heading line plus the following lines up to (but not
|
|
including) the next '## Q<n>' heading line.
|
|
"""
|
|
blocks = []
|
|
current = None
|
|
for i, line in enumerate(lines):
|
|
m = HEADING_RE.match(line)
|
|
if m:
|
|
if current is not None:
|
|
blocks.append(current)
|
|
current = {
|
|
"line": i,
|
|
"qid": "Q" + m.group(1),
|
|
"title": m.group(2).strip(),
|
|
"body": [],
|
|
}
|
|
else:
|
|
if current is not None:
|
|
current["body"].append((i, line))
|
|
if current is not None:
|
|
blocks.append(current)
|
|
return blocks
|
|
|
|
|
|
def derive(block):
|
|
status = ""
|
|
done_implies = ""
|
|
for _, line in block["body"]:
|
|
if status == "":
|
|
sm = STATUS_RE.match(line)
|
|
if sm:
|
|
status = _sentence(sm.group(1))
|
|
if done_implies == "":
|
|
dm = DONE_RE.match(line)
|
|
if dm:
|
|
done_implies = dm.group(1).strip()
|
|
return status, done_implies
|
|
|
|
|
|
def extract(lines):
|
|
rows = []
|
|
for block in split_blocks(lines):
|
|
status, done_implies = derive(block)
|
|
rows.append(
|
|
{
|
|
"q": block["qid"],
|
|
"title": block["title"],
|
|
"status": status,
|
|
"done_implies": done_implies,
|
|
"line": block["line"] + 1,
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
def _sha256(data):
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def _git(args, cwd):
|
|
"""Run git, returning stdout stripped, or None when git or the object is absent."""
|
|
try:
|
|
proc = subprocess.run(["git", "-C", cwd] + args,
|
|
capture_output=True, text=True)
|
|
except OSError:
|
|
return None
|
|
if proc.returncode != 0:
|
|
return None
|
|
return proc.stdout.strip()
|
|
|
|
|
|
def pin_source(repo_root, raw):
|
|
"""Return (provenance dict, error). The pin is the HEAD blob or a refusal."""
|
|
blob_sha = _git(["rev-parse", "HEAD:" + "WORK-QUEUE.md"], repo_root)
|
|
rev = _git(["rev-parse", "HEAD"], repo_root)
|
|
if blob_sha is None or rev is None:
|
|
return None, ("cannot read WORK-QUEUE.md at HEAD (git missing, not a "
|
|
"repository, or the path is untracked) -- pass "
|
|
"--allow-dirty to snapshot the worktree instead")
|
|
try:
|
|
blob = subprocess.run(["git", "-C", repo_root, "cat-file", "blob", blob_sha],
|
|
capture_output=True)
|
|
except OSError:
|
|
return None, "git cat-file is unavailable -- pass --allow-dirty"
|
|
if blob.returncode != 0:
|
|
return None, "cannot read blob %s -- pass --allow-dirty" % blob_sha
|
|
if _sha256(blob.stdout) != _sha256(raw):
|
|
return None, ("the worktree's WORK-QUEUE.md is not the HEAD blob (%s): "
|
|
"commit it first, or pass --allow-dirty to snapshot the "
|
|
"worktree and record that the pin is absent" % blob_sha[:12])
|
|
return {"source_rev": rev, "source_blob_sha": blob_sha,
|
|
"source_matches_head_blob": True}, None
|
|
|
|
|
|
USAGE = """usage: extract_workqueue.py [--allow-dirty]
|
|
|
|
(no flag) pin the source to the HEAD blob; refuse (rc 1) if the
|
|
worktree's WORK-QUEUE.md is not that blob
|
|
--allow-dirty snapshot the worktree anyway; the manifest records
|
|
source_blob_sha: null and source_matches_head_blob: false
|
|
-h, --help this text
|
|
|
|
Writes workqueue-census.tsv, workqueue-census.md and extract_manifest.json to
|
|
this directory. Exit 0 clean, 1 refused (source not pinned and not allowed to
|
|
be dirty), 2 usage error.
|
|
"""
|
|
|
|
|
|
def main():
|
|
argv = sys.argv[1:]
|
|
unknown = [a for a in argv if a not in ("--allow-dirty", "-h", "--help")]
|
|
if unknown:
|
|
print("usage error: unknown argument(s) %r" % unknown, file=sys.stderr)
|
|
print(USAGE, file=sys.stderr)
|
|
return 2
|
|
if "-h" in argv or "--help" in argv:
|
|
print(USAGE)
|
|
return 0
|
|
allow_dirty = "--allow-dirty" in argv
|
|
with open(SRC, "rb") as fh:
|
|
raw = fh.read()
|
|
lines = raw.decode("utf-8").splitlines()
|
|
|
|
prov, err = ({"source_rev": None, "source_blob_sha": None,
|
|
"source_matches_head_blob": False}, None)
|
|
if not allow_dirty:
|
|
prov, err = pin_source(REPO_ROOT, raw)
|
|
if err is not None:
|
|
print("REFUSED: " + err, file=sys.stderr)
|
|
return 1
|
|
else:
|
|
prov["source_rev"] = _git(["rev-parse", "HEAD"], REPO_ROOT)
|
|
assert prov is not None
|
|
|
|
rows = extract(lines)
|
|
|
|
tsv_path = os.path.join(OUT, "workqueue-census.tsv")
|
|
md_path = os.path.join(OUT, "workqueue-census.md")
|
|
json_path = os.path.join(OUT, "extract_manifest.json")
|
|
|
|
data = "".join(
|
|
f"{r['q']}\t{r['title']}\t{r['status']}\t{r['done_implies']}\n" for r in rows
|
|
)
|
|
with open(tsv_path, "w", encoding="utf-8") as fh:
|
|
fh.write("q\ttitle\tstatus\tdone_implies\n")
|
|
fh.write(data)
|
|
|
|
distinct = sorted({r["q"] for r in rows})
|
|
reopened = sorted(q for q in distinct if sum(1 for r in rows if r["q"] == q) > 1)
|
|
|
|
with open(md_path, "w", encoding="utf-8") as fh:
|
|
fh.write("# Work queue census v2 (extract lane)\n\n")
|
|
fh.write(
|
|
"One row per '## Q<n> ...' heading in WORK-QUEUE.md. Extracted from "
|
|
f"{len(rows)} headings; {len(distinct)} distinct q labels; "
|
|
f"{len(reopened)} reopened ({', '.join(reopened)}). "
|
|
"extracted_at_utc is the measurement moment, not a constant.\n\n"
|
|
)
|
|
fh.write("| q | title | status | done_implies |\n")
|
|
fh.write("|---|---|---|---|\n")
|
|
for r in rows:
|
|
fh.write(f"| {r['q']} | {r['title']} | {r['status']} | {r['done_implies']} |\n")
|
|
|
|
manifest = {
|
|
"schema": "workqueue-census-v2",
|
|
"source": "WORK-QUEUE.md",
|
|
"source_rev": prov["source_rev"],
|
|
"source_blob_sha": prov["source_blob_sha"],
|
|
"source_matches_head_blob": prov["source_matches_head_blob"],
|
|
"source_sha256": _sha256(raw),
|
|
"source_lines": len(lines),
|
|
"rows": len(rows),
|
|
"columns": ["q", "title", "status", "done_implies"],
|
|
"row_digest": _sha256(data.encode("utf-8")),
|
|
"distinct_q_labels": len(distinct),
|
|
"reopened_q_labels": reopened,
|
|
"extracted_at_utc": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"generated_by": "extract_workqueue.py",
|
|
"verification": "CHECK (independent re-derivation, does not import extractor)",
|
|
}
|
|
with open(json_path, "w", encoding="utf-8") as fh:
|
|
json.dump(manifest, fh, indent=2)
|
|
|
|
print(f"rows: {len(rows)}")
|
|
print(f"distinct labels: {len(distinct)}; reopened: {reopened}")
|
|
print(f"source pin: rev={prov['source_rev']} blob={prov['source_blob_sha']}"
|
|
if prov["source_blob_sha"] else
|
|
"source pin: ABSENT (--allow-dirty; the manifest says so)")
|
|
print(f"written: {tsv_path}, {md_path}, {json_path}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|