File size: 7,473 Bytes
b298b38 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 |
# 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))
|