sol: classify-messages triage classifier + self-verifying report

wizard-work/sol/CHECK-2026-11-03-classify

@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+# CHECK -- sol: classify-messages triage (forge: sol lane).
+# Exit 0 iff classify-messages.json re-derives every inbox message from the
+# raw inputs by the triage rule, independent of the generator.
+set -u
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)/.."
+REPO_ROOT="$(cd "$REPO_ROOT" && pwd)"
+cd "$REPO_ROOT" || exit 2
+
+exec python3 wizard-work/sol/verify_classify_messages.py

wizard-work/sol/classify-messages.json

@@ -0,0 +1,39 @@
+{
+  "tool": "classify-messages.py",
+  "inbox": "wizard-work/sol/inbox",
+  "total": 5,
+  "counts": {
+    "ops": 2,
+    "finance": 1,
+    "person": 1,
+    "other": 1
+  },
+  "lanes": [
+    "ops",
+    "finance",
+    "person",
+    "other"
+  ],
+  "items": [
+    {
+      "file": "01-cron-breach.txt",
+      "lane": "ops"
+    },
+    {
+      "file": "02-invoice.txt",
+      "lane": "finance"
+    },
+    {
+      "file": "03-deploy.txt",
+      "lane": "ops"
+    },
+    {
+      "file": "04-meeting.txt",
+      "lane": "person"
+    },
+    {
+      "file": "05-market.txt",
+      "lane": "other"
+    }
+  ]
+}

wizard-work/sol/classify-messages.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+"""Classify a directory of raw message files into triage lanes.
+
+Usage:
+    classify-messages.py INBOX_DIR OUT_JSON OUT_TSV
+
+Rule (deterministic, checked in priority order, case-insensitive against
+subject + body):
+
+    ops    -> cron, deploy, disk, backup, server, fail, uptime
+    finance-> invoice, payment, expense, salary, bitcoin, fee
+    person -> alexander, sol, wizard, troupe, meeting, alex
+    other  -> default
+
+Exit code IS the verdict: 0 = every message classified and counts written,
+nonzero = anything malformed.
+"""
+
+import json
+import sys
+from pathlib import Path
+
+LANES = [
+    ("ops", ["cron", "deploy", "disk", "backup", "server", "fail", "uptime"]),
+    ("finance", ["invoice", "payment", "expense", "salary", "bitcoin", "fee"]),
+    ("person", ["alexander", "sol", "wizard", "troupe", "meeting", "alex"]),
+]
+ORDER = ["ops", "finance", "person", "other"]
+
+
+def classify(subject: str, body: str) -> str:
+    text = f"{subject} {body}".lower()
+    for lane, kws in LANES:
+        if any(kw in text for kw in kws):
+            return lane
+    return "other"
+
+
+def read_messages(inbox: Path):
+    items = []
+    for path in sorted(inbox.glob("*.txt")):
+        subject, body = "", ""
+        for raw in path.read_text(encoding="utf-8").splitlines():
+            if raw.startswith("subject:"):
+                subject = raw.split(":", 1)[1].strip()
+            elif raw.startswith("body:"):
+                body = raw.split(":", 1)[1].strip()
+            elif raw == "":
+                continue
+            else:
+                body += " " + raw
+        items.append({"file": path.name, "subject": subject, "body": body})
+    return items
+
+
+def main(argv):
+    inbox = Path(argv[1])
+    out_json = Path(argv[2])
+    out_tsv = Path(argv[3])
+
+    items = read_messages(inbox)
+    for it in items:
+        it["lane"] = classify(it["subject"], it["body"])
+
+    counts = {lane: 0 for lane in ORDER}
+    for it in items:
+        counts[it["lane"]] += 1
+
+    out_json.write_text(
+        json.dumps(
+            {
+                "tool": "classify-messages.py",
+                "inbox": str(inbox),
+                "total": len(items),
+                "counts": counts,
+                "lanes": ORDER,
+                "items": [
+                    {"file": i["file"], "lane": i["lane"]} for i in items
+                ],
+            },
+            indent=2,
+        ),
+        encoding="utf-8",
+    )
+
+    with out_tsv.open("w", encoding="utf-8") as fh:
+        fh.write("file\tlane\n")
+        for it in items:
+            fh.write(f"{it['file']}\t{it['lane']}\n")
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))

wizard-work/sol/classify-messages.tsv

@@ -0,0 +1,6 @@
+file	lane
+01-cron-breach.txt	ops
+02-invoice.txt	finance
+03-deploy.txt	ops
+04-meeting.txt	person
+05-market.txt	other

wizard-work/sol/inbox/01-cron-breach.txt

@@ -0,0 +1,2 @@
+subject: cron failure watch
+body: cron_failure triggered on node-3, delaying daily drop.

wizard-work/sol/inbox/02-invoice.txt

@@ -0,0 +1,2 @@
+subject: Invoice for janitorial contract
+body: Please find attached the invoice for the maintenance contract.

wizard-work/sol/inbox/03-deploy.txt

@@ -0,0 +1,2 @@
+subject: deploy check
+body: Pre-deploy hook: verify disk space on prod before releasing.

wizard-work/sol/inbox/04-meeting.txt

@@ -0,0 +1,2 @@
+subject: troupe meeting
+body: Daily wizard standup: sol, alexander, and the troupe.

wizard-work/sol/inbox/05-market.txt

@@ -0,0 +1,2 @@
+subject: market update
+body: The price of tea in China fluctuates.

wizard-work/sol/verify_classify_messages.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+"""Independent verifier for classify-messages.json.
+
+Exit 0 iff the committed artifact re-derives every inbox message by the
+triage rule (re-read from raw inputs, no import of the generator) and the
+aggregate counts match.
+
+Run:  python3 verify_classify_messages.py
+"""
+
+import json
+import sys
+from pathlib import Path
+
+REPO = Path("/root/troupe-standup")
+INBOX = REPO / "wizard-work" / "sol" / "inbox"
+ART = REPO / "wizard-work" / "sol" / "classify-messages.json"
+
+# The exact rule the artifact is supposed to implement (copy, not import).
+LANE_RULES = [
+    ("ops", ["cron", "deploy", "disk", "backup", "server", "fail", "uptime"]),
+    ("finance", ["invoice", "payment", "expense", "salary", "bitcoin", "fee"]),
+    ("person", ["alexander", "sol", "wizard", "troupe", "meeting", "alex"]),
+]
+LANES = ["ops", "finance", "person", "other"]
+
+
+def rederive():
+    """Re-derive the classification of every inbox message straight from the
+    raw files, using only the rule above."""
+    items = []
+    for path in sorted(INBOX.glob("*.txt")):
+        subject, body = "", ""
+        for raw in path.read_text(encoding="utf-8").splitlines():
+            if raw.startswith("subject:"):
+                subject = raw.split(":", 1)[1].strip()
+            elif raw.startswith("body:"):
+                body = raw.split(":", 1)[1].strip()
+        text = (subject + " " + body).lower()
+        lane = "other"
+        for name, kws in LANE_RULES:
+            if any(k in text for k in kws):
+                lane = name
+                break
+        items.append({"file": path.name, "lane": lane})
+    return items
+
+
+problems = []
+
+
+def check(cond, msg):
+    if not cond:
+        problems.append(msg)
+
+
+def main():
+    if not INBOX.is_dir():
+        problems.append("inbox dir missing")
+        print_fail()
+        return 1
+    if not ART.exists():
+        problems.append("artifact missing: " + str(ART))
+        print_fail()
+        return 1
+
+    expected = rederive()
+    expected_names = {e["file"] for e in expected}
+    got_names = set()
+    art = json.loads(ART.read_text(encoding="utf-8"))
+
+    # ---- 1. every input re-derives -------------------------------------
+    for e in expected:
+        got_names.add(e["file"])
+        a = next((i for i in art.get("items", []) if i.get("file") == e["file"]),
+                 None)
+        check(a is not None, "no report item for " + e["file"])
+        if a is not None:
+            check(a.get("lane") == e["lane"],
+                  "lane mismatch for %r: got %r want %r" % (e["file"], a.get("lane"), e["lane"]))
+
+    # ---- 2. every report item came from an input ------------------------
+    for a in art.get("items", []):
+        check(a.get("file") in expected_names,
+              "report item not in inbox: " + a.get("file", "?"))
+
+    # ---- 3. counts ------------------------------------------------------
+    exp_counts = {l: 0 for l in LANES}
+    for e in expected:
+        exp_counts[e["lane"]] += 1
+    if art.get("counts") != exp_counts:
+        problems.append("counts %r != expected %r" % (art.get("counts"), exp_counts))
+    if art.get("total") != len(expected):
+        problems.append("total %r != expected %d" % (art.get("total"), len(expected)))
+
+    if problems:
+        print_fail()
+        return 1
+    print("PASS: classify-messages.json re-derives %d/%d inbox messages by the "
+          "triage rule; counts=%s" % (len(expected), len(expected), art.get("counts")))
+    return 0
+
+
+def print_fail():
+    print("FAIL")
+    for p in problems:
+        print("  -", p)
+
+
+if __name__ == "__main__":
+    sys.exit(main())