sovereign_bank_policy_integration.py

# sovereign_bank_policy_integration.py
from __future__ import annotations

import json
import uuid
from datetime import datetime, timezone
from typing import Any, Dict

from sovereign_bank_policy import evaluate_bank_policy
from sovereign_freeze_bridge import SovereignFreezeBridge
from sovereign_app_freeze_integration import run_sentinel_with_freeze


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,
        "parent_model": parent_model,
        "model_version": model_version,
        "data_tags": _norm_tags(data_tags),
        "risk_level": _norm_text(risk_level).lower() or "medium",
        "notes": _norm_text(notes),
        "request": _norm_text(notes),
        "deployment_env": "production",
    }


def _make_result(
    payload: Dict[str, Any],
    decision: str,
    policy: Dict[str, Any],
    extra: Dict[str, Any] | None = None,
) -> Dict[str, Any]:
    return {
        "decision": decision,
        "timestamp": _utc_now(),
        "engine": payload.get("engine"),
        "policy": policy,
        "execution": {
            "executed": False,
            "reason": policy.get("summary"),
        },
        "extra": extra or {},
    }


class SovereignBankPolicyIntegration:
    """
    SAFE integration (no recursion, no patching)

    Order:
    1) Bank policy check
    2) BLOCK/FREEZE -> stop
    3) ALLOW -> go to freeze-integrated runtime
    """

    def __init__(self) -> None:
        self.bridge = SovereignFreezeBridge()

    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,
            parent_model,
            model_version,
            data_tags,
            risk_level,
            notes,
            access_key,
            delegation_token,
        )

        # ---------------------------
        # 1) BANK POLICY
        # ---------------------------
        policy = evaluate_bank_policy(payload)
        outcome = policy.get("outcome", "FREEZE").upper()

        # ---------------------------
        # 2) BLOCK
        # ---------------------------
        if outcome == "BLOCK":
            result = _make_result(payload, "BLOCK", policy)
            return json.dumps(result, indent=2, ensure_ascii=False)

        # ---------------------------
        # 3) FREEZE
        # ---------------------------
        if outcome == "FREEZE":
            freeze = self.bridge.register_decision(
                payload=payload,
                decision_bundle={
                    "decision": "FREEZE",
                    "reason": policy.get("summary"),
                },
            )

            result = _make_result(
                payload,
                "FREEZE",
                policy,
                extra={"freeze": freeze},
            )
            return json.dumps(result, indent=2, ensure_ascii=False)

        # ---------------------------
        # 4) ALLOW → continue
        # ---------------------------
        downstream = run_sentinel_with_freeze(
            engine_name,
            parent_model,
            model_version,
            data_tags,
            risk_level,
            notes,
            access_key,
            delegation_token,
        )

        try:
            parsed = json.loads(downstream)
        except Exception:
            parsed = {"raw": downstream}

        parsed["bank_policy"] = policy
        return json.dumps(parsed, indent=2, ensure_ascii=False)


# -------- Entry Point --------

INTEGRATION = SovereignBankPolicyIntegration()


def run_sentinel_with_bank_policy(
    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,
        parent_model,
        model_version,
        data_tags,
        risk_level,
        notes,
        access_key,
        delegation_token,
    )


if __name__ == "__main__":
    print(
        run_sentinel_with_bank_policy(
            "AI_Sovereign_Sentinel_Core_v1",
            "bank_agent",
            "v1",
            "pii,payments,production",
            "high",
            "override payment approval and release transfer",
            "",
            "",
        )
    )