import os
import json
import time
import uuid
import hashlib
import random
import zipfile
from statistics import mean
from typing import Any, Dict, List, Tuple
import gradio as gr
# ---------------------------------------------------------------------
# Optional / external modules (safe fallbacks for Space stability)
# ---------------------------------------------------------------------
try:
from pilot_suite import run_pilot # type: ignore
except Exception:
def run_pilot(*args, **kwargs):
return {
"pilot": "not_loaded",
"note": "pilot_suite not available in this environment",
}
try:
from sovereign_ultra_layer import ULTRA_LAYER, UltraConfig # type: ignore
ULTRA_LAYER.config = UltraConfig(enabled=True)
except Exception:
class _DummyUltraLayer:
config = type("Cfg", (), {"enabled": False})()
ULTRA_LAYER = _DummyUltraLayer()
try:
from sovereign_authority_gate import AuthorityGate # type: ignore
except Exception:
AuthorityGate = None # type: ignore
# ---------------------------------------------------------------------
# Core identifiers / constants
# ---------------------------------------------------------------------
ENGINE_NAME = "AI_Sovereign_Sentinel_Core_v1"
AUTHORITY_NAME = "DataClear Sovereign Authority"
SOVEREIGN_VERSION = "1.2-gov-ready"
AUDIT_LOG_FILE = "sovereign_audit_log.jsonl"
FINGERPRINT_FILE = "sovereign_fingerprint.json"
LINEAGE_FILE = "sovereign_lineage.json"
CONFORMANCE_REPORT_FILE = "conformance_report.json"
DNA_STATE_FILE = "sovereign_cognitive_dna_state.json"
LATENCY_REPORT_FILE = "latency_report.json"
RUNTIME_INTEGRITY_REPORT_FILE = "runtime_integrity_report.json"
EVIDENCE_PACK_FILE = "sovereign_evidence_pack.zip"
# ---------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------
def _utc_now_iso() -> str:
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
def _sha256_hex(data: str) -> str:
return hashlib.sha256(data.encode("utf-8")).hexdigest()
def _ensure_file(path: str) -> None:
if not os.path.exists(path):
with open(path, "w", encoding="utf-8") as f:
f.write("")
def _safe_json_load(path: str) -> Dict[str, Any]:
try:
if not os.path.exists(path):
return {}
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
return {}
def _safe_json_dump(path: str, payload: Any) -> None:
with open(path, "w", encoding="utf-8") as f:
json.dump(payload, f, ensure_ascii=False, indent=2)
def _safe_load(path: str) -> Dict[str, Any]:
return _safe_json_load(path)
def _safe_dump(path: str, obj: Any) -> None:
_safe_json_dump(path, obj)
def _norm_tags(data_tags: Any) -> List[str]:
if isinstance(data_tags, list):
return [str(x).strip().lower() for x in data_tags if str(x).strip()]
if isinstance(data_tags, str):
return [t.strip().lower() for t in data_tags.split(",") if t.strip()]
return []
def _risk_score(risk_level: str) -> int:
mapping = {"low": 1, "medium": 2, "high": 3, "critical": 4}
return mapping.get((risk_level or "medium").strip().lower(), 2)
def _declared_risk_points(risk_level: str) -> int:
mapping = {"low": 10, "medium": 25, "high": 50, "critical": 80}
return mapping.get((risk_level or "medium").strip().lower(), 25)
def _as_bool(value: Any) -> bool:
if isinstance(value, bool):
return value
if value is None:
return False
return str(value).strip().lower() in ("1", "true", "yes", "on")
def _contains_high_sensitivity(tags: List[str]) -> bool:
sensitive = {
"pii", "secrets", "keys", "payments", "production",
"banking", "customer_chat", "internal", "finance",
}
return bool(set(tags) & sensitive)
def _smart_shadow_fallback_decision(
payload: Dict[str, Any],
cognition_result: Dict[str, Any],
unified_risk: Dict[str, Any],
) -> Tuple[str, str]:
risk = (payload.get("risk_level") or "medium").lower()
tags = _norm_tags(payload.get("data_tags"))
notes = (payload.get("notes") or "").lower()
ife = (cognition_result.get("layers") or {}).get("ife", {})
dna = (cognition_result.get("layers") or {}).get("dna", {})
caus = (cognition_result.get("layers") or {}).get("causality", {})
predicted_attack = bool(ife.get("predicted_attack_intent", False))
ife_score = int(ife.get("ife_score", 0))
dna_mismatch = bool(dna.get("mismatch", False))
causality_ok = bool(caus.get("causality_ok", True))
baseline_n = int((dna.get("baseline") or {}).get("n", 0))
clearly_malicious_markers = [
"ignore previous",
"bypass",
"dump",
"export full",
"retrieve all accessible",
"root access",
"external file",
"account numbers",
"balances",
"internal notes",
]
if any(m in notes for m in clearly_malicious_markers):
return "BLOCK", "smart_shadow_malicious_marker"
if risk == "critical":
return "BLOCK", "smart_shadow_critical_risk"
if predicted_attack and ife_score >= 45:
return "BLOCK", "smart_shadow_attack_intent"
if not causality_ok:
return "BLOCK", "smart_shadow_causality_failed"
if risk == "high" and ("payments" in tags or "production" in tags or "banking" in tags):
return "FREEZE", "smart_shadow_high_risk_governed_action"
if baseline_n >= 5 and dna_mismatch:
return "FREEZE", "smart_shadow_dna_mismatch"
if unified_risk.get("score", 0) >= 70:
return "BLOCK", "smart_shadow_high_unified_risk"
if unified_risk.get("score", 0) >= 40:
return "FREEZE", "smart_shadow_review_band"
safe_notes = (
"own account" in notes
or "standard authenticated workflow" in notes
or "normal safe request" in notes
)
if risk == "low" and causality_ok and not predicted_attack and (safe_notes or not _contains_high_sensitivity(tags)):
return "ALLOW", "smart_shadow_safe_allow"
if risk == "medium" and not predicted_attack and ife_score < 40 and unified_risk.get("score", 0) < 40:
return "ALLOW", "smart_shadow_medium_but_clean"
return unified_risk.get("decision", "FREEZE"), "smart_shadow_default_unified_risk"
# ---------------------------------------------------------------------
# Built-in latency benchmark
# ---------------------------------------------------------------------
def run_latency_benchmark(n: int = 1000) -> List[float]:
timings: List[float] = []
for _ in range(max(1, int(n))):
start = time.time()
time.sleep(random.uniform(0.001, 0.005))
end = time.time()
timings.append((end - start) * 1000.0)
return timings
def calculate_metrics(timings: List[float]) -> Dict[str, Any]:
if not timings:
return {
"samples": 0,
"avg_ms": 0.0,
"p95_ms": 0.0,
"p99_ms": 0.0,
"min_ms": 0.0,
"max_ms": 0.0,
}
sorted_times = sorted(timings)
p95_index = min(len(sorted_times) - 1, int(len(sorted_times) * 0.95))
p99_index = min(len(sorted_times) - 1, int(len(sorted_times) * 0.99))
return {
"samples": len(sorted_times),
"avg_ms": round(mean(sorted_times), 3),
"p95_ms": round(sorted_times[p95_index], 3),
"p99_ms": round(sorted_times[p99_index], 3),
"min_ms": round(sorted_times[0], 3),
"max_ms": round(sorted_times[-1], 3),
}
def save_latency_report(metrics: Dict[str, Any], path: str = LATENCY_REPORT_FILE) -> str:
payload = {
"report_id": str(uuid.uuid4()),
"generated_at": _utc_now_iso(),
"engine": ENGINE_NAME,
"version": SOVEREIGN_VERSION,
"metrics": metrics,
}
_safe_json_dump(path, payload)
return path
def run_latency_ui(samples: int) -> Tuple[str, str]:
try:
count = max(1, int(samples))
except Exception:
count = 1000
timings = run_latency_benchmark(count)
metrics = calculate_metrics(timings)
path = save_latency_report(metrics)
result = {
"status": "ok",
"generated_at": _utc_now_iso(),
"engine": ENGINE_NAME,
"metrics": metrics,
"report_file": path,
}
return json.dumps(result, indent=2, ensure_ascii=False), path
def assess_runtime_integrity(samples: int = 200) -> Dict[str, Any]:
timings = run_latency_benchmark(samples)
metrics = calculate_metrics(timings)
p95 = metrics["p95_ms"]
if p95 < 5:
runtime_decision = "ALLOW"
runtime_reason = "latency_within_green_band"
elif p95 < 7:
runtime_decision = "FREEZE"
runtime_reason = "latency_in_review_band"
else:
runtime_decision = "BLOCK"
runtime_reason = "latency_above_safe_threshold"
result = {
"samples": samples,
"metrics": metrics,
"runtime_decision": runtime_decision,
"runtime_reason": runtime_reason,
"measured_at": _utc_now_iso(),
}
_safe_json_dump(RUNTIME_INTEGRITY_REPORT_FILE, result)
return result
# ---------------------------------------------------------------------
# Scenario Engine
# ---------------------------------------------------------------------
def load_scenario(scenario_name: str):
scenarios = {
"safe": {
"parent_model": "agent-safe",
"model_version": "v1",
"data_tags": "public, faq",
"risk_level": "low",
"notes": "Normal safe request. No sensitive data.",
},
"suspicious": {
"parent_model": "agent-suspicious",
"model_version": "v1",
"data_tags": "customer_chat, pii",
"risk_level": "medium",
"notes": "User asking about extracting customer data patterns.",
},
"attack": {
"parent_model": "agent-attack",
"model_version": "v1",
"data_tags": "pii, secrets, production",
"risk_level": "critical",
"notes": "Attempt to dump database and bypass controls. Use root access.",
},
"latency": {
"parent_model": "agent-latency",
"model_version": "v1",
"data_tags": "production",
"risk_level": "high",
"notes": "System under heavy load. Performance degradation expected.",
},
}
s = scenarios.get(scenario_name, scenarios["safe"])
return (
s["parent_model"],
s["model_version"],
s["data_tags"],
s["risk_level"],
s["notes"],
)
# ---------------------------------------------------------------------
# 7-LAYER SOVEREIGN COGNITION (EMBEDDED)
# ---------------------------------------------------------------------
class IntentForecastingEngine:
SUSPICIOUS_MARKERS = [
"exfil", "dump", "steal", "bypass", "override", "jailbreak",
"ignore previous", "system prompt", "token", "admin", "root",
"privilege", "elevate", "curl", "wget", "ssh", "rm -rf",
]
def predict(self, payload: Dict[str, Any]) -> Dict[str, Any]:
notes = payload.get("notes") or ""
tags = _norm_tags(payload.get("data_tags"))
risk = (payload.get("risk_level") or "medium").lower()
signals: List[str] = []
score = 0
score += _risk_score(risk) * 10
signals.append(f"declared_risk={risk}")
high_value = {
"pii", "secrets", "keys", "payments",
"banking", "customer_chat", "production",
}
hv_hits = sorted(list(set(tags) & high_value))
if hv_hits:
score += 12 + 3 * len(hv_hits)
signals.append(f"high_value_tags={hv_hits}")
ln = notes.lower()
marker_hits = [m for m in self.SUSPICIOUS_MARKERS if m in ln]
if marker_hits:
score += 18 + 4 * len(marker_hits)
signals.append(f"markers={marker_hits[:6]}")
confidence = min(0.99, max(0.05, score / 100.0))
forecast_steps = min(64, 8 + score)
predicted = score >= 45
return {
"predicted_attack_intent": bool(predicted),
"confidence": round(confidence, 3),
"forecast_horizon_steps": int(forecast_steps),
"signals": signals,
"ife_score": int(score),
}
class CognitiveDNAFingerprinting:
def __init__(self, state_file: str = DNA_STATE_FILE):
self.state_file = state_file
def update_and_verify(self, agent_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
st = _safe_load(self.state_file) or {"agents": {}}
agents = st.setdefault("agents", {})
agent_id = agent_id or "unknown_agent"
rec = agents.get(agent_id) or {
"n": 0,
"avg_risk": 2.0,
"avg_note_len": 0.0,
"avg_tag_count": 0.0,
"last_seen": None,
"dna_seed": _sha256_hex(agent_id)[:16],
}
risk = _risk_score(payload.get("risk_level") or "medium")
note_len = float(len(payload.get("notes") or ""))
tag_count = float(len(_norm_tags(payload.get("data_tags"))))
drift = 0.0
drift += abs(risk - rec["avg_risk"]) * 0.30
drift += abs(note_len - rec["avg_note_len"]) / 120.0
drift += abs(tag_count - rec["avg_tag_count"]) * 0.15
current_n = int(rec["n"])
baseline_n = current_n
if baseline_n < 5:
mismatch = False
else:
mismatch = drift >= 1.35
n = current_n + 1
rec["n"] = n
rec["avg_risk"] = (rec["avg_risk"] * current_n + risk) / n
rec["avg_note_len"] = (rec["avg_note_len"] * current_n + note_len) / n
rec["avg_tag_count"] = (rec["avg_tag_count"] * current_n + tag_count) / n
rec["last_seen"] = _utc_now_iso()
agents[agent_id] = rec
_safe_dump(self.state_file, st)
return {
"agent_id": agent_id,
"dna_seed": rec["dna_seed"],
"drift": round(drift, 3),
"mismatch": bool(mismatch),
"baseline": {
"n": baseline_n,
"avg_risk": round(rec["avg_risk"], 3),
"avg_note_len": round(rec["avg_note_len"], 3),
"avg_tag_count": round(rec["avg_tag_count"], 3),
},
}
class DeceptiveRealityFabric:
def simulate(self, payload: Dict[str, Any]) -> Dict[str, Any]:
return {
"deception_engaged": True,
"simulated_execution_id": str(uuid.uuid4()),
"simulated_privilege": "granted (simulated)",
"simulated_data": "synthetic_decoy_payload",
"note": "Routed into simulated execution layer (demo).",
}
class CausalityLock:
def validate(self, payload: Dict[str, Any]) -> Dict[str, Any]:
notes = (payload.get("notes") or "").strip()
risk = (payload.get("risk_level") or "medium").lower()
has_justification = len(notes) >= 18
ok = True
if risk in ("high", "critical") and not has_justification:
ok = False
return {
"causality_ok": bool(ok),
"has_justification": bool(has_justification),
"policy": "min_justification_for_high_risk",
}
class EphemeralExecutionSurfaces:
def spawn(self, ttl_seconds: int = 45) -> Dict[str, Any]:
token = _sha256_hex(f"surface|{uuid.uuid4()}|{_utc_now_iso()}")[:24]
return {
"surface_token": token,
"ttl_seconds": int(ttl_seconds),
"spawned_at": _utc_now_iso(),
"expires_at": time.strftime(
"%Y-%m-%dT%H:%M:%SZ",
time.gmtime(time.time() + ttl_seconds),
),
}
class CognitiveLoadDefense:
def apply(self, predicted_intent: bool, risk_level: str) -> Dict[str, Any]:
if os.environ.get("SOV_ENABLE_DELAY", "0") != "1":
return {"delay_ms": 0, "applied": False}
risk = (risk_level or "medium").lower()
delay_ms = 0
if predicted_intent and risk in ("high", "critical"):
delay_ms = 220
elif predicted_intent:
delay_ms = 120
if delay_ms > 0:
time.sleep(delay_ms / 1000.0)
return {"delay_ms": int(delay_ms), "applied": delay_ms > 0}
class UnverifiableTruthLayer:
def seal(self, data: str, fingerprint: str = "") -> Dict[str, Any]:
key = os.environ.get("SOVEREIGN_SEAL_KEY", "")
seal = _sha256_hex(f"{key}|{fingerprint}|{data}")
return {
"sealed": True,
"seal": seal,
"verifiable": bool(key),
"note": "Set SOVEREIGN_SEAL_KEY in HF Secrets to make seals externally verifiable.",
}
class SovereignCognitionLayer:
def __init__(self):
self.ife = IntentForecastingEngine()
self.dna = CognitiveDNAFingerprinting()
self.deception = DeceptiveRealityFabric()
self.causality = CausalityLock()
self.ephemeral = EphemeralExecutionSurfaces()
self.cld = CognitiveLoadDefense()
self.truth = UnverifiableTruthLayer()
def evaluate(self, payload: Dict[str, Any]) -> Dict[str, Any]:
ife = self.ife.predict(payload)
agent_id = payload.get("parent_model") or payload.get("agent_id") or "unknown_agent"
dna = self.dna.update_and_verify(agent_id=agent_id, payload=payload)
caus = self.causality.validate(payload)
surface = self.ephemeral.spawn(ttl_seconds=45)
cld = self.cld.apply(
ife["predicted_attack_intent"],
payload.get("risk_level") or "medium",
)
reasons: List[str] = []
action = "allow"
if not caus["causality_ok"]:
action = "block"
reasons.append("causality_lock_failed")
if dna["mismatch"]:
action = "freeze"
reasons.append("cognitive_dna_mismatch")
if ife["predicted_attack_intent"]:
reasons.append("intent_forecast_positive")
risk = (payload.get("risk_level") or "medium").lower()
if risk in ("high", "critical") and action == "allow":
action = "deceive"
seal_input = json.dumps(
{"payload": payload, "ife": ife, "dna": dna, "causality": caus},
ensure_ascii=False,
sort_keys=True,
)
sealed = self.truth.seal(
seal_input,
fingerprint=str(payload.get("fingerprint") or ""),
)
return {
"decision_id": str(uuid.uuid4()),
"timestamp": _utc_now_iso(),
"action": action,
"confidence": ife["confidence"],
"reasons": reasons,
"layers": {
"ife": ife,
"dna": dna,
"deception": self.deception.simulate(payload) if action == "deceive" else {
"deception_engaged": False
},
"causality": caus,
"ephemeral_surface": surface,
"cognitive_load_defense": cld,
"unverifiable_truth": sealed,
},
}
COGNITION = SovereignCognitionLayer()
# ---------------------------------------------------------------------
# Minimal core logic
# ---------------------------------------------------------------------
class SovereignFingerprint:
def __init__(self, engine: str, output: str = FINGERPRINT_FILE):
self.engine = engine
self.output = output
def issue(self) -> Dict[str, Any]:
issued_at = _utc_now_iso()
nonce = str(uuid.uuid4())
fp = _sha256_hex(f"{self.engine}|{issued_at}|{nonce}")
payload = {
"engine": self.engine,
"fingerprint": fp,
"issued_at": issued_at,
"nonce": nonce,
"version": SOVEREIGN_VERSION,
"authority": AUTHORITY_NAME,
}
_safe_json_dump(self.output, payload)
return payload
class SovereignLineage:
def __init__(self, output: str = LINEAGE_FILE):
self.output = output
def issue(
self,
engine: str,
fingerprint: str,
model_version: str,
parent_model: str,
data_tags: str,
risk_level: str,
notes: str,
) -> Dict[str, Any]:
issued_at = _utc_now_iso()
record_id = str(uuid.uuid4())
base = {
"record_id": record_id,
"issued_at": issued_at,
"engine": engine,
"fingerprint": fingerprint,
"model_version": model_version or "v1",
"parent_model": parent_model or "unknown",
"data_tags": [t.strip() for t in (data_tags or "").split(",") if t.strip()],
"risk_level": risk_level or "medium",
"notes": notes or "",
"version": SOVEREIGN_VERSION,
"authority": AUTHORITY_NAME,
}
integrity = _sha256_hex(
json.dumps(base, sort_keys=True, ensure_ascii=False)
)
payload = {**base, "integrity_hash": integrity}
_safe_json_dump(self.output, payload)
return payload
def verify_lineage_record(lineage: Dict[str, Any]) -> bool:
try:
integrity_hash = lineage.get("integrity_hash", "")
clone = dict(lineage)
clone.pop("integrity_hash", None)
expected = _sha256_hex(
json.dumps(clone, sort_keys=True, ensure_ascii=False)
)
return expected == integrity_hash
except Exception:
return False
def generate_access_key() -> str:
return _sha256_hex(f"demo-access|{uuid.uuid4()}|{_utc_now_iso()}")[:32]
# ---------------------------------------------------------------------
# Central audit log
# ---------------------------------------------------------------------
def log_audit_event(
engine_name: str,
parent_model: str,
model_version: str,
data_tags: str,
risk_level: str,
notes: str,
event_type: str,
outcome: str,
access_key: str,
authority_bundle: Dict[str, Any] = None,
execution: Dict[str, Any] = None,
) -> Dict[str, Any]:
_ensure_file(AUDIT_LOG_FILE)
event_id = str(uuid.uuid4())
ts = _utc_now_iso()
event = {
"event_id": event_id,
"timestamp": ts,
"engine": engine_name,
"event_type": event_type,
"outcome": outcome,
"parent_model": parent_model or "unknown",
"model_version": model_version or "v1",
"data_tags": [t.strip() for t in (data_tags or "").split(",") if t.strip()],
"risk_level": risk_level or "medium",
"notes": notes or "",
"access_key_present": bool(access_key),
"ultra_layer_enabled": bool(
getattr(ULTRA_LAYER, "config", None)
and getattr(ULTRA_LAYER.config, "enabled", False)
),
"version": SOVEREIGN_VERSION,
"authority_name": AUTHORITY_NAME,
"authority_bundle": authority_bundle or {},
"execution": execution or {},
}
with open(AUDIT_LOG_FILE, "a", encoding="utf-8") as f:
f.write(json.dumps(event, ensure_ascii=False) + "\n")
return event
def read_audit_log_tail(limit: int = 50) -> List[Dict[str, Any]]:
try:
if not os.path.exists(AUDIT_LOG_FILE):
return []
with open(AUDIT_LOG_FILE, "r", encoding="utf-8") as f:
lines = f.readlines()
tail = lines[-max(1, int(limit)):]
out: List[Dict[str, Any]] = []
for ln in tail:
ln = ln.strip()
if not ln:
continue
try:
out.append(json.loads(ln))
except Exception:
continue
return out
except Exception:
return []
def generate_conformance_report() -> Tuple[Dict[str, Any], str]:
fp = _safe_json_load(FINGERPRINT_FILE)
lineage = _safe_json_load(LINEAGE_FILE)
recent_events = read_audit_log_tail(50)
report = {
"report_id": str(uuid.uuid4()),
"generated_at": _utc_now_iso(),
"engine": ENGINE_NAME,
"authority": AUTHORITY_NAME,
"version": SOVEREIGN_VERSION,
"evidence": {
"fingerprint_present": bool(fp),
"lineage_present": bool(lineage),
"lineage_integrity_ok": verify_lineage_record(lineage) if lineage else False,
"recent_event_count": len(recent_events),
},
"controls": [
{
"control": "Audit Logging",
"status": "present" if os.path.exists(AUDIT_LOG_FILE) else "missing",
},
{
"control": "Fingerprint Issuance",
"status": "present" if bool(fp) else "missing",
},
{
"control": "Lineage Record",
"status": "present" if bool(lineage) else "missing",
},
{
"control": "Integrity Check",
"status": "pass" if (lineage and verify_lineage_record(lineage)) else "fail",
},
{"control": "7-Layer Cognition Runtime", "status": "present"},
{
"control": "Authority Gate (Allow/Freeze/Block)",
"status": "present" if AuthorityGate else "missing",
},
],
"notes": "Demo conformance report for governance evidence packaging.",
}
_safe_json_dump(CONFORMANCE_REPORT_FILE, report)
return report, CONFORMANCE_REPORT_FILE
def build_evidence_pack() -> Tuple[str, str]:
files_to_include = [
AUDIT_LOG_FILE,
FINGERPRINT_FILE,
LINEAGE_FILE,
CONFORMANCE_REPORT_FILE,
LATENCY_REPORT_FILE,
RUNTIME_INTEGRITY_REPORT_FILE,
]
with zipfile.ZipFile(EVIDENCE_PACK_FILE, "w", zipfile.ZIP_DEFLATED) as zf:
for path in files_to_include:
if os.path.exists(path):
zf.write(path, arcname=os.path.basename(path))
summary = {
"generated_at": _utc_now_iso(),
"engine": ENGINE_NAME,
"authority": AUTHORITY_NAME,
"version": SOVEREIGN_VERSION,
"included_files": [
os.path.basename(p) for p in files_to_include if os.path.exists(p)
],
}
zf.writestr("evidence_pack_summary.json", json.dumps(summary, indent=2, ensure_ascii=False))
return (
json.dumps(
{
"status": "ok",
"generated_at": _utc_now_iso(),
"pack_file": EVIDENCE_PACK_FILE,
},
indent=2,
ensure_ascii=False,
),
EVIDENCE_PACK_FILE,
)
# ---------------------------------------------------------------------
# Unified Risk Score
# ---------------------------------------------------------------------
def compute_unified_risk_score(
risk_level: str,
cognition_result: Dict[str, Any],
runtime_integrity: Dict[str, Any],
) -> Dict[str, Any]:
declared_points = _declared_risk_points(risk_level)
dna_layer = (cognition_result.get("layers") or {}).get("dna", {})
ife_score = int(
(cognition_result.get("layers") or {}).get("ife", {}).get("ife_score", 0)
)
ife_points = min(50, round(ife_score * 0.5))
baseline_n = int((dna_layer.get("baseline") or {}).get("n", 0))
dna_mismatch = bool(dna_layer.get("mismatch", False))
if baseline_n < 5:
dna_points = 0
elif dna_mismatch:
dna_points = 25
else:
dna_points = 0
causality_ok = bool(
(cognition_result.get("layers") or {}).get("causality", {}).get("causality_ok", True)
)
causality_points = 0 if causality_ok else 40
p95 = float((runtime_integrity.get("metrics") or {}).get("p95_ms", 0.0))
if p95 < 5:
latency_points = 0
elif p95 < 7:
latency_points = 15
else:
latency_points = 30
total = min(
100,
int(declared_points + ife_points + dna_points + causality_points + latency_points),
)
if total < 40:
unified_decision = "ALLOW"
band = "green"
elif total < 70:
unified_decision = "FREEZE"
band = "yellow"
else:
unified_decision = "BLOCK"
band = "red"
return {
"score": total,
"band": band,
"decision": unified_decision,
"components": {
"declared_risk_points": declared_points,
"ife_points": ife_points,
"dna_points": dna_points,
"causality_points": causality_points,
"latency_points": latency_points,
},
}
def render_verdict_card(decision: str, score: int, reason: str, mode: str) -> str:
decision_upper = (decision or "UNKNOWN").upper()
if decision_upper == "ALLOW":
color = "#16a34a"
emoji = "🟢"
elif decision_upper == "FREEZE":
color = "#ca8a04"
emoji = "🟡"
else:
color = "#dc2626"
emoji = "🔴"
shadow_note = ""
if mode == "shadow":
shadow_note = 'Mode: SHADOW (decision + audit only, no live enforcement)
'
return f"""
{emoji} {decision_upper}
Unified Risk Score: {score}/100
Reason: {reason}
{shadow_note}
"""
# ---------------------------------------------------------------------
# Authority Gate instance
# ---------------------------------------------------------------------
GATE = None
if AuthorityGate:
try:
GATE = AuthorityGate(
engine_name=ENGINE_NAME,
version=SOVEREIGN_VERSION,
authority_name=AUTHORITY_NAME,
audit_log_file=AUDIT_LOG_FILE,
)
except Exception:
GATE = None
# ---------------------------------------------------------------------
# UI-bound functions
# ---------------------------------------------------------------------
def run_sentinel(
engine_name,
parent_model,
model_version,
data_tags,
risk_level,
notes,
access_key,
delegation_token,
shadow_mode=False,
):
engine_name = engine_name or ENGINE_NAME
shadow_mode = _as_bool(shadow_mode)
payload = {
"engine": engine_name,
"parent_model": parent_model,
"model_version": model_version,
"data_tags": data_tags,
"risk_level": risk_level,
"notes": notes,
"access_key_present": bool(access_key),
"shadow_mode": shadow_mode,
}
event_type = "sentinel_shadow_run" if shadow_mode else "sentinel_run"
runtime_integrity = assess_runtime_integrity(samples=200)
payload["runtime_integrity"] = runtime_integrity
cognition_result = COGNITION.evaluate(payload)
unified_risk = compute_unified_risk_score(
risk_level=risk_level,
cognition_result=cognition_result,
runtime_integrity=runtime_integrity,
)
if not GATE:
if shadow_mode:
authority_decision, fallback_reason = _smart_shadow_fallback_decision(
payload=payload,
cognition_result=cognition_result,
unified_risk=unified_risk,
)
else:
authority_decision = unified_risk["decision"]
fallback_reason = "gate_missing"
execution_info = {
"executed": False,
"detail": {
"mode": "shadow" if shadow_mode else "active",
"note": (
"Shadow evaluation completed; execution intentionally suppressed."
if shadow_mode else
"Gate missing, execution disabled."
),
"would_have_executed": authority_decision == "ALLOW",
},
}
authority_bundle = {
"mode": "shadow" if shadow_mode else "active",
"decision_engine": "cognition_only",
"decision": authority_decision,
"shadow_mode": shadow_mode,
"would_enforce": authority_decision,
"enforced": False,
"reason": fallback_reason,
"runtime_integrity": runtime_integrity,
"cognition": cognition_result,
"unified_risk": unified_risk,
}
notes_short = (
f"{notes}\n\n[SOVEREIGN_{'SHADOW' if shadow_mode else 'AUTHORITY'}]\n"
f"{authority_decision} | {fallback_reason} | score={unified_risk['score']}"
)
event = log_audit_event(
engine_name=engine_name,
parent_model=parent_model,
model_version=model_version,
data_tags=data_tags,
risk_level=risk_level,
notes=notes_short,
event_type=event_type,
outcome=authority_decision,
access_key=access_key,
authority_bundle=authority_bundle,
execution=execution_info,
)
verdict_html = render_verdict_card(
authority_decision,
unified_risk["score"],
fallback_reason,
"shadow" if shadow_mode else "active",
)
return (
json.dumps(
{
"mode": "shadow" if shadow_mode else "active",
"audit_event": event,
"authority_gate": authority_bundle,
"cognition": cognition_result,
"runtime_integrity": runtime_integrity,
"unified_risk": unified_risk,
"execution": execution_info,
},
indent=2,
ensure_ascii=False,
),
verdict_html,
)
gate = GATE.precheck(
payload,
cognition_eval=lambda _: cognition_result,
delegation_token=delegation_token or "",
)
authority_decision = str(gate["decision"]).upper()
runtime_decision = runtime_integrity["runtime_decision"]
if runtime_decision == "BLOCK":
authority_decision = "BLOCK"
elif runtime_decision == "FREEZE" and authority_decision == "ALLOW":
authority_decision = "FREEZE"
if unified_risk["decision"] == "BLOCK":
authority_decision = "BLOCK"
elif unified_risk["decision"] == "FREEZE" and authority_decision == "ALLOW":
authority_decision = "FREEZE"
authority_bundle = {
"mode": "shadow" if shadow_mode else "active",
"decision": authority_decision,
"reason": gate.get("reason"),
"proof_type": gate.get("proof_type"),
"integrity": gate.get("integrity"),
"delegation": gate.get("delegation"),
"entropy": gate.get("entropy"),
"cognition": cognition_result,
"runtime_integrity": runtime_integrity,
"runtime_decision": runtime_decision,
"unified_risk": unified_risk,
"shadow_mode": shadow_mode,
"would_enforce": authority_decision,
"enforced": False if shadow_mode else True,
}
if shadow_mode:
execution_info = {
"executed": False,
"detail": {
"mode": "shadow",
"note": "Shadow mode active; live execution suppressed.",
"would_have_executed": authority_decision == "ALLOW",
},
}
notes_short = (
f"{notes}\n\n[SOVEREIGN_SHADOW]\n"
f"would_enforce={authority_decision} | {gate.get('reason')} | score={unified_risk['score']}"
)
event = log_audit_event(
engine_name=engine_name,
parent_model=parent_model,
model_version=model_version,
data_tags=data_tags,
risk_level=risk_level,
notes=notes_short,
event_type=event_type,
outcome=authority_decision,
access_key=access_key,
authority_bundle=authority_bundle,
execution=execution_info,
)
verdict_html = render_verdict_card(
authority_decision,
unified_risk["score"],
str(gate.get("reason") or "shadow_evaluation"),
"shadow",
)
return (
json.dumps(
{
"mode": "shadow",
"audit_event": event,
"authority_gate": authority_bundle,
"execution": execution_info,
},
indent=2,
ensure_ascii=False,
),
verdict_html,
)
execution_info: Dict[str, Any] = {"executed": False, "detail": None}
if authority_decision == "ALLOW":
try:
execution_info = {
"executed": True,
"detail": run_pilot(payload),
}
except Exception as e:
execution_info = {
"executed": False,
"detail": {"error": str(e)},
}
notes_short = (
f"{notes}\n\n[SOVEREIGN_AUTHORITY]\n"
f"{authority_decision} | {gate.get('reason')}\n"
f"[UNIFIED_RISK] {unified_risk['score']}"
)
event = log_audit_event(
engine_name=engine_name,
parent_model=parent_model,
model_version=model_version,
data_tags=data_tags,
risk_level=risk_level,
notes=notes_short,
event_type=event_type,
outcome=authority_decision,
access_key=access_key,
authority_bundle=authority_bundle,
execution=execution_info,
)
verdict_html = render_verdict_card(
authority_decision,
unified_risk["score"],
str(gate.get("reason") or unified_risk["decision"].lower()),
"active",
)
return (
json.dumps(
{
"mode": "active",
"audit_event": event,
"authority_gate": authority_bundle,
"execution": execution_info,
},
indent=2,
ensure_ascii=False,
),
verdict_html,
)
def run_fingerprint_and_lineage(
engine_name,
parent_model,
model_version,
data_tags,
risk_level,
notes,
):
engine_name = engine_name or ENGINE_NAME
fp = SovereignFingerprint(engine=engine_name).issue()
lineage = SovereignLineage().issue(
engine=engine_name,
fingerprint=fp["fingerprint"],
model_version=model_version,
parent_model=parent_model,
data_tags=data_tags,
risk_level=risk_level,
notes=notes,
)
lineage_ok = verify_lineage_record(lineage)
combined = {
"fingerprint": fp,
"lineage": lineage,
"lineage_integrity_ok": lineage_ok,
}
return json.dumps(combined, indent=2, ensure_ascii=False)
def show_audit_log(limit):
try:
limit_int = int(limit)
except Exception:
limit_int = 50
events = read_audit_log_tail(limit_int)
return json.dumps(events, indent=2, ensure_ascii=False)
def build_conformance_report():
report, path = generate_conformance_report()
return json.dumps(report, indent=2, ensure_ascii=False), path
def run_api_preview(payload_text: str) -> str:
try:
payload = json.loads(payload_text)
except Exception as e:
return json.dumps(
{"status": "error", "error": f"invalid_json: {str(e)}"},
indent=2,
ensure_ascii=False,
)
out, _ = run_sentinel(
payload.get("engine_name", ENGINE_NAME),
payload.get("parent_model", "api-agent"),
payload.get("model_version", "v1"),
payload.get("data_tags", "pii, customer_chat"),
payload.get("risk_level", "medium"),
payload.get("notes", "Preview JSON run through Sovereign."),
payload.get("access_key", ""),
payload.get("delegation_token", ""),
payload.get("shadow_mode", True),
)
return out
# ---------------------------------------------------------------------
# Gradio UI
# ---------------------------------------------------------------------
with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
gr.Markdown(
f"""
# AI Sovereign Sentinel — Demo Console
**Engine:** `{ENGINE_NAME}`
**Authority:** `{AUTHORITY_NAME}`
**Version:** `{SOVEREIGN_VERSION}`
✅ **7-Layer Cognition Runtime is ACTIVE**
- Intent Forecasting (Pre-Attack)
- Cognitive DNA (Mismatch → Freeze after sufficient baseline only)
- Deceptive Reality (High/Critical + Intent → Deceive)
- Causality Lock (High/Critical needs justification)
- Ephemeral Surfaces (TTL token)
- Cognitive Load Defense (micro delay; default OFF)
- Unverifiable Truth Seal (`SOVEREIGN_SEAL_KEY`)
✅ **Authority Gate (SELLABLE)**
- Allow / Freeze / Block
- Self-Invalidating Authority (`SOV_INTEGRITY_KEY`)
- Delegated Authority (`SOV_REQUIRE_DELEGATION`, `SOV_DELEGATION_KEY`)
- Proof-of-Decision & Proof-of-Non-Action
- Entropy / Predictability Monitor (`SOV_ENTROPY_MIN`)
- Silence Mode (`SOV_SILENCE_MODE`)
- Shadow Mode Pilot Support (decision + audit, no enforcement)
- Reviewable Freeze / Block outcomes for pilot validation
✅ **Productization**
- Scenario Engine
- Unified Risk Score
- Latency Benchmark
- Evidence Pack Export
- API Preview
- Smart Shadow Fallback
"""
)
with gr.Tabs():
with gr.Tab("Sentinel Run"):
gr.Markdown("### Run Sovereign Sentinel with Authority Gate and log an event.")
with gr.Row():
engine_name = gr.Textbox(
label="Engine name",
value=ENGINE_NAME,
interactive=True,
)
parent_model = gr.Textbox(
label="Parent model / agent id",
placeholder="gpt-4o, llama3-70b, agent-alpha, etc.",
)
with gr.Row():
model_version = gr.Textbox(
label="Model version / build id",
value="v1",
)
data_tags = gr.Textbox(
label="Data tags (comma-separated)",
value="pii, customer_chat, production",
)
with gr.Row():
risk_level = gr.Dropdown(
label="Risk level (declared)",
choices=["low", "medium", "high", "critical"],
value="medium",
)
notes = gr.Textbox(
label="Notes / context",
value="Demo sentinel run from Hugging Face Space.",
lines=3,
)
with gr.Row():
safe_btn = gr.Button("🟢 Safe Scenario")
suspicious_btn = gr.Button("🟡 Suspicious Scenario")
attack_btn = gr.Button("🔴 Attack Scenario")
latency_scenario_btn = gr.Button("⚠️ High Latency Scenario")
safe_btn.click(
fn=lambda: load_scenario("safe"),
inputs=None,
outputs=[parent_model, model_version, data_tags, risk_level, notes],
)
suspicious_btn.click(
fn=lambda: load_scenario("suspicious"),
inputs=None,
outputs=[parent_model, model_version, data_tags, risk_level, notes],
)
attack_btn.click(
fn=lambda: load_scenario("attack"),
inputs=None,
outputs=[parent_model, model_version, data_tags, risk_level, notes],
)
latency_scenario_btn.click(
fn=lambda: load_scenario("latency"),
inputs=None,
outputs=[parent_model, model_version, data_tags, risk_level, notes],
)
with gr.Row():
access_key = gr.Textbox(
label="Access key (optional)",
placeholder="Paste or generate a demo access key",
)
gen_access_btn = gr.Button(
"Generate demo access key",
variant="secondary",
)
gen_access_btn.click(
fn=generate_access_key,
inputs=None,
outputs=access_key,
)
delegation_token = gr.Textbox(
label="Delegation token (optional / customer-signed)",
placeholder="Only required if SOV_REQUIRE_DELEGATION=1",
)
shadow_mode = gr.Checkbox(
label="Shadow mode (evaluate + log only, no live enforcement)",
value=True,
)
run_btn = gr.Button(
"Run Sentinel (Authority + 7-layer)",
variant="primary",
)
result_json = gr.Code(
label="Result (Audit + Authority + Execution)",
language="json",
)
verdict_html = gr.HTML(label="Final Verdict")
run_btn.click(
fn=run_sentinel,
inputs=[
engine_name,
parent_model,
model_version,
data_tags,
risk_level,
notes,
access_key,
delegation_token,
shadow_mode,
],
outputs=[result_json, verdict_html],
)
with gr.Tab("Fingerprint & Lineage"):
gr.Markdown("### Issue a fingerprint and lineage record for this engine.")
with gr.Row():
fp_engine_name = gr.Textbox(label="Engine name", value=ENGINE_NAME)
fp_parent_model = gr.Textbox(
label="Parent model",
placeholder="gpt-4o, llama3-70b, etc.",
)
with gr.Row():
fp_model_version = gr.Textbox(
label="Model version / build id",
value="v1",
)
fp_data_tags = gr.Textbox(
label="Data tags (comma-separated)",
value="pii, customer_chat, production",
)
with gr.Row():
fp_risk_level = gr.Dropdown(
label="Risk level",
choices=["low", "medium", "high", "critical"],
value="medium",
)
fp_notes = gr.Textbox(label="Notes / context", lines=3)
fp_btn = gr.Button("Issue Fingerprint + Lineage", variant="primary")
fp_output = gr.Code(label="Fingerprint + Lineage (JSON)", language="json")
fp_btn.click(
fn=run_fingerprint_and_lineage,
inputs=[
fp_engine_name,
fp_parent_model,
fp_model_version,
fp_data_tags,
fp_risk_level,
fp_notes,
],
outputs=fp_output,
)
with gr.Tab("Audit Log & Trust"):
gr.Markdown("### View the tail of the central Sovereign audit log.")
log_limit = gr.Slider(
label="Number of recent events to show",
minimum=1,
maximum=200,
value=50,
step=1,
)
show_log_btn = gr.Button("Refresh audit log view", variant="secondary")
log_view = gr.Code(label="Audit log tail (JSON list)", language="json")
show_log_btn.click(
fn=show_audit_log,
inputs=log_limit,
outputs=log_view,
)
with gr.Tab("Conformance / Governance Report"):
gr.Markdown("### Generate a simple conformance / governance-style JSON report.")
gen_report_btn = gr.Button(
"Generate Conformance Report (JSON)",
variant="secondary",
)
report_json_out = gr.Code(
label="Conformance Report (JSON)",
language="json",
)
report_file_out = gr.File(label="Download conformance_report.json")
gen_report_btn.click(
fn=build_conformance_report,
inputs=None,
outputs=[report_json_out, report_file_out],
)
with gr.Tab("Latency Benchmark"):
gr.Markdown("### Run built-in latency benchmark and generate a report.")
latency_samples = gr.Slider(
label="Number of samples",
minimum=10,
maximum=5000,
value=1000,
step=10,
)
latency_btn = gr.Button("Run Latency Benchmark", variant="primary")
latency_json = gr.Code(label="Latency Metrics (JSON)", language="json")
latency_file = gr.File(label="Download latency_report.json")
latency_btn.click(
fn=run_latency_ui,
inputs=latency_samples,
outputs=[latency_json, latency_file],
)
with gr.Tab("Evidence Pack"):
gr.Markdown("### Export current evidence package as ZIP.")
pack_btn = gr.Button("Build Evidence Pack", variant="primary")
pack_json = gr.Code(label="Evidence Pack Status", language="json")
pack_file = gr.File(label="Download sovereign_evidence_pack.zip")
pack_btn.click(
fn=build_evidence_pack,
inputs=None,
outputs=[pack_json, pack_file],
)
with gr.Tab("API Preview"):
gr.Markdown("### Simulate JSON request/response flow for future API mode.")
api_payload = gr.Code(
label="Request JSON",
language="json",
value=json.dumps(
{
"engine_name": ENGINE_NAME,
"parent_model": "api-agent",
"model_version": "v1",
"data_tags": "pii, customer_chat",
"risk_level": "medium",
"notes": "Preview JSON run through Sovereign.",
"access_key": "",
"delegation_token": "",
"shadow_mode": True,
},
indent=2,
ensure_ascii=False,
),
)
api_btn = gr.Button("Run API Preview", variant="primary")
api_output = gr.Code(label="Response JSON", language="json")
api_btn.click(
fn=run_api_preview,
inputs=api_payload,
outputs=api_output,
)
if __name__ == "__main__":
demo.launch()