Update app.py

app.py

@@ -1,20 +1,126 @@
 import gradio as gr
-import asyncio
 import json
 import logging
-import traceback
-import os
 import numpy as np
 import pandas as pd
 from datetime import datetime
-from typing import Dict, Any, List, Optional
+from typing import Dict, Any, List, Optional, Tuple
 import threading
-import urllib.request
 import time
+import os
+import sqlite3
+import contextlib
 from scipy.stats import beta
+import plotly.graph_objects as go
+
+# ----------------------------------------------------------------------
+# Configuration from environment variables
+# ----------------------------------------------------------------------
+LOW_THRESHOLD = float(os.getenv("ARF_LOW_THRESHOLD", "0.2"))
+HIGH_THRESHOLD = float(os.getenv("ARF_HIGH_THRESHOLD", "0.8"))
+ALPHA_PRIOR = float(os.getenv("ARF_ALPHA_PRIOR", "1.0"))
+BETA_PRIOR = float(os.getenv("ARF_BETA_PRIOR", "1.0"))
+DB_PATH = os.getenv("ARF_DB_PATH", "/data/arf_decisions.db")
 
 # ----------------------------------------------------------------------
-# Memory monitoring (no external dependencies)
+# Logging
+# ----------------------------------------------------------------------
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+# ----------------------------------------------------------------------
+# SQLite persistence
+# ----------------------------------------------------------------------
+def init_db():
+    """Create the decisions table if it doesn't exist."""
+    with contextlib.closing(sqlite3.connect(DB_PATH)) as conn:
+        cursor = conn.cursor()
+        cursor.execute('''
+            CREATE TABLE IF NOT EXISTS decisions (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                timestamp TEXT NOT NULL,
+                decision_json TEXT NOT NULL,
+                risk REAL NOT NULL
+            )
+        ''')
+        conn.commit()
+    logger.info(f"Database initialized at {DB_PATH}")
+
+def save_decision_to_db(decision: dict, risk: float):
+    """Insert a decision into the database."""
+    try:
+        with contextlib.closing(sqlite3.connect(DB_PATH)) as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                "INSERT INTO decisions (timestamp, decision_json, risk) VALUES (?, ?, ?)",
+                (decision["timestamp"], json.dumps(decision), risk)
+            )
+            conn.commit()
+    except Exception as e:
+        logger.error(f"Failed to save decision to DB: {e}")
+
+def load_recent_decisions(limit: int = 100) -> List[Tuple[str, dict, float]]:
+    """Load the most recent decisions from the database."""
+    decisions = []
+    try:
+        with contextlib.closing(sqlite3.connect(DB_PATH)) as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                "SELECT timestamp, decision_json, risk FROM decisions ORDER BY timestamp DESC LIMIT ?",
+                (limit,)
+            )
+            rows = cursor.fetchall()
+            for ts, json_str, risk in rows:
+                decisions.append((ts, json.loads(json_str), risk))
+        decisions.reverse()  # oldest first
+    except Exception as e:
+        logger.error(f"Failed to load decisions from DB: {e}")
+    return decisions
+
+def vacuum_db():
+    """Run VACUUM on the database (periodic maintenance)."""
+    try:
+        with contextlib.closing(sqlite3.connect(DB_PATH)) as conn:
+            conn.execute("VACUUM")
+        logger.info("Database vacuumed")
+    except Exception as e:
+        logger.error(f"Vacuum failed: {e}")
+
+# ----------------------------------------------------------------------
+# Thread‑safe history (in‑memory + DB backup)
+# ----------------------------------------------------------------------
+decision_history = []
+risk_history = []
+history_lock = threading.Lock()
+
+def update_dashboard_data(decision: dict, risk: float):
+    """Thread‑safe update of both in‑memory history and database."""
+    with history_lock:
+        decision_history.append((datetime.utcnow().isoformat(), decision, risk))
+        risk_history.append((datetime.utcnow().isoformat(), risk))
+        # Keep only last 100 in memory
+        if len(decision_history) > 100:
+            decision_history.pop(0)
+        if len(risk_history) > 100:
+            risk_history.pop(0)
+    save_decision_to_db(decision, risk)
+
+def refresh_history_from_db():
+    """Load recent history from database (called at startup)."""
+    global decision_history, risk_history
+    decisions = load_recent_decisions(100)
+    with history_lock:
+        decision_history.clear()
+        risk_history.clear()
+        for ts, dec, risk in decisions:
+            decision_history.append((ts, dec, risk))
+            risk_history.append((ts, risk))
+
+# ----------------------------------------------------------------------
+# Memory monitoring (daemon thread)
 # ----------------------------------------------------------------------
 def get_memory_usage():
     """Return current process memory usage in MB (RSS)."""
@@ -37,30 +143,24 @@ def get_memory_usage():
             pass
     return None
 
-def log_memory_usage():
-    mem_mb = get_memory_usage()
-    if mem_mb is not None:
-        logging.info(f"Process memory: {mem_mb:.1f} MB")
-    else:
-        logging.info("Process memory: unknown")
-    threading.Timer(60, log_memory_usage).start()
-
-# ----------------------------------------------------------------------
-# Plotly
-# ----------------------------------------------------------------------
-import plotly.graph_objects as go
-
-# ----------------------------------------------------------------------
-# Logging
-# ----------------------------------------------------------------------
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
-logger = logging.getLogger(__name__)
+def memory_monitor_loop():
+    """Periodically log memory usage. Runs in a daemon thread."""
+    while True:
+        try:
+            mem_mb = get_memory_usage()
+            if mem_mb is not None:
+                logger.info(f"Process memory: {mem_mb:.1f} MB")
+            else:
+                logger.info("Process memory: unknown")
+        except Exception as e:
+            logger.error(f"Memory logging error: {e}")
+        time.sleep(60)
 
 # ----------------------------------------------------------------------
 # Bayesian Risk Engine (Beta‑Binomial)
 # ----------------------------------------------------------------------
 class BayesianRiskEngine:
-    def __init__(self, alpha=1.0, beta=1.0):
+    def __init__(self, alpha=ALPHA_PRIOR, beta=BETA_PRIOR):
         self.alpha = alpha
         self.beta = beta
 
@@ -77,10 +177,12 @@ class BayesianRiskEngine:
         return lo, hi
 
 # ----------------------------------------------------------------------
-# Policy Engine
+# Policy Engine (now configurable)
 # ----------------------------------------------------------------------
 class PolicyEngine:
-    def __init__(self, thresholds={"low": 0.2, "high": 0.8}):
+    def __init__(self, thresholds: Dict[str, float] = None):
+        if thresholds is None:
+            thresholds = {"low": LOW_THRESHOLD, "high": HIGH_THRESHOLD}
         self.thresholds = thresholds
 
     def evaluate(self, risk):
@@ -92,18 +194,53 @@ class PolicyEngine:
             return "escalate", f"Risk in escalation zone ({self.thresholds['low']}-{self.thresholds['high']})"
 
 # ----------------------------------------------------------------------
-# History
+# Infrastructure analysis (synchronous, with error handling)
 # ----------------------------------------------------------------------
-decision_history = []
-risk_history = []
-
-def update_dashboard_data(decision, risk):
-    decision_history.append((datetime.utcnow().isoformat(), decision, risk))
-    risk_history.append((datetime.utcnow().isoformat(), risk))
-    if len(decision_history) > 100:
-        decision_history.pop(0)
-    if len(risk_history) > 100:
-        risk_history.pop(0)
+def handle_infra_with_governance(fault_type: str, context_window: int, session_state: dict):
+    try:
+        fault_map = {
+            "none": (1, 99),
+            "switch_down": (20, 80),
+            "server_overload": (35, 65),
+            "cascade": (60, 40)
+        }
+        failures, successes = fault_map.get(fault_type, (1, 99))
+
+        risk_engine = BayesianRiskEngine()
+        risk_engine.update(failures, successes)
+        risk = risk_engine.risk()
+        ci_low, ci_high = risk_engine.risk_interval(0.95)
+
+        policy_engine = PolicyEngine()
+        action, reason = policy_engine.evaluate(risk)
+        control_decision = autonomous_control_decision(risk, risk_engine, policy_engine)
+
+        analysis_result = {
+            "risk": risk,
+            "risk_ci": [ci_low, ci_high],
+            "decision": action,
+            "justification": reason,
+            "healing_actions": ["restart"] if action == "deny" else ["monitor"],
+            "posterior_parameters": {
+                "alpha": risk_engine.alpha,
+                "beta": risk_engine.beta
+            }
+        }
+        output = {
+            **analysis_result,
+            "governance": {
+                "policy_evaluation": {
+                    "action": action,
+                    "reason": reason,
+                    "thresholds": policy_engine.thresholds
+                },
+                "control_plane_decision": control_decision
+            }
+        }
+        return output, session_state
+    except Exception as e:
+        logger.exception("Error in handle_infra_with_governance")
+        return {"error": str(e)}, session_state
 
 def autonomous_control_decision(risk, risk_engine, policy_engine):
     action, reason = policy_engine.evaluate(risk)
@@ -118,53 +255,7 @@ def autonomous_control_decision(risk, risk_engine, policy_engine):
     return decision
 
 # ----------------------------------------------------------------------
-# Infrastructure analysis
-# ----------------------------------------------------------------------
-async def handle_infra_with_governance(fault_type, context_window, session_state):
-    fault_map = {
-        "none": (1, 99),
-        "switch_down": (20, 80),
-        "server_overload": (35, 65),
-        "cascade": (60, 40)
-    }
-    failures, successes = fault_map.get(fault_type, (1, 99))
-    severity = "low" if failures < 10 else "medium" if failures < 40 else "high"
-
-    risk_engine = BayesianRiskEngine(alpha=1, beta=1)
-    risk_engine.update(failures, successes)
-    risk = risk_engine.risk()
-    ci_low, ci_high = risk_engine.risk_interval(0.95)
-
-    policy_engine = PolicyEngine(thresholds={"low": 0.2, "high": 0.8})
-    action, reason = policy_engine.evaluate(risk)
-    control_decision = autonomous_control_decision(risk, risk_engine, policy_engine)
-
-    analysis_result = {
-        "risk": risk,
-        "risk_ci": [ci_low, ci_high],
-        "decision": action,
-        "justification": reason,
-        "healing_actions": ["restart"] if action == "deny" else ["monitor"],
-        "posterior_parameters": {
-            "alpha": risk_engine.alpha,
-            "beta": risk_engine.beta
-        }
-    }
-    output = {
-        **analysis_result,
-        "governance": {
-            "policy_evaluation": {
-                "action": action,
-                "reason": reason,
-                "thresholds": policy_engine.thresholds
-            },
-            "control_plane_decision": control_decision
-        }
-    }
-    return output, session_state
-
-# ----------------------------------------------------------------------
-# MCMC (Metropolis‑Hastings)
+# MCMC (Metropolis‑Hastings) with input validation and timeout
 # ----------------------------------------------------------------------
 class MHMCMC:
     def __init__(self, log_target, proposal_sd=0.1):
@@ -189,47 +280,60 @@ class MHMCMC:
         acceptance_rate = accepted / (n_samples + burn_in)
         return samples, acceptance_rate
 
-def run_hmc_mcmc(samples, warmup):
-    # Generate data: 10 observations with mean 0.5, std 0.2
-    data = np.random.normal(0.5, 0.2, 10)
-    def log_prior(mu):
-        return -0.5 * (mu ** 2)   # prior N(0,1)
-    def log_likelihood(mu):
-        return -0.5 * np.sum(((data - mu) / 0.2) ** 2)
-    def log_posterior(mu):
-        return log_prior(mu) + log_likelihood(mu)
-
-    sampler = MHMCMC(log_posterior, proposal_sd=0.05)
-    mu_samples, acceptance = sampler.sample(samples, initial_state=[0.0], burn_in=warmup)
-    mu_samples = mu_samples.flatten()
-
-    mean = np.mean(mu_samples)
-    median = np.median(mu_samples)
-    credible_interval = np.percentile(mu_samples, [2.5, 97.5])
-
-    fig_trace = go.Figure()
-    fig_trace.add_trace(go.Scatter(y=mu_samples, mode='lines', name='μ', line=dict(width=1)))
-    fig_trace.update_layout(title="Trace of μ (Metropolis-Hastings)", xaxis_title="Iteration", yaxis_title="μ")
-
-    fig_hist = go.Figure()
-    fig_hist.add_trace(go.Histogram(x=mu_samples, nbinsx=50, name='Posterior'))
-    fig_hist.update_layout(title="Posterior Distribution of μ", xaxis_title="μ", yaxis_title="Density")
-
-    summary = {
-        "mean": mean,
-        "median": median,
-        "credible_interval_95": f"[{credible_interval[0]:.3f}, {credible_interval[1]:.3f}]",
-        "acceptance_rate": f"{acceptance:.2%}"
-    }
-    return summary, fig_trace, fig_hist
+def run_hmc_mcmc(samples: int, warmup: int):
+    try:
+        # Input validation
+        samples = max(500, min(10000, int(samples)))
+        warmup = max(100, min(2000, int(warmup)))
+
+        # Generate data: 10 observations with mean 0.5, std 0.2
+        np.random.seed(42)  # for reproducibility
+        data = np.random.normal(0.5, 0.2, 10)
+
+        def log_prior(mu):
+            return -0.5 * (mu ** 2)   # prior N(0,1)
+
+        def log_likelihood(mu):
+            return -0.5 * np.sum(((data - mu) / 0.2) ** 2)
+
+        def log_posterior(mu):
+            return log_prior(mu) + log_likelihood(mu)
+
+        sampler = MHMCMC(log_posterior, proposal_sd=0.05)
+        mu_samples, acceptance = sampler.sample(samples, initial_state=[0.0], burn_in=warmup)
+        mu_samples = mu_samples.flatten()
+
+        mean = np.mean(mu_samples)
+        median = np.median(mu_samples)
+        credible_interval = np.percentile(mu_samples, [2.5, 97.5])
+
+        fig_trace = go.Figure()
+        fig_trace.add_trace(go.Scatter(y=mu_samples, mode='lines', name='μ', line=dict(width=1)))
+        fig_trace.update_layout(title="Trace of μ (Metropolis-Hastings)", xaxis_title="Iteration", yaxis_title="μ")
+
+        fig_hist = go.Figure()
+        fig_hist.add_trace(go.Histogram(x=mu_samples, nbinsx=50, name='Posterior'))
+        fig_hist.update_layout(title="Posterior Distribution of μ", xaxis_title="μ", yaxis_title="Density")
+
+        summary = {
+            "mean": mean,
+            "median": median,
+            "credible_interval_95": f"[{credible_interval[0]:.3f}, {credible_interval[1]:.3f}]",
+            "acceptance_rate": f"{acceptance:.2%}"
+        }
+        return summary, fig_trace, fig_hist
+    except Exception as e:
+        logger.exception("MCMC computation failed")
+        return {"error": str(e)}, go.Figure(), go.Figure()
 
 # ----------------------------------------------------------------------
-# Dashboard plots
+# Dashboard plots (thread‑safe)
 # ----------------------------------------------------------------------
 def generate_risk_gauge():
-    if not risk_history:
-        return go.Figure()
-    latest_risk = risk_history[-1][1]
+    with history_lock:
+        if not risk_history:
+            return go.Figure()
+        latest_risk = risk_history[-1][1]
     fig = go.Figure(go.Indicator(
         mode="gauge+number",
         value=latest_risk,
@@ -238,27 +342,29 @@ def generate_risk_gauge():
             'axis': {'range': [0, 1]},
             'bar': {'color': "darkblue"},
             'steps': [
-                {'range': [0, 0.2], 'color': "lightgreen"},
-                {'range': [0.2, 0.8], 'color': "yellow"},
-                {'range': [0.8, 1], 'color': "red"}
+                {'range': [0, LOW_THRESHOLD], 'color': "lightgreen"},
+                {'range': [LOW_THRESHOLD, HIGH_THRESHOLD], 'color': "yellow"},
+                {'range': [HIGH_THRESHOLD, 1], 'color': "red"}
             ]
         }))
     return fig
 
 def generate_decision_pie():
-    if not decision_history:
-        return go.Figure()
-    approved = sum(1 for _, d, _ in decision_history if d.get("approved", False))
-    blocked = len(decision_history) - approved
+    with history_lock:
+        if not decision_history:
+            return go.Figure()
+        approved = sum(1 for _, d, _ in decision_history if d.get("approved", False))
+        blocked = len(decision_history) - approved
     fig = go.Figure(data=[go.Pie(labels=["Approved", "Blocked"], values=[approved, blocked])])
     fig.update_layout(title="Policy Decisions")
     return fig
 
 def generate_action_timeline():
-    if not decision_history:
-        return go.Figure()
-    times = [d["timestamp"] for _, d, _ in decision_history]
-    approvals = [1 if d.get("approved", False) else 0 for _, d, _ in decision_history]
+    with history_lock:
+        if not decision_history:
+            return go.Figure()
+        times = [d["timestamp"] for _, d, _ in decision_history]
+        approvals = [1 if d.get("approved", False) else 0 for _, d, _ in decision_history]
     fig = go.Figure()
     fig.add_trace(go.Scatter(x=times, y=approvals, mode='markers+lines', name='Approvals'))
     fig.update_layout(title="Autonomous Actions Timeline", xaxis_title="Time", yaxis_title="Approved (1) / Blocked (0)")
@@ -295,24 +401,43 @@ oss_caps = {
 }
 
 # ----------------------------------------------------------------------
-# Start memory monitoring
+# Health endpoint (custom route)
 # ----------------------------------------------------------------------
-log_memory_usage()
+async def health_endpoint():
+    return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}
+
+# ----------------------------------------------------------------------
+# Startup
+# ----------------------------------------------------------------------
+# Ensure data directory exists
+os.makedirs(os.path.dirname(DB_PATH) if os.path.dirname(DB_PATH) else ".", exist_ok=True)
+init_db()
+refresh_history_from_db()
+
+# Start memory monitor daemon thread
+mem_thread = threading.Thread(target=memory_monitor_loop, daemon=True)
+mem_thread.start()
+
+# Start periodic vacuum (once a day)
+def vacuum_scheduler():
+    while True:
+        time.sleep(86400)  # 24 hours
+        vacuum_db()
+vacuum_thread = threading.Thread(target=vacuum_scheduler, daemon=True)
+vacuum_thread.start()
 
 # ----------------------------------------------------------------------
 # Gradio UI
 # ----------------------------------------------------------------------
 with gr.Blocks(title="ARF v4 – Bayesian Risk Scoring Demo") as demo:
-    gr.Markdown("""
+    gr.Markdown(f"""
     # 🧠 ARF v4 – Bayesian Risk Scoring for AI Reliability (Demo)
     **Mathematically rigorous risk estimation using conjugate priors and MCMC**
-
     This demo showcases:
     - **Bayesian conjugate prior (Beta-Binomial)** – online risk update from observed failures/successes.
-    - **Policy thresholds** – approve (<0.2), escalate (0.2‑0.8), deny (>0.8).
+    - **Policy thresholds** – approve (<{LOW_THRESHOLD}), escalate ({LOW_THRESHOLD}‑{HIGH_THRESHOLD}), deny (>{HIGH_THRESHOLD}).
     - **Metropolis-Hastings MCMC** – sampling from a posterior distribution (simulating HMC concepts).
     - **Autonomous control decisions** – based on the current risk estimate.
-
     All components are implemented with only `numpy`, `scipy`, and standard libraries.
     """)
 
@@ -326,8 +451,8 @@ with gr.Blocks(title="ARF v4 – Bayesian Risk Scoring Demo") as demo:
                         "version": oss_caps["version"],
                         "governance_mode": "advisory",
                         "policies_loaded": 2,
-                        "risk_threshold_low": 0.2,
-                        "risk_threshold_high": 0.8
+                        "risk_threshold_low": LOW_THRESHOLD,
+                        "risk_threshold_high": HIGH_THRESHOLD
                     })
                 with gr.Column():
                     control_stats = gr.JSON(label="Control Statistics", value={
@@ -377,9 +502,9 @@ with gr.Blocks(title="ARF v4 – Bayesian Risk Scoring Demo") as demo:
         with gr.TabItem("Policy Management"):
             gr.Markdown("### 📋 Execution Policies")
             policies_json = [
-                {"name": "Low Risk Policy", "conditions": ["risk < 0.2"], "action": "approve", "priority": 1},
-                {"name": "Medium Risk Policy", "conditions": ["0.2 ≤ risk ≤ 0.8"], "action": "escalate", "priority": 2},
-                {"name": "High Risk Policy", "conditions": ["risk > 0.8"], "action": "deny", "priority": 3}
+                {"name": "Low Risk Policy", "conditions": [f"risk < {LOW_THRESHOLD}"], "action": "approve", "priority": 1},
+                {"name": "Medium Risk Policy", "conditions": [f"{LOW_THRESHOLD} ≤ risk ≤ {HIGH_THRESHOLD}"], "action": "escalate", "priority": 2},
+                {"name": "High Risk Policy", "conditions": [f"risk > {HIGH_THRESHOLD}"], "action": "deny", "priority": 3}
             ]
             gr.JSON(label="Active Policies", value=policies_json)
 
@@ -404,7 +529,7 @@ with gr.Blocks(title="ARF v4 – Bayesian Risk Scoring Demo") as demo:
 
     # Wire events
     infra_btn.click(
-        fn=lambda f, w, s: handle_infra_with_governance(f, w, s),
+        fn=handle_infra_with_governance,
         inputs=[infra_fault, gr.State(50), infra_state],
         outputs=[infra_output, infra_state]
     )
@@ -415,5 +540,13 @@ with gr.Blocks(title="ARF v4 – Bayesian Risk Scoring Demo") as demo:
         outputs=[hmc_summary, hmc_trace_plot, hmc_pair_plot]
     )
 
+# Add health endpoint
+demo.add_http_route(
+    path="/health",
+    method="GET",
+    endpoint=health_endpoint
+)
+
 if __name__ == "__main__":
-    demo.launch(theme="soft")
+    demo.queue()
+    demo.launch(theme="soft", server_name="0.0.0.0", server_port=7860)