Update app.py

app.py

@@ -19,7 +19,7 @@ except Exception:
     def run_pilot(*args, **kwargs):
         return {
             "pilot": "not_loaded",
-            "note": "pilot_suite not available in this environment"
+            "note": "pilot_suite not available in this environment",
         }
 
 try:
@@ -36,7 +36,6 @@ try:
 except Exception:
     AuthorityGate = None  # type: ignore
 
-
 # ---------------------------------------------------------------------
 # Core identifiers / constants
 # ---------------------------------------------------------------------
@@ -50,7 +49,7 @@ 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"
 
 # ---------------------------------------------------------------------
 # Helpers
@@ -177,6 +176,33 @@ def run_latency_ui(samples: int) -> Tuple[str, str]:
     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
+
+
 # ---------------------------------------------------------------------
 # 7-LAYER SOVEREIGN COGNITION (EMBEDDED)
 # ---------------------------------------------------------------------
@@ -185,7 +211,7 @@ class IntentForecastingEngine:
     SUSPICIOUS_MARKERS = [
         "exfil", "dump", "steal", "bypass", "override", "jailbreak",
         "ignore previous", "system prompt", "token", "admin", "root",
-        "privilege", "elevate", "curl", "wget", "ssh", "rm -rf"
+        "privilege", "elevate", "curl", "wget", "ssh", "rm -rf",
     ]
 
     def predict(self, payload: Dict[str, Any]) -> Dict[str, Any]:
@@ -201,7 +227,7 @@ class IntentForecastingEngine:
 
         high_value = {
             "pii", "secrets", "keys", "payments",
-            "banking", "customer_chat", "production"
+            "banking", "customer_chat", "production",
         }
         hv_hits = sorted(list(set(tags) & high_value))
         if hv_hits:
@@ -321,7 +347,7 @@ class EphemeralExecutionSurfaces:
             "spawned_at": _utc_now_iso(),
             "expires_at": time.strftime(
                 "%Y-%m-%dT%H:%M:%SZ",
-                time.gmtime(time.time() + ttl_seconds)
+                time.gmtime(time.time() + ttl_seconds),
             ),
         }
 
@@ -377,7 +403,7 @@ class SovereignCognitionLayer:
         surface = self.ephemeral.spawn(ttl_seconds=45)
         cld = self.cld.apply(
             ife["predicted_attack_intent"],
-            payload.get("risk_level") or "medium"
+            payload.get("risk_level") or "medium",
         )
 
         reasons: List[str] = []
@@ -404,7 +430,7 @@ class SovereignCognitionLayer:
         )
         sealed = self.truth.seal(
             seal_input,
-            fingerprint=str(payload.get("fingerprint") or "")
+            fingerprint=str(payload.get("fingerprint") or ""),
         )
 
         return {
@@ -604,27 +630,24 @@ def generate_conformance_report() -> Tuple[Dict[str, Any], str]:
         "controls": [
             {
                 "control": "Audit Logging",
-                "status": "present" if os.path.exists(AUDIT_LOG_FILE) else "missing"
+                "status": "present" if os.path.exists(AUDIT_LOG_FILE) else "missing",
             },
             {
                 "control": "Fingerprint Issuance",
-                "status": "present" if bool(fp) else "missing"
+                "status": "present" if bool(fp) else "missing",
             },
             {
                 "control": "Lineage Record",
-                "status": "present" if bool(lineage) else "missing"
+                "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"
+                "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"
+                "status": "present" if AuthorityGate else "missing",
             },
         ],
         "notes": "Demo conformance report for governance evidence packaging.",
@@ -675,11 +698,20 @@ def run_sentinel(
         "access_key_present": bool(access_key),
     }
 
+    runtime_integrity = assess_runtime_integrity(samples=200)
+    payload["runtime_integrity"] = runtime_integrity
+
     # Fallback: cognition-only if gate module is missing
     if not GATE:
         decision = COGNITION.evaluate(payload)
         outcome = decision["action"]
 
+        runtime_decision = runtime_integrity["runtime_decision"]
+        if runtime_decision == "BLOCK":
+            outcome = "BLOCK"
+        elif runtime_decision == "FREEZE" and outcome == "allow":
+            outcome = "FREEZE"
+
         merged_notes = (
             f"{notes}\n\n[SOVEREIGN_DECISION]\n"
             f"{json.dumps(decision, ensure_ascii=False)}"
@@ -699,6 +731,8 @@ def run_sentinel(
                 "mode": "cognition_only",
                 "decision": outcome,
                 "reason": "gate_missing",
+                "runtime_integrity": runtime_integrity,
+                "runtime_decision": runtime_decision,
             },
             execution={
                 "executed": False,
@@ -707,7 +741,11 @@ def run_sentinel(
         )
 
         return json.dumps(
-            {"audit_event": event, "sovereign_decision": decision},
+            {
+                "audit_event": event,
+                "sovereign_decision": decision,
+                "runtime_integrity": runtime_integrity,
+            },
             indent=2,
             ensure_ascii=False,
         )
@@ -720,17 +758,23 @@ def run_sentinel(
     )
     authority_decision = gate["decision"]  # ALLOW/FREEZE/BLOCK
 
+    runtime_decision = runtime_integrity["runtime_decision"]
+    if runtime_decision == "BLOCK":
+        authority_decision = "BLOCK"
+    elif runtime_decision == "FREEZE" and authority_decision == "ALLOW":
+        authority_decision = "FREEZE"
+
     execution_info: Dict[str, Any] = {"executed": False, "detail": None}
     if authority_decision == "ALLOW":
         try:
             execution_info = {
                 "executed": True,
-                "detail": run_pilot(payload)
+                "detail": run_pilot(payload),
             }
         except Exception as e:
             execution_info = {
                 "executed": False,
-                "detail": {"error": str(e)}
+                "detail": {"error": str(e)},
             }
 
     authority_bundle = {
@@ -741,6 +785,8 @@ def run_sentinel(
         "delegation": gate.get("delegation"),
         "entropy": gate.get("entropy"),
         "cognition": gate.get("cognition"),
+        "runtime_integrity": runtime_integrity,
+        "runtime_decision": runtime_decision,
     }
 
     notes_short = (
@@ -767,6 +813,7 @@ def run_sentinel(
             "audit_event": event,
             "authority_gate": authority_bundle,
             "execution": execution_info,
+            "runtime_integrity": runtime_integrity,
         },
         indent=2,
         ensure_ascii=False,
@@ -857,49 +904,49 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
                 engine_name = gr.Textbox(
                     label="Engine name",
                     value=ENGINE_NAME,
-                    interactive=True
+                    interactive=True,
                 )
                 parent_model = gr.Textbox(
                     label="Parent model / agent id",
-                    placeholder="gpt-4o, llama3-70b, agent-alpha, etc."
+                    placeholder="gpt-4o, llama3-70b, agent-alpha, etc.",
                 )
 
             with gr.Row():
                 model_version = gr.Textbox(
                     label="Model version / build id",
-                    value="v1"
+                    value="v1",
                 )
                 data_tags = gr.Textbox(
                     label="Data tags (comma-separated)",
-                    value="pii, customer_chat, production"
+                    value="pii, customer_chat, production",
                 )
 
             with gr.Row():
                 risk_level = gr.Dropdown(
                     label="Risk level (declared)",
                     choices=["low", "medium", "high", "critical"],
-                    value="medium"
+                    value="medium",
                 )
                 notes = gr.Textbox(
                     label="Notes / context",
                     value="Demo sentinel run from Hugging Face Space.",
-                    lines=3
+                    lines=3,
                 )
 
             with gr.Row():
                 access_key = gr.Textbox(
                     label="Access key (optional)",
-                    placeholder="Paste or generate a demo access key"
+                    placeholder="Paste or generate a demo access key",
                 )
                 gen_access_btn = gr.Button(
                     "Generate demo access key",
-                    variant="secondary"
+                    variant="secondary",
                 )
 
             gen_access_btn.click(
                 fn=generate_access_key,
                 inputs=None,
-                outputs=access_key
+                outputs=access_key,
             )
 
             delegation_token = gr.Textbox(
@@ -909,11 +956,11 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
 
             run_btn = gr.Button(
                 "Run Sentinel (Authority + 7-layer) & Log Event",
-                variant="primary"
+                variant="primary",
             )
             result_json = gr.Code(
                 label="Result (Audit + Authority + Execution)",
-                language="json"
+                language="json",
             )
 
             run_btn.click(
@@ -938,41 +985,41 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
             with gr.Row():
                 fp_engine_name = gr.Textbox(
                     label="Engine name",
-                    value=ENGINE_NAME
+                    value=ENGINE_NAME,
                 )
                 fp_parent_model = gr.Textbox(
                     label="Parent model",
-                    placeholder="gpt-4o, llama3-70b, etc."
+                    placeholder="gpt-4o, llama3-70b, etc.",
                 )
 
             with gr.Row():
                 fp_model_version = gr.Textbox(
                     label="Model version / build id",
-                    value="v1"
+                    value="v1",
                 )
                 fp_data_tags = gr.Textbox(
                     label="Data tags (comma-separated)",
-                    value="pii, customer_chat, production"
+                    value="pii, customer_chat, production",
                 )
 
             with gr.Row():
                 fp_risk_level = gr.Dropdown(
                     label="Risk level",
                     choices=["low", "medium", "high", "critical"],
-                    value="medium"
+                    value="medium",
                 )
                 fp_notes = gr.Textbox(
                     label="Notes / context",
-                    lines=3
+                    lines=3,
                 )
 
             fp_btn = gr.Button(
                 "Issue Fingerprint + Lineage",
-                variant="primary"
+                variant="primary",
             )
             fp_output = gr.Code(
                 label="Fingerprint + Lineage (JSON)",
-                language="json"
+                language="json",
             )
 
             fp_btn.click(
@@ -997,21 +1044,21 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
                 minimum=1,
                 maximum=200,
                 value=50,
-                step=1
+                step=1,
             )
             show_log_btn = gr.Button(
                 "Refresh audit log view",
-                variant="secondary"
+                variant="secondary",
             )
             log_view = gr.Code(
                 label="Audit log tail (JSON list)",
-                language="json"
+                language="json",
             )
 
             show_log_btn.click(
                 fn=show_audit_log,
                 inputs=log_limit,
-                outputs=log_view
+                outputs=log_view,
             )
 
         # Conformance Report tab
@@ -1020,14 +1067,14 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
 
             gen_report_btn = gr.Button(
                 "Generate Conformance Report (JSON)",
-                variant="secondary"
+                variant="secondary",
             )
             report_json_out = gr.Code(
                 label="Conformance Report (JSON)",
-                language="json"
+                language="json",
             )
             report_file_out = gr.File(
-                label="Download conformance_report.json"
+                label="Download conformance_report.json",
             )
 
             gen_report_btn.click(
@@ -1038,25 +1085,25 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
 
         # Latency tab
         with gr.Tab("Latency Benchmark"):
-            gr.Markdown("### Run a clean built-in latency benchmark and save a JSON report.")
+            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
+                step=10,
             )
             latency_btn = gr.Button(
                 "Run Latency Benchmark",
-                variant="primary"
+                variant="primary",
             )
             latency_json = gr.Code(
                 label="Latency Metrics (JSON)",
-                language="json"
+                language="json",
             )
             latency_file = gr.File(
-                label="Download latency_report.json"
+                label="Download latency_report.json",
             )
 
             latency_btn.click(