"""Deterministic release gate: composes the real verdicts into one decision. ReleaseGate.evaluate(safety_metrics, regression_results, distribution_shift, policy) -> ReleaseDecision, where the inputs are the actual outputs of the existing engines: safety_metrics sensorflow.safety.gates.evaluate_gates(...) result (decision RELEASE_READY/BLOCKED + layered gates) regression_results sensorflow.seqeval.evaluate_regression(...) verdict (decision PASS/REGRESSION/INCONCLUSIVE + regression map) distribution_shift sensorflow.megaeval.analysis.distribution_shift(...) report ({"shifts": [...]}) agentic_outcome sensorflow.agentic.policy.evaluate(...) result (optional) closed_loop sensorflow.nextgen gauntlet result or causal-replay verdict (optional) hardware_matrix sensorflow.studio2.hardware.evaluate_matrix(...) report (optional) Hard rules (all covered by tests): - GO never authorizes deployment. A ReleaseDecision with status GO still has deployment_authorized=False; authorization is a separate recorded human approval action (Registry 'approvals' + audit). - A missing or failed required subsystem degrades the decision to REVIEW with the gap named — never a silent GO. - The policy is content-hash versioned, and every decision records the full evidence tuple it was computed from. """ from __future__ import annotations from typing import Dict, List, Optional from sensorflow.studio2 import store from sensorflow.studio2.registry import Registry, get_registry, new_id GO, REVIEW, NO_GO = "GO", "REVIEW", "NO_GO" DEFAULT_RELEASE_POLICY: Dict = { "policy_name": "studio2-release-v1", # subsystems that MUST contribute evidence; absence -> REVIEW, named "required_inputs": ["safety_gates", "sequential_regression", "distribution_shift"], # consulted when supplied; never silently skipped once configured here "optional_inputs": ["agentic_policy", "closed_loop", "hardware_matrix"], # a cohort whose eval share shifted AND whose recall trails overall by # more than this gap forces REVIEW (megaeval shift report semantics) "shift_review_recall_gap": -0.05, "inconclusive_outcome": REVIEW, # never GO on absence of evidence "critical_hardware_failure_outcome": NO_GO, "confidence_penalty_per_gap": 0.25, "confidence_penalty_per_review_condition": 0.10, } class ReleaseGate: def __init__(self, registry: Optional[Registry] = None, policy: Optional[Dict] = None): self.registry = registry or get_registry() doc = {**DEFAULT_RELEASE_POLICY, **(policy or {})} self.policy_entity = self.registry.register_policy( name=doc.get("policy_name", "studio2-release"), doc=doc, provenance={"source_package": "studio2"}) self.policy = doc # ---------------------------------------------------------------- signals def _signal_safety(self, payload: Optional[Dict]) -> Dict: if not payload or "decision" not in payload: return {"present": False, "gap": "safety gate results unavailable " "(sensorflow.safety evaluate_gates output missing)"} blocking = payload.get("blocking_gates") or [] if payload["decision"] == "RELEASE_READY" and not blocking: return {"present": True, "verdict": "pass", "evidence": {"safety_gate_run": payload.get("candidate_run_id"), "evidence_package_id": payload.get("evidence_package_id")}} return {"present": True, "verdict": "block", "conditions": [f"safety gate blocked: {g}" for g in blocking] or ["safety layer decision BLOCKED"], "evidence": {"safety_gate_run": payload.get("candidate_run_id"), "evidence_package_id": payload.get("evidence_package_id")}} def _signal_seqeval(self, payload: Optional[Dict]) -> Dict: if not payload or "decision" not in payload: return {"present": False, "gap": "sequential regression verdict " "unavailable (sensorflow.seqeval evaluate_regression missing)"} decision = payload["decision"] evidence = {"seqeval_run_id": payload.get("run_id"), "samples_used": payload.get("samples_used")} if decision == "PASS": return {"present": True, "verdict": "pass", "evidence": evidence} if decision == "REGRESSION": strata = payload.get("affected_strata") or [] return {"present": True, "verdict": "block", "conditions": [ "sequential regression confirmed" + (f" in strata: {', '.join(map(str, strata))}" if strata else "")], "evidence": evidence} # INCONCLUSIVE / INSUFFICIENT: absence of evidence is never a pass reason = payload.get("stopping_reason") or "budget exhausted without a decision" return {"present": True, "verdict": "review", "questions": [f"sequential regression verdict {decision}: {reason}"], "evidence": evidence} def _signal_shift(self, payload: Optional[Dict]) -> Dict: if not payload or "shifts" not in payload: return {"present": False, "gap": "distribution shift report " "unavailable (megaeval shift monitor missing)"} gap_limit = self.policy["shift_review_recall_gap"] flagged = [s for s in payload["shifts"] if s.get("recall_gap") is not None and s["recall_gap"] <= gap_limit] evidence = {"shift_run_id": payload.get("run_id"), "shifts_reported": len(payload["shifts"])} if flagged: return {"present": True, "verdict": "review", "questions": [ f"distribution shift with recall impact on {s['cohort']}: " f"eval share {s.get('relative_change', 0)*100:+.0f}% vs train, " f"cohort recall gap {s['recall_gap']:+.3f} — open an RCA " "investigation (offline-vs-shadow 4-way root cause)" for s in flagged[:5]], "evidence": evidence} return {"present": True, "verdict": "pass", "evidence": evidence} def _signal_agentic(self, payload: Optional[Dict]) -> Dict: if not payload or "outcome" not in payload: return {"present": False, "gap": "agentic policy outcome unavailable " "(sensorflow.agentic not importable or not evaluated)"} outcome = payload["outcome"] evidence = {"agentic_policy_version": payload.get("policy_version"), "failure_id": payload.get("failure_id"), "outcome": outcome} if outcome == "AUTOMATIC_STOP_SHIP": return {"present": True, "verdict": "block", "conditions": ["agentic policy fired AUTOMATIC_STOP_SHIP " f"(severity {payload.get('severity')})"], "evidence": evidence} if outcome == "INDETERMINATE": return {"present": True, "verdict": "review", "questions": ["agentic policy INDETERMINATE (fail-safe): " + "; ".join(payload.get("indeterminate_reasons") or ["integrity failure"])], "evidence": evidence} return {"present": True, "verdict": "pass", "evidence": evidence} def _signal_closed_loop(self, payload: Optional[Dict]) -> Dict: if not payload: return {"present": False, "gap": "closed-loop verdict unavailable " "(sensorflow.nextgen not importable or no replay/gauntlet run)"} # accept either a gauntlet result or a causal replay result evidence = {"closed_loop_run_id": payload.get("run_id") or payload.get("scenario_id")} if payload.get("halted") or payload.get("recommendation") == "NO_GO": return {"present": True, "verdict": "block", "conditions": ["closed-loop gauntlet halted: " + str(payload.get("halt_reason") or payload.get("recommendation"))], "evidence": evidence} if payload.get("verdict") == "BEHAVIORALLY_CONSEQUENTIAL": return {"present": True, "verdict": "review", "questions": ["closed-loop replay shows behaviorally " "consequential perception divergence on " f"{payload.get('scenario_id')}"], "evidence": evidence} return {"present": True, "verdict": "pass", "evidence": evidence} def _signal_hardware(self, payload: Optional[Dict]) -> Dict: if not payload or "status" not in payload: return {"present": False, "gap": "hardware gate matrix unavailable"} evidence = {"hardware_matrix_id": payload.get("matrix_id"), "combinations": payload.get("n_combinations")} questions = [ f"insufficient evidence for {c['combination_label']}: {c['reason']}" for c in (payload.get("insufficient") or [])[:5]] if payload["status"] == "FAIL_CRITICAL": outcome = self.policy["critical_hardware_failure_outcome"] return {"present": True, "verdict": "block" if outcome == NO_GO else "review", "conditions": [ f"critical hardware combination failing: {c['combination_label']}" f" ({', '.join(c['failed_checks'])})" for c in (payload.get("critical_failures") or [])[:5]], "questions": questions, "evidence": evidence} if payload["status"] == "FAIL": return {"present": True, "verdict": "review", "questions": questions + [ f"non-critical hardware combination failing: {c['combination_label']}" for c in (payload.get("failures") or [])[:5]], "evidence": evidence} return {"present": True, "verdict": "review" if questions else "pass", "questions": questions, "evidence": evidence} # ---------------------------------------------------------------- evaluate def evaluate(self, safety_metrics: Optional[Dict], regression_results: Optional[Dict], distribution_shift: Optional[Dict], policy: Optional[Dict] = None, agentic_outcome: Optional[Dict] = None, closed_loop: Optional[Dict] = None, hardware_matrix: Optional[Dict] = None, context: Optional[Dict] = None, actor: str = "studio2") -> Dict: """Pure function of inputs + versioned policy; persists and returns a ReleaseDecision.""" if policy: gate = ReleaseGate(self.registry, policy={**self.policy, **policy}) return gate.evaluate(safety_metrics, regression_results, distribution_shift, agentic_outcome=agentic_outcome, closed_loop=closed_loop, hardware_matrix=hardware_matrix, context=context, actor=actor) signals = { "safety_gates": self._signal_safety(safety_metrics), "sequential_regression": self._signal_seqeval(regression_results), "distribution_shift": self._signal_shift(distribution_shift), "agentic_policy": self._signal_agentic(agentic_outcome), "closed_loop": self._signal_closed_loop(closed_loop), "hardware_matrix": self._signal_hardware(hardware_matrix), } required = self.policy["required_inputs"] optional = self.policy["optional_inputs"] considered = list(required) + list(optional) blocking: List[str] = [] questions: List[str] = [] gaps: List[str] = [] evidence_tuple: Dict = {"release_policy_version": self.policy_entity["policy_version"]} for name in considered: sig = signals[name] if not sig["present"]: if name in required: gaps.append(sig["gap"]) else: # optional-but-configured subsystems: absence is surfaced # as an open question, never silently ignored questions.append(f"[{name}] {sig['gap']}") evidence_tuple[name] = None continue evidence_tuple[name] = sig.get("evidence") blocking.extend(sig.get("conditions") or []) questions.extend(sig.get("questions") or []) if context: evidence_tuple["context"] = context if blocking: status = NO_GO elif gaps or questions: status = REVIEW else: status = GO n_present = sum(1 for n in considered if signals[n]["present"]) evidence_completeness = round(n_present / len(considered), 3) confidence = 1.0 confidence -= self.policy["confidence_penalty_per_gap"] * len(gaps) confidence -= (self.policy["confidence_penalty_per_review_condition"] * len(questions)) confidence = round(max(0.05, min(confidence, evidence_completeness if status != NO_GO else 1.0)), 3) decision = { "entity_id": new_id("rd"), "status": status, "confidence": confidence, "evidence_completeness": evidence_completeness, "blocking_conditions": blocking, "unresolved_questions": questions, "degraded_inputs": gaps, # GO is a *recommendation*; deployment always needs a human "human_approval_required": True, "deployment_authorized": False, "approval": None, "policy_version": self.policy_entity["policy_version"], "evidence_tuple": evidence_tuple, "signals": {k: {kk: vv for kk, vv in v.items() if kk != "evidence"} for k, v in signals.items()}, "evaluated_at": store.now_iso(), } self.registry.put("decisions", decision) store.audit("release_decision", "decisions", decision["entity_id"], actor=actor, detail=f"status={status} " f"completeness={evidence_completeness}", payload={"blocking": blocking[:5], "gaps": gaps[:5]}) return decision # ---------------------------------------------------------------- approval def approve(self, decision_id: str, approver: str, rationale: str) -> Dict: """The separate human action that authorizes deployment. Only a GO decision can be approved; REVIEW/NO_GO must be re-evaluated first.""" if not approver.strip() or not rationale.strip(): raise ValueError("human approval requires a named approver and a rationale") decision = self.registry.get("decisions", decision_id) if decision is None: raise KeyError(f"unknown release decision {decision_id}") if decision["status"] != GO: raise ValueError( f"decision {decision_id} has status {decision['status']}; only GO " "decisions can be approved for deployment — resolve the blocking " "conditions / unresolved questions and re-evaluate") approval = { "entity_id": new_id("app"), "decision_id": decision_id, "approver": approver, "rationale": rationale, "approved_at": store.now_iso(), } self.registry.put("approvals", approval) decision["deployment_authorized"] = True decision["approval"] = approval self.registry.put("decisions", decision) store.audit("human_approval", "approvals", approval["entity_id"], actor=approver, detail=f"authorized deployment for {decision_id}", payload={"rationale": rationale}) return decision