sol: triage classifier artifact (classify.py, demands.json, expected.csv)

wizard-work/sol/actual.csv

@@ -0,0 +1,9 @@
+id,text,category,score,priority,reason
+D1,"Should we approve the migration of the logs pipeline to the new cluster, and who owns the rollback plan?",decision,2,1,"matched approve, should we"
+D2,Please fix the 502 error on the ingress gateway and re-run the smoke suite before close of day.,action,2,2,"matched fix, run"
+D3,Summarize the free-tier costs across the AWS and self-hosted stacks for the quarter.,info,0,3,no rule matched; defaulted to info
+D4,Send the updated dispatch manifest to ops.,action,3,2,"matched dispatch, send, update"
+D5,Closed the stale ticket and turned off the dev snapshot.,done,1,4,matched closed
+D6,How does the Stackchain library handle a stale Hermes profile?,info,1,3,matched how does
+D7,Choose and commit to the single provider for the nightly cron jobs.,decision,2,1,"matched choose, commit"
+D8,Deploy the new release to staging and watch the canary.,action,1,2,matched deploy

wizard-work/sol/actual.json

@@ -0,0 +1,77 @@
+{
+  "classified": [
+    {
+      "id": "D1",
+      "text": "Should we approve the migration of the logs pipeline to the new cluster, and who owns the rollback plan?",
+      "category": "decision",
+      "score": 2,
+      "priority": 1,
+      "reason": "matched approve, should we"
+    },
+    {
+      "id": "D2",
+      "text": "Please fix the 502 error on the ingress gateway and re-run the smoke suite before close of day.",
+      "category": "action",
+      "score": 2,
+      "priority": 2,
+      "reason": "matched fix, run"
+    },
+    {
+      "id": "D3",
+      "text": "Summarize the free-tier costs across the AWS and self-hosted stacks for the quarter.",
+      "category": "info",
+      "score": 0,
+      "priority": 3,
+      "reason": "no rule matched; defaulted to info"
+    },
+    {
+      "id": "D4",
+      "text": "Send the updated dispatch manifest to ops.",
+      "category": "action",
+      "score": 3,
+      "priority": 2,
+      "reason": "matched dispatch, send, update"
+    },
+    {
+      "id": "D5",
+      "text": "Closed the stale ticket and turned off the dev snapshot.",
+      "category": "done",
+      "score": 1,
+      "priority": 4,
+      "reason": "matched closed"
+    },
+    {
+      "id": "D6",
+      "text": "How does the Stackchain library handle a stale Hermes profile?",
+      "category": "info",
+      "score": 1,
+      "priority": 3,
+      "reason": "matched how does"
+    },
+    {
+      "id": "D7",
+      "text": "Choose and commit to the single provider for the nightly cron jobs.",
+      "category": "decision",
+      "score": 2,
+      "priority": 1,
+      "reason": "matched choose, commit"
+    },
+    {
+      "id": "D8",
+      "text": "Deploy the new release to staging and watch the canary.",
+      "category": "action",
+      "score": 1,
+      "priority": 2,
+      "reason": "matched deploy"
+    }
+  ],
+  "summary": {
+    "total": 8,
+    "by_category": {
+      "decision": 2,
+      "info": 2,
+      "action": 3,
+      "done": 1
+    }
+  }
+}

wizard-work/sol/classify.py

@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+"""Triage classifier for the sol lane.
+
+Classifies a list of demands/requests into triage categories
+(decision | info | action | done) and emits a CSV + JSON payload.
+
+Rules are keyword/phrase regexes, scored per category.  Highest score wins;
+ties broken by a fixed priority order.  No third-party deps, stdlib only.
+"""
+import argparse
+import csv
+import json
+import re
+import sys
+from pathlib import Path
+
+CATS = ["decision", "info", "action", "done"]
+# fixed priority for tie-breaking (higher = elected first on a draw)
+PRIORITY = {"decision": 4, "action": 3, "info": 2, "done": 1}
+
+RULES = {
+    "decision": [
+        r"decision", r"\bwhich\b", r"\bchoose\b", r"\bpick\b",
+        r"\bshould we\b", r"\bapprove\b", r"\bconfirm\b", r"\bagree\b",
+        r"\bpolicy\b", r"\breconcile\b", r"\bresolve\b", r"\bftr eve\b",
+        r"\barchitect\b", r"\bcommit\b",
+    ],
+    "info": [
+        r"what is", r"how does", r"explain", r"overview", r"summary",
+        r"define", r"meaning", r"difference", r"compare", r"describe",
+        r"purpose", r"why", r"when",
+    ],
+    "action": [
+        r"fix", r"implement", r"send", r"create", r"deploy", r"update",
+        r"clean", r"setup", r"\brun\b", r"\benable\b", r"\binstall\b",
+        r"\bmigrate\b", r"\bbuild\b", r"\bprocess\b", r"\bdispatch\b",
+    ],
+    "done": [
+        r"\bdone\b", r"\bclosed\b", r"\bcompleted\b", r"\bresolved\b",
+        r"\bfinished\b", r"\bprocessed\b", r"\bsent\b", r"\bok\b",
+        r"\bworkaround\b",
+    ],
+}
+
+TIE_PRIORITY = {"decision": 4, "action": 3, "info": 2, "done": 1}
+
+
+def classify(text):
+    """Return (category, score, matched) for a single demand string."""
+    low = text.lower()
+    scores = {}
+    matched = {}
+    for cat, patterns in RULES.items():
+        s = 0
+        ms = []
+        for p in patterns:
+            m = re.search(p, low)
+            if m:
+                s += 1
+                ms.append(m.group(0))
+        scores[cat] = s
+        matched[cat] = ms
+
+    best = max(scores.values())
+    if best == 0:
+        cat = "info"
+        reason = "no rule matched; defaulted to info"
+    else:
+        # pick highest score, tie-break by fixed priority
+        winners = [c for c in CATS if scores[c] == best]
+        cat = max(winners, key=lambda c: TIE_PRIORITY[c])
+        parts = []
+        for c in winners:
+            parts.append(f"{c}:{scores[c]}")
+        reason = "matched " + ", ".join(sorted(set(x for c in winners for x in matched[c])))
+    return cat, best, reason
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description="Triage classifier")
+    ap.add_argument("--input", required=True, help="JSON file with a 'demands' array")
+    ap.add_argument("--output", required=True, help="CSV output path")
+    ap.add_argument("--pretty", action="store_true", help="also write .json payload")
+    args = ap.parse_args()
+
+    data = json.loads(Path(args.input).read_text(encoding="utf-8"))
+    demands = data.get("demands", data if isinstance(data, list) else [])
+
+    rows = []
+    for d in demands:
+        text = d.get("text", "")
+        priority = d.get("priority")
+        cat, score, reason = classify(text)
+        if priority is None:
+            priority = {"decision": 1, "action": 2, "info": 3, "done": 4}[cat]
+        rows.append({
+            "id": d.get("id", ""),
+            "text": text,
+            "category": cat,
+            "score": score,
+            "priority": priority,
+            "reason": reason,
+        })
+
+    fieldnames = ["id", "text", "category", "score", "priority", "reason"]
+    with open(args.output, "w", encoding="utf-8", newline="") as f:
+        w = csv.DictWriter(f, fieldnames=fieldnames, lineterminator="\n")
+        w.writeheader()
+        w.writerows(rows)
+
+    if args.pretty:
+        out_json = args.output.rsplit(".", 1)[0] + ".json"
+        Path(out_json).write_text(
+            json.dumps({"classified": rows, "summary": _summary(rows)}, indent=2),
+            encoding="utf-8",
+        )
+    return 0
+
+
+def _summary(rows):
+    counts = {c: 0 for c in CATS}
+    for r in rows:
+        counts[r["category"]] += 1
+    return {"total": len(rows), "by_category": counts}
+
+
+if __name__ == "__main__":
+    sys.exit(main())

wizard-work/sol/demands.json

@@ -0,0 +1,12 @@
+{
+  "demands": [
+    {"id": "D1", "text": "Should we approve the migration of the logs pipeline to the new cluster, and who owns the rollback plan?"},
+    {"id": "D2", "text": "Please fix the 502 error on the ingress gateway and re-run the smoke suite before close of day."},
+    {"id": "D3", "text": "Summarize the free-tier costs across the AWS and self-hosted stacks for the quarter."},
+    {"id": "D4", "text": "Send the updated dispatch manifest to ops."},
+    {"id": "D5", "text": "Closed the stale ticket and turned off the dev snapshot."},
+    {"id": "D6", "text": "How does the Stackchain library handle a stale Hermes profile?"},
+    {"id": "D7", "text": "Choose and commit to the single provider for the nightly cron jobs."},
+    {"id": "D8", "text": "Deploy the new release to staging and watch the canary."}
+  ]
+}

wizard-work/sol/expected.csv

@@ -0,0 +1,9 @@
+id,text,category,score,priority,reason
+D1,"Should we approve the migration of the logs pipeline to the new cluster, and who owns the rollback plan?",decision,2,1,"matched approve, should we"
+D2,Please fix the 502 error on the ingress gateway and re-run the smoke suite before close of day.,action,2,2,"matched fix, run"
+D3,Summarize the free-tier costs across the AWS and self-hosted stacks for the quarter.,info,0,3,no rule matched; defaulted to info
+D4,Send the updated dispatch manifest to ops.,action,3,2,"matched dispatch, send, update"
+D5,Closed the stale ticket and turned off the dev snapshot.,done,1,4,matched closed
+D6,How does the Stackchain library handle a stale Hermes profile?,info,1,3,matched how does
+D7,Choose and commit to the single provider for the nightly cron jobs.,decision,2,1,"matched choose, commit"
+D8,Deploy the new release to staging and watch the canary.,action,1,2,matched deploy

wizard-work/sol/verify.sh

@@ -0,0 +1,24 @@
+#!/usr/bin/env bash
+# Verify the triage classifier artifact end-to-end.
+set -u
+cd "$(dirname "$0")" || exit 1
+
+# 1) regenerate the artifact
+python3 classify.py --input demands.json --output actual.csv --pretty
+test $? -eq 0 || { echo "FAIL: classification run"; exit 1; }
+
+# 2) JSON summary assertion (no -c one-liner)
+python3 - <<'PY'
+import json
+d = json.load(open("actual.json"))
+want = {"total": 8, "by_category": {"decision": 2, "info": 2, "action": 3, "done": 1}}
+if d.get("summary") != want:
+    raise SystemExit("summary mismatch: %r" % (d.get("summary"),))
+PY
+test $? -eq 0 || { echo "FAIL: summary assertion"; exit 1; }
+
+# 3) CSV diff against known-good output
+diff -u expected.csv actual.csv
+test $? -eq 0 || { echo "FAIL: CSV diff"; exit 1; }
+
+echo "CHECK OK"