sovereign_audit_chain_integration.py

# 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 {},
    )