Update app.py

app.py

@@ -4,6 +4,8 @@ import json
 import time
 import uuid
 import hashlib
+import random
+from statistics import mean
 from typing import Any, Dict, List, Tuple
 
 import gradio as gr
@@ -15,7 +17,10 @@ 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"}
+        return {
+            "pilot": "not_loaded",
+            "note": "pilot_suite not available in this environment"
+        }
 
 try:
     from sovereign_ultra_layer import ULTRA_LAYER, UltraConfig  # type: ignore
@@ -23,16 +28,15 @@ try:
 except Exception:
     class _DummyUltraLayer:
         config = type("Cfg", (), {"enabled": False})()
+
     ULTRA_LAYER = _DummyUltraLayer()
 
-# ---------------------------------------------------------------------
-# Authority Gate module (new)
-# ---------------------------------------------------------------------
 try:
-    from sovereign_authority_gate import AuthorityGate  # new module
-except Exception as e:
+    from sovereign_authority_gate import AuthorityGate  # type: ignore
+except Exception:
     AuthorityGate = None  # type: ignore
 
+
 # ---------------------------------------------------------------------
 # Core identifiers / constants
 # ---------------------------------------------------------------------
@@ -44,8 +48,9 @@ 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"
+
 
 # ---------------------------------------------------------------------
 # Helpers
@@ -53,14 +58,17 @@ DNA_STATE_FILE = "sovereign_cognitive_dna_state.json"
 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):
@@ -70,16 +78,20 @@ def _safe_json_load(path: str) -> Dict[str, Any]:
     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()]
@@ -87,9 +99,83 @@ def _norm_tags(data_tags: Any) -> List[str]:
         return [t.strip().lower() for t in data_tags.split(",") if t.strip()]
     return []
 
+
 def _risk_score(risk_level: str) -> int:
-    m = {"low": 1, "medium": 2, "high": 3, "critical": 4}
-    return m.get((risk_level or "medium").strip().lower(), 2)
+    mapping = {"low": 1, "medium": 2, "high": 3, "critical": 4}
+    return mapping.get((risk_level or "medium").strip().lower(), 2)
+
+
+# ---------------------------------------------------------------------
+# 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
+
 
 # ---------------------------------------------------------------------
 # 7-LAYER SOVEREIGN COGNITION (EMBEDDED)
@@ -103,7 +189,7 @@ class IntentForecastingEngine:
     ]
 
     def predict(self, payload: Dict[str, Any]) -> Dict[str, Any]:
-        notes = (payload.get("notes") or "")
+        notes = payload.get("notes") or ""
         tags = _norm_tags(payload.get("data_tags"))
         risk = (payload.get("risk_level") or "medium").lower()
 
@@ -113,7 +199,10 @@ class IntentForecastingEngine:
         score += _risk_score(risk) * 10
         signals.append(f"declared_risk={risk}")
 
-        high_value = {"pii", "secrets", "keys", "payments", "banking", "customer_chat", "production"}
+        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)
@@ -137,6 +226,7 @@ class IntentForecastingEngine:
             "ife_score": int(score),
         }
 
+
 class CognitiveDNAFingerprinting:
     # Layer 2
     def __init__(self, state_file: str = DNA_STATE_FILE):
@@ -157,7 +247,7 @@ class CognitiveDNAFingerprinting:
         }
 
         risk = _risk_score(payload.get("risk_level") or "medium")
-        note_len = float(len((payload.get("notes") or "")))
+        note_len = float(len(payload.get("notes") or ""))
         tag_count = float(len(_norm_tags(payload.get("data_tags"))))
 
         drift = 0.0
@@ -190,6 +280,7 @@ class CognitiveDNAFingerprinting:
             },
         }
 
+
 class DeceptiveRealityFabric:
     # Layer 3
     def simulate(self, payload: Dict[str, Any]) -> Dict[str, Any]:
@@ -201,6 +292,7 @@ class DeceptiveRealityFabric:
             "note": "Routed into simulated execution layer (demo).",
         }
 
+
 class CausalityLock:
     # Layer 4
     def validate(self, payload: Dict[str, Any]) -> Dict[str, Any]:
@@ -218,6 +310,7 @@ class CausalityLock:
             "policy": "min_justification_for_high_risk",
         }
 
+
 class EphemeralExecutionSurfaces:
     # Layer 5
     def spawn(self, ttl_seconds: int = 45) -> Dict[str, Any]:
@@ -226,9 +319,13 @@ class EphemeralExecutionSurfaces:
             "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)),
+            "expires_at": time.strftime(
+                "%Y-%m-%dT%H:%M:%SZ",
+                time.gmtime(time.time() + ttl_seconds)
+            ),
         }
 
+
 class CognitiveLoadDefense:
     # Layer 6 — default OFF to avoid slowing the system
     def apply(self, predicted_intent: bool, risk_level: str) -> Dict[str, Any]:
@@ -237,6 +334,7 @@ class CognitiveLoadDefense:
 
         risk = (risk_level or "medium").lower()
         delay_ms = 0
+
         if predicted_intent and risk in ("high", "critical"):
             delay_ms = 220
         elif predicted_intent:
@@ -247,6 +345,7 @@ class CognitiveLoadDefense:
 
         return {"delay_ms": int(delay_ms), "applied": delay_ms > 0}
 
+
 class UnverifiableTruthLayer:
     # Layer 7
     def seal(self, data: str, fingerprint: str = "") -> Dict[str, Any]:
@@ -259,6 +358,7 @@ class UnverifiableTruthLayer:
             "note": "Set SOVEREIGN_SEAL_KEY in HF Secrets to make seals externally verifiable.",
         }
 
+
 class SovereignCognitionLayer:
     def __init__(self):
         self.ife = IntentForecastingEngine()
@@ -275,7 +375,10 @@ class SovereignCognitionLayer:
         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")
+        cld = self.cld.apply(
+            ife["predicted_attack_intent"],
+            payload.get("risk_level") or "medium"
+        )
 
         reasons: List[str] = []
         action = "allow"
@@ -299,7 +402,10 @@ class SovereignCognitionLayer:
             ensure_ascii=False,
             sort_keys=True,
         )
-        sealed = self.truth.seal(seal_input, fingerprint=str(payload.get("fingerprint") or ""))
+        sealed = self.truth.seal(
+            seal_input,
+            fingerprint=str(payload.get("fingerprint") or "")
+        )
 
         return {
             "decision_id": str(uuid.uuid4()),
@@ -310,7 +416,9 @@ class SovereignCognitionLayer:
             "layers": {
                 "ife": ife,
                 "dna": dna,
-                "deception": self.deception.simulate(payload) if action == "deceive" else {"deception_engaged": False},
+                "deception": self.deception.simulate(payload) if action == "deceive" else {
+                    "deception_engaged": False
+                },
                 "causality": caus,
                 "ephemeral_surface": surface,
                 "cognitive_load_defense": cld,
@@ -318,8 +426,10 @@ class SovereignCognitionLayer:
             },
         }
 
+
 COGNITION = SovereignCognitionLayer()
 
+
 # ---------------------------------------------------------------------
 # Minimal core logic (self-contained for the demo Space)
 # ---------------------------------------------------------------------
@@ -343,6 +453,7 @@ class SovereignFingerprint:
         _safe_json_dump(self.output, payload)
         return payload
 
+
 class SovereignLineage:
     def __init__(self, output: str = LINEAGE_FILE):
         self.output = output
@@ -359,6 +470,7 @@ class SovereignLineage:
     ) -> Dict[str, Any]:
         issued_at = _utc_now_iso()
         record_id = str(uuid.uuid4())
+
         base = {
             "record_id": record_id,
             "issued_at": issued_at,
@@ -372,24 +484,32 @@ class SovereignLineage:
             "version": SOVEREIGN_VERSION,
             "authority": AUTHORITY_NAME,
         }
-        integrity = _sha256_hex(json.dumps(base, sort_keys=True, ensure_ascii=False))
+
+        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))
+        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
 # ---------------------------------------------------------------------
@@ -416,14 +536,17 @@ def log_audit_event(
         "timestamp": ts,
         "engine": engine_name,
         "event_type": event_type,
-        "outcome": outcome,  # ALLOW/FREEZE/BLOCK (or legacy if used elsewhere)
+        "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)),
+        "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 {},
@@ -435,14 +558,18 @@ def log_audit_event(
 
     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)) :]
+
+        tail = lines[-max(1, int(limit)):]
         out: List[Dict[str, Any]] = []
+
         for ln in tail:
             ln = ln.strip()
             if not ln:
@@ -451,10 +578,12 @@ def read_audit_log_tail(limit: int = 50) -> List[Dict[str, Any]]:
                 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)
@@ -473,12 +602,30 @@ def generate_conformance_report() -> Tuple[Dict[str, Any], str]:
             "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"},
+            {
+                "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.",
     }
@@ -486,22 +633,36 @@ def generate_conformance_report() -> Tuple[Dict[str, Any], str]:
     _safe_json_dump(CONFORMANCE_REPORT_FILE, report)
     return report, CONFORMANCE_REPORT_FILE
 
+
 # ---------------------------------------------------------------------
-# Authority Gate instance (new)
+# Authority Gate instance
 # ---------------------------------------------------------------------
 GATE = None
 if AuthorityGate:
-    GATE = AuthorityGate(
-        engine_name=ENGINE_NAME,
-        version=SOVEREIGN_VERSION,
-        authority_name=AUTHORITY_NAME,
-        audit_log_file=AUDIT_LOG_FILE,
-    )
+    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):
+def run_sentinel(
+    engine_name,
+    parent_model,
+    model_version,
+    data_tags,
+    risk_level,
+    notes,
+    access_key,
+    delegation_token,
+):
     engine_name = engine_name or ENGINE_NAME
 
     payload = {
@@ -514,11 +675,16 @@ def run_sentinel(engine_name, parent_model, model_version, data_tags, risk_level
         "access_key_present": bool(access_key),
     }
 
-    # If gate module missing, fall back to cognition-only (should not happen if you add the file)
+    # Fallback: cognition-only if gate module is missing
     if not GATE:
         decision = COGNITION.evaluate(payload)
         outcome = decision["action"]
-        merged_notes = f"{notes}\n\n[SOVEREIGN_DECISION]\n{json.dumps(decision, ensure_ascii=False)}"
+
+        merged_notes = (
+            f"{notes}\n\n[SOVEREIGN_DECISION]\n"
+            f"{json.dumps(decision, ensure_ascii=False)}"
+        )
+
         event = log_audit_event(
             engine_name=engine_name,
             parent_model=parent_model,
@@ -529,22 +695,43 @@ def run_sentinel(engine_name, parent_model, model_version, data_tags, risk_level
             event_type="sentinel_run",
             outcome=outcome,
             access_key=access_key,
-            authority_bundle={"mode": "cognition_only", "decision": outcome, "reason": "gate_missing"},
-            execution={"executed": False, "detail": {"note": "gate missing, execution disabled"}},
+            authority_bundle={
+                "mode": "cognition_only",
+                "decision": outcome,
+                "reason": "gate_missing",
+            },
+            execution={
+                "executed": False,
+                "detail": {"note": "gate missing, execution disabled"}
+            },
         )
-        return json.dumps({"audit_event": event, "sovereign_decision": decision}, indent=2, ensure_ascii=False)
 
-    # ✅ 6) Unilateral Freeze Doctrine: authority gate runs BEFORE execution
-    gate = GATE.precheck(payload, cognition_eval=COGNITION.evaluate, delegation_token=delegation_token or "")
+        return json.dumps(
+            {"audit_event": event, "sovereign_decision": decision},
+            indent=2,
+            ensure_ascii=False,
+        )
+
+    # Unilateral Freeze Doctrine: authority gate runs BEFORE execution
+    gate = GATE.precheck(
+        payload,
+        cognition_eval=COGNITION.evaluate,
+        delegation_token=delegation_token or "",
+    )
     authority_decision = gate["decision"]  # ALLOW/FREEZE/BLOCK
 
-    execution_info = {"executed": False, "detail": None}
+    execution_info: Dict[str, Any] = {"executed": False, "detail": None}
     if authority_decision == "ALLOW":
-        # Only now we allow any "execution"
         try:
-            execution_info = {"executed": True, "detail": run_pilot(payload)}
+            execution_info = {
+                "executed": True,
+                "detail": run_pilot(payload)
+            }
         except Exception as e:
-            execution_info = {"executed": False, "detail": {"error": str(e)}}
+            execution_info = {
+                "executed": False,
+                "detail": {"error": str(e)}
+            }
 
     authority_bundle = {
         "decision": authority_decision,
@@ -556,7 +743,11 @@ def run_sentinel(engine_name, parent_model, model_version, data_tags, risk_level
         "cognition": gate.get("cognition"),
     }
 
-    notes_short = f"{notes}\n\n[SOVEREIGN_AUTHORITY]\n{authority_decision} | {gate.get('reason')}"
+    notes_short = (
+        f"{notes}\n\n[SOVEREIGN_AUTHORITY]\n"
+        f"{authority_decision} | {gate.get('reason')}"
+    )
+
     event = log_audit_event(
         engine_name=engine_name,
         parent_model=parent_model,
@@ -572,12 +763,24 @@ def run_sentinel(engine_name, parent_model, model_version, data_tags, risk_level
     )
 
     return json.dumps(
-        {"audit_event": event, "authority_gate": authority_bundle, "execution": execution_info},
+        {
+            "audit_event": event,
+            "authority_gate": authority_bundle,
+            "execution": execution_info,
+        },
         indent=2,
-        ensure_ascii=False
+        ensure_ascii=False,
     )
 
-def run_fingerprint_and_lineage(engine_name, parent_model, model_version, data_tags, risk_level, notes):
+
+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()
@@ -599,18 +802,22 @@ def run_fingerprint_and_lineage(engine_name, parent_model, model_version, data_t
     }
     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
 
+
 # ---------------------------------------------------------------------
 # Gradio UI
 # ---------------------------------------------------------------------
@@ -647,34 +854,80 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
             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.")
+                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")
+                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)
+                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():
-                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)
+                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",
             )
 
-            run_btn = gr.Button("Run Sentinel (Authority + 7-layer) & Log Event", variant="primary")
-            result_json = gr.Code(label="Result (Audit + Authority + Execution)", language="json")
+            run_btn = gr.Button(
+                "Run Sentinel (Authority + 7-layer) & Log Event",
+                variant="primary"
+            )
+            result_json = gr.Code(
+                label="Result (Audit + Authority + Execution)",
+                language="json"
+            )
 
             run_btn.click(
                 fn=run_sentinel,
-                inputs=[engine_name, parent_model, model_version, data_tags, risk_level, notes, access_key, delegation_token],
+                inputs=[
+                    engine_name,
+                    parent_model,
+                    model_version,
+                    data_tags,
+                    risk_level,
+                    notes,
+                    access_key,
+                    delegation_token,
+                ],
                 outputs=result_json,
             )
 
@@ -683,23 +936,55 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
             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.")
+                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")
+                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_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],
+                inputs=[
+                    fp_engine_name,
+                    fp_parent_model,
+                    fp_model_version,
+                    fp_data_tags,
+                    fp_risk_level,
+                    fp_notes,
+                ],
                 outputs=fp_output,
             )
 
@@ -707,19 +992,43 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
         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")
+            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)
+            show_log_btn.click(
+                fn=show_audit_log,
+                inputs=log_limit,
+                outputs=log_view
+            )
 
         # Conformance Report tab
         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 = 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,
@@ -727,34 +1036,36 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
                 outputs=[report_json_out, report_file_out],
             )
 
-if __name__ == "__main__":
-    demo.launch()
-
-
-from sovereign_latency_benchmark import run_latency_benchmark, calculate_metrics, save_report
-
-timings = run_latency_benchmark(1000)
-metrics = calculate_metrics(timings)
-save_report(metrics)
-print("LATENCY:", metrics)
-
-print("STARTING LATENCY TEST...")
-
-from sovereign_latency_benchmark import run_latency_benchmark, calculate_metrics, save_report
-
-timings = run_latency_benchmark(1000)
-metrics = calculate_metrics(timings)
-save_report(metrics)
+        # Latency tab
+        with gr.Tab("Latency Benchmark"):
+            gr.Markdown("### Run a clean built-in latency benchmark and save a JSON report.")
 
-print("LATENCY:", metrics)
-
-print("STARTING LATENCY TEST...")
+            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"
+            )
 
-from sovereign_latency_benchmark import run_latency_benchmark, calculate_metrics, save_report
+            latency_btn.click(
+                fn=run_latency_ui,
+                inputs=latency_samples,
+                outputs=[latency_json, latency_file],
+            )
 
-timings = run_latency_benchmark(1000)
-metrics = calculate_metrics(timings)
-save_report(metrics)
 
-import run_latency
+if __name__ == "__main__":
+    demo.launch()