sovereign_app_freeze_integration.py

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