sovereign_freeze_ops_api.py

# sovereign_freeze_ops_api.py
from __future__ import annotations

import json
from typing import Any, Dict, List, Optional

from sovereign_freeze_bridge import SovereignFreezeBridge
from sovereign_freeze_review import SovereignFreezeReview


class SovereignFreezeOpsAPI:
    """
    Operational API for bank-grade freeze management.

    Capabilities:
    - list freeze cases
    - get freeze case
    - review freeze case
    - release freeze case
    - escalate freeze case to block
    - check effective execution state for a payload
    """

    def __init__(self) -> None:
        self.bridge = SovereignFreezeBridge()
        self.review = SovereignFreezeReview()

    # ---------------------------------------------------------
    # Query operations
    # ---------------------------------------------------------
    def list_cases(self) -> Dict[str, Any]:
        cases: List[Dict[str, Any]] = []

        # Source of truth here is bridge storage
        raw_cases = self.bridge._load_cases()
        for freeze_case_id, obj in raw_cases.items():
            if not isinstance(obj, dict):
                continue
            cases.append(
                {
                    "freeze_case_id": freeze_case_id,
                    "status": obj.get("status"),
                    "created_at": obj.get("created_at"),
                    "updated_at": obj.get("updated_at"),
                    "engine": obj.get("engine"),
                    "parent_model": obj.get("parent_model"),
                    "model_version": obj.get("model_version"),
                    "deployment_env": obj.get("deployment_env"),
                    "risk_level": obj.get("risk_level"),
                    "reason": obj.get("reason"),
                    "review_required": obj.get("review_required", True),
                }
            )

        cases = sorted(cases, key=lambda x: str(x.get("created_at") or ""), reverse=True)

        return {
            "ok": True,
            "count": len(cases),
            "cases": cases,
        }

    def get_case(self, freeze_case_id: str) -> Dict[str, Any]:
        obj = self.bridge.get_case(freeze_case_id)
        if not obj:
            return {
                "ok": False,
                "error": "freeze_case_not_found",
                "freeze_case_id": freeze_case_id,
            }

        return {
            "ok": True,
            "freeze_case_id": freeze_case_id,
            "case": obj,
        }

    def review_case(self, freeze_case_id: str) -> Dict[str, Any]:
        """
        Reads bridge case and returns a bank-review friendly summary.
        """
        obj = self.bridge.get_case(freeze_case_id)
        if not obj:
            return {
                "ok": False,
                "error": "freeze_case_not_found",
                "freeze_case_id": freeze_case_id,
            }

        return {
            "ok": True,
            "freeze_case_id": freeze_case_id,
            "review": {
                "freeze_case_id": freeze_case_id,
                "status": obj.get("status"),
                "reason": obj.get("reason"),
                "engine": obj.get("engine"),
                "parent_model": obj.get("parent_model"),
                "model_version": obj.get("model_version"),
                "deployment_env": obj.get("deployment_env"),
                "data_tags": obj.get("data_tags", []),
                "risk_level": obj.get("risk_level"),
                "decision_id": obj.get("decision_id"),
                "authority_decision": obj.get("authority_decision"),
                "source": obj.get("source"),
                "review_required": obj.get("review_required", True),
                "created_at": obj.get("created_at"),
                "updated_at": obj.get("updated_at"),
                "released_at": obj.get("released_at"),
                "released_by": obj.get("released_by"),
                "release_note": obj.get("release_note"),
                "escalated_at": obj.get("escalated_at"),
                "escalated_by": obj.get("escalated_by"),
                "escalation_note": obj.get("escalation_note"),
                "audit_trail": obj.get("audit_trail", []),
            },
        }

    # ---------------------------------------------------------
    # Mutation operations
    # ---------------------------------------------------------
    def release_case(self, freeze_case_id: str, reviewer: str, note: str) -> Dict[str, Any]:
        bridge_out = self.bridge.release_case(
            freeze_case_id=freeze_case_id,
            released_by=reviewer,
            note=note,
        )

        # Also try to reflect in review storage if freeze_id names overlap.
        # Best-effort only; bridge is the source of truth from now on.
        try:
            self.review.release(
                freeze_id=freeze_case_id,
                reviewer=reviewer,
                reason=note,
            )
        except Exception:
            pass

        return {
            "ok": bool(bridge_out.get("ok")),
            "operation": "release_case",
            "freeze_case_id": freeze_case_id,
            "result": bridge_out,
        }

    def escalate_case(self, freeze_case_id: str, reviewer: str, note: str) -> Dict[str, Any]:
        bridge_out = self.bridge.escalate_case(
            freeze_case_id=freeze_case_id,
            escalated_by=reviewer,
            note=note,
        )

        # Best-effort sync
        try:
            self.review.block(
                freeze_id=freeze_case_id,
                reviewer=reviewer,
                reason=note,
            )
        except Exception:
            pass

        return {
            "ok": bool(bridge_out.get("ok")),
            "operation": "escalate_case",
            "freeze_case_id": freeze_case_id,
            "result": bridge_out,
        }

    # ---------------------------------------------------------
    # Effective execution state checks
    # ---------------------------------------------------------
    def effective_state_for_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        state = self.bridge.effective_state(payload)
        return {
            "ok": True,
            "payload_check": True,
            "state": state,
        }

    def can_execute(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        allowed, state = self.bridge.can_execute(payload)
        return {
            "ok": True,
            "allowed": allowed,
            "state": state,
        }


OPS = SovereignFreezeOpsAPI()


def list_freeze_cases() -> Dict[str, Any]:
    return OPS.list_cases()


def get_freeze_case(freeze_case_id: str) -> Dict[str, Any]:
    return OPS.get_case(freeze_case_id)


def review_freeze_case(freeze_case_id: str) -> Dict[str, Any]:
    return OPS.review_case(freeze_case_id)


def release_freeze_case(freeze_case_id: str, reviewer: str, note: str) -> Dict[str, Any]:
    return OPS.release_case(freeze_case_id, reviewer, note)


def escalate_freeze_case(freeze_case_id: str, reviewer: str, note: str) -> Dict[str, Any]:
    return OPS.escalate_case(freeze_case_id, reviewer, note)


def effective_state_for_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
    return OPS.effective_state_for_payload(payload)


def can_execute_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
    return OPS.can_execute(payload)


if __name__ == "__main__":
    print(json.dumps(list_freeze_cases(), indent=2, ensure_ascii=False))