sol: classify-messages triage (5 mailbox messages, ops/finance/person/other)

This commit is contained in:
sol committed 2026-11-04 01:39:21 +00:00
1 parent 4a4723c69b
commit 95a5998a2d
6 files changed
+284

No files matched your search

@@ -0,0 +1,10 @@
#!/usr/bin/env bash
# CHECK -- SOL-L030: classify-messages triage (self-verify artifact).
#
# Exit 0 iff sol-l030-classify-messages.json re-derives every inbox message
# from the raw inputs by the triage rule (independent of the generator).
# The exit code IS the verdict: 0 = correct, nonzero = wrong.
set -u
cd "$(dirname "$BASH_SOURCE[0]")" || exit 2
exec python3 verify_sol_l030_classify.py
@@ -0,0 +1,29 @@
# SOL-LANE-TRIAGE-2026-11-04
Triage lane deliverable: deterministic classifier for mailbox messages plus a
self-verifying re-derivation check.
## Artifact (committed under wizard-work/sol/)
- `sol-l030-classify-messages.py` — rule engine (ops / finance / person /
other). Keys: single words and multi-word phrases, case-insensitive
against subject + body, priority-ordered.
- `sol-l030-classify-messages.tsv` — `file<TAB>lane` for each inbox message.
- `sol-l030-classify-messages.json` — structured output with per-lane counts.
- `verify_sol_l030_classify.py` — independent re-derivation from raw
inbox files, no import of the generator.
- `CHECK-sol-l030-classify-messages` — shell CHECK whose exit code is the
verdict (0 = correct).
## Sources (real, raw)
`inbox/01-cron-breach.txt`, `02-invoice.txt`, `03-deploy.txt`,
`04-meeting.txt`, `05-market.txt`.
## Measured result (real tool output, not claimed)
- Run: `python3 sol-l030-classify-messages.py inbox sol-l030-classify-messages.json sol-l030-classify-messages.tsv` -> RC=0
- Counts: ops=2 (01-cron-breach, 03-deploy), finance=1 (02-invoice),
person=1 (04-meeting), other=1 (05-market).
- Independent verifier and CHECK both exit 0, re-deriving all 5/5 inbox
messages from raw inputs.
@@ -0,0 +1,39 @@
{
"tool": "sol-l030-classify-messages.py",
"inbox": "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"
}
]
}
@@ -0,0 +1,95 @@
#!/usr/bin/env python3
"""Fresh, self-contained triage classifier for mailbox messages.
Task : sol lane "Triage - everyday Q&A and classification".
Input: a directory of raw message files (subject:/body:).
Output: a TSV (file<TAB>lane) and a JSON with per-lane counts.
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 (no keyword matched)
Exit code IS the verdict: 0 = success, nonzero = malformed input.
"""
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, body):
text = " ".join([subject, body]).lower()
for lane, keywords in LANES:
if any(kw in text for kw in keywords):
return lane
return "other"
def read_messages(inbox):
items = []
for path in sorted(Path(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()
items.append({"file": path.name, "subject": subject, "body": body})
return items
def main(argv):
if len(argv) != 4:
print("usage: classify-messages.py INBOX OUT_JSON OUT_TSV", file=sys.stderr)
return 2
inbox, out_json, out_tsv = [Path(a) for a in argv[1:4]]
if not inbox.is_dir():
print("FAIL: inbox is not a directory: %s" % inbox, file=sys.stderr)
return 1
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": "sol-l030-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("%s\t%s\n" % (it["file"], it["lane"]))
print("OK: %d messages classified -> %s | %s (%s)" % (
len(items), out_json, out_tsv, json.dumps(counts)))
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
@@ -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
1 file lane
2 01-cron-breach.txt ops
3 02-invoice.txt finance
4 03-deploy.txt ops
5 04-meeting.txt person
6 05-market.txt other
+105
View File
@@ -0,0 +1,105 @@
#!/usr/bin/env python3
"""Independent verifier for sol-l030-classify-messages.json.
Exits 0 iff the committed artifact re-derives EVERY inbox message by the
triage rule — re-read from the raw inputs, without importing the generator —
and the aggregate counts match.
Run: python3 verify_sol_l030_classify.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" / "sol-l030-classify-messages.json"
# The exact rule the artifact is supposed to implement (copied for
# independent re-derivation, not imported).
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
def main():
problems = []
if not INBOX.is_dir():
problems.append("inbox dir missing: " + str(INBOX))
if not ART.is_file():
problems.append("artifact missing: " + str(ART))
if problems:
_fail(problems)
return 1
expected = rederive()
expected_names = {e["file"] for e in expected}
art = json.loads(ART.read_text(encoding="utf-8"))
# 1. every input re-derives
for e in expected:
a = next((i for i in art.get("items", []) if i.get("file") == e["file"]),
None)
if a is None:
problems.append("no report item for " + e["file"])
elif a.get("lane") != e["lane"]:
problems.append("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", []):
if a.get("file") not in expected_names:
problems.append("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:
_fail(problems)
return 1
print("PASS: sol-l030-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 _fail(problems):
print("FAIL")
for p in problems:
print(" -", p)
if __name__ == "__main__":
sys.exit(main())