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