feat(models): cross-plane promotion decision + RunReceipt (fail-closed, replayable)

.github/workflows/tools-tests.yml

@@ -0,0 +1,16 @@
+name: tools-tests
+on:
+  push:
+    paths: ['tools/**']
+  pull_request:
+    paths: ['tools/**']
+jobs:
+  pytest:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with: { python-version: '3.11' }
+      - run: python -m pip install pytest
+      - name: Run tools/ tests (promotion, surface feeds)
+        run: python -m pytest tools/ -q

tools/promote_model.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+"""Cross-plane champion → challenger promotion, fail-closed, with a sealed RunReceipt.
+
+One promotion contract both planes (cloud lattice + local agent-machine) call. A
+challenger is promoted ONLY when it beats the champion AND has passed the eval gate;
+a regulated target additionally requires examiner sign-off. Every decision emits a
+RunReceipt whose hash is deterministic over the decision inputs — so a promotion is
+replayable: re-run the same inputs, get the same receipt. Nothing self-promotes:
+eval_passed and signed_off default False.
+"""
+from __future__ import annotations
+
+import datetime
+import hashlib
+import json
+from typing import Any, Dict, List
+
+DRIFT_THRESHOLD = 0.10  # PSI above this flags the target for review
+
+
+def _seal(payload: Dict[str, Any]) -> str:
+    return "sha256:" + hashlib.sha256(
+        json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    ).hexdigest()
+
+
+def decide_promotion(
+    target: str,
+    champion: Dict[str, Any],
+    challenger: Dict[str, Any],
+    *,
+    drift: float = 0.0,
+    eval_passed: bool = False,
+    regulated: bool = False,
+    signed_off: bool = False,
+) -> Dict[str, Any]:
+    """Return {verdict, reason, flags, receipt}. Verdict ∈ promote|hold|shadow|blocked."""
+    improved = challenger.get("val", 0) > champion.get("val", 0)
+    delta = round(challenger.get("val", 0) - champion.get("val", 0), 4)
+
+    if not eval_passed:
+        verdict = "shadow" if improved else "hold"
+        reason = ("challenger leads but has not passed the eval gate — held in shadow"
+                  if improved else "no improvement and eval gate not passed")
+    elif not improved:
+        verdict, reason = "hold", (
+            f"challenger {challenger.get('metric')} {challenger.get('val')} "
+            f"does not beat champion {champion.get('val')}")
+    elif regulated and not signed_off:
+        verdict, reason = "blocked", "regulated target — examiner sign-off required before promotion"
+    else:
+        verdict, reason = "promote", f"challenger beats champion by {delta} and passed the eval gate"
+
+    flags: List[str] = []
+    if drift > DRIFT_THRESHOLD:
+        flags.append(f"PSI drift {drift} exceeds {DRIFT_THRESHOLD} — target flagged for review")
+
+    decision_core = {
+        "kind": "RunReceipt",
+        "recordType": "ModelPromotion",
+        "target": target,
+        "champion": {"name": champion.get("name"), "ver": champion.get("ver"), "val": champion.get("val")},
+        "challenger": {"name": challenger.get("name"), "ver": challenger.get("ver"), "val": challenger.get("val")},
+        "delta": delta,
+        "drift": drift,
+        "eval_passed": eval_passed,
+        "regulated": regulated,
+        "signed_off": signed_off,
+        "verdict": verdict,
+    }
+    receipt = dict(decision_core)
+    receipt["receipt_hash"] = _seal(decision_core)  # deterministic → replayable
+    receipt["occurred_at"] = datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+    return {"verdict": verdict, "reason": reason, "flags": flags, "receipt": receipt}

tools/test_promote_model.py

@@ -0,0 +1,45 @@
+from promote_model import decide_promotion, _seal
+
+CH = {"name": "GBM-fraud", "ver": "v4", "metric": "AUC", "val": 0.962}
+CL = {"name": "GNN-fraud", "ver": "v1", "metric": "AUC", "val": 0.971}
+WORSE = {"name": "X", "ver": "v1", "metric": "AUC", "val": 0.95}
+
+
+def test_promote_when_better_and_eval_passed_unregulated():
+    d = decide_promotion("fraud", CH, CL, eval_passed=True)
+    assert d["verdict"] == "promote"
+    assert d["receipt"]["receipt_hash"].startswith("sha256:")
+
+
+def test_regulated_blocks_without_signoff():
+    d = decide_promotion("credit", CH, CL, eval_passed=True, regulated=True)
+    assert d["verdict"] == "blocked"
+
+
+def test_regulated_promotes_with_signoff():
+    d = decide_promotion("credit", CH, CL, eval_passed=True, regulated=True, signed_off=True)
+    assert d["verdict"] == "promote"
+
+
+def test_hold_when_no_improvement():
+    assert decide_promotion("fraud", CH, WORSE, eval_passed=True)["verdict"] == "hold"
+
+
+def test_shadow_when_better_but_no_eval():
+    assert decide_promotion("fraud", CH, CL)["verdict"] == "shadow"
+
+
+def test_fail_closed_default_never_promotes():
+    # defaults: eval_passed=False → never promote, even for a clear winner
+    assert decide_promotion("fraud", CH, CL)["verdict"] != "promote"
+
+
+def test_drift_is_flagged():
+    d = decide_promotion("credit", CH, CL, eval_passed=True, drift=0.11)
+    assert d["flags"] and "drift" in d["flags"][0]
+
+
+def test_receipt_is_deterministic_replayable():
+    a = decide_promotion("fraud", CH, CL, eval_passed=True)["receipt"]["receipt_hash"]
+    b = decide_promotion("fraud", CH, CL, eval_passed=True)["receipt"]["receipt_hash"]
+    assert a == b  # same inputs → same seal (occurred_at excluded from the hash)