Phase 3 pilot: ingest srdcheck --trace as qc.verdict provenance records

SKILL.md

@@ -48,6 +48,8 @@ tablekit checkin --seat vesh                # noticed a quiet seat
 tablekit turn --seat bram [--wait N]        # a seat got the floor
 tablekit qc [--json]                        # run the checks now (exit 1 on findings)
 tablekit pairs [--json]                     # what's still open
+srdcheck query ... --trace 2>&1 | tablekit verdict [--seat S]
+                                            # record the rules rail's verdict provenance
 ```
 
 Try it with zero setup: `python3 examples/demo_session.py` runs a synthetic
@@ -95,9 +97,11 @@ in-fiction, not as a form.
 
 - **Session retro:** the session JSONL is dmcheck-compatible — run
   `dmcheck run <session>.jsonl --charter <table charter>` after close-out.
-- **Live ruling:** when srdcheck adjudicates mid-turn, record the ruling as
-  a beat and its dice as `roll`/`consumed` pairs; the ledger becomes the
-  precedent trail.
+- **Live ruling:** when srdcheck adjudicates mid-turn, pipe its `--trace`
+  stream through `tablekit verdict` — the ledger stores the verdict's
+  provenance (tool@version, adapter pins, sha256 verdict id) as a
+  `qc.verdict` record — then narrate the ruling as a beat and its dice as
+  `roll`/`consumed` pairs. The ledger becomes the precedent trail.
 - **Session start:** charactercheck `seatpack --for-dm` supplies each
   seat's numbers; tablekit tracks the evening those numbers live through.
 
@@ -108,6 +112,9 @@ Family contract: [FAMILY.md](https://github.com/chaoz23/srdcheck/blob/main/FAMIL
 - **0.2.0:** bang markers CUT — replaced by inferred signals + a
   plain-English close-out. If you remember `!markers`, that's stale and
   tests reject reintroducing them.
+- **0.5.x:** `tablekit verdict` ingests a rules rail's `--trace`
+  observability stream as `qc.verdict` provenance records (pilot: srdcheck;
+  advisory always, never scored).
 - **0.4.x:** five lanes stable (qa/qc/ux/uxr/out); uxr inferred from
   ordinary speech during play, asked about conversationally at close.
 

tablekit/cli.py

@@ -43,6 +43,8 @@
     tablekit signal --seat S --kind pacing --quote "<their words>"
                                                record an inferred signal
     tablekit park "<what looked off>" [--topic X]   park it, keep playing
+    srdcheck query ... --trace 2>&1 | tablekit verdict [--seat S]
+                                               record the rail's verdict provenance
     tablekit park --list | --done "<detail>"    the standing issues list
     tablekit qc [--state FILE] [--json]        run and record the checks
     tablekit pairs                             what is still open
@@ -416,6 +418,74 @@ def cmd_turn(args):
     return 0
 
 
+def cmd_verdict(args):
+    """Ingest a rules rail's --trace stream into the ledger.
+
+    Reads stdin, keeps every line that is a JSON object shaped like an
+    srdcheck observability `request.completed` event (verdict_id + engine +
+    exit_code present), and appends one `qc.verdict` record per line. All
+    other lines — the human-facing verdict JSON, `request.started`, prose —
+    are skipped silently, so `srdcheck query ... --trace 2>&1 | tablekit
+    verdict` works as a single pipe.
+
+    This records provenance, not authority: the verdict stays advisory and
+    the ruling stays the GM's. Nothing here is ever aggregated into a score.
+    """
+    seat = _flag(args, "--seat")
+    cfg, led = _ctx(args)
+    _no_positionals(args, "verdict")
+    sid = _seat_id(cfg, seat) if seat else None
+    if seat:
+        _known_seat(cfg, seat)
+    ingested = 0
+    for line in sys.stdin:
+        line = line.strip()
+        if not line or not line.startswith("{"):
+            continue
+        try:
+            ev = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+        if not isinstance(ev, dict) or ev.get("event") != "request.completed":
+            continue
+        engine = ev.get("engine") or {}
+        verdict_id = ev.get("verdict_id")
+        exit_code = ev.get("exit_code")
+        if (not verdict_id or not engine.get("name")
+                or isinstance(exit_code, bool)
+                or not isinstance(exit_code, int)):
+            continue
+        adapters = ",".join(
+            f"{a.get('name')}@{a.get('version')}"
+            for a in ev.get("adapters") or [] if isinstance(a, dict))
+        led.append(
+            "qc.verdict",
+            tool=f"{engine.get('name')}@{engine.get('version')}",
+            query_type=str(ev.get("query_type") or "unknown"),
+            outcome=str(ev.get("outcome") or "unknown"),
+            exit_code=exit_code,
+            verdict_id=str(verdict_id),
+            request_id=(str(ev["request_id"])
+                        if ev.get("request_id") is not None else None),
+            adapters=adapters or None,
+            duration_ms=(ev["duration_ms"]
+                         if isinstance(ev.get("duration_ms"), (int, float))
+                         and not isinstance(ev.get("duration_ms"), bool)
+                         else None),
+            schema_version=(str(ev["schema_version"])
+                            if ev.get("schema_version") is not None else None),
+            seat=sid,
+        )
+        ingested += 1
+    if not ingested:
+        print("verdict: no request.completed events on stdin — pipe the "
+              "rail's --trace stream (e.g. srdcheck query ... --trace 2>&1 "
+              "| tablekit verdict)", file=sys.stderr)
+        return 2
+    print(f"{ingested} verdict(s) recorded")
+    return 0
+
+
 def cmd_park(args):
     """Park something for below-the-table investigation.
 
@@ -611,7 +681,8 @@ def cmd_migrate_events(args):
     "inbound": cmd_inbound, "roll": cmd_roll, "consumed": cmd_consumed,
     "checkin": cmd_checkin, "turn": cmd_turn, "signal": cmd_signal,
     "debrief": cmd_debrief, "qc": cmd_qc, "pairs": cmd_pairs,
-    "sweep": cmd_sweep, "park": cmd_park, "report": cmd_report,
+    "sweep": cmd_sweep, "park": cmd_park, "verdict": cmd_verdict,
+    "report": cmd_report,
     "schema": cmd_schema, "contract": cmd_contract,
     "migrate-events": cmd_migrate_events,
 }

tablekit/events.py

@@ -85,6 +85,12 @@
     "qa.delta":      ("topic", "detail"),
     # --- qc: is the refereeing correct --------------------------------
     "qc.finding": ("check", "detail"),
+    # An external rules rail's verdict, ingested from its --trace
+    # observability stream (srdcheck-shaped: request.completed lines).
+    # Stored so a ruling's provenance survives the evening and session
+    # retro can see which calls the rail actually decided. Never a score.
+    "qc.verdict": ("tool", "query_type", "outcome", "exit_code",
+                   "verdict_id"),
     "qc.pass":    ("checks",),
     # Written ONLY by detector.record(), so "was this session checked?" cannot
     # be satisfied by a finding that some other code path happened to emit.
@@ -218,6 +224,10 @@ def _invalid_json_constant(value):
     "qa.command": {"cmd": _string, "ok": _boolean},
     "qa.delta": {"topic": _string, "detail": _string},
     "qc.finding": {"check": _string, "detail": _string},
+    "qc.verdict": {"tool": _string, "query_type": _string,
+                   "outcome": _string,
+                   "exit_code": _integer,
+                   "verdict_id": _string},
     "qc.pass": {"checks": _integer},
     "qc.run": {"findings": _integer},
     "qc.mark": {"narrated_through": _integer},
@@ -233,6 +243,13 @@ def _invalid_json_constant(value):
 }
 
 OPTIONAL_FIELD_VALIDATORS = {
+    "qc.verdict": {
+        "request_id": _string,
+        "adapters": _string,
+        "seat": _string,
+        "duration_ms": _number,
+        "schema_version": _string,
+    },
     "qc.finding": {
         "severity": lambda v, f, t: _enum(v, f, t, ("defect", "attention")),
         "status": lambda v, f, t: _enum(

tests/test_verdict_ingest.py

@@ -0,0 +1,127 @@
+"""qc.verdict ingestion — the rules rail's provenance lands in the ledger.
+
+unittest on purpose; see test_tablekit.py's module docstring.
+"""
+
+import io
+import json
+import os
+import sys
+import tempfile
+import unittest
+from contextlib import redirect_stdout
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from tablekit import cli  # noqa: E402
+from tablekit.events import Ledger, SchemaError, make  # noqa: E402
+
+COMPLETED = {
+    "event": "request.completed",
+    "schema_version": "srdcheck.observability/1.0",
+    "engine": {"name": "srdcheck", "version": "0.9.0"},
+    "adapters": [{"name": "srd-5.2.1", "version": "0.2.1"}],
+    "query_type": "concentration.check",
+    "exit_code": 0,
+    "outcome": "legal",
+    "duration_ms": 0.046,
+    "request_id": "turn-047",
+    "verdict_id": "sha256:6e29f15cc3f918114f0f97e06b6f5897fcad9b514f2",
+}
+
+STARTED = {
+    "event": "request.started",
+    "schema_version": "srdcheck.observability/1.0",
+    "engine": {"name": "srdcheck", "version": "0.9.0"},
+    "query_type": "concentration.check",
+    "request_id": "turn-047",
+}
+
+
+class VerdictIngest(unittest.TestCase):
+    def setUp(self):
+        d = tempfile.mkdtemp(prefix="tk-verdict-")
+        self.path = os.path.join(d, "session.jsonl")
+
+    def _rows(self):
+        with open(self.path) as fh:
+            return [json.loads(line) for line in fh if line.strip()]
+
+    def _run(self, stdin_text, *extra):
+        old = sys.stdin
+        sys.stdin = io.StringIO(stdin_text)
+        try:
+            with redirect_stdout(io.StringIO()):
+                return cli.main(["verdict", *extra, "--ledger", self.path])
+        finally:
+            sys.stdin = old
+
+    def test_ingests_completed_events_only(self):
+        stream = "\n".join([
+            json.dumps(STARTED),
+            json.dumps(COMPLETED),
+            '{ not json',
+            '{"verdict": "legal", "exit_code": 0}',  # human verdict, no id
+            "  ",
+        ])
+        self.assertEqual(self._run(stream), 0)
+        rows = self._rows()
+        self.assertEqual(len(rows), 1)
+        rec = rows[0]
+        self.assertEqual(rec["type"], "qc.verdict")
+        self.assertEqual(rec["tool"], "srdcheck@0.9.0")
+        self.assertEqual(rec["query_type"], "concentration.check")
+        self.assertEqual(rec["outcome"], "legal")
+        self.assertEqual(rec["exit_code"], 0)
+        self.assertEqual(rec["adapters"], "srd-5.2.1@0.2.1")
+        self.assertEqual(rec["request_id"], "turn-047")
+        self.assertTrue(rec["verdict_id"].startswith("sha256:"))
+
+    def test_mixed_pipe_with_pretty_verdict_json(self):
+        # `srdcheck query ... --trace 2>&1` interleaves pretty-printed verdict
+        # JSON (multi-line, so no line parses as an object with an event key)
+        # with the observability lines. Only the completed line lands.
+        pretty = json.dumps({"verdict": "legal", "exit_code": 0}, indent=2)
+        stream = json.dumps(COMPLETED) + "\n" + pretty + "\n"
+        self.assertEqual(self._run(stream), 0)
+        self.assertEqual(len(self._rows()), 1)
+
+    def test_empty_stream_refuses(self):
+        err = io.StringIO()
+        old = sys.stderr
+        sys.stderr = err
+        try:
+            self.assertEqual(self._run("no events here\n"), 2)
+        finally:
+            sys.stderr = old
+        self.assertIn("request.completed", err.getvalue())
+        self.assertFalse(os.path.exists(self.path))
+
+    def test_exit_two_verdicts_are_first_class(self):
+        refused = dict(COMPLETED, exit_code=2, outcome="cannot-adjudicate",
+                       verdict_id="sha256:beef")
+        self.assertEqual(self._run(json.dumps(refused)), 0)
+        rec = self._rows()[0]
+        self.assertEqual(rec["exit_code"], 2)
+        self.assertEqual(rec["outcome"], "cannot-adjudicate")
+
+    def test_schema_refuses_negative_exit_code(self):
+        with self.assertRaises(SchemaError):
+            make("qc.verdict", tool="x@1", query_type="q", outcome="legal",
+                 exit_code=-1, verdict_id="sha256:x")
+
+    def test_dmcheck_compat_lane_is_qc(self):
+        # dmcheck reads only play-lane types; qc.verdict must not be one.
+        from tablekit.events import PLAY_TYPES, lane_of
+        self.assertNotIn("qc.verdict", PLAY_TYPES)
+        self.assertEqual(lane_of("qc.verdict"), "qc")
+
+    def test_ledger_append_roundtrip(self):
+        led = Ledger(self.path)
+        led.append("qc.verdict", tool="srdcheck@0.9.0", query_type="q",
+                   outcome="illegal", exit_code=1, verdict_id="sha256:aa")
+        self.assertEqual(self._rows()[0]["exit_code"], 1)
+
+
+if __name__ == "__main__":
+    unittest.main()