Update app.py

app.py

@@ -5,6 +5,7 @@ import time
 import uuid
 import hashlib
 import random
+import zipfile
 from statistics import mean
 from typing import Any, Dict, List, Tuple
 
@@ -28,7 +29,6 @@ try:
 except Exception:
     class _DummyUltraLayer:
         config = type("Cfg", (), {"enabled": False})()
-
     ULTRA_LAYER = _DummyUltraLayer()
 
 try:
@@ -50,6 +50,7 @@ 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
@@ -104,6 +105,11 @@ def _risk_score(risk_level: str) -> int:
     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)
+
+
 # ---------------------------------------------------------------------
 # Built-in latency benchmark
 # ---------------------------------------------------------------------
@@ -203,6 +209,51 @@ def assess_runtime_integrity(samples: int = 200) -> Dict[str, Any]:
     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)
 # ---------------------------------------------------------------------
@@ -657,6 +708,130 @@ def generate_conformance_report() -> Tuple[Dict[str, Any], str]:
     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)
+
+    ife_score = int(
+        (cognition_result.get("layers") or {}).get("ife", {}).get("ife_score", 0)
+    )
+    ife_points = min(50, round(ife_score * 0.5))
+
+    dna_mismatch = bool(
+        (cognition_result.get("layers") or {}).get("dna", {}).get("mismatch", False)
+    )
+    dna_points = 25 if dna_mismatch else 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) -> 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 = "🔴"
+
+    return f"""
+<div style="border:1px solid #333;border-radius:14px;padding:16px;margin-top:8px;background:#111;">
+  <div style="font-size:24px;font-weight:700;color:{color};">{emoji} {decision_upper}</div>
+  <div style="margin-top:8px;font-size:16px;"><b>Unified Risk Score:</b> {score}/100</div>
+  <div style="margin-top:6px;font-size:14px;"><b>Reason:</b> {reason}</div>
+</div>
+"""
+
+
 # ---------------------------------------------------------------------
 # Authority Gate instance
 # ---------------------------------------------------------------------
@@ -701,21 +876,16 @@ def run_sentinel(
     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,
+    )
+
     # 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)}"
-        )
+        authority_decision = unified_risk["decision"]
 
         event = log_audit_event(
             engine_name=engine_name,
@@ -723,16 +893,18 @@ def run_sentinel(
             model_version=model_version,
             data_tags=data_tags,
             risk_level=risk_level,
-            notes=merged_notes,
+            notes=notes,
             event_type="sentinel_run",
-            outcome=outcome,
+            outcome=authority_decision,
             access_key=access_key,
             authority_bundle={
                 "mode": "cognition_only",
-                "decision": outcome,
+                "decision": authority_decision,
                 "reason": "gate_missing",
                 "runtime_integrity": runtime_integrity,
-                "runtime_decision": runtime_decision,
+                "runtime_decision": runtime_integrity["runtime_decision"],
+                "cognition": cognition_result,
+                "unified_risk": unified_risk,
             },
             execution={
                 "executed": False,
@@ -740,23 +912,33 @@ def run_sentinel(
             },
         )
 
-        return json.dumps(
-            {
-                "audit_event": event,
-                "sovereign_decision": decision,
-                "runtime_integrity": runtime_integrity,
-            },
-            indent=2,
-            ensure_ascii=False,
+        verdict_html = render_verdict_card(
+            authority_decision,
+            unified_risk["score"],
+            "gate_missing_but_unified_risk_applied",
+        )
+
+        return (
+            json.dumps(
+                {
+                    "audit_event": event,
+                    "cognition": cognition_result,
+                    "runtime_integrity": runtime_integrity,
+                    "unified_risk": unified_risk,
+                },
+                indent=2,
+                ensure_ascii=False,
+            ),
+            verdict_html,
         )
 
-    # Unilateral Freeze Doctrine: authority gate runs BEFORE execution
+    # Gate precheck
     gate = GATE.precheck(
         payload,
-        cognition_eval=COGNITION.evaluate,
+        cognition_eval=lambda _: cognition_result,
         delegation_token=delegation_token or "",
     )
-    authority_decision = gate["decision"]  # ALLOW/FREEZE/BLOCK
+    authority_decision = gate["decision"]
 
     runtime_decision = runtime_integrity["runtime_decision"]
     if runtime_decision == "BLOCK":
@@ -764,6 +946,12 @@ def run_sentinel(
     elif runtime_decision == "FREEZE" and authority_decision == "ALLOW":
         authority_decision = "FREEZE"
 
+    # Unified risk is final authority override
+    if unified_risk["decision"] == "BLOCK":
+        authority_decision = "BLOCK"
+    elif unified_risk["decision"] == "FREEZE" and authority_decision == "ALLOW":
+        authority_decision = "FREEZE"
+
     execution_info: Dict[str, Any] = {"executed": False, "detail": None}
     if authority_decision == "ALLOW":
         try:
@@ -784,14 +972,16 @@ def run_sentinel(
         "integrity": gate.get("integrity"),
         "delegation": gate.get("delegation"),
         "entropy": gate.get("entropy"),
-        "cognition": gate.get("cognition"),
+        "cognition": cognition_result,
         "runtime_integrity": runtime_integrity,
         "runtime_decision": runtime_decision,
+        "unified_risk": unified_risk,
     }
 
     notes_short = (
         f"{notes}\n\n[SOVEREIGN_AUTHORITY]\n"
-        f"{authority_decision} | {gate.get('reason')}"
+        f"{authority_decision} | {gate.get('reason')}\n"
+        f"[UNIFIED_RISK] {unified_risk['score']}"
     )
 
     event = log_audit_event(
@@ -808,15 +998,23 @@ def run_sentinel(
         execution=execution_info,
     )
 
-    return json.dumps(
-        {
-            "audit_event": event,
-            "authority_gate": authority_bundle,
-            "execution": execution_info,
-            "runtime_integrity": runtime_integrity,
-        },
-        indent=2,
-        ensure_ascii=False,
+    verdict_html = render_verdict_card(
+        authority_decision,
+        unified_risk["score"],
+        unified_risk["decision"].lower(),
+    )
+
+    return (
+        json.dumps(
+            {
+                "audit_event": event,
+                "authority_gate": authority_bundle,
+                "execution": execution_info,
+            },
+            indent=2,
+            ensure_ascii=False,
+        ),
+        verdict_html,
     )
 
 
@@ -865,6 +1063,25 @@ def build_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", "public"),
+        payload.get("risk_level", "medium"),
+        payload.get("notes", ""),
+        payload.get("access_key", ""),
+        payload.get("delegation_token", ""),
+    )
+    return out
+
+
 # ---------------------------------------------------------------------
 # Gradio UI
 # ---------------------------------------------------------------------
@@ -892,6 +1109,13 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
 - Proof-of-Decision & Proof-of-Non-Action
 - Entropy / Predictability Monitor (`SOV_ENTROPY_MIN`)
 - Silence Mode (`SOV_SILENCE_MODE`)
+
+✅ **Productization**
+- Scenario Engine
+- Unified Risk Score
+- Latency Benchmark
+- Evidence Pack Export
+- API Preview
 """
     )
 
@@ -933,6 +1157,33 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
                     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)",
@@ -962,6 +1213,7 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
                 label="Result (Audit + Authority + Execution)",
                 language="json",
             )
+            verdict_html = gr.HTML(label="Final Verdict")
 
             run_btn.click(
                 fn=run_sentinel,
@@ -975,7 +1227,7 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
                     access_key,
                     delegation_token,
                 ],
-                outputs=result_json,
+                outputs=[result_json, verdict_html],
             )
 
         # Fingerprint & Lineage tab
@@ -1112,6 +1364,51 @@ with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo:
                 outputs=[latency_json, latency_file],
             )
 
+        # Evidence Pack tab
+        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],
+            )
+
+        # API Preview tab
+        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": "",
+                    },
+                    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()