troupe-standup/tools/append_only_owners.py

#!/usr/bin/env python3
"""append_only_owners.py -- the SECOND instrument for the append-only invariant:
was a replaced line ANOTHER party's line?

WHY THIS EXISTS
  tools/append_only_audit.py answers "was an existing line replaced?" (LOSS). That
  is not the rule. The ledger's rule is "append to another party's file, never
  overwrite it" -- the violated thing is another PARTY's line, not merely a line.
  A count of 74 LOSS rows on append-sanctioned paths is therefore a count of
  candidates, and this tool adjudicates each one on the only evidence that bears
  on it: the author of the change that removed the line, against the author of the
  line that was removed.

THE PROBLEM, STATED RATHER THAN HIDDEN: a line has no author in a blob diff.
  "The author of the line that was removed" needs blame, which this tool does not
  run (it would need a `git blame` per row at every parent, and blame on a
  multi-author appended file mis-attributes long lines anyway). Two mechanical
  proxies are used and reported SEPARATELY, each with its blind spot named:

    PREV_TOUCHER  the author of this path's immediately preceding change -- the
                  party that last wrote the file. STRONG on a single-author file
                  (checkins/<date>-<party>.md, replies/...), WEAK on a shared
                  multi-author file (WORK-QUEUE.md), where the previous toucher is
                  one of several interleaved writers and not the line's author.
    CREATOR       the author of this path's first change -- the file's introducer.

  A third, filename-derived owner is deliberately NOT used: replies/ and
  recoveries/ filenames do not name their author, so a filename rule would invent
  owners. The filename split is available ONLY where the filename is the author
  (checkins/<date>-<party>.md) and is reported as a scoping note, never as the
  adjudication.

AUTHOR-NAME NORMALISATION is mandatory and is itself a measured fact of this
history: git's author field carries several spellings of one party (Timmy/timmy,
Zephyrus Sparkborn/zephyrus-sparkborn -- 2 folded parties out of 10 raw strings at
f5403b2). Compared raw, one party's own edit is miscounted as another party's. The
fold is lowercase alphanumerics; every raw spelling is printed beside its fold so
the fold cannot hide a genuinely distinct party.

LOST-LINE SHAPE is classified mechanically from the break line's text, so a reader
can see whether the rule is being bent over status metadata or over prose:
  STATUS (a `**Status` line) / TABLE_ROW (`|`) / SIGNOFF (`- `, `-- `, `— `) /
  TIMED_CLAIM (carries a clock or ISO timestamp) / SHORT (<40 chars) / PROSE / EMPTY.

USAGE
  python3 append_only_owners.py AUDIT.json [--json OUT] [--selftest]   # 15 cells
    AUDIT.json is a --json output of tools/append_only_audit.py at the same rev.
"""
import argparse
import json
import re
import sys
from collections import Counter, defaultdict

SANC = ("WORK-QUEUE.md", "checkins/", "replies/", "recoveries/", "audits/", "inbox/")
TIME_RE = re.compile(r"\b\d{1,2}:\d{2}(:\d{2})?\s*(Z|UTC)\b|\b\d{4}-\d{2}-\d{2}T\d{2}:\d{2}")


def sanctioned(path):
    return any(path == s or path.startswith(s) for s in SANC)


def fold(name):
    """Fold a git author string to comparable party form: lowercase alphanumerics."""
    return re.sub(r"[^a-z0-9]", "", (name or "").lower())


def shape(text):
    t = (text or "").strip()
    if not t:
        return "EMPTY"
    if t.startswith("**Status") or t.startswith("Status"):
        return "STATUS"
    if t.startswith("|"):
        return "TABLE_ROW"
    if t.startswith("— ") or t.startswith("-- ") or t.startswith("- "):
        return "SIGNOFF"
    if TIME_RE.search(t):
        return "TIMED_CLAIM"
    if len(t) < 40:
        return "SHORT"
    return "PROSE"


def adjudicate(rows):
    """Pure core: audit rows -> the owner adjudication. No I/O, so it is testable."""
    spellings = Counter(r["author"] for r in rows)
    by_fold = defaultdict(Counter)
    for name, n in spellings.items():
        by_fold[fold(name)][name] += n

    by_path = defaultdict(list)
    for r in rows:
        by_path[r["path"]].append(r)
    for p in by_path:
        by_path[p].sort(key=lambda r: (r["date"], r["commit"]))

    creator = {p: fold(v[0]["author"]) for p, v in by_path.items()}
    out = []
    for p, v in by_path.items():
        last = None
        for r in v:
            if sanctioned(p) and r["line_class"] == "LOSS":
                rec = {
                    "path": p, "commit": r["commit"][:8], "date": r["date"][:19],
                    "author": fold(r["author"]), "raw_author": r["author"],
                    "prev_toucher": last, "creator": creator[p],
                    "missing_parent_lines": r["missing_parent_lines"],
                    "added_child_lines": r["added_child_lines"],
                    "loss_kind": r["loss_kind"],
                    "parent_bytes": r["parent_bytes"], "child_bytes": r["child_bytes"],
                    "break_text": r["order_break_line_text"],
                    "shape": shape(r["order_break_line_text"]),
                }
                rec["vs_prev"] = ("FIRST_CHANGE_ON_PATH" if last is None
                                  else "SAME_PARTY" if last == rec["author"] else "OTHER_PARTY")
                rec["vs_creator"] = ("SAME_PARTY" if creator[p] == rec["author"] else "OTHER_PARTY")
                out.append(rec)
            last = fold(r["author"])
    return {"spellings": spellings, "folded": {k: dict(v) for k, v in by_fold.items()},
            "losses": out}


def report(res, out=sys.stdout):
    sp = res["spellings"]
    folded = Counter({k: sum(v.values()) for k, v in res["folded"].items()})
    print("== git author spellings: %d raw -> %d folded ==" % (len(sp), len(folded)), file=out)
    for f, raw in sorted(res["folded"].items()):
        if len(raw) > 1:
            print("  VARIANT %-22s -> %s" % (f, ", ".join("%s(%d)" % (k, v)
                  for k, v in sorted(raw.items(), key=lambda kv: -kv[1]))), file=out)
    losses = res["losses"]
    print("\n== sanctioned LOSS candidates: %d ==" % len(losses), file=out)
    print("\n-- remover vs PREV_TOUCHER --", file=out)
    for k, v in Counter(r["vs_prev"] for r in losses).most_common():
        print("  %-24s %d" % (k, v), file=out)
    print("\n-- remover vs CREATOR --", file=out)
    for k, v in Counter(r["vs_creator"] for r in losses).most_common():
        print("  %-24s %d" % (k, v), file=out)
    print("\n-- lost-line SHAPE --", file=out)
    for k, v in Counter(r["shape"] for r in losses).most_common():
        print("  %-14s %d" % (k, v), file=out)
    print("\n-- loss_kind --", file=out)
    for k, v in Counter(r["loss_kind"] for r in losses).most_common():
        print("  %-22s %d" % (k, v), file=out)
    print("\n-- rows where the remover is NOT the path's previous toucher --", file=out)
    others = [r for r in losses if r["vs_prev"] == "OTHER_PARTY"]
    for r in others:
        print("  %s %s %-42s by %-18s prev=%-16s bytes %d->%d miss=%d [%s] %s"
              % (r["commit"], r["date"], r["path"], r["raw_author"], r["prev_toucher"],
                 r["parent_bytes"], r["child_bytes"], r["missing_parent_lines"],
                 r["shape"], r["break_text"][:60]), file=out)
    print("  total: %d" % len(others), file=out)
    print("\n-- per committed party (folded) --", file=out)
    for k, v in Counter(r["author"] for r in losses).most_common():
        same = sum(1 for r in losses if r["author"] == k and r["vs_prev"] == "SAME_PARTY")
        oth = sum(1 for r in losses if r["author"] == k and r["vs_prev"] == "OTHER_PARTY")
        first = sum(1 for r in losses if r["author"] == k and r["vs_prev"] == "FIRST_CHANGE_ON_PATH")
        print("  %-18s %3d  (own-file %d, other-file %d, first-change %d)"
              % (k, v, same, oth, first), file=out)


# --------------------------------------------------------------------- selftest

def _row(path, date, author, line_class="LOSS", break_text="", **kw):
    d = {"path": path, "date": date, "commit": kw.pop("commit", date),
         "author": author, "line_class": line_class, "loss_kind": "REWRITE",
         "missing_parent_lines": 1, "added_child_lines": 1,
         "parent_bytes": 10, "child_bytes": 20, "order_break_line_text": break_text}
    d.update(kw)
    return d


def selftest():
    # A hand-built history. Same folds must be one party; different folds another.
    rows = [
        _row("README.md", "2026-01-01T00:00:00Z", "alice", line_class="NEW"),
        # Alice's own file, Alice edits it -> SAME_PARTY
        _row("checkins/a.md", "2026-01-01T00:00:00Z", "Alice", line_class="NEW"),
        _row("checkins/a.md", "2026-01-02T00:00:00Z", "alice",
             break_text="**Status: OPEN — a", commit="c2"),
        # a case-variant spelling of the SAME party must fold to SAME_PARTY
        _row("checkins/b.md", "2026-01-01T00:00:00Z", "Bob", line_class="NEW"),
        _row("checkins/b.md", "2026-01-02T00:00:00Z", "bob", break_text="| 1 | x |",
             commit="c4"),
        # a genuinely different party -> OTHER_PARTY
        _row("WORK-QUEUE.md", "2026-01-01T00:00:00Z", "alice", line_class="NEW"),
        _row("WORK-QUEUE.md", "2026-01-02T00:00:00Z", "bob", break_text="plain prose line here",
             commit="c6"),
        # a LOSS with no preceding change on the path -> FIRST_CHANGE_ON_PATH
        _row("replies/x.md", "2026-01-03T00:00:00Z", "carol",
             break_text="2026-01-03T00:00Z reading", commit="c7"),
        # non-sanctioned path: must be excluded entirely
        _row("tools/t.py", "2026-01-04T00:00:00Z", "dave", break_text="gone", commit="c8"),
        # a non-LOSS row on a sanctioned path: must be excluded
        _row("checkins/a.md", "2026-01-05T00:00:00Z", "alice", line_class="SUBSEQUENCE",
             commit="c9"),
    ]
    res = adjudicate(rows)
    L = res["losses"]
    idx = {(r["path"], r["commit"]): r for r in L}

    cells = [
        ("sanctioned LOSS rows only", len(L) == 4),
        ("non-sanctioned path excluded", all(r["path"] != "tools/t.py" for r in L)),
        ("non-LOSS row excluded", all(r["commit"] != "c9" for r in L)),
        ("case variant folds to SAME_PARTY",
         idx[("checkins/a.md", "c2")]["vs_prev"] == "SAME_PARTY"),
        ("Bob/bob is ONE folded party",
         idx[("checkins/b.md", "c4")]["vs_prev"] == "SAME_PARTY"),
        ("genuine other party is OTHER_PARTY",
         idx[("WORK-QUEUE.md", "c6")]["vs_prev"] == "OTHER_PARTY"),
        ("no preceding change -> FIRST_CHANGE_ON_PATH",
         idx[("replies/x.md", "c7")]["vs_prev"] == "FIRST_CHANGE_ON_PATH"),
        ("creator is the path's introducer",
         idx[("checkins/a.md", "c2")]["creator"] == fold("Alice")),
        ("shape STATUS", shape("**Status: OPEN — x") == "STATUS"),
        ("shape TABLE_ROW", shape("| a | b |") == "TABLE_ROW"),
        ("shape SIGNOFF", shape("- **x:** y") == "SIGNOFF"),
        ("shape TIMED_CLAIM", shape("taken 06:51Z from this host") == "TIMED_CLAIM"),
        ("shape PROSE", shape("a" * 60) == "PROSE"),
        ("shape SHORT", shape("gone") == "SHORT"),
        ("raw spellings preserved",
         set(res["folded"]["bob"]) == {"Bob", "bob"}),
    ]
    ok = 0
    for name, passed in cells:
        print("%-42s %s" % (name, "PASS" if passed else "FAIL"))
        ok += bool(passed)
    print("\nselftest: %d cells, %d passed, %d failed" % (len(cells), ok, len(cells) - ok))
    return 0 if ok == len(cells) else 1


def main():
    ap = argparse.ArgumentParser(
        description="Adjudicate append-sanctioned LOSS rows: whose line was replaced?")
    ap.add_argument("audit_json", nargs="?", help="a --json output of append_only_audit.py")
    ap.add_argument("--json", help="write the adjudication as JSON")
    ap.add_argument("--selftest", action="store_true")
    a = ap.parse_args()
    if a.selftest:
        return selftest()
    if not a.audit_json:
        ap.error("need AUDIT.json (or --selftest)")
    # A missing, non-UTF-8 or non-audit file is a PRECONDITION failure: rc 2
    # with a named message. This line used to be a bare
    # `json.load(open(...))`, so all three classes escaped as a traceback with
    # rc 1 -- the code this tool never otherwise uses, and the same code the
    # repo's graders use for a real fault.
    try:
        with open(a.audit_json, encoding="utf-8") as fh:
            d = json.load(fh)
    except (OSError, ValueError) as exc:
        print("cannot read audit %s: %s" % (a.audit_json, exc), file=sys.stderr)
        return 2
    if not isinstance(d, dict) or not isinstance(d.get("rows"), list):
        print("not an append_only_audit --json file (no 'rows' list): %s"
              % a.audit_json, file=sys.stderr)
        return 2
    res = adjudicate(d["rows"])
    report(res)
    if a.json:
        with open(a.json, "w") as f:
            json.dump(res, f, indent=1)
        print("\njson written: %s" % a.json, file=sys.stderr)
    return 0


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