ledger: extract Hermes profile census to table

wizard-work/ledger/build_profile_census.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+"""Build the Hermes profile census table.
+
+Artifact: profiles-census.csv
+Columns:
+  profile   - Hermes profile directory name under /root/.hermes/profiles
+  archives  - number of *.txt activity transcripts in that profile dir
+  is_ledger - 1 if the profile is named "ledger", else 0
+
+The table is generated from the real filesystem (not fabricated), so it is
+fully reproducible by re-running this script.
+"""
+import csv
+import os
+
+PROFILES_DIR = "/root/.hermes/profiles"
+OUT = "/root/.hermes/wizard-outputs/ledger/profiles-census.csv"
+
+
+def archive_count(profile_dir):
+    count = 0
+    for root, dirs, files in os.walk(profile_dir):
+        for f in files:
+            if f.endswith(".txt"):
+                count += 1
+    return count
+
+
+def build_rows():
+    rows = []
+    for name in sorted(os.listdir(PROFILES_DIR)):
+        pdir = os.path.join(PROFILES_DIR, name)
+        if not os.path.isdir(pdir):
+            continue
+        rows.append(
+            {
+                "profile": name,
+                "archives": archive_count(pdir),
+                "is_ledger": 1 if name == "ledger" else 0,
+            }
+        )
+    return rows
+
+
+def main():
+    os.makedirs(os.path.dirname(OUT), exist_ok=True)
+    with open(OUT, "w", newline="") as fh:
+        writer = csv.DictWriter(fh, fieldnames=["profile", "archives", "is_ledger"])
+        writer.writeheader()
+        writer.writerows(build_rows())
+    print("wrote %s (%d profiles)" % (OUT, len(build_rows())))
+
+
+if __name__ == "__main__":
+    main()

wizard-work/ledger/profiles-census.csv

@@ -0,0 +1,30 @@
+profile,archives,is_ledger
+.deleted,0,0
+argus,2,0
+beam,741,0
+buzzpilot,1,0
+courtney,2,0
+forge,30,0
+home,2,0
+hou3,1,0
+inference-lab,2,0
+julie,7,0
+kendra,2,0
+ledger,17,1
+lilly,3,0
+merlin,5,0
+nightprowl,125,0
+null,614,0
+praetorian,7,0
+quill,2,0
+scribe,0,0
+sentinel,180,0
+sherry,2,0
+sol,30,0
+spark,12,0
+spark-snippet-wizard,0,0
+sphinx,53,0
+timmy,2,0
+vincent,2,0
+volt,51167,0
+worker,0,0

wizard-work/ledger/verify_profile_census.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+"""Verify the Hermes profile census artifact (Extract lane).
+
+Checks:
+  1. The CSV header is exactly: profile,archives,is_ledger
+  2. Every data row has 3 fields, numeric archives/is_ledger
+  3. The profile set and counts match a fresh walk of /root/.hermes/profiles
+  4. is_ledger flag is correct for every row
+
+Exit code 0 = correct, nonzero = wrong.
+"""
+import csv
+import os
+import sys
+
+DEFAULT_CSV = "/root/.hermes/wizard-outputs/ledger/profiles-census.csv"
+PROFILES_DIR = "/root/.hermes/profiles"
+
+
+def expected_rows():
+    rows = {}
+    for name in os.listdir(PROFILES_DIR):
+        pdir = os.path.join(PROFILES_DIR, name)
+        if not os.path.isdir(pdir):
+            continue
+        n = 0
+        for root, dirs, files in os.walk(pdir):
+            for f in files:
+                if f.endswith(".txt"):
+                    n += 1
+        rows[name] = (n, 1 if name == "ledger" else 0)
+    return rows
+
+
+def main():
+    csv_path = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_CSV
+    with open(csv_path, newline="") as fh:
+        reader = csv.reader(fh)
+        rows = list(reader)
+
+    errors = []
+
+    if not rows or rows[0] != ["profile", "archives", "is_ledger"]:
+        errors.append("bad header: %r" % (rows[0] if rows else None))
+
+    data = rows[1:]
+    if not data:
+        errors.append("no data rows")
+
+    expected = expected_rows()
+    if len(data) != len(expected):
+        errors.append("row count %d != expected %d" % (len(data), len(expected)))
+
+    for r in data:
+        if len(r) != 3:
+            errors.append("row has %d fields: %r" % (len(r), r))
+            continue
+        name, a, l = r
+        if not (a.isdigit() and l.isdigit()):
+            errors.append("non-numeric fields in row: %r" % r)
+            continue
+        want = expected.get(name)
+        if want is None:
+            errors.append("unexpected profile %r" % name)
+        elif (int(a), int(l)) != want:
+            errors.append("mismatch for %r: got (%s,%s) want %s" % (name, a, l, want))
+
+    extra = set(expected) - {r[0] for r in data}
+    for name in sorted(extra):
+        errors.append("missing profile %r" % name)
+
+    if errors:
+        print("FAIL")
+        for e in errors:
+            print(" -", e)
+        sys.exit(1)
+
+    total = sum(int(r[1]) for r in data)
+    print("OK: %d profiles, %d total txts, ledger=%s" % (len(data), total, [r for r in data if r[2] == "1"]))
+    sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()