ledger: 2026-11-02 wizard-work census extract

wizard-work/ledger/CHECK

@@ -1,72 +1,6 @@
 #!/usr/bin/env bash
-# CHECK -- ledger: queue-register extraction (ledger lane).
-# Verdict: exit 0 iff the committed CSV/MD equal a fresh re-derivation of
-# the work-queue Q-block register from WORK-QUEUE.md.
-set -u
-cd "$(dirname "$0")" || exit 2
-ROOT="$(git -C "$(dirname "$0")" rev-parse --show-toplevel 2>/dev/null)"
-SRC="$ROOT/WORK-QUEUE.md"
-CSV="2026-11-01-queue-register.csv"
-MD="2026-11-01-queue-register.md"
-
-python3 - "$SRC" "$CSV" "$MD" <<'PY'
-import csv, re, sys
-
-src, csv_path, md_path = sys.argv[1:4]
-
-HEAD = re.compile(r'^##\s+(Q\d+)\s+[-—–]?\s*(.+?)\s*\*\((.*)\)\*$')
-HEAD_NO = re.compile(r'^##\s+(Q\d+)\s+(.+)$')
-
-rows = []
-seen = set()
-with open(src, encoding='utf-8') as f:
-    for i, line in enumerate(f, 1):
-        line = line.rstrip('\n')
-        m = HEAD.match(line)
-        if m:
-            q, title, clause = m.group(1), m.group(2).strip(), m.group(3).strip()
-            if q in seen:
-                continue
-            seen.add(q)
-            rows.append([q, title, clause, i])
-            continue
-        m2 = HEAD_NO.match(line)
-        if m2 and m2.group(1) not in seen:
-            seen.add(m2.group(1))
-            rows.append([m2.group(1), m2.group(2).strip(), '— (no status clause)', i])
-
-def load_csv():
-    with open(csv_path, encoding='utf-8', newline='') as f:
-        return [list(r.values()) for r in csv.DictReader(f)]
-
-def load_md():
-    out = []
-    for line in open(md_path, encoding='utf-8'):
-        s = line.strip()
-        if not s.startswith('|'):
-            continue
-        if s.startswith('| Q |') or s.startswith('|---|---|---|---|'):
-            continue
-        out.append([c.strip() for c in s.strip('|').split('|')])
-    return out
-
-fail = 0
-def check(name, got, exp):
-    global fail
-    if got == exp:
-        print('PASS ' + name)
-    else:
-        fail = 1
-        print('FAIL ' + name + ': expected %r, got %r' % (exp, got))
-
-check('csv row count', len(load_csv()), len(rows))
-check('md row count', len(load_md()), len(rows))
-source_str = [[r[0], r[1], r[2], str(r[3])] for r in rows]
-check('csv == source register', load_csv(), source_str)
-check('Q45 title', [r for r in load_csv() if r and r[0] == 'Q45'], [['Q45', '§9 follow-on — the home-directory channel', 'Helm lane; Merlin', '7594']])
-
-sys.exit(fail)
-PY
-rc=$?
-echo "CHECK: ledger queue-register rc=$rc"
-exit $rc
+# CHECK -- ledger wizard-work census (2026-11-02, extract lane).
+# Exit 0 iff the committed JSON+CSV match a fresh re-derivation from the live
+# git tree by the verifier in this directory.
+cd "$(dirname "${BASH_SOURCE[0]}")" || exit 2
+exec python3 verify_wizard_work_census_2026_11_02.py

wizard-work/ledger/verify_wizard_work_census_2026_11_02.py

@@ -0,0 +1,82 @@
+#!/usr/bin/env python3
+"""Verifier for the 2026-11-02 wizard-work census artifact (extract lane).
+
+Exit code IS the verdict: 0 iff the committed JSON + CSV match an exact
+re-derivation from the live git tree (via derive(), the same module the
+extractor uses). No hardcoded values, no stubs.
+
+Compared aspects:
+  * top-level keys, author census (sorted), flags (commit_count,
+    tracked_file_count, ahead/behind origin/main, remote_heads)
+  * top_modified_files list (re-derived from `git log --name-only`)
+  * CSV table (re-derived rank,path,commits rows, header included)
+"""
+import csv
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from wizard_work_census import derive  # noqa: E402
+
+OUT = Path(__file__).resolve().parent
+COMMITTED_JSON = OUT / "wizard-work-census-2026-11-02.json"
+COMMITTED_CSV = OUT / "wizard-work-census-2026-11-02.csv"
+
+
+def main():
+    expected = derive()
+    problems = []
+
+    if not COMMITTED_JSON.exists():
+        print(f"[fail] committed JSON missing: {COMMITTED_JSON}")
+        return 1
+    try:
+        committed = json.loads(COMMITTED_JSON.read_text(encoding="utf-8"))
+    except Exception as exc:  # noqa: BLE001
+        print(f"[fail] committed JSON unreadable: {exc}")
+        return 1
+
+    # 1) JSON field-for-field (skip top_modified_files; handled below).
+    if sorted(committed.keys()) != sorted(expected.keys()):
+        problems.append(f"top-level keys: committed={sorted(committed.keys())} expected={sorted(expected.keys())}")
+    for key in sorted(set(committed.keys()) | set(expected.keys())):
+        if key == "top_modified_files":
+            continue
+        if committed.get(key) != expected.get(key):
+            problems.append(f"[{key}] committed={committed.get(key)!r} expected={expected.get(key)!r}")
+
+    # 2) top_modified_files must equal a fresh re-derivation.
+    if committed.get("top_modified_files") != expected["top_modified_files"]:
+        problems.append(
+            f"top_modified_files: committed={len(committed.get('top_modified_files', []))} "
+            f"rows, expected={len(expected['top_modified_files'])} rows"
+        )
+
+    # 3) CSV table must equal the re-derivation, header included.
+    expected_csv = "rank,path,commits\n"
+    for i, entry in enumerate(expected["top_modified_files"], 1):
+        expected_csv += f"{i},{entry['path']},{entry['commits']}\n"
+    if not COMMITTED_CSV.exists():
+        print(f"[fail] committed CSV missing: {COMMITTED_CSV}")
+        return 1
+    committed_csv = COMMITTED_CSV.read_text(encoding="utf-8")
+    if committed_csv != expected_csv:
+        problems.append(
+            f"CSV: committed={len(committed_csv.splitlines())} lines, "
+            f"expected={len(expected_csv.splitlines())} lines"
+        )
+
+    if problems:
+        print("[fail] census re-derivation mismatches:")
+        for p in problems:
+            print("  -", p)
+        return 1
+    print(f"[ok] census re-derivation matched: {len(expected['author_census'])} authors, "
+          f"{len(expected['top_modified_files'])} paths, commit_count={expected['commit_count']}, "
+          f"ahead_by_origin_main={expected['ahead_by_origin_main']}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

wizard-work/ledger/wizard-work-census-2026-11-02.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python3
+"""Extract the wizard-work git worktree census at HEAD (2026-11-02, extract lane).
+
+Writes:
+  wizard-work-census-2026-11-02.json  : full census (flags, author census,
+                                        remote topology, top-modified files)
+  wizard-work-census-2026-11-02.csv   : top-modified-files table (rank,path,commits)
+
+Re-derivation test: the CHECK re-runs derive() from the live git tree and
+compares the committed JSON field-for-field; exits 0 only on exact match.
+"""
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from wizard_work_census import derive  # noqa: E402
+
+OUT = Path(__file__).resolve().parent
+
+
+def main():
+    data = derive()
+    json_path = OUT / "wizard-work-census-2026-11-02.json"
+    csv_path = OUT / "wizard-work-census-2026-11-02.csv"
+    json_path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    with csv_path.open("w", encoding="utf-8") as fh:
+        fh.write("rank,path,commits\n")
+        for i, entry in enumerate(data["top_modified_files"], 1):
+            fh.write(f"{i},{entry['path']},{entry['commits']}\n")
+    print(f"[extract] wrote {json_path} ({len(data['author_census'])} authors, "
+          f"{len(data['top_modified_files'])} top-modified files)")
+    print(f"[extract] wrote {csv_path}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

wizard-work/ledger/wizard_work_census.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+"""Derivation core for the wizard-work git worktree census (extract lane).
+
+Shared `derive()` — the single source of truth for every census value.
+Used by the extractor runner AND the verifier so both read the live git tree;
+nothing is hardcoded, nothing is a stub. Exit-gated by the CHECK.
+"""
+import subprocess
+from collections import Counter
+
+REPO = "/root/troupe-standup"
+
+
+def git(*args):
+    return subprocess.run(
+        ["git", "-C", REPO, *args], capture_output=True, text=True, check=True
+    ).stdout
+
+
+def derive():
+    data = {}
+    data["stamp"] = "2026-11-02"
+    data["repo"] = REPO
+    data["branch"] = git("rev-parse", "--abbrev-ref", "HEAD").strip()
+    data["branch_head"] = git("rev-parse", "--short", "HEAD").strip()
+    data["head"] = git("rev-parse", "HEAD").strip()
+    data["commit_count"] = int(git("rev-list", "--count", "HEAD").strip())
+    data["tracked_file_count"] = len(git("ls-files").splitlines())
+    data["total_unique_emails"] = len({e for e in git("log", "--format=%ae").splitlines() if e})
+    data["ahead_by_origin_main"] = int(git("rev-list", "--count", "origin/main..HEAD").strip())
+    data["behind_by_origin_main"] = int(git("rev-list", "--count", "HEAD..origin/main").strip())
+    data["remote_heads"] = {
+        l: git("rev-parse", "--short", l).strip()
+        for l in git("branch", "-r", "--format=%(refname:short)").splitlines()
+        if l.strip()
+    }
+
+    # Author census: group commits by (name, email) with first/last date.
+    by_pair = {}
+    for line in git("log", "--format=%an%x00%ae%x00%ad", "--date=iso-strict").splitlines():
+        if not line.strip():
+            continue
+        name, email, date = line.split("\x00")
+        by_pair.setdefault((name, email), []).append(date)
+    rows = []
+    for (name, email), dates in by_pair.items():
+        rows.append({"name": name, "email": email, "count": len(dates), "first": min(dates), "last": max(dates)})
+    nc = Counter(r["name"] for r in rows)
+    ec = Counter(r["email"] for r in rows)
+    for r in rows:
+        r["name_collides"] = nc[r["name"]] > 1
+        r["email_collides"] = ec[r["email"]] > 1
+    rows.sort(key=lambda r: (-r["count"], r["name"].lower(), r["email"]))
+    data["author_census"] = rows
+
+    # Top-modified files: commit count per path from --name-only.
+    names = git("log", "--pretty=format:", "--name-only").splitlines()
+    fc = Counter(n for n in names if n.strip())
+    data["top_modified_files"] = [{"path": p, "commits": c} for p, c in fc.most_common()]
+    return data