feat(rollout): promotion gate — enforce a consumed APPROVE review verdict, seal to evidence

.gitignore

@@ -12,6 +12,9 @@ node_modules/
 # OS cruft
 .DS_Store
 
+# Promotion-gate runtime evidence (emitted at rollout, not committed; dir kept via .gitkeep)
+artifacts/gate-decisions/*.json
+
 # Porter control-plane build outputs (compiled locally, not committed)
 artifacts/porter-shim/shim
 artifacts/cloudshell-hardened-pack/culler/culler

Makefile

@@ -1,6 +1,6 @@
 # SourceOS Continuum — lifecycle entry points.
 # Control-plane targets delegate to Makefile.porter (the rehomed Porter control plane).
-.PHONY: validate onboard dev-up dev-down shim-test test rollout
+.PHONY: validate onboard dev-up dev-down shim-test test rollout promotion-gate
 
 validate: ## repo hygiene + CapD validity
 	python3 tools/validate.py
@@ -20,5 +20,9 @@ shim-test: ## test the porter-shim control plane
 test: ## cloud-native test: ephemeral preview env + GitOps PR checks + evidence bundle
 	@echo "[continuum] test — scaffold: PR-driven preview environments + evidence"
 
-rollout: ## promote local → scale-up cluster (hyperswarm), signed images
-	@echo "[continuum] rollout — scaffold: promote via caps.infra.cluster-scaleup.hyperswarm"
+promotion-gate: ## rollout gate: require an APPROVE review verdict (fail-closed, evidence-emitting)
+	@test -n "$(VERDICT)" || (echo "[continuum] promotion-gate BLOCKED: set VERDICT=<review-receipt.json>" && exit 1)
+	python3 tools/promotion_gate.py --verdict $(VERDICT)
+
+rollout: promotion-gate ## promote local → scale-up cluster (hyperswarm), gated on an APPROVE review verdict
+	@echo "[continuum] rollout — promote via caps.infra.cluster-scaleup.hyperswarm (promotion gate passed)"

artifacts/gate-decisions/.gitkeep

@@ -0,0 +1 @@
+# Per-action promotion-gate evidence bundle (tools/promotion_gate.py writes decisions here).

capd/continuum.local-paas.capd.json

@@ -21,6 +21,7 @@
     "control_plane_primitives": "caps.infra.paas.porter-local@0.1.0",
     "runtime_release": "SocioProphet/lattice-forge",
     "workspace_controller": "SocioProphet/sociosphere",
+    "review_gate": "SocioProphet/prophet-platform:tools/review_gate.py",
     "scales_up_to": "caps.infra.cluster-scaleup.hyperswarm@0.1.0"
   },
   "policy": {

docs/LIFECYCLE.md

@@ -30,6 +30,11 @@ Targets: `SocioProphet/lattice-forge`, GitOps checks.
 
 ## 4. Rollout
 Promote the proven workload up and out.
+- Promotion gate: promotion requires an `APPROVE` review verdict from the review gate consumed
+  from `prophet-platform` (this repo's CapD `links.integration_target`; pin-not-vendor).
+  `tools/promotion_gate.py` verifies the verdict's seal and, fail-closed, blocks on anything but
+  APPROVE. Every decision — allow or block — is emitted to the per-action evidence bundle under
+  `artifacts/gate-decisions/` (policy `evidence_emitting`).
 - Promote local → composable cluster via the pinned scale-up wrapper
   (`caps.infra.cluster-scaleup.hyperswarm`).
 - Signed images, drain/rollback gates, promotion evidence.

tools/promotion_gate.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Rollout promotion gate — require an APPROVE review verdict before continuum promotes.
+
+Continuum owns the rollout promotion gate and its per-action evidence (LIFECYCLE.md §4). The
+review VERDICT it gates on is produced by the review capability CONSUMED from prophet-platform
+(`tools/review_gate.py`, named in this repo's CapD `links.integration_target`) — this file
+reviews nothing. It enforces the consumed verdict and seals the decision into the evidence
+bundle, honouring the CapD `policy.evidence_emitting`.
+
+Fail-closed: promotion is allowed only when the verdict is APPROVE AND its seal recomputes.
+A tampered verdict, a non-APPROVE verdict, or a missing seal all block promotion — and the
+block is itself written to evidence, because a refused promotion is a decision worth keeping.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+_ROOT = Path(__file__).resolve().parent.parent
+APPROVE = "APPROVE"
+
+
+def _recompute_seal(verdict: dict) -> str:
+    """Recompute the reviewer's seal over the verdict body. This is an integrity check on
+    consumed evidence, not a reimplementation of the review — it only proves the verdict
+    bytes were not altered between the reviewer and this gate. The seal is a sha256 over the
+    canonical JSON of every field except the seal itself; that contract is shared with the
+    producer (prophet-platform review_gate._seal_review)."""
+    body = {k: v for k, v in verdict.items() if k != "review_digest"}
+    return "sha256:" + hashlib.sha256(
+        json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()
+
+
+def gate(verdict: dict, evidence_dir: Path, write_evidence: bool = True) -> tuple[bool, dict]:
+    claimed = verdict.get("review_digest")
+    seal_ok = bool(claimed) and claimed == _recompute_seal(verdict)
+    approved = verdict.get("verdict") == APPROVE
+    promote = seal_ok and approved
+
+    decision = {
+        "gate": "sourceos-continuum.promotion_gate.v1",
+        "reviewed": verdict.get("reviewed"),
+        "review_tool": verdict.get("tool"),
+        "review_verdict": verdict.get("verdict"),
+        "seal_ok": seal_ok,
+        "promotion": "allow" if promote else "block",
+        "decided_at": datetime.now(timezone.utc).isoformat(),
+    }
+    if not seal_ok:
+        decision["reason"] = "review verdict seal did not recompute — not trustworthy evidence"
+    elif not approved:
+        decision["reason"] = f"review verdict is {verdict.get('verdict')!r}; promotion requires {APPROVE}"
+
+    if write_evidence:
+        evidence_dir.mkdir(parents=True, exist_ok=True)
+        reviewed = verdict.get("reviewed") or {}
+        key = str(reviewed.get("idempotency_key", "unknown")).replace("/", "_").replace("@", "-at-")
+        stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+        path = evidence_dir / f"{key}.{stamp}.json"
+        path.write_text(json.dumps(decision, indent=2, sort_keys=True) + "\n")
+        decision["evidence"] = str(path.relative_to(_ROOT)) if path.is_relative_to(_ROOT) else str(path)
+
+    return promote, decision
+
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser(description="Rollout promotion gate over a consumed review verdict.")
+    ap.add_argument("--verdict", type=Path, required=True,
+                    help="the sealed review receipt produced by the consumed reviewer")
+    ap.add_argument("--evidence-dir", type=Path, default=_ROOT / "artifacts" / "gate-decisions")
+    ap.add_argument("--no-write", action="store_true", help="do not emit an evidence artifact (dry-run)")
+    args = ap.parse_args(argv)
+
+    try:
+        verdict = json.loads(args.verdict.read_text())
+    except (OSError, json.JSONDecodeError) as exc:
+        print(f"BLOCK: cannot read a review verdict from {args.verdict}: {exc}", file=sys.stderr)
+        return 1
+
+    promote, decision = gate(verdict, args.evidence_dir, write_evidence=not args.no_write)
+    print(json.dumps(decision, indent=2, sort_keys=True))
+    if not promote:
+        print(f"BLOCK: {decision.get('reason', 'promotion refused')}", file=sys.stderr)
+    return 0 if promote else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/test_promotion_gate.py

@@ -0,0 +1,84 @@
+"""Coverage for tools/promotion_gate.py — continuum's rollout promotion gate.
+
+The gate consumes a sealed review verdict (produced by the prophet-platform reviewer) and
+decides promotion. Fail-closed: only APPROVE with an intact seal promotes; REJECT,
+NEEDS_HUMAN, and a tampered verdict all block — and every decision, allow or block, is
+written to the per-action evidence bundle.
+"""
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _load():
+    spec = importlib.util.spec_from_file_location("promotion_gate", ROOT / "tools" / "promotion_gate.py")
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules["promotion_gate"] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
+pg = _load()
+
+
+def _sealed(verdict_value: str, key: str = "engine-at-0.4.45") -> dict:
+    v = {
+        "tool": "prophet-platform.review_gate.v1",
+        "reviewed": {"idempotency_key": key, "to_version": "0.4.45"},
+        "verdict": verdict_value,
+        "checks": [],
+    }
+    v["review_digest"] = pg._recompute_seal(v)
+    return v
+
+
+def test_approve_promotes_and_writes_evidence(tmp_path):
+    promote, dec = pg.gate(_sealed("APPROVE"), tmp_path)
+    assert promote is True and dec["promotion"] == "allow"
+    written = list(tmp_path.glob("*.json"))
+    assert len(written) == 1
+    assert json.loads(written[0].read_text())["promotion"] == "allow"
+
+
+def test_reject_blocks(tmp_path):
+    promote, dec = pg.gate(_sealed("REJECT"), tmp_path)
+    assert promote is False and dec["promotion"] == "block"
+    assert "requires APPROVE" in dec["reason"]
+
+
+def test_needs_human_blocks(tmp_path):
+    promote, _ = pg.gate(_sealed("NEEDS_HUMAN"), tmp_path)
+    assert promote is False
+
+
+def test_tampered_verdict_blocks_even_when_it_says_approve(tmp_path):
+    v = _sealed("REJECT")          # seal is computed over REJECT
+    v["verdict"] = "APPROVE"       # flip to APPROVE without re-sealing
+    promote, dec = pg.gate(v, tmp_path)
+    assert promote is False and dec["seal_ok"] is False
+    assert "seal did not recompute" in dec["reason"]
+
+
+def test_missing_seal_blocks(tmp_path):
+    promote, dec = pg.gate({"tool": "x", "reviewed": {}, "verdict": "APPROVE"}, tmp_path)
+    assert promote is False and dec["seal_ok"] is False
+
+
+def test_block_is_also_evidenced(tmp_path):
+    pg.gate(_sealed("REJECT"), tmp_path)
+    written = list(tmp_path.glob("*.json"))
+    assert len(written) == 1 and json.loads(written[0].read_text())["promotion"] == "block"
+
+
+def test_main_exit_codes(tmp_path):
+    good = tmp_path / "good.json"
+    good.write_text(json.dumps(_sealed("APPROVE")))
+    bad = tmp_path / "bad.json"
+    bad.write_text(json.dumps(_sealed("REJECT")))
+    assert pg.main(["--verdict", str(good), "--evidence-dir", str(tmp_path / "ev")]) == 0
+    assert pg.main(["--verdict", str(bad), "--evidence-dir", str(tmp_path / "ev")]) == 1

tools/validate.py

@@ -18,6 +18,7 @@
     "docs/LIFECYCLE.md",
     "docs/CONTINUUM_SCOPE.md",
     "capd/continuum.local-paas.capd.json",
+    "tools/promotion_gate.py",
 ]
 CAPD_KEYS = ("capability_id", "kind", "status", "links", "composes_with", "policy")