File size: 3,663 Bytes
67c5e10 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | # 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()
|