sitecustomize.py

# sitecustomize.py
"""
Auto-loaded by Python at interpreter startup.
We use it to inject the Sovereign Authority Layer WITHOUT touching app.py/core/ultra.
"""

import os
import json
import time

def _safe_print(msg: str) -> None:
    try:
        print(msg)
    except Exception:
        pass


def _patch_run_pilot():
    """
    Monkey-patch pilot_suite.run_pilot so every run is post-processed by:
      - SovereignAuthorityLayer (7 mandatory capabilities)
      - Policy manifest snapshot (sealed)
      - Build fingerprint / SBOM-lite (startup)
      - Evidence package export (zip/folder)
    """
    try:
        import pilot_suite  # the module app.py imports from
    except Exception as e:
        _safe_print(f"[SovereignPatch] pilot_suite import failed: {e}")
        return

    if not hasattr(pilot_suite, "run_pilot"):
        _safe_print("[SovereignPatch] pilot_suite.run_pilot not found. No patch applied.")
        return

    original_run_pilot = pilot_suite.run_pilot

    # Avoid double-patching
    if getattr(original_run_pilot, "_sovereign_patched", False):
        _safe_print("[SovereignPatch] run_pilot already patched.")
        return

    try:
        from sovereign_authority_layer import SovereignAuthorityLayer
        from sovereign_evidence_packager import (
            ensure_build_fingerprint,
            write_policy_manifest,
            export_evidence_package,
        )
    except Exception as e:
        _safe_print(f"[SovereignPatch] import layer/packager failed: {e}")
        return

    # Generate build fingerprint once at startup (capability #9)
    try:
        ensure_build_fingerprint()
    except Exception as e:
        _safe_print(f"[SovereignPatch] ensure_build_fingerprint failed: {e}")

    layer = SovereignAuthorityLayer()

    def patched_run_pilot(*args, **kwargs):
        started = time.time()
        result = original_run_pilot(*args, **kwargs)

        # Defensive: ensure dict-like result
        if result is None:
            result = {}
        if not isinstance(result, dict):
            # keep original but wrap into dict
            result = {"raw_result": str(result)}

        # Apply 7 mandatory authority capabilities (capabilities #1-#7)
        try:
            result = layer.apply(result)
        except Exception as e:
            _safe_print(f"[SovereignPatch] authority layer apply failed: {e}")

        # Policy manifest snapshot (capability #8)
        policy_manifest_path = None
        try:
            policy_manifest_path = write_policy_manifest(result)
        except Exception as e:
            _safe_print(f"[SovereignPatch] write_policy_manifest failed: {e}")

        # Evidence package export (capability #10)
        try:
            export_evidence_package(result, policy_manifest_path=policy_manifest_path)
        except Exception as e:
            _safe_print(f"[SovereignPatch] export_evidence_package failed: {e}")

        # Optional: add runtime marker
        try:
            result.setdefault("sovereign_patch", {})
            result["sovereign_patch"].update({
                "authority_layer": "enabled",
                "policy_manifest": bool(policy_manifest_path),
                "evidence_package": True,
                "patch_latency_ms": int((time.time() - started) * 1000),
            })
        except Exception:
            pass

        return result

    patched_run_pilot._sovereign_patched = True  # type: ignore[attr-defined]
    pilot_suite.run_pilot = patched_run_pilot
    _safe_print("[SovereignPatch] ✅ run_pilot patched (Zero-touch Authority Layer enabled).")


# Apply patch on interpreter startup
_patch_run_pilot()