#!/usr/bin/env python3 """Corrected, complete census of every numbered work-queue (Q) item. Extract lane: data extraction and tables. One row per "## Q ..." 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 '## ...' where the '' 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' (dupes kept as separate rows). * title : verbatim heading text after '## ' (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' 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 ...' 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())