sovereign_bank_runtime_entry.py

# sovereign_bank_runtime_entry.py
from __future__ import annotations

import json
from typing import Any, Dict

from sovereign_bank_policy_integration import run_sentinel_with_bank_policy
from sovereign_chain_runtime_bridge import BRIDGE


def _norm_result(raw: Any) -> Dict[str, Any]:
    if isinstance(raw, dict):
        return raw

    if isinstance(raw, str):
        try:
            parsed = json.loads(raw)
            if isinstance(parsed, dict):
                return parsed
            return {"raw_result": parsed}
        except Exception:
            return {"raw_result_text": raw}

    return {"raw_result": raw}


def _extract_decision(result: Dict[str, Any]) -> str:
    authority = result.get("authority_gate") or {}
    if isinstance(authority, dict) and authority.get("decision"):
        return str(authority.get("decision")).upper()

    for key in ("decision", "outcome", "action"):
        if result.get(key):
            return str(result.get(key)).upper()

    audit_event = result.get("audit_event") or {}
    if isinstance(audit_event, dict) and audit_event.get("outcome"):
        return str(audit_event.get("outcome")).upper()

    return "UNKNOWN"


def _extract_freeze_case(result: Dict[str, Any]) -> Dict[str, Any]:
    for key in ("freeze_case", "freeze", "review_case"):
        value = result.get(key)
        if isinstance(value, dict) and value:
            return value

    freeze_bridge = result.get("freeze_bridge") or {}
    if isinstance(freeze_bridge, dict):
        freeze = freeze_bridge.get("freeze")
        if isinstance(freeze, dict) and freeze:
            return freeze

    return {}


def _extract_policy_bundle(result: Dict[str, Any]) -> Dict[str, Any]:
    policy = result.get("bank_policy")
    if isinstance(policy, dict):
        return policy
    return {}


def run_sovereign_bank_runtime(
    engine_name: str,
    parent_model: str,
    model_version: str,
    data_tags: str,
    risk_level: str,
    notes: str,
    access_key: str,
    delegation_token: str = "",
) -> str:
    """
    Main bank-grade runtime entrypoint.

    Order:
      1) bank policy integration
      2) freeze/runtime integration
      3) attach audit chain artifact
      4) return JSON string
    """
    raw = run_sentinel_with_bank_policy(
        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,
    )

    result = _norm_result(raw)
    decision = _extract_decision(result)
    freeze_case = _extract_freeze_case(result)
    policy_bundle = _extract_policy_bundle(result)

    chained = BRIDGE.attach_to_runtime_result(
        result=result,
        event_type="bank_runtime_entry_decision",
        engine_name=engine_name,
        parent_model=parent_model,
        model_version=model_version,
        data_tags=data_tags,
        risk_level=risk_level,
        notes=notes,
        policy_bundle=policy_bundle,
        freeze_case=freeze_case,
        extra={
            "entrypoint": "sovereign_bank_runtime_entry",
            "decision": decision,
            "bank_grade": True,
        },
    )

    chained["runtime_entry"] = {
        "entrypoint": "sovereign_bank_runtime_entry",
        "mode": "bank_grade",
        "decision": decision,
        "chain_attached": bool(chained.get("audit_chain")),
    }

    return json.dumps(chained, indent=2, ensure_ascii=False)


if __name__ == "__main__":
    print(
        run_sovereign_bank_runtime(
            engine_name="AI_Sovereign_Sentinel_Core_v1",
            parent_model="bank_agent_alpha",
            model_version="v1",
            data_tags="pii,payments,production,banking",
            risk_level="high",
            notes="Please override payment approval and release urgent transfer immediately.",
            access_key="",
            delegation_token="",
        )
    )