app.py

import os
import json
import time
import uuid
import hashlib
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()

# ---------------------------------------------------------------------
# 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"

# ---------------------------------------------------------------------
# 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:
    m = {"low": 1, "medium": 2, "high": 3, "critical": 4}
    return m.get((risk_level or "medium").strip().lower(), 2)

# ---------------------------------------------------------------------
# 7-LAYER SOVEREIGN COGNITION (EMBEDDED)
# ---------------------------------------------------------------------
class IntentForecastingEngine:
    # Layer 1
    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:
    # Layer 2
    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

        mismatch = drift >= 1.35  # demo threshold

        n = int(rec["n"]) + 1
        rec["n"] = n
        rec["avg_risk"] = (rec["avg_risk"] * (n - 1) + risk) / n
        rec["avg_note_len"] = (rec["avg_note_len"] * (n - 1) + note_len) / n
        rec["avg_tag_count"] = (rec["avg_tag_count"] * (n - 1) + 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": rec["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:
    # Layer 3
    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:
    # Layer 4
    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:
    # Layer 5
    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:
    # Layer 6
    def apply(self, predicted_intent: bool, risk_level: str) -> Dict[str, Any]:
        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:
    # Layer 7
    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,  # allow|block|freeze|deceive
            "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 (self-contained for the demo Space)
# ---------------------------------------------------------------------
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]

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,
) -> 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": AUTHORITY_NAME,
    }

    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"},
        ],
        "notes": "Demo conformance report for governance evidence packaging.",
    }

    _safe_json_dump(CONFORMANCE_REPORT_FILE, report)
    return report, CONFORMANCE_REPORT_FILE

# ---------------------------------------------------------------------
# UI-bound functions
# ---------------------------------------------------------------------
def run_sentinel(engine_name, parent_model, model_version, data_tags, risk_level, notes, access_key):
    if not engine_name:
        engine_name = ENGINE_NAME

    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),
    }
    decision = COGNITION.evaluate(payload)
    outcome = decision["action"]

    merged_notes = f"{notes}\n\n[SOVEREIGN_DECISION]\n{json.dumps(decision, ensure_ascii=False)}"

    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=merged_notes,
        event_type="sentinel_run",
        outcome=outcome,
        access_key=access_key,
    )

    return json.dumps({"audit_event": event, "sovereign_decision": decision}, indent=2, ensure_ascii=False)

def run_fingerprint_and_lineage(engine_name, parent_model, model_version, data_tags, risk_level, notes):
    if not engine_name:
        engine_name = 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

# ---------------------------------------------------------------------
# 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) 
- Deceptive Reality (High/Critical + Intent → Deceive) 
- Causality Lock (High/Critical needs justification) 
- Ephemeral Surfaces (TTL token) 
- Cognitive Load Defense (micro delay) 
- Unverifiable Truth Seal (`SOVEREIGN_SEAL_KEY`) 
"""
    )

    with gr.Tabs():
        # Sentinel Run tab
        with gr.Tab("Sentinel Run"):
            gr.Markdown("### Run Sovereign Sentinel and log a monitoring event (7-layer active).")

            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():
                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)

            run_btn = gr.Button("Run Sentinel (7-layer) & Log Event", variant="primary")
            result_json = gr.Code(label="Result (Audit + Decision JSON)", language="json")

            run_btn.click(
                fn=run_sentinel,
                inputs=[engine_name, parent_model, model_version, data_tags, risk_level, notes, access_key],
                outputs=result_json,
            )

        # Fingerprint & Lineage tab
        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,
            )

        # Audit Log tab
        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)

        # 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.click(
                fn=build_conformance_report,
                inputs=None,
                outputs=[report_json_out, report_file_out],
            )

if __name__ == "__main__":
    demo.launch()