troupe-standup/wizard-work/ledger/check_git_author_census.py

#!/usr/bin/env python3
"""Git author census: extract author identities + commit counts from a snapshot of the
worktree git history, write a table, and (as 'check') verify the artifact against a fresh
recompile from `git log` of the pre-census snapshot (the commit the artifact represents).

The census is a report ON the repo, committed after it. To keep the check honest (not a
self-referential tautology), the artifact is verified against the git snapshot it describes
(HEAD^), not against HEAD which includes the census commit itself.
"""
import csv
import os
import subprocess
import sys

REPO = "/root/troupe-standup"
LEDGER_DIR = REPO + "/wizard-work/ledger"
CSV_NAME = "2026-10-30-git-author-census.csv"
MD_NAME = "2026-10-30-git-author-census.md"
ASN_DATE = "2026-10-30"
# Snapshot the census artifact describes: the pre-census tip of the worktree, i.e. the
# commit the artifact was generated from (so the census commit's own author is not
# counted in itself). Pinned to a fixed hash so the check stays stable as the repo
# keeps evolving.
SNAPSHOT = "a8f8c39"


def git_log(base):
    proc = subprocess.run(
        ["git", "log", base, "--format=%an|%ae"],
        cwd=REPO,
        check=True,
        capture_output=True,
        text=True,
    )
    return proc.stdout


def git_author_census(base):
    """Recompute from git; {author_identity: count}, identity = 'Name <email>'."""
    counts = {}
    for raw in git_log(base).splitlines():
        if not raw:
            continue
        name, _, email = raw.partition("|")
        key = "{} <{}>".format(name.strip(), email.strip())
        counts[key] = counts.get(key, 0) + 1
    return counts


def write_artifacts(counts):
    total = sum(counts.values())
    rows = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))
    os.makedirs(LEDGER_DIR, exist_ok=True)

    csv_path = LEDGER_DIR + "/" + CSV_NAME
    with open(csv_path, "w", newline="") as f:
        w = csv.writer(f)
        w.writerow(["rank", "author", "commits"])
        for i, (k, v) in enumerate(rows, 1):
            w.writerow([i, k, v])

    md_path = LEDGER_DIR + "/" + MD_NAME
    with open(md_path, "w") as f:
        f.write("# Git author census ({date})\n\n".format(date=ASN_DATE))
        f.write("Snapshot: `git log` of the troupe worktree at `{snapshot}` ({total} commits\n".format(
            snapshot=SNAPSHOT, total=total))
        f.write("reachable from HEAD before this census was committed). Extracted {date}.\n\n".format(
            date=ASN_DATE))
        f.write("Identity key is the author name plus email in `Name <email>` form as\n")
        f.write("recorded by git; no canonicalization (e.g. `Timmy` vs `timmy`, or a\n")
        f.write("stackchain.ai vs agents.invalid address, stays a distinct row).\n\n")
        f.write("| rank | author | commits | pct |\n")
        f.write("|---:|---|---:|---:|\n")
        for i, (k, v) in enumerate(rows, 1):
            pct = 100.0 * v / total if total else 0.0
            f.write("| {rank} | {author} | {commits} | {pct:.2f} |\n".format(
                rank=i, author=k, commits=v, pct=pct))
        f.write("\n**Total: {total} commits** across {n} distinct author identities.\n".format(
            total=total, n=len(rows)))
    return csv_path, md_path, total


def verify(csv_path, md_path):
    expected = git_author_census(SNAPSHOT)
    total = sum(expected.values())
    with open(csv_path, newline="") as f:
        rows = list(csv.reader(f))
    errs = []
    if not rows or rows[0] != ["rank", "author", "commits"]:
        errs.append("CSV header wrong: {!r}".format(rows[0] if rows else None))
    body = rows[1:]
    if len(body) != len(expected):
        errs.append("row count {} != author count {}".format(len(body), len(expected)))
    for i, row in enumerate(body, 1):
        if len(row) != 3:
            errs.append("bad row width: {!r}".format(row)); continue
        rank, author, commits = row
        if str(i) != rank:
            errs.append("rank mismatch at row {}: {!r}".format(i, rank))
        exp = expected.get(author)
        if exp is None:
            errs.append("author not in recomputed census: {!r}".format(author))
        elif str(exp) != commits:
            errs.append("commits mismatch for {!r}: artifact {} expected {}".format(author, commits, exp))
    # independent total from the same snapshot
    proc = subprocess.run(
        ["git", "rev-list", "--count", SNAPSHOT], cwd=REPO,
        check=True, capture_output=True, text=True)
    if proc.stdout.strip() != str(total):
        errs.append("snapshot commit total {} != census total {}".format(proc.stdout.strip(), total))
    md = open(md_path).read()
    if "Total: {} commits".format(total) not in md:
        errs.append("MD missing total line for {total}".format(total=total))

    if errs:
        print("FAIL:")
        for e in errs:
            print("  - " + e)
        return 1
    print("PASS: snapshot `{snapshot}` = {total} commits, {n} author identities; "
          "{rows} CSV rows + MD match a fresh recompile from `git log {snapshot}`.".format(
              snapshot=SNAPSHOT, total=total, n=len(expected), rows=len(body)))
    return 0


def main():
    mode = sys.argv[1] if len(sys.argv) > 1 else "check"
    if mode == "generate":
        csv_path, md_path, total = write_artifacts(git_author_census(SNAPSHOT))
        print("wrote", csv_path)
        print("wrote", md_path)
        print("snapshot {snapshot} total commits:".format(snapshot=SNAPSHOT), total)
        return 0
    if mode == "check":
        csv_path = LEDGER_DIR + "/" + CSV_NAME
        md_path = LEDGER_DIR + "/" + MD_NAME
        return verify(csv_path, md_path)
    print("usage: check_git_author_census.py [generate|check]", file=sys.stderr)
    return 2


if __name__ == "__main__":
    sys.exit(main())