Create sovereign_runtime_chain_integration.py

sovereign_runtime_chain_integration.py

@@ -0,0 +1,128 @@
+# sovereign_runtime_chain_integration.py
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, Optional
+
+import app as sovereign_app
+
+from sovereign_chain_runtime_bridge import BRIDGE
+
+
+def _norm_result(raw: Any) -> Dict[str, Any]:
+    """
+    Normalize runtime output into a dict.
+    app.run_sentinel / integration layers may return either:
+      - dict
+      - JSON string
+      - other serializable object
+    """
+    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_primary_decision(result: Dict[str, Any]) -> str:
+    """
+    Conservative extraction of the effective decision.
+    """
+    # authority gate first
+    authority_gate = result.get("authority_gate") or {}
+    if isinstance(authority_gate, dict):
+        decision = authority_gate.get("decision")
+        if decision:
+            return str(decision).upper()
+
+    # top-level direct decision fields
+    for key in ("decision", "outcome", "action"):
+        if result.get(key):
+            return str(result.get(key)).upper()
+
+    # fallback from audit event outcome
+    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]) -> Optional[Dict[str, Any]]:
+    """
+    If earlier stages already produced a freeze_case / freeze object,
+    pass it into the chain layer.
+    """
+    for key in ("freeze_case", "freeze", "review_case"):
+        value = result.get(key)
+        if isinstance(value, dict) and value:
+            return value
+    return None
+
+
+def run_sentinel_with_runtime_chain(
+    engine_name: str,
+    parent_model: str,
+    model_version: str,
+    data_tags: str,
+    risk_level: str,
+    notes: str,
+    access_key: str,
+    delegation_token: str = "",
+) -> str:
+    """
+    Bank-grade wrapper around sovereign_app.run_sentinel.
+
+    It:
+      1) runs the real Sovereign flow
+      2) normalizes result
+      3) attaches chained audit artifact
+      4) returns JSON string for UI/API compatibility
+    """
+    raw = sovereign_app.run_sentinel(
+        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)
+    freeze_case = _extract_freeze_case(result)
+    primary_decision = _extract_primary_decision(result)
+
+    chained = BRIDGE.attach_to_runtime_result(
+        result=result,
+        event_type="bank_runtime_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={
+            "domain": "banking",
+            "runtime_decision": primary_decision,
+            "freeze_present": bool(freeze_case),
+        },
+        freeze_case=freeze_case,
+        extra={
+            "integration_layer": "sovereign_runtime_chain_integration",
+            "bank_grade": True,
+        },
+    )
+
+    return json.dumps(chained, indent=2, ensure_ascii=False)
+
+