sensorflow/studio2/release_gate.py

"""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