Create sovereign_app_freeze_integration.py

sovereign_app_freeze_integration.py

@@ -0,0 +1,347 @@
+
+# sovereign_app_freeze_integration.py
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional, Tuple
+
+import app as sovereign_app
+from sovereign_freeze_bridge import SovereignFreezeBridge
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _norm_text(v: Any) -> str:
+    return str(v or "").strip()
+
+
+def _norm_tags(v: Any) -> list[str]:
+    if isinstance(v, list):
+        return [str(x).strip().lower() for x in v if str(x).strip()]
+    if isinstance(v, str):
+        return [x.strip().lower() for x in v.split(",") if x.strip()]
+    return []
+
+
+def _build_payload(
+    engine_name: str,
+    parent_model: str,
+    model_version: str,
+    data_tags: Any,
+    risk_level: str,
+    notes: str,
+    access_key: str,
+    delegation_token: str,
+) -> Dict[str, Any]:
+    return {
+        "engine": engine_name,
+        "engine_name": engine_name,
+        "parent_model": parent_model,
+        "parent_model_id": parent_model,
+        "agent_id": parent_model,
+        "model_version": model_version,
+        "build_id": model_version,
+        "data_tags": _norm_tags(data_tags),
+        "risk_level": _norm_text(risk_level).lower() or "medium",
+        "notes": _norm_text(notes),
+        "access_key_present": bool(_norm_text(access_key)),
+        "delegation_token_present": bool(_norm_text(delegation_token)),
+        "deployment_env": "production",
+        "environment": "production",
+    }
+
+
+def _safe_parse_json(raw: Any) -> Dict[str, Any]:
+    if isinstance(raw, dict):
+        return raw
+    if isinstance(raw, str):
+        try:
+            parsed = json.loads(raw)
+            return parsed if isinstance(parsed, dict) else {"raw": raw}
+        except Exception:
+            return {"raw": raw}
+    return {"raw": raw}
+
+
+def _extract_decision(bundle: Dict[str, Any]) -> str:
+    for key in ("decision", "action", "outcome", "authority_decision"):
+        v = bundle.get(key)
+        if isinstance(v, str) and v.strip():
+            return v.strip().upper()
+
+    auth = bundle.get("authority_gate") or {}
+    if isinstance(auth, dict):
+        v = auth.get("decision")
+        if isinstance(v, str) and v.strip():
+            return v.strip().upper()
+
+    return "UNKNOWN"
+
+
+def _extract_reason(bundle: Dict[str, Any]) -> str:
+    v = bundle.get("reason")
+    if isinstance(v, str) and v.strip():
+        return v.strip()
+
+    reasons = bundle.get("reasons")
+    if isinstance(reasons, list) and reasons:
+        return " | ".join(str(x) for x in reasons[:6])
+
+    auth = bundle.get("authority_gate") or {}
+    if isinstance(auth, dict):
+        v = auth.get("reason")
+        if isinstance(v, str) and v.strip():
+            return v.strip()
+
+    return "no_explicit_reason"
+
+
+def _make_override_result(
+    payload: Dict[str, Any],
+    forced_decision: str,
+    reason: str,
+    freeze_state: Dict[str, Any],
+) -> Dict[str, Any]:
+    decision_id = str(uuid.uuid4())
+    ts = _utc_now()
+
+    audit_event = {
+        "event_id": decision_id,
+        "timestamp": ts,
+        "engine": payload.get("engine"),
+        "event_type": "sentinel_run",
+        "outcome": forced_decision,
+        "parent_model": payload.get("parent_model"),
+        "model_version": payload.get("model_version"),
+        "data_tags": payload.get("data_tags", []),
+        "risk_level": payload.get("risk_level"),
+        "notes": f"[FREEZE_OVERRIDE] {reason}",
+        "version": getattr(sovereign_app, "SOVEREIGN_VERSION", "unknown"),
+        "authority_name": getattr(sovereign_app, "AUTHORITY_NAME", "unknown"),
+        "authority_bundle": {
+            "decision": forced_decision,
+            "reason": reason,
+            "proof_type": "proof_of_non_action",
+            "freeze_state": freeze_state,
+        },
+        "execution": {
+            "executed": False,
+            "detail": {
+                "blocked_by_freeze_bridge": True,
+                "freeze_state": freeze_state,
+            },
+        },
+    }
+
+    return {
+        "audit_event": audit_event,
+        "authority_gate": {
+            "decision": forced_decision,
+            "reason": reason,
+            "proof_type": "proof_of_non_action",
+            "freeze_state": freeze_state,
+        },
+        "execution": {
+            "executed": False,
+            "detail": {
+                "blocked_by_freeze_bridge": True,
+                "freeze_state": freeze_state,
+            },
+        },
+        "freeze_bridge": {
+            "bridge_status": "pre_execution_override",
+            "decision": forced_decision,
+            "freeze": freeze_state,
+        },
+    }
+
+
+def _persist_override_audit(result: Dict[str, Any]) -> None:
+    """
+    Best-effort write into app.py audit log so freeze override is visible centrally.
+    """
+    try:
+        event = result.get("audit_event")
+        if not isinstance(event, dict):
+            return
+
+        fn = getattr(sovereign_app, "log_audit_event", None)
+        if callable(fn):
+            fn(
+                engine_name=event.get("engine", "unknown_engine"),
+                parent_model=event.get("parent_model", "unknown_parent"),
+                model_version=event.get("model_version", "unknown_version"),
+                data_tags=",".join(event.get("data_tags", []) or []),
+                risk_level=event.get("risk_level", "medium"),
+                notes=event.get("notes", ""),
+                event_type=event.get("event_type", "sentinel_run"),
+                outcome=event.get("outcome", "FREEZE"),
+                access_key="",
+                authority_bundle=event.get("authority_bundle", {}),
+                execution=event.get("execution", {}),
+            )
+    except Exception:
+        pass
+
+
+class SovereignAppFreezeIntegration:
+    """
+    Bank-grade freeze integration wrapper around app.run_sentinel.
+
+    Flow:
+    1) build normalized payload
+    2) check existing freeze state before execution
+    3) if active freeze/escalated block exists => do not execute
+    4) otherwise call original app.run_sentinel
+    5) parse decision and register freeze if result == FREEZE
+    6) return unified JSON result
+    """
+
+    def __init__(self) -> None:
+        self.bridge = SovereignFreezeBridge()
+        self.original_run_sentinel = sovereign_app.run_sentinel
+
+    def _precheck(self, payload: Dict[str, Any]) -> Tuple[bool, Dict[str, Any]]:
+        return self.bridge.can_execute(payload)
+
+    def run(
+        self,
+        engine_name: str,
+        parent_model: str,
+        model_version: str,
+        data_tags: Any,
+        risk_level: str,
+        notes: str,
+        access_key: str,
+        delegation_token: str,
+    ) -> str:
+        payload = _build_payload(
+            engine_name=engine_name,
+            parent_model=parent_model,
+            model_version=model_version,
+            data_tags=data_tags,
+            risk_level=risk_level,
+            notes=notes,
+            access_key=access_key,
+            delegation_token=delegation_token,
+        )
+
+        # 1) Pre-execution freeze check
+        can_exec, state = self._precheck(payload)
+        if not can_exec:
+            forced = "BLOCK" if state.get("state") == "escalated_block" else "FREEZE"
+            result = _make_override_result(
+                payload=payload,
+                forced_decision=forced,
+                reason=state.get("reason", "freeze_active"),
+                freeze_state=state,
+            )
+            _persist_override_audit(result)
+            return json.dumps(result, indent=2, ensure_ascii=False)
+
+        # 2) Normal app execution
+        raw = self.original_run_sentinel(
+            engine_name,
+            parent_model,
+            model_version,
+            data_tags if isinstance(data_tags, str) else ",".join(_norm_tags(data_tags)),
+            risk_level,
+            notes,
+            access_key,
+            delegation_token,
+        )
+
+        parsed = _safe_parse_json(raw)
+
+        # 3) Detect main decision
+        decision = _extract_decision(parsed)
+        reason = _extract_reason(parsed)
+
+        # 4) Register or link freeze case if decision == FREEZE
+        bridge_result = self.bridge.register_decision(
+            payload=payload,
+            decision_bundle={
+                "decision_id": (
+                    (parsed.get("audit_event") or {}).get("event_id")
+                    or (parsed.get("authority_gate") or {}).get("decision_id")
+                    or str(uuid.uuid4())
+                ),
+                "decision": decision,
+                "reason": reason,
+                "authority_gate": parsed.get("authority_gate", {}),
+                "execution": parsed.get("execution", {}),
+            },
+        )
+
+        parsed["freeze_bridge"] = bridge_result
+
+        # 5) If bridge says an existing freeze overrides execution, force output consistency
+        override_decision = bridge_result.get("decision")
+        if bridge_result.get("bridge_status") == "existing_freeze_overrides_execution" and override_decision in {"FREEZE", "BLOCK"}:
+            parsed["authority_gate"] = dict(parsed.get("authority_gate") or {})
+            parsed["authority_gate"]["decision"] = override_decision
+            parsed["authority_gate"]["reason"] = bridge_result.get("freeze", {}).get("reason", "existing_freeze")
+            parsed["authority_gate"]["proof_type"] = "proof_of_non_action"
+
+            parsed["execution"] = {
+                "executed": False,
+                "detail": {
+                    "blocked_by_freeze_bridge": True,
+                    "freeze_state": bridge_result.get("freeze", {}),
+                },
+            }
+
+        return json.dumps(parsed, indent=2, ensure_ascii=False)
+
+    def patch_app(self) -> None:
+        """
+        Monkey-patch app.run_sentinel so current Gradio / app flow starts using freeze integration.
+        """
+        sovereign_app.run_sentinel = self.run
+
+
+INTEGRATION = SovereignAppFreezeIntegration()
+
+
+def run_sentinel_with_freeze(
+    engine_name: str,
+    parent_model: str,
+    model_version: str,
+    data_tags: Any,
+    risk_level: str,
+    notes: str,
+    access_key: str,
+    delegation_token: str,
+) -> str:
+    return INTEGRATION.run(
+        engine_name=engine_name,
+        parent_model=parent_model,
+        model_version=model_version,
+        data_tags=data_tags,
+        risk_level=risk_level,
+        notes=notes,
+        access_key=access_key,
+        delegation_token=delegation_token,
+    )
+
+
+def patch_app_run_sentinel() -> None:
+    INTEGRATION.patch_app()
+
+
+if __name__ == "__main__":
+    sample = run_sentinel_with_freeze(
+        engine_name="AI_Sovereign_Sentinel_Core_v1",
+        parent_model="agent_alpha",
+        model_version="v1",
+        data_tags="pii,payments,customer_chat",
+        risk_level="high",
+        notes="test transfer request with elevated ambiguity",
+        access_key="",
+        delegation_token="",
+    )
+    print(sample)