Create sovereign_audit_chain_integration.py

sovereign_audit_chain_integration.py

@@ -0,0 +1,167 @@
+# sovereign_audit_chain_integration.py
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional
+
+from sovereign_audit_chain import append_audit_event
+
+
+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_chain_event(
+    *,
+    event_type: str,
+    decision: str,
+    engine_name: str,
+    parent_model: str = "",
+    model_version: str = "",
+    data_tags: Any = None,
+    risk_level: str = "",
+    notes: str = "",
+    execution: Optional[Dict[str, Any]] = None,
+    authority_bundle: Optional[Dict[str, Any]] = None,
+    policy_bundle: Optional[Dict[str, Any]] = None,
+    freeze_case: Optional[Dict[str, Any]] = None,
+    extra: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    return {
+        "timestamp": _utc_now(),
+        "event_type": _norm_text(event_type),
+        "decision": _norm_text(decision).upper(),
+        "engine_name": _norm_text(engine_name),
+        "parent_model": _norm_text(parent_model),
+        "model_version": _norm_text(model_version),
+        "data_tags": _norm_tags(data_tags),
+        "risk_level": _norm_text(risk_level).lower(),
+        "notes": _norm_text(notes),
+        "execution": execution or {},
+        "authority_bundle": authority_bundle or {},
+        "policy_bundle": policy_bundle or {},
+        "freeze_case": freeze_case or {},
+        "extra": extra or {},
+    }
+
+
+def write_chain_event(
+    *,
+    event_type: str,
+    decision: str,
+    engine_name: str,
+    parent_model: str = "",
+    model_version: str = "",
+    data_tags: Any = None,
+    risk_level: str = "",
+    notes: str = "",
+    execution: Optional[Dict[str, Any]] = None,
+    authority_bundle: Optional[Dict[str, Any]] = None,
+    policy_bundle: Optional[Dict[str, Any]] = None,
+    freeze_case: Optional[Dict[str, Any]] = None,
+    extra: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    event = build_chain_event(
+        event_type=event_type,
+        decision=decision,
+        engine_name=engine_name,
+        parent_model=parent_model,
+        model_version=model_version,
+        data_tags=data_tags,
+        risk_level=risk_level,
+        notes=notes,
+        execution=execution,
+        authority_bundle=authority_bundle,
+        policy_bundle=policy_bundle,
+        freeze_case=freeze_case,
+        extra=extra,
+    )
+    result = append_audit_event(event)
+    return {
+        "ok": True,
+        "event": event,
+        "chain": result,
+    }
+
+
+def attach_chain_to_result(
+    result: Dict[str, Any],
+    *,
+    event_type: str,
+    engine_name: Optional[str] = None,
+    parent_model: str = "",
+    model_version: str = "",
+    data_tags: Any = None,
+    risk_level: str = "",
+    notes: str = "",
+    policy_bundle: Optional[Dict[str, Any]] = None,
+    freeze_case: Optional[Dict[str, Any]] = None,
+    extra: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    result = result or {}
+
+    decision = str(
+        result.get("decision")
+        or result.get("outcome")
+        or result.get("action")
+        or "UNKNOWN"
+    ).upper()
+
+    authority_bundle = result.get("authority_gate") or result.get("authority_bundle") or {}
+    execution = result.get("execution") or {}
+
+    chain_write = write_chain_event(
+        event_type=event_type,
+        decision=decision,
+        engine_name=engine_name or result.get("engine_name") or result.get("engine") or "unknown_engine",
+        parent_model=parent_model or result.get("parent_model") or "",
+        model_version=model_version or result.get("model_version") or "",
+        data_tags=data_tags if data_tags is not None else result.get("data_tags") or [],
+        risk_level=risk_level or result.get("risk_level") or "",
+        notes=notes or result.get("notes") or "",
+        execution=execution,
+        authority_bundle=authority_bundle,
+        policy_bundle=policy_bundle,
+        freeze_case=freeze_case,
+        extra=extra,
+    )
+
+    result["audit_chain"] = chain_write["chain"]
+    return result
+
+
+def chain_event_from_freeze_case(
+    freeze_case: Dict[str, Any],
+    *,
+    event_type: str,
+    decision: str,
+    extra: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    freeze_case = freeze_case or {}
+
+    return write_chain_event(
+        event_type=event_type,
+        decision=decision,
+        engine_name=freeze_case.get("engine_name") or "unknown_engine",
+        parent_model=freeze_case.get("parent_model") or "",
+        model_version=freeze_case.get("model_version") or "",
+        data_tags=freeze_case.get("data_tags") or [],
+        risk_level=freeze_case.get("risk_level") or "",
+        notes=freeze_case.get("notes") or "",
+        freeze_case=freeze_case,
+        extra=extra or {},
+    )
+