Create sovereign_freeze_review.py

sovereign_freeze_review.py

@@ -0,0 +1,129 @@
+
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+from typing import Dict, Any, Optional
+
+
+def _utc():
+    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+
+
+class SovereignFreezeReview:
+    """
+    Bank-grade freeze lifecycle manager
+
+    Handles:
+    - review
+    - approve (release)
+    - reject (block)
+    """
+
+    def __init__(self, freeze_dir: str = ".sovereign_runtime/freezes"):
+        self.freeze_dir = Path(freeze_dir)
+        self.freeze_dir.mkdir(parents=True, exist_ok=True)
+
+    def _path(self, freeze_id: str) -> Path:
+        return self.freeze_dir / f"{freeze_id}.json"
+
+    def load(self, freeze_id: str) -> Dict[str, Any]:
+        path = self._path(freeze_id)
+        if not path.exists():
+            raise FileNotFoundError(f"Freeze not found: {freeze_id}")
+
+        return json.loads(path.read_text())
+
+    def save(self, freeze_id: str, data: Dict[str, Any]):
+        path = self._path(freeze_id)
+        path.write_text(json.dumps(data, indent=2))
+
+    # ---------------------------------------------------------
+    # REVIEW (read-only)
+    # ---------------------------------------------------------
+    def review(self, freeze_id: str) -> Dict[str, Any]:
+        data = self.load(freeze_id)
+
+        return {
+            "freeze_id": freeze_id,
+            "status": data.get("status"),
+            "reason": data.get("reason"),
+            "risk_score": data.get("decision_payload", {}).get("risk_score"),
+            "policy": data.get("decision_payload", {}).get("policy_id"),
+            "created_at": data.get("created_at"),
+            "request": data.get("request_payload"),
+        }
+
+    # ---------------------------------------------------------
+    # RELEASE (approve execution)
+    # ---------------------------------------------------------
+    def release(
+        self,
+        freeze_id: str,
+        reviewer: str,
+        reason: str
+    ) -> Dict[str, Any]:
+
+        data = self.load(freeze_id)
+
+        if data.get("status") != "FROZEN":
+            raise Exception("Freeze is not active")
+
+        data["status"] = "RELEASED"
+        data["released_at"] = _utc()
+        data["released_by"] = reviewer
+        data["release_reason"] = reason
+
+        self.save(freeze_id, data)
+
+        return {
+            "status": "RELEASED",
+            "freeze_id": freeze_id,
+            "released_by": reviewer
+        }
+
+    # ---------------------------------------------------------
+    # ESCALATE → BLOCK
+    # ---------------------------------------------------------
+    def block(
+        self,
+        freeze_id: str,
+        reviewer: str,
+        reason: str
+    ) -> Dict[str, Any]:
+
+        data = self.load(freeze_id)
+
+        if data.get("status") != "FROZEN":
+            raise Exception("Freeze is not active")
+
+        data["status"] = "BLOCKED"
+        data["blocked_at"] = _utc()
+        data["blocked_by"] = reviewer
+        data["block_reason"] = reason
+
+        self.save(freeze_id, data)
+
+        return {
+            "status": "BLOCKED",
+            "freeze_id": freeze_id,
+            "blocked_by": reviewer
+        }
+
+
+# ---------------------------------------------------------
+# test
+# ---------------------------------------------------------
+if __name__ == "__main__":
+    fr = SovereignFreezeReview()
+
+    # Example usage:
+    # assumes freeze file already exists
+
+    try:
+        fz = fr.review("frz_test")
+        print("REVIEW:", json.dumps(fz, indent=2))
+    except:
+        print("No test freeze found")
+