From adfb95d357aeaf00049939688786f6509e66d1da Mon Sep 17 00:00:00 2001 From: Michael Heller Date: Mon, 3 Aug 2026 06:35:09 -0400 Subject: [PATCH] feat(models): cross-plane promotion decision + RunReceipt (fail-closed, replayable) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit decide_promotion() is the one promotion contract both planes call: a challenger promotes only if it beats the champion AND passes the eval gate; regulated targets need examiner sign-off; nothing self-promotes (eval_passed/signed_off default False). Emits a RunReceipt whose hash is deterministic over the decision inputs — a promotion is examiner-replayable. 8 tests + a tools-tests CI workflow (gates promotion + surface-feed tests, previously ungated). --- .github/workflows/tools-tests.yml | 16 +++++++ tools/promote_model.py | 74 +++++++++++++++++++++++++++++++ tools/test_promote_model.py | 45 +++++++++++++++++++ 3 files changed, 135 insertions(+) create mode 100644 .github/workflows/tools-tests.yml create mode 100644 tools/promote_model.py create mode 100644 tools/test_promote_model.py diff --git a/.github/workflows/tools-tests.yml b/.github/workflows/tools-tests.yml new file mode 100644 index 0000000..4c5841a --- /dev/null +++ b/.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 diff --git a/tools/promote_model.py b/tools/promote_model.py new file mode 100644 index 0000000..957e069 --- /dev/null +++ b/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} diff --git a/tools/test_promote_model.py b/tools/test_promote_model.py new file mode 100644 index 0000000..79fae51 --- /dev/null +++ b/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)