#!/usr/bin/env python3 """ append_only_audit.py -- measure the repo-wide append-only invariant from git objects. THE INVARIANT, stated so it can be falsified. Two tests are applied to every (commit, path) change, because the ledger's rule ("append to another party's file, never overwrite it") is one claim and a stricter byte rule is a different one: BYTE TEST pure append iff the parent blob's bytes are an exact prefix of the child blob's bytes and the child is strictly longer. classes: NEW / IDENTICAL / APPEND / EDIT LINE TEST non-destructive iff every parent line still occurs in the child, in order (the parent line sequence is an order-preserving subsequence of the child's). This is the rule the ledger actually states: no existing text is lost. An insertion at a mid-file anchor breaks the BYTE test and passes the LINE test; a rewrite or deletion fails both. classes: NO_PARENT / IDENTICAL / SUBSEQUENCE / LOSS LOSS rows report the 1-based index of the first parent line that is gone, how many parent lines are unmatched, and that line's text. UNITS AND WHAT COUNTS AS EVIDENCE. Blob bytes are read through `git cat-file --batch` (binary-safe); sizes come from the object, never from a text-mode `git show`, which folds CRLF and under-reports size by the number of CR pairs. Line counts are of the blob's own lines. ONE INSTRUMENT, ONE HALF OF THE RULE. This tool answers "was an existing line replaced?" and prints that count as CANDIDATE ROWS, never as violations: the rule forbids ANOTHER PARTY's line, and a LOSS row does not carry an author. The second half -- whose line it was -- is `tools/append_only_owners.py`, which reads this tool's --json and adjudicates each row on the remover's identity. At f5403b2 the candidates were 74 and the adjudication read 65 same-party / 9 other-party. Neither number is the other's restatement. OUTPUT: --json PATH writes the full per-change record; stdout is a per-file table plus the LOSS list restricted to append-sanctioned paths. Read-only: the only write is the --json file. USAGE: python3 append_only_audit.py --repo /root/troupe-standup --json out.json python3 append_only_audit.py --selftest # synthetic repo, 18 cells """ import argparse import json import os import re import shutil import subprocess import sys import tempfile from collections import Counter, defaultdict EMPTY_TREE = "4b825dc642cb6eb9a060e54bf8d69288fbee4904" # git's well-known empty tree ESCAPE_RE = re.compile(rb"\\u([0-9a-fA-F]{4})") # Paths the ledger's sharing rule covers (append to another party's file, never # overwrite it). Everything else is a file its owner may rewrite freely. APPEND_SANCTIONED = ("WORK-QUEUE.md", "checkins/", "replies/", "recoveries/", "audits/", "inbox/") def is_sanctioned(path): return any(path == p or path.startswith(p) for p in APPEND_SANCTIONED) def run(cmd, cwd, binary=False, check=True): kw = {} if binary else {"text": True} p = subprocess.run(cmd, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, **kw) if check and p.returncode != 0: raise RuntimeError("%r failed rc=%d: %s" % (cmd, p.returncode, p.stderr[:400])) return p.stdout if p.returncode == 0 else b"" class BlobReader: """Binary-safe streaming blob reader over a persistent `git cat-file --batch`.""" def __init__(self, repo): self.proc = subprocess.Popen( ["git", "cat-file", "--batch"], cwd=repo, stdin=subprocess.PIPE, stdout=subprocess.PIPE, ) def get(self, oid): if oid is None: return b"" self.proc.stdin.write((oid + "\n").encode()) self.proc.stdin.flush() header = self.proc.stdout.readline().decode().strip() parts = header.split() if len(parts) == 2 and parts[1] == "missing": raise RuntimeError("blob %s missing" % oid) if len(parts) < 3: raise RuntimeError("bad batch header: %r" % header) size = int(parts[2]) data = self.proc.stdout.read(size) self.proc.stdout.read(1) # trailing newline return data def close(self): try: self.proc.stdin.close() except Exception: pass self.proc.terminate() def first_parent(repo, sha): out = run(["git", "rev-parse", "--verify", "--quiet", sha + "^1"], repo, binary=True, check=False) return out.decode().strip() or None def changed_paths(repo, parent, sha): """[(status, path)] for the first-parent diff; a root commit makes every path A.""" base = parent if parent else EMPTY_TREE out = run(["git", "diff", "--name-status", "-M", "-z", base, sha], repo, binary=True) fields = [f.decode("utf-8", "replace") for f in out.split(b"\x00") if f] rows, i = [], 0 while i < len(fields): st = fields[i] if st.startswith("R") or st.startswith("C"): rows.append((st, fields[i + 1], fields[i + 2])) i += 3 else: rows.append((st, fields[i + 1])) i += 2 return rows def blob_oid(repo, sha, path): out = run(["git", "rev-parse", "--verify", "--quiet", "%s:%s" % (sha, path)], repo, binary=True, check=False) return out.decode().strip() or None def byte_class(parent_bytes, child_bytes): """Pure byte append iff parent is a strict byte prefix of child.""" if not parent_bytes: return "NEW", 0 if parent_bytes == child_bytes: return "IDENTICAL", len(parent_bytes) n = min(len(parent_bytes), len(child_bytes)) cpl = 0 while cpl < n and parent_bytes[cpl] == child_bytes[cpl]: cpl += 1 if cpl == len(parent_bytes) and len(child_bytes) > len(parent_bytes): return "APPEND", cpl return "EDIT", cpl def line_test(parent_bytes, child_bytes): """(class, first_order_break_1based, that_line_text, missing_parent_lines, added_child_lines) class LOSS means the parent's line sequence is not an order-preserving subsequence of the child's: a line that existed is no longer there in order. first_order_break is where the greedy scan lost the thread (a locator). missing_parent_lines is the exact multiset difference |parent - child|: the number of parent lines that do not occur anywhere in the child. The two are different units and are reported separately (the 7th line after a break can still be present in the child). """ if not parent_bytes: return "NO_PARENT", None, "", 0, 0 pl = parent_bytes.split(b"\n") cl = child_bytes.split(b"\n") if pl == cl: return "IDENTICAL", None, "", 0, 0 missing = sum((Counter(pl) - Counter(cl)).values()) added = sum((Counter(cl) - Counter(pl)).values()) i = _subseq_match_count(pl, cl) if i == len(pl): return "SUBSEQUENCE", None, "", 0, added return "LOSS", i + 1, pl[i].decode("utf-8", "replace")[:120], missing, added def escape_unescape(b): """Replace literal backslash-uXXXX escape text with the character it denotes. One commit in this history (c597a133, Merlin 2026-10-15) is exactly this: 29 literal `\\u2014`-style escapes in WORK-QUEUE.md and 43 in a check-in were rewritten as the characters they denote. That replaces existing lines without changing what they say, so it must be told apart from text actually lost. """ return ESCAPE_RE.sub(lambda m: chr(int(m.group(1), 16)).encode("utf-8"), b) def _subseq_match_count(parent_lines, child_lines): """Greedy count of parent_lines that match, in order, within child_lines. Returns the number of parent_lines consumed by the forward scan: len(parent_lines) means the parent is a subsequence of the child; a smaller value means the scan broke after that many matches, and parent_lines[i] is the first parent line that did not match. """ i = 0 for ln in child_lines: if i < len(parent_lines) and ln == parent_lines[i]: i += 1 return i def is_line_subsequence(a_bytes, b_bytes): """True iff a_bytes's lines are an order-preserving subsequence of b_bytes's.""" al = a_bytes.split(b"\n") bl = b_bytes.split(b"\n") return _subseq_match_count(al, bl) == len(al) def loss_kind(parent_bytes, child_bytes): """ESCAPE_NORMALISATION if unescaping the parent alone explains the change, else REWRITE (text that existed is not there in the same form for any other reason). Only meaningful for rows the line test classed LOSS.""" if is_line_subsequence(escape_unescape(parent_bytes), child_bytes): return "ESCAPE_NORMALISATION" return "REWRITE" def audit(repo, progress=False): head = run(["git", "rev-parse", "HEAD"], repo).strip() shas = run(["git", "rev-list", "--reverse", "HEAD"], repo).split() reader = BlobReader(repo) rows, per_file = [], defaultdict(lambda: { "changes": 0, "NEW": 0, "APPEND": 0, "IDENTICAL": 0, "EDIT": 0, "NO_PARENT": 0, "SUBSEQUENCE": 0, "LOSS": 0, "renames": 0, "losses": [], "edits": [], "first_seen": None, "last_seen": None, }) cache = {} try: for n, sha in enumerate(shas): parent = first_parent(repo, sha) meta = run(["git", "log", "-1", "--format=%H%x09%an%x09%ad%x09%s", "--date=iso-strict", sha], repo).rstrip("\n").split("\t") for row in changed_paths(repo, parent, sha): if row[0] == "D": continue newpath = row[-1] is_rename = row[0].startswith("R") p_oid = blob_oid(repo, parent, newpath) if parent else None c_oid = blob_oid(repo, sha, newpath) if p_oid and p_oid not in cache: cache[p_oid] = reader.get(p_oid) if c_oid and c_oid not in cache: cache[c_oid] = reader.get(c_oid) pb = cache.get(p_oid, b"") cb = cache.get(c_oid, b"") bcls, cpl = byte_class(pb, cb) if is_rename: bcls = "RENAME" lcls, brk_no, brk_txt, miss_p, add_c = line_test(pb, cb) kind = loss_kind(pb, cb) if lcls == "LOSS" else "" rec = { "commit": sha, "author": meta[1], "date": meta[2], "subject": meta[3] if len(meta) > 3 else "", "path": newpath, "status": row[0], "byte_class": bcls, "common_prefix": cpl, "line_class": lcls, "loss_kind": kind, "first_order_break_line": brk_no, "order_break_line_text": brk_txt, "missing_parent_lines": miss_p, "added_child_lines": add_c, "parent_bytes": len(pb), "child_bytes": len(cb), } rows.append(rec) agg = per_file[newpath] agg["changes"] += 1 if bcls == "RENAME": agg["renames"] += 1 else: agg[bcls] += 1 agg[lcls] += 1 if bcls == "EDIT": agg["edits"].append({k: rec[k] for k in ("commit", "author", "date", "parent_bytes", "child_bytes", "common_prefix", "status")}) if lcls == "LOSS": agg["LOSS_" + kind] = agg.get("LOSS_" + kind, 0) + 1 agg["losses"].append({k: rec[k] for k in ("commit", "author", "date", "parent_bytes", "child_bytes", "loss_kind", "first_order_break_line", "order_break_line_text", "missing_parent_lines", "added_child_lines", "status")}) agg["first_seen"] = agg["first_seen"] or meta[2] agg["last_seen"] = meta[2] if progress and (n + 1) % 200 == 0: print(" ... %d/%d commits" % (n + 1, len(shas)), file=sys.stderr) if len(cache) > 4000: cache = {k: v for k, v in cache.items() if len(v) < 200000} finally: reader.close() return {"repo": repo, "head": head, "commits": len(shas), "rows": rows, "per_file": dict(per_file), "sanctioned_definition": list(APPEND_SANCTIONED)} def report(res, out=sys.stdout): rows, per_file = res["rows"], res["per_file"] bcount, lcount = defaultdict(int), defaultdict(int) for r in rows: bcount[r["byte_class"]] += 1 lcount[r["line_class"]] += 1 print("append-only audit -- repo %s" % res["repo"], file=out) print("HEAD %s ; %d commits ; %d (commit,path) changes" % (res["head"], res["commits"], len(rows)), file=out) print("BYTE test : " + " ".join("%s=%d" % (k, bcount[k]) for k in ("NEW", "APPEND", "IDENTICAL", "EDIT", "RENAME")), file=out) print("LINE test : " + " ".join("%s=%d" % (k, lcount[k]) for k in ("NO_PARENT", "SUBSEQUENCE", "IDENTICAL", "LOSS")), file=out) print(" (NEW in the byte test == NO_PARENT in the line test, two names for one rule)", file=out) print("", file=out) clean = sorted(p for p, a in per_file.items() if a["LOSS"] == 0) lossy = sorted((p, a) for p, a in per_file.items() if a["LOSS"] > 0) print("files with ZERO LOSS rows (no existing line ever lost): %d of %d" % (len(clean), len(per_file)), file=out) print("", file=out) print("files whose history contains LOSS rows (an existing line changed or removed):", file=out) for p, a in sorted(lossy, key=lambda kv: (-kv[1]["LOSS"], kv[0])): flag = " <-- APPEND-SANCTIONED" if is_sanctioned(p) else "" print(" %-52s LOSS=%-3d (of %d changes) byte-EDIT=%-3d%s" % (p, a["LOSS"], a["changes"], a["EDIT"], flag), file=out) print("", file=out) bad = [(p, a) for p, a in lossy if is_sanctioned(p)] if bad: total = sum(a["LOSS"] for _, a in bad) nesc = sum(a.get("LOSS_ESCAPE_NORMALISATION", 0) for _, a in bad) nrew = sum(a.get("LOSS_REWRITE", 0) for _, a in bad) print("CANDIDATE ROWS on append-sanctioned paths (%d file(s), %d LOSS rows)." % (len(bad), total), file=out) print(" A CANDIDATE is not a violation: the rule forbids ANOTHER PARTY's line, and a", file=out) print(" LOSS row only says a line was replaced. tools/append_only_owners.py adjudicates", file=out) print(" each row on the remover's identity; at f5403b2 it read 65 same-party / 9 other.", file=out) print(" of those: %d fully explained by unicode-escape normalisation (the parent's" % nesc, file=out) print(" lines reappear unchanged once literal \\uXXXX text is decoded); %d are REWRITE —" % nrew, file=out) print(" text that existed does not reappear in the same form for any encoding reason.", file=out) for p, a in sorted(bad, key=lambda kv: -kv[1]["LOSS"]): print("", file=out) print(" %s (%d changes, %d byte-APPEND, %d byte-EDIT, %d LOSS)" % (p, a["changes"], a["APPEND"], a["EDIT"], a["LOSS"]), file=out) for e in a["losses"]: print(" %s %s %s [%s]" % (e["commit"][:8], e["date"][:19], e["author"], e["loss_kind"]), file=out) print(" %d -> %d B ; parent lines gone: %d ; parent lines added: %d ; order break at line #%s" % (e["parent_bytes"], e["child_bytes"], e["missing_parent_lines"], e["added_child_lines"], e["first_order_break_line"]), file=out) if e["order_break_line_text"]: print(" line where the order scan broke: %r" % e["order_break_line_text"], file=out) else: print("CANDIDATE ROWS on append-sanctioned paths: none", file=out) print("", file=out) print("terms: BYTE-APPEND = parent blob is a byte-exact prefix of child and child is", file=out) print("longer. LOSS = the parent's line sequence is not an order-preserving subsequence", file=out) print("of the child's, i.e. a line that existed is no longer there in order. A mid-file", file=out) print("insertion is byte-EDIT and line-SUBSEQUENCE: it moves later text without losing", file=out) print("any. Sizes come from git objects via `git cat-file --batch`.", file=out) # --------------------------------------------------------------------------- selftest def _g(repo, *args): return subprocess.run(["git", "-C", repo] + list(args), check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).stdout.decode() def selftest(): tmp = tempfile.mkdtemp(prefix="aoa-selftest-") try: repo = os.path.join(tmp, "r") os.makedirs(repo) _g(repo, "init", "-q", "-b", "main") _g(repo, "config", "user.email", "t@test.invalid") _g(repo, "config", "user.name", "T") p = lambda *a: os.path.join(repo, *a) cells = [] # --- synthetic history: append, mid-file insertion, in-place edit, deletion, rename with open(p("a.txt"), "w") as f: f.write("one\ntwo\n") with open(p("b.txt"), "w") as f: f.write("keep\n") with open(p("mid.txt"), "w") as f: f.write("head\nbody\ntail\n") _g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c1 new") with open(p("a.txt"), "a") as f: f.write("three\n") _g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c2 append") with open(p("mid.txt"), "w") as f: # insert in the middle f.write("head\nINSERTED\nbody\ntail\n") _g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c3 mid insertion") with open(p("mid.txt"), "w") as f: # lose the word 'body' f.write("head\nINSERTED\ntail\n") _g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c4 line loss") with open(p("a.txt"), "w") as f: # in-place byte edit, same length f.write("ONE\ntwo\nthree\n") _g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c5 in-place edit") with open(p("c.txt"), "w") as f: f.write("new file\n") with open(p("d.txt"), "w") as f: f.write("keep\n") _g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c6 new files") os.rename(p("b.txt"), p("b2.txt")) _g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c7 rename") with open(p("esc.txt"), "w") as f: # literal escape text f.write("dash \\u2014 here\nplain\n") _g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c8 escape text") with open(p("esc.txt"), "w") as f: # rewritten as the character f.write("dash \u2014 here\nplain\n") _g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c9 escape normalised") res = audit(repo) rows = res["rows"] by = defaultdict(list) for r in rows: by[r["path"]].append(r) def cls(path, commit_subject_prefix, key): for r in rows: if r["path"] == path and r["subject"].startswith(commit_subject_prefix): return r[key] return None cells = [ ("c1 a NEW byte", cls("a.txt", "c1", "byte_class") == "NEW"), ("c1 b NEW byte", cls("b.txt", "c1", "byte_class") == "NEW"), ("c2 append byte", cls("a.txt", "c2", "byte_class") == "APPEND"), ("c3 insertion is byte-EDIT", cls("mid.txt", "c3", "byte_class") == "EDIT"), ("c3 insertion is line-SUBSEQUENCE", cls("mid.txt", "c3", "line_class") == "SUBSEQUENCE"), ("c4 loss is byte-EDIT", cls("mid.txt", "c4", "byte_class") == "EDIT"), ("c4 loss flagged LOSS", cls("mid.txt", "c4", "line_class") == "LOSS"), ("c4 loss names break at line 3", cls("mid.txt", "c4", "first_order_break_line") == 3), ("c4 loss counts 1 missing parent line", cls("mid.txt", "c4", "missing_parent_lines") == 1), ("c4 loss adds 0 child lines", cls("mid.txt", "c4", "added_child_lines") == 0), ("c4 break line text quoted", cls("mid.txt", "c4", "order_break_line_text") == "body"), ("c3 insertion adds 1 child line", cls("mid.txt", "c3", "added_child_lines") == 1), ("c5 in-place byte edit (cpl 0)", any( r["path"] == "a.txt" and r["byte_class"] == "EDIT" and r["common_prefix"] == 0 for r in rows)), ("c5 in-place edit is line-LOSS", cls("a.txt", "c5", "line_class") == "LOSS"), ("c7 rename detected", cls("b2.txt", "c7", "byte_class") == "RENAME"), ("c9 escape rewrite is LOSS", cls("esc.txt", "c9", "line_class") == "LOSS"), ("c9 classified ESCAPE_NORMALISATION", cls("esc.txt", "c9", "loss_kind") == "ESCAPE_NORMALISATION"), ("c4 real loss is not called normalisation", cls("mid.txt", "c4", "loss_kind") == "REWRITE"), ] ok = 0 for name, passed in cells: print("%-40s %s" % (name, "PASS" if passed else "FAIL")) ok += bool(passed) print("selftest: %d cells, %d passed, %d failed" % (len(cells), ok, len(cells) - ok)) return 0 if ok == len(cells) else 1 finally: shutil.rmtree(tmp, ignore_errors=True) def main(): ap = argparse.ArgumentParser() ap.add_argument("--repo", default=".") ap.add_argument("--json") ap.add_argument("--selftest", action="store_true") ap.add_argument("--progress", action="store_true") a = ap.parse_args() if a.selftest: return selftest() # A repo that is not there, or is not a git work tree, is a PRECONDITION # failure: rc 2 with a named message. Before 2026-10-17 this path escaped # as a FileNotFoundError traceback with rc 1 out of the first `git` call. if not os.path.isdir(a.repo): print("cannot read repo: %s is not a directory" % a.repo, file=sys.stderr) return 2 try: res = audit(a.repo, progress=a.progress) except (OSError, RuntimeError, subprocess.SubprocessError) as exc: print("cannot read repo %s: %s" % (a.repo, exc), file=sys.stderr) return 2 report(res) if a.json: try: with open(a.json, "w") as f: json.dump(res, f, indent=1, sort_keys=True) except OSError as exc: print("cannot write --json " + a.json + ": " + str(exc), file=sys.stderr) return 2 print("\njson written: %s (%d bytes)" % (a.json, os.path.getsize(a.json))) return 0 if __name__ == "__main__": sys.exit(main())