Create sovereign_chain_runtime_bridge.py

sovereign_chain_runtime_bridge.py

@@ -0,0 +1,173 @@
+
+# sovereign_chain_runtime_bridge.py
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+from sovereign_audit_chain_integration import (
+    attach_chain_to_result,
+    chain_event_from_freeze_case,
+)
+
+try:
+    from sovereign_freeze_review import SovereignFreezeReview
+except Exception:
+    SovereignFreezeReview = None  # type: ignore
+
+
+class SovereignChainRuntimeBridge:
+    """
+    Bridges the audit-chain layer into:
+      1) runtime decision results
+      2) freeze lifecycle events
+
+    Goal:
+      Every bank-grade decision path gets a chained audit artifact.
+    """
+
+    def __init__(self) -> None:
+        self.freeze_review = SovereignFreezeReview() if SovereignFreezeReview else None
+
+    # ------------------------------------------------------------------
+    # Runtime result chaining
+    # ------------------------------------------------------------------
+    def attach_to_runtime_result(
+        self,
+        result: Dict[str, Any],
+        *,
+        event_type: str = "sentinel_runtime_decision",
+        engine_name: Optional[str] = None,
+        parent_model: str = "",
+        model_version: str = "",
+        data_tags: Any = None,
+        risk_level: str = "",
+        notes: str = "",
+        policy_bundle: Optional[Dict[str, Any]] = None,
+        freeze_case: Optional[Dict[str, Any]] = None,
+        extra: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        return attach_chain_to_result(
+            result=result,
+            event_type=event_type,
+            engine_name=engine_name,
+            parent_model=parent_model,
+            model_version=model_version,
+            data_tags=data_tags,
+            risk_level=risk_level,
+            notes=notes,
+            policy_bundle=policy_bundle,
+            freeze_case=freeze_case,
+            extra=extra,
+        )
+
+    # ------------------------------------------------------------------
+    # Freeze event chaining helpers
+    # ------------------------------------------------------------------
+    def chain_freeze_created(
+        self,
+        freeze_case: Dict[str, Any],
+        *,
+        extra: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        return chain_event_from_freeze_case(
+            freeze_case,
+            event_type="freeze_created",
+            decision="FREEZE",
+            extra=extra or {},
+        )
+
+    def chain_freeze_released(
+        self,
+        freeze_case: Dict[str, Any],
+        *,
+        extra: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        return chain_event_from_freeze_case(
+            freeze_case,
+            event_type="freeze_released",
+            decision="ALLOW",
+            extra=extra or {},
+        )
+
+    def chain_freeze_escalated(
+        self,
+        freeze_case: Dict[str, Any],
+        *,
+        extra: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        return chain_event_from_freeze_case(
+            freeze_case,
+            event_type="freeze_escalated",
+            decision="BLOCK",
+            extra=extra or {},
+        )
+
+    # ------------------------------------------------------------------
+    # Freeze lifecycle wrappers
+    # ------------------------------------------------------------------
+    def release_case(
+        self,
+        case_id: str,
+        *,
+        reviewer: str,
+        release_notes: str = "",
+        extra: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        if not self.freeze_review:
+            raise RuntimeError("SovereignFreezeReview is not available")
+
+        result = self.freeze_review.release_case(
+            case_id=case_id,
+            reviewer=reviewer,
+            release_notes=release_notes,
+        )
+
+        freeze_case = result.get("case") or {}
+        chain = self.chain_freeze_released(
+            freeze_case,
+            extra={
+                "reviewer": reviewer,
+                "release_notes": release_notes,
+                **(extra or {}),
+            },
+        )
+        result["audit_chain"] = chain.get("chain")
+        return result
+
+    def escalate_case(
+        self,
+        case_id: str,
+        *,
+        reviewer: str,
+        escalation_reason: str = "",
+        extra: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        if not self.freeze_review:
+            raise RuntimeError("SovereignFreezeReview is not available")
+
+        result = self.freeze_review.escalate_case(
+            case_id=case_id,
+            reviewer=reviewer,
+            escalation_reason=escalation_reason,
+        )
+
+        freeze_case = result.get("case") or {}
+        chain = self.chain_freeze_escalated(
+            freeze_case,
+            extra={
+                "reviewer": reviewer,
+                "escalation_reason": escalation_reason,
+                **(extra or {}),
+            },
+        )
+        result["audit_chain"] = chain.get("chain")
+        return result
+
+    def get_case(self, case_id: str) -> Dict[str, Any]:
+        if not self.freeze_review:
+            raise RuntimeError("SovereignFreezeReview is not available")
+        return self.freeze_review.get_case(case_id)
+
+
+BRIDGE = SovereignChainRuntimeBridge()
+