sovereign_chain_runtime_bridge.py

# 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()