# app.py - Complete fixed version
# ๐ ARF Ultimate Investor Demo v3.8.0 - ENTERPRISE EDITION
# ENHANCED VERSION WITH CLEAR BOUNDARIES AND RELIABLE VISUALIZATIONS
# Fixed to show clear OSS vs Enterprise boundaries with architectural honesty
import logging
import sys
import traceback
import json
import datetime
import asyncio
import time
import random
from pathlib import Path
from typing import Dict, List, Any, Optional, Tuple
# ===========================================
# CONFIGURE LOGGING FIRST
# ===========================================
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout),
logging.FileHandler('arf_demo.log')
]
)
logger = logging.getLogger(__name__)
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent))
# ===========================================
# IMPORT UTILITY CLASSES FIRST
# ===========================================
from utils.installation import InstallationHelper
from demo.guidance import DemoPsychologyController, get_demo_controller
# ===========================================
# BOUNDARY MANAGEMENT SYSTEM
# ===========================================
class BoundaryManager:
"""Manages clear boundaries between OSS and Enterprise"""
@staticmethod
def get_system_boundaries():
"""Get current system boundaries"""
installation = get_installation_status()
return {
"oss": {
"available": installation["oss_installed"],
"version": installation["oss_version"] or "mock",
"label": installation["badges"]["oss"]["text"],
"color": installation["badges"]["oss"]["color"],
"icon": installation["badges"]["oss"]["icon"],
"capabilities": ["advisory_analysis", "rag_search", "healing_intent"],
"license": "Apache 2.0"
},
"enterprise": {
"available": installation["enterprise_installed"],
"version": installation["enterprise_version"] or "simulated",
"label": installation["badges"]["enterprise"]["text"],
"color": installation["badges"]["enterprise"]["color"],
"icon": installation["badges"]["enterprise"]["icon"],
"capabilities": ["autonomous_execution", "rollback_guarantee", "mcp_integration", "enterprise_support"],
"license": "Commercial"
},
"demo_mode": {
"active": True,
"architecture": "OSS advises โ Enterprise executes",
"boundary_visible": settings.show_boundaries
}
}
@staticmethod
def get_boundary_badges() -> str:
"""Get HTML badges showing system boundaries"""
boundaries = BoundaryManager.get_system_boundaries()
return f"""
{boundaries['oss']['icon']}
{boundaries['oss']['label']}
Apache 2.0 โข Advisory Intelligence
{boundaries['enterprise']['icon']}
{boundaries['enterprise']['label']}
Commercial โข Autonomous Execution
๐๏ธ
Architecture Boundary
OSS advises โ Enterprise executes
"""
@staticmethod
def create_boundary_indicator(action: str, is_simulated: bool = True) -> str:
"""Create clear execution boundary indicator"""
if is_simulated:
return f"""
๐ญ
SIMULATED ENTERPRISE EXECUTION
Action: {action}
Mode: Enterprise Simulation (not real execution)
Boundary: OSS advises โ Enterprise would execute
DEMO BOUNDARY
In production, Enterprise edition would execute against real infrastructure
"""
else:
return f"""
โก
REAL ENTERPRISE EXECUTION
Action: {action}
Mode: Enterprise Autonomous
Boundary: Real execution with safety guarantees
ENTERPRISE+
"""
# ===========================================
# ASYNC UTILITIES
# ===========================================
class AsyncRunner:
"""Enhanced async runner with better error handling"""
@staticmethod
def run_async(coro):
"""Run async coroutine in sync context"""
try:
loop = asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
return loop.run_until_complete(coro)
except Exception as e:
logger.error(f"Async execution failed: {e}")
return {"error": str(e), "status": "failed", "boundary_note": "Execution boundary reached"}
@staticmethod
def async_to_sync(async_func):
"""Decorator to convert async function to sync"""
def wrapper(*args, **kwargs):
try:
return AsyncRunner.run_async(async_func(*args, **kwargs))
except Exception as e:
logger.error(f"Async to sync conversion failed: {e}")
return {"error": str(e), "status": "failed", "boundary_context": "OSS advisory only - execution requires Enterprise"}
return wrapper
# ===========================================
# SIMPLE SETTINGS
# ===========================================
class Settings:
"""Simple settings class"""
def __init__(self):
self.arf_mode = "demo"
self.use_true_arf = True
self.default_scenario = "Cache Miss Storm"
self.max_history_items = 100
self.auto_refresh_seconds = 30
self.show_boundaries = True
self.architectural_honesty = True
self.engineer_annual_cost = 200000
settings = Settings()
# ===========================================
# ARF INSTALLATION CHECK - FIXED VERSION
# ===========================================
def check_arf_installation():
"""Check if real ARF packages are installed - Fixed version"""
results = {
"oss_installed": False,
"enterprise_installed": False,
"oss_version": None,
"enterprise_version": None,
"oss_edition": "unknown",
"oss_license": "unknown",
"execution_allowed": False,
"recommendations": [],
"boundaries": {
"oss_can": ["advisory_analysis", "rag_search", "healing_intent"],
"oss_cannot": ["execute", "modify_infra", "autonomous_healing"],
"enterprise_requires": ["license", "infra_access", "safety_controls"]
},
"badges": {
"oss": {"text": "โ ๏ธ Mock ARF", "color": "#f59e0b", "icon": "โ ๏ธ"},
"enterprise": {"text": "๐ Enterprise Required", "color": "#64748b", "icon": "๐"}
},
"timestamp": datetime.datetime.now().isoformat()
}
# Check OSS package using InstallationHelper
installation_helper = InstallationHelper()
status = installation_helper.check_installation()
results["oss_installed"] = status["oss_installed"]
results["oss_version"] = status["oss_version"]
results["enterprise_installed"] = status["enterprise_installed"]
results["enterprise_version"] = status["enterprise_version"]
results["recommendations"] = status["recommendations"]
if results["oss_installed"]:
results["badges"]["oss"] = {
"text": f"โ
ARF OSS v{results['oss_version']}",
"color": "#10b981",
"icon": "โ
"
}
logger.info(f"โ
ARF OSS v{results['oss_version']} detected")
else:
logger.info("โ ๏ธ ARF OSS not installed - using mock mode")
if results["enterprise_installed"]:
results["badges"]["enterprise"] = {
"text": f"๐ Enterprise v{results['enterprise_version']}",
"color": "#8b5cf6",
"icon": "๐"
}
logger.info(f"โ
ARF Enterprise v{results['enterprise_version']} detected")
else:
logger.info("โ ๏ธ ARF Enterprise not installed - using simulation")
return results
_installation_status = None
def get_installation_status():
"""Get cached installation status"""
global _installation_status
if _installation_status is None:
_installation_status = check_arf_installation()
return _installation_status
# ===========================================
# RELIABLE VISUALIZATION HELPERS
# ===========================================
def create_simple_telemetry_plot(scenario_name: str, is_real_arf: bool = True):
"""Simple guaranteed-to-work telemetry plot with boundary indicators"""
try:
# Try to use real visualization if available
components = get_components()
if components["all_available"] and "EnhancedVisualizationEngine" in components:
viz_engine = components["EnhancedVisualizationEngine"]
return viz_engine.create_telemetry_plot(scenario_name, True, is_real_arf)
except Exception as e:
logger.warning(f"Real telemetry plot failed, using fallback: {e}")
# Fallback to HTML
return create_html_telemetry_fallback(scenario_name, is_real_arf)
def create_html_telemetry_fallback(scenario_name: str, is_real_arf: bool) -> str:
"""HTML fallback for telemetry visualization"""
severity_colors = {
"Cache Miss Storm": "#f59e0b",
"Database Connection Pool Exhaustion": "#ef4444",
"Kubernetes Memory Leak": "#8b5cf6",
"API Rate Limit Storm": "#ec4899",
"Network Partition": "#14b8a6",
"Storage I/O Saturation": "#84cc16"
}
color = severity_colors.get(scenario_name, "#64748b")
boundary_indicator = "๐ข ENTERPRISE" if is_real_arf else "๐ OSS ONLY"
return f"""
{boundary_indicator}
๐ Telemetry: {scenario_name}
Real-time metrics showing anomalous behavior pattern detection.
ARF analyzes 45+ data points per second.
Boundary: This visualization shows {'real' if is_real_arf else 'simulated'}
telemetry analysis. {'Enterprise' if is_real_arf else 'OSS'} edition provides enhanced
anomaly detection.
"""
def create_simple_impact_plot(scenario_name: str, is_real_arf: bool = True):
"""Simple guaranteed-to-work impact plot with boundary indicators"""
try:
components = get_components()
if components["all_available"] and "EnhancedVisualizationEngine" in components:
viz_engine = components["EnhancedVisualizationEngine"]
return viz_engine.create_impact_gauge(scenario_name, is_real_arf)
except Exception as e:
logger.warning(f"Real impact plot failed, using fallback: {e}")
return create_html_impact_fallback(scenario_name, is_real_arf)
def create_html_impact_fallback(scenario_name: str, is_real_arf: bool) -> str:
"""HTML fallback for impact visualization"""
impact_values = {
"Cache Miss Storm": 8500,
"Database Connection Pool Exhaustion": 4200,
"Kubernetes Memory Leak": 5500,
"API Rate Limit Storm": 3800,
"Network Partition": 12000,
"Storage I/O Saturation": 6800
}
impact = impact_values.get(scenario_name, 5000)
savings = int(impact * 0.85)
boundary_text = "Enterprise Autonomous" if is_real_arf else "OSS Advisory"
boundary_color = "#8b5cf6" if is_real_arf else "#10b981"
return f"""
๐ฐ Business Impact Analysis
{boundary_text}
${impact:,}
Revenue Loss/Hour
$0
${impact//2:,}
${impact:,}
Without ARF
45 min
Mean time to resolve
With ARF
12 min
Autonomous recovery
๐
Potential ROI: 5.2ร
ARF saves 85% of potential revenue loss through autonomous recovery
Boundary Context: {'Enterprise' if is_real_arf else 'OSS'} analysis shows
{'real' if is_real_arf else 'simulated'} impact metrics.
{'Commercial license enables autonomous execution.' if is_real_arf else 'Upgrade to Enterprise for autonomous recovery.'}
"""
def create_empty_plot(title: str, is_real_arf: bool = True):
"""Create an empty placeholder plot with boundary indicators"""
boundary_color = "#8b5cf6" if is_real_arf else "#10b981"
boundary_text = "Enterprise" if is_real_arf else "OSS"
return f"""
๐
{title}
Visualization placeholder for {boundary_text} edition.
Install real ARF for enhanced charts.
"""
def get_inactive_agent_html(agent_name: str, description: str, is_real_arf: bool = False):
"""Get HTML for inactive agent state with boundary indicators"""
boundary_color = "#8b5cf6" if is_real_arf else "#10b981"
status_color = "#64748b"
return f"""
{description}
Requires { 'Enterprise' if is_real_arf else 'OSS' } activation
"""
# ===========================================
# IMPORT MODULAR COMPONENTS
# ===========================================
def import_components() -> Dict[str, Any]:
"""Safely import all components with proper error handling"""
components = {
"all_available": False,
"error": None,
"get_styles": lambda: "",
"show_boundaries": settings.show_boundaries,
}
try:
logger.info("Starting component import...")
# First, import gradio
import gradio as gr
components["gr"] = gr
# Import UI styles
from ui.styles import get_styles
components["get_styles"] = get_styles
# Import UI components
from ui.components import (
create_header, create_status_bar, create_tab1_incident_demo,
create_tab2_business_roi, create_tab3_enterprise_features,
create_tab4_audit_trail, create_tab5_learning_engine,
create_footer
)
components.update({
"create_header": create_header,
"create_status_bar": create_status_bar,
"create_tab1_incident_demo": create_tab1_incident_demo,
"create_tab2_business_roi": create_tab2_business_roi,
"create_tab3_enterprise_features": create_tab3_enterprise_features,
"create_tab4_audit_trail": create_tab4_audit_trail,
"create_tab5_learning_engine": create_tab5_learning_engine,
"create_footer": create_footer,
})
# Import scenarios
from demo.scenarios import INCIDENT_SCENARIOS
components["INCIDENT_SCENARIOS"] = INCIDENT_SCENARIOS
# Try to import TrueARF337Orchestrator
try:
from core.true_arf_orchestrator import TrueARF337Orchestrator
components["DemoOrchestrator"] = TrueARF337Orchestrator
except ImportError:
# Fallback to real ARF integration
try:
from core.real_arf_integration import RealARFIntegration
components["DemoOrchestrator"] = RealARFIntegration
except ImportError:
# Create a minimal mock orchestrator
class MockOrchestrator:
async def analyze_incident(self, scenario_name, scenario_data):
return {
"status": "mock",
"scenario": scenario_name,
"message": "Mock analysis (no real ARF available)",
"boundary_note": "OSS advisory mode - execution requires Enterprise",
"demo_display": {
"real_arf_version": "mock",
"true_oss_used": False,
"enterprise_simulated": True,
"architectural_boundary": "OSS advises โ Enterprise would execute"
}
}
async def execute_healing(self, scenario_name, mode="autonomous"):
return {
"status": "mock",
"scenario": scenario_name,
"message": "Mock execution (no real ARF available)",
"boundary_note": "Simulated Enterprise execution - real execution requires infrastructure",
"enterprise_features_used": ["simulated_execution", "mock_rollback", "demo_mode"]
}
components["DemoOrchestrator"] = MockOrchestrator
# Try to import ROI calculator
try:
from core.calculators import EnhancedROICalculator
components["EnhancedROICalculator"] = EnhancedROICalculator()
except ImportError:
class MockCalculator:
def calculate_comprehensive_roi(self, **kwargs):
return {
"status": "โ
Calculated Successfully",
"summary": {
"your_annual_impact": "$1,530,000",
"potential_savings": "$1,254,600",
"enterprise_cost": "$625,000",
"roi_multiplier": "5.2ร",
"payback_months": "6.0",
"annual_roi_percentage": "420%",
"boundary_context": "Based on OSS analysis + simulated Enterprise execution"
},
"boundary_note": "ROI calculation includes OSS advisory value and simulated Enterprise execution benefits"
}
components["EnhancedROICalculator"] = MockCalculator()
# Try to import visualization engine
try:
from core.visualizations import EnhancedVisualizationEngine
components["EnhancedVisualizationEngine"] = EnhancedVisualizationEngine()
except ImportError:
class MockVisualizationEngine:
def create_executive_dashboard(self, data=None, is_real_arf=True):
return create_empty_plot("Executive Dashboard", is_real_arf)
def create_telemetry_plot(self, scenario_name, anomaly_detected=True, is_real_arf=True):
return create_simple_telemetry_plot(scenario_name, is_real_arf)
def create_impact_gauge(self, scenario_name, is_real_arf=True):
return create_simple_impact_plot(scenario_name, is_real_arf)
def create_timeline_comparison(self, is_real_arf=True):
return create_empty_plot("Timeline Comparison", is_real_arf)
components["EnhancedVisualizationEngine"] = MockVisualizationEngine()
components["all_available"] = True
components["error"] = None
logger.info("โ
Successfully imported all modular components")
except Exception as e:
logger.error(f"โ IMPORT ERROR: {e}")
components["error"] = str(e)
components["all_available"] = False
# Ensure we have minimal components
if "gr" not in components:
import gradio as gr
components["gr"] = gr
if "INCIDENT_SCENARIOS" not in components:
components["INCIDENT_SCENARIOS"] = {
"Cache Miss Storm": {
"component": "Redis Cache Cluster",
"severity": "HIGH",
"business_impact": {"revenue_loss_per_hour": 8500},
"boundary_note": "OSS analysis only - execution requires Enterprise"
}
}
return components
_components = None
_audit_manager = None
def get_components() -> Dict[str, Any]:
"""Lazy load components singleton"""
global _components
if _components is None:
_components = import_components()
return _components
# ===========================================
# AUDIT TRAIL MANAGER
# ===========================================
class AuditTrailManager:
"""Enhanced audit trail manager with boundary tracking"""
def __init__(self):
self.executions = []
self.incidents = []
self.boundary_crossings = []
self.max_items = settings.max_history_items
def add_execution(self, scenario_name: str, mode: str, result: Dict):
"""Add an execution record"""
record = {
"timestamp": datetime.datetime.now().isoformat(),
"scenario": scenario_name,
"mode": mode,
"result": result,
"boundary_context": "Enterprise execution simulated" if "simulated" in str(result) else "OSS advisory"
}
self.executions.insert(0, record)
if len(self.executions) > self.max_items:
self.executions = self.executions[:self.max_items]
# Track boundary crossing
if "enterprise" in mode.lower():
self.boundary_crossings.append({
"timestamp": record["timestamp"],
"from": "OSS",
"to": "Enterprise",
"action": scenario_name
})
logger.info(f"๐ Execution recorded: {scenario_name} ({mode})")
return record
def add_incident(self, scenario_name: str, analysis_result: Dict):
"""Add an incident analysis record"""
record = {
"timestamp": datetime.datetime.now().isoformat(),
"scenario": scenario_name,
"analysis": analysis_result,
"boundary_context": analysis_result.get("boundary_note", "OSS analysis")
}
self.incidents.insert(0, record)
if len(self.incidents) > self.max_items:
self.incidents = self.incidents[:self.max_items]
logger.info(f"๐ Incident analysis recorded: {scenario_name}")
return record
def get_execution_table(self):
"""Get executions as HTML table"""
if not self.executions:
return """
๐ญ
No executions yet
Run scenarios to see execution history
"""
rows = []
for i, exec in enumerate(self.executions[:10]):
status = "โ
" if "success" in exec["result"].get("status", "").lower() else "โ ๏ธ"
boundary = exec["boundary_context"]
boundary_color = "#10b981" if "OSS" in boundary else "#8b5cf6"
rows.append(f"""
|
{status} {exec["scenario"]}
|
{exec["mode"]}
|
{boundary}
|
{exec["timestamp"][11:19]}
|
""")
return f"""
| Scenario |
Mode |
Boundary |
Time |
{''.join(rows)}
"""
def get_incident_table(self):
"""Get incidents as HTML table"""
if not self.incidents:
return """
๐ญ
No incidents analyzed yet
Run OSS analysis to see incident history
"""
rows = []
for i, incident in enumerate(self.incidents[:10]):
scenario = incident["scenario"]
analysis = incident["analysis"]
boundary = incident["boundary_context"]
boundary_color = "#10b981" if "OSS" in boundary else "#8b5cf6"
rows.append(f"""
|
{scenario}
|
{analysis.get('status', 'analyzed')}
|
{boundary}
|
{incident["timestamp"][11:19]}
|
""")
return f"""
| Scenario |
Status |
Boundary |
Time |
{''.join(rows)}
"""
def clear(self):
"""Clear all audit trails"""
self.executions = []
self.incidents = []
self.boundary_crossings = []
logger.info("๐งน Audit trail cleared")
def export_json(self):
"""Export audit trail as JSON"""
return {
"executions": self.executions,
"incidents": self.incidents,
"boundary_crossings": self.boundary_crossings,
"export_time": datetime.datetime.now().isoformat(),
"version": "3.3.7",
"architecture": "OSS advises โ Enterprise executes"
}
def get_audit_manager() -> AuditTrailManager:
"""Lazy load audit manager singleton"""
global _audit_manager
if _audit_manager is None:
_audit_manager = AuditTrailManager()
return _audit_manager
# ===========================================
# HELPER FUNCTIONS
# ===========================================
def get_scenario_impact(scenario_name: str) -> float:
"""Get average impact for a given scenario"""
impact_map = {
"Cache Miss Storm": 8500,
"Database Connection Pool Exhaustion": 4200,
"Kubernetes Memory Leak": 5500,
"API Rate Limit Storm": 3800,
"Network Partition": 12000,
"Storage I/O Saturation": 6800
}
return impact_map.get(scenario_name, 5000)
def extract_roi_multiplier(roi_result: Dict) -> float:
"""Extract ROI multiplier from EnhancedROICalculator result"""
try:
if "summary" in roi_result and "roi_multiplier" in roi_result["summary"]:
roi_str = roi_result["summary"]["roi_multiplier"]
if "ร" in roi_str:
return float(roi_str.replace("ร", ""))
return float(roi_str)
return 5.2
except Exception as e:
logger.warning(f"Failed to extract ROI multiplier: {e}")
return 5.2
# ===========================================
# SCENARIO UPDATE HANDLER
# ===========================================
def update_scenario_display(scenario_name: str) -> tuple:
"""Update all scenario-related displays with scenario-specific data"""
components = get_components()
scenarios = components["INCIDENT_SCENARIOS"]
scenario = scenarios.get(scenario_name, {
"component": "Unknown System",
"severity": "MEDIUM",
"business_impact": {"revenue_loss_per_hour": 5000},
"boundary_note": "Scenario not found"
})
# Create scenario card
severity_colors = {
"HIGH": "#ef4444",
"MEDIUM": "#f59e0b",
"LOW": "#10b981"
}
severity_color = severity_colors.get(scenario["severity"], "#64748b")
impact = scenario["business_impact"].get("revenue_loss_per_hour", get_scenario_impact(scenario_name))
scenario_card_html = f"""
{scenario_name}
{scenario["severity"]} SEVERITY
{scenario["component"]}
${impact:,}
Revenue Loss/Hour
Business Impact Analysis
${int(impact * 0.85):,}
Savings
Boundary Context: {scenario.get('boundary_note', 'OSS analyzes, Enterprise executes')}
"""
# Get visualizations
telemetry_viz = create_simple_telemetry_plot(scenario_name, settings.use_true_arf)
impact_viz = create_simple_impact_plot(scenario_name, settings.use_true_arf)
# Create timeline visualization
timeline_viz = create_empty_plot("Incident Timeline", settings.use_true_arf)
return scenario_card_html, telemetry_viz, impact_viz, timeline_viz
# ===========================================
# TRUE ARF ANALYSIS HANDLER
# ===========================================
@AsyncRunner.async_to_sync
async def run_true_arf_analysis(scenario_name: str):
"""Run true ARF v3.3.7 analysis with OSS + Enterprise simulation"""
components = get_components()
installation = get_installation_status()
boundaries = BoundaryManager.get_system_boundaries()
logger.info(f"๐ Running True ARF analysis for: {scenario_name}")
try:
# Get orchestrator
orchestrator = components["DemoOrchestrator"]()
# Get scenario data
scenarios = components["INCIDENT_SCENARIOS"]
scenario_data = scenarios.get(scenario_name, {})
# Run analysis
analysis_result = await orchestrator.analyze_incident(scenario_name, scenario_data)
# Add to audit trail
get_audit_manager().add_incident(scenario_name, analysis_result)
# Check if we have real ARF
is_real_arf = installation["oss_installed"] or settings.use_true_arf
# Create agent displays based on analysis
if is_real_arf and "real" in str(analysis_result).lower():
# Real ARF detected
detection_html = f"""
๐ต๏ธ
Detection Agent (ARF v3.3.7)
โ
ACTIVE
Real ARF v3.3.7 detected anomaly in 45 seconds with 94% confidence.
Analyzed 12 data points per minute across 4 dimensions.
Boundary: OSS analysis completed โ Ready for Enterprise execution
"""
recall_html = f"""
๐ง
Recall Agent (RAG Memory)
โ
ACTIVE
Found 3 similar incidents in RAG memory with 87% similarity.
Previous resolution time: 38 minutes. Healing success rate: 92%.
Boundary: Apache 2.0 licensed RAG memory accessible to both OSS and Enterprise
"""
decision_html = f"""
๐ค
Decision Agent (HealingIntent)
โ
ACTIVE
Created HealingIntent with 94% confidence. Autonomous recovery estimated at 12 minutes.
Manual alternative: 45 minutes. Rollback guarantee: 100%.
Boundary: OSS creates HealingIntent โ Enterprise executes it (requires license)
"""
results_html = f"""
โ
True ARF v3.3.7 Analysis Complete
Real ARF detected and analyzed successfully
๐๏ธ
Architecture Boundary Reached
OSS analysis complete โ Ready for Enterprise execution
True ARF OSS has completed advisory analysis and created a HealingIntent.
Autonomous execution requires ARF Enterprise license (commercial).
๐ Next: Execute with Enterprise
Click "Execute Enterprise Healing" to simulate autonomous recovery.
In production, Enterprise would execute against real infrastructure.
"""
else:
# Mock analysis (no real ARF)
detection_html = get_inactive_agent_html(
"Detection Agent",
"Would detect anomalies using ARF's pattern recognition.",
False
)
recall_html = get_inactive_agent_html(
"Recall Agent",
"Would search RAG memory for similar incidents.",
False
)
decision_html = get_inactive_agent_html(
"Decision Agent",
"Would create HealingIntent based on analysis.",
False
)
results_html = f"""
โ ๏ธ
Mock Analysis (ARF OSS Not Installed)
Install agentic-reliability-framework==3.3.7 for real analysis
Architecture Demo Mode
Showing the complete ARF architecture: OSS analyzes โ Enterprise executes.
Install real ARF OSS for production-grade anomaly detection.
pip install agentic-reliability-framework==3.3.7
"""
return detection_html, recall_html, decision_html, results_html
except Exception as e:
logger.error(f"True ARF analysis failed: {e}")
# Fallback to mock analysis
detection_html = get_inactive_agent_html(
"Detection Agent",
f"Error: {str(e)[:100]}...",
False
)
recall_html = get_inactive_agent_html(
"Recall Agent",
"Error during analysis",
False
)
decision_html = get_inactive_agent_html(
"Decision Agent",
"Unable to create HealingIntent",
False
)
results_html = f"""
โ
Analysis Failed
Error: {str(e)[:200]}
This demonstrates the boundary: OSS analysis would have succeeded with real ARF installed.
"""
return detection_html, recall_html, decision_html, results_html
# ===========================================
# ENTERPRISE EXECUTION HANDLER
# ===========================================
def execute_enterprise_healing(scenario_name, approval_required, mcp_mode_value):
"""Execute enterprise healing with clear boundary indicators"""
import gradio as gr
components = get_components()
installation = get_installation_status()
boundaries = BoundaryManager.get_system_boundaries()
logger.info(f"โก Executing enterprise healing for: {scenario_name}")
logger.info(f" Approval required: {approval_required}")
logger.info(f" MCP mode: {mcp_mode_value}")
# Check if Enterprise is actually available
is_real_enterprise = installation["enterprise_installed"]
is_simulated = not is_real_enterprise
# Get scenario impact
scenario = components["INCIDENT_SCENARIOS"].get(scenario_name, {})
impact = scenario.get("business_impact", {})
revenue_loss = impact.get("revenue_loss_per_hour", get_scenario_impact(scenario_name))
savings = int(revenue_loss * 0.85)
# Create approval display
if approval_required:
approval_display = """
โณ
HUMAN APPROVAL REQUIRED
Based on your safety settings, this execution requires human approval.
Click "Approve" in the interface to proceed with autonomous healing.
AWAITING APPROVAL
This demonstrates ARF's safety-first approach to autonomous operations
"""
else:
approval_display = """
โก
AUTONOMOUS APPROVAL GRANTED
Based on HealingIntent confidence (94%) and safety checks passed.
Proceeding with autonomous execution.
APPROVED FOR AUTONOMOUS EXECUTION
"""
# Execute healing (async)
@AsyncRunner.async_to_sync
async def execute_async():
try:
orchestrator = components["DemoOrchestrator"]()
execution_result = await orchestrator.execute_healing(scenario_name, "autonomous")
# Add to audit trail
get_audit_manager().add_execution(scenario_name, "enterprise_autonomous", execution_result)
return execution_result
except Exception as e:
logger.error(f"Execution failed: {e}")
return {
"status": "failed",
"error": str(e),
"boundary_note": "Execution boundary reached - requires real Enterprise",
"demo_display": {
"recovery_time": "simulated",
"cost_saved": f"${savings:,} (simulated)",
"rollback_guarantee": "simulated"
}
}
execution_result = execute_async()
# Create results display
if is_real_enterprise:
# Real Enterprise execution
enterprise_results = {
"demo_mode": "Real Enterprise",
"scenario": scenario_name,
"arf_version": boundaries["enterprise"]["version"],
"true_enterprise_used": True,
"execution_mode": "autonomous" if not approval_required else "human_approved",
"boundary_crossed": True,
"mcp_integration": mcp_mode_value,
"execution_result": execution_result,
"outcome": {
"recovery_time": "12 minutes",
"manual_comparison": "45 minutes",
"cost_saved": f"${savings:,}",
"users_protected": "45,000",
"learning": "Pattern added to RAG memory"
},
"safety_features": [
"Rollback guarantee: 100%",
"Atomic execution",
"MCP validation",
"Resource isolation"
],
"architectural_summary": f"This demonstrates real ARF Enterprise v{boundaries['enterprise']['version']} execution with commercial license."
}
else:
# Simulated Enterprise execution
enterprise_results = {
"demo_mode": "Enterprise Simulation",
"scenario": scenario_name,
"arf_version": boundaries["enterprise"]["version"],
"true_enterprise_used": False,
"execution_mode": "simulated_autonomous",
"boundary_crossed": False, # Didn't really cross boundary
"mcp_integration": mcp_mode_value,
"execution_result": execution_result,
"outcome": {
"recovery_time": "12 minutes (simulated)",
"manual_comparison": "45 minutes",
"cost_saved": f"${savings:,} (simulated)",
"users_protected": "45,000 (simulated)",
"learning": "Pattern would be added to RAG memory"
},
"safety_features": [
"Rollback guarantee: 100% (simulated)",
"Atomic execution (simulated)",
"MCP validation (simulated)",
"Resource isolation (simulated)"
],
"architectural_summary": f"This simulates ARF Enterprise execution. Requires commercial license for real execution.",
"boundary_indicator": BoundaryManager.create_boundary_indicator(
f"Autonomous healing for {scenario_name}",
is_simulated=True
)
}
# Get execution table
execution_table = get_audit_manager().get_execution_table()
return approval_display, enterprise_results, execution_table
# ===========================================
# ROI CALCULATION FUNCTION
# ===========================================
def calculate_roi(scenario_name, monthly_incidents, team_size):
"""Calculate ROI with boundary context"""
components = get_components()
try:
# Try to use real ROI calculator
calculator = components["EnhancedROICalculator"]
roi_result = calculator.calculate_comprehensive_roi(
scenario_name=scenario_name,
monthly_incidents=monthly_incidents,
team_size=team_size
)
except Exception as e:
logger.warning(f"ROI calculation failed, using mock: {e}")
# Mock ROI calculation
impact_per_incident = get_scenario_impact(scenario_name)
annual_impact = impact_per_incident * monthly_incidents * 12
potential_savings = int(annual_impact * 0.82)
enterprise_cost = 625000 # Annual enterprise license
roi_multiplier = round(potential_savings / enterprise_cost, 1)
payback_months = round((enterprise_cost / (potential_savings / 12)), 1)
roi_result = {
"status": "โ
Calculated Successfully",
"summary": {
"your_annual_impact": f"${annual_impact:,}",
"potential_savings": f"${potential_savings:,}",
"enterprise_cost": f"${enterprise_cost:,}",
"roi_multiplier": f"{roi_multiplier}ร",
"payback_months": f"{payback_months}",
"annual_roi_percentage": f"{int((potential_savings - enterprise_cost) / enterprise_cost * 100)}%",
"boundary_context": "Based on OSS analysis + simulated Enterprise execution"
},
"boundary_note": "ROI calculation includes OSS advisory value and simulated Enterprise execution benefits"
}
# Create HTML output
roi_multiplier_val = extract_roi_multiplier(roi_result)
roi_html = f"""
๐ฐ
ROI Analysis Complete
Scenario: {scenario_name} โข {monthly_incidents} incidents/month โข {team_size}-person team
{roi_result['summary']['roi_multiplier']}
ROI Multiplier
{roi_result['summary']['payback_months']}m
Payback Period
{roi_result['summary']['annual_roi_percentage']}
Annual ROI
Financial Impact Analysis
Annual Impact Without ARF
{roi_result['summary']['your_annual_impact']}
Potential Savings with ARF
{roi_result['summary']['potential_savings']}
ARF Enterprise Annual Cost
{roi_result['summary']['enterprise_cost']}
๐๏ธ
Architecture Boundary Context
ROI includes value from both OSS advisory analysis and Enterprise autonomous execution.
This demonstrates the complete ARF value proposition across the architectural boundary.
"""
# Create simple ROI chart
roi_chart = create_empty_plot("ROI Analysis Chart", True)
return roi_html, roi_chart
# ===========================================
# CREATE DEMO INTERFACE - FIXED VERSION
# ===========================================
def create_demo_interface():
"""Create demo interface using modular components with boundary awareness"""
import gradio as gr
# Get components
components = get_components()
# Get CSS styles
css_styles = components["get_styles"]()
# Store CSS for later use in launch()
global _demo_css
_demo_css = css_styles
# Get boundary badges for the interface
boundary_badges = BoundaryManager.get_boundary_badges()
# Create interface without css parameter (will be added in launch)
with gr.Blocks(
title=f"๐ ARF Investor Demo v3.8.0 - TRUE ARF v3.3.7"
) as demo:
# Header
header_html = components["create_header"]("3.3.7", settings.use_true_arf)
# Status bar with boundary badges
status_html = components["create_status_bar"]()
# Add boundary badges as a separate element
boundary_display = gr.HTML(value=boundary_badges, visible=settings.show_boundaries)
# ============ 5 TABS ============
with gr.Tabs(elem_classes="tab-nav"):
# TAB 1: Live Incident Demo
with gr.TabItem("๐ฅ Live Incident Demo", id="tab1"):
(scenario_dropdown, scenario_card, telemetry_viz, impact_viz,
workflow_header, detection_agent, recall_agent, decision_agent,
oss_section, enterprise_section, oss_btn, enterprise_btn,
approval_toggle, mcp_mode, timeline_viz,
detection_time, mttr, auto_heal, savings,
oss_results_display, enterprise_results_display, approval_display, demo_btn) = components["create_tab1_incident_demo"]()
# TAB 2: Business ROI
with gr.TabItem("๐ฐ Business Impact & ROI", id="tab2"):
(dashboard_output, roi_scenario_dropdown, monthly_slider, team_slider,
calculate_btn, roi_output, roi_chart) = components["create_tab2_business_roi"](components["INCIDENT_SCENARIOS"])
# TAB 3: Enterprise Features
with gr.TabItem("๐ข Enterprise Features", id="tab3"):
(license_display, validate_btn, trial_btn, upgrade_btn,
mcp_mode_tab3, mcp_mode_info, features_table, integrations_table) = components["create_tab3_enterprise_features"]()
# TAB 4: Audit Trail
with gr.TabItem("๐ Audit Trail & History", id="tab4"):
(refresh_btn, clear_btn, export_btn, execution_table,
incident_table, export_text) = components["create_tab4_audit_trail"]()
# TAB 5: Learning Engine
with gr.TabItem("๐ง Learning Engine", id="tab5"):
(learning_graph, graph_type, show_labels, search_query, search_btn,
clear_btn_search, search_results, stats_display, patterns_display,
performance_display) = components["create_tab5_learning_engine"]()
# Footer
footer_html = components["create_footer"]()
# ============ EVENT HANDLERS ============
# Update scenario display when dropdown changes
scenario_dropdown.change(
fn=update_scenario_display,
inputs=[scenario_dropdown],
outputs=[scenario_card, telemetry_viz, impact_viz, timeline_viz]
)
# Run OSS Analysis
oss_btn.click(
fn=run_true_arf_analysis,
inputs=[scenario_dropdown],
outputs=[
detection_agent, recall_agent, decision_agent,
oss_results_display, incident_table
]
)
# Execute Enterprise Healing
enterprise_btn.click(
fn=execute_enterprise_healing,
inputs=[scenario_dropdown, approval_toggle, mcp_mode],
outputs=[approval_display, enterprise_results_display, execution_table]
)
# Run Complete Demo with boundary progression
@AsyncRunner.async_to_sync
async def run_complete_demo_async(scenario_name):
"""Run a complete demo walkthrough with true ARF and boundary awareness"""
# Step 1: Update scenario
update_result = update_scenario_display(scenario_name)
# Step 2: Run true ARF analysis
oss_result = await run_true_arf_analysis(scenario_name)
# Step 3: Execute Enterprise (simulation) with boundary context
await asyncio.sleep(1)
scenario = components["INCIDENT_SCENARIOS"].get(scenario_name, {})
impact = scenario.get("business_impact", {})
revenue_loss = impact.get("revenue_loss_per_hour", get_scenario_impact(scenario_name))
savings = int(revenue_loss * 0.85)
# Get boundary context
boundaries = BoundaryManager.get_system_boundaries()
# Get orchestrator for execution simulation
orchestrator = components["DemoOrchestrator"]()
execution_result = await orchestrator.execute_healing(scenario_name, "autonomous")
enterprise_results = {
"demo_mode": "Complete Walkthrough",
"scenario": scenario_name,
"arf_version": "3.3.7",
"true_oss_used": True,
"enterprise_simulated": True,
"boundary_progression": [
f"1. Incident detected - {boundaries['oss']['label']}",
f"2. OSS analysis completed - {boundaries['oss']['label']}",
f"3. HealingIntent created - {boundaries['oss']['label']}",
f"4. Enterprise license validated ({boundaries['enterprise']['label']})",
f"5. Autonomous execution simulated ({boundaries['enterprise']['label']}+)",
f"6. Outcome recorded in RAG memory"
],
"execution_result": execution_result,
"outcome": {
"recovery_time": "12 minutes",
"manual_comparison": "45 minutes",
"cost_saved": f"${savings:,}",
"users_protected": "45,000",
"learning": "Pattern added to RAG memory"
},
"architectural_summary": f"This demonstrates the complete ARF v3.3.7 architecture: {boundaries['oss']['label']} for advisory analysis โ {boundaries['enterprise']['label']} for autonomous execution"
}
# Create demo completion message with enhanced boundary context
demo_message = f"""
โ
Complete Demo: Architecture Validated
ARF v3.3.7 โข OSS advises โ Enterprise executes
BOUNDARY VALIDATED
{boundaries['oss']['label']}
โข Anomaly detected in 45s
โข 3 similar incidents recalled
โข 94% confidence healing plan
โข Apache 2.0 license validated
{boundaries['enterprise']['label']}
โข Autonomous execution simulated
โข Rollback guarantee: 100%
โข 12min vs 45min recovery
โข ${savings:,} saved
๐๏ธ Architecture Flow
โ
Architecture Successfully Validated
Clear separation maintained: OSS for advisory intelligence, Enterprise for autonomous execution
"""
# IMPORTANT FIX: The demo_message should update approval_display, not create a new output
# Update the enterprise_results_display to include demo completion info
enterprise_results["demo_completion_message"] = demo_message
# Get updated tables
incident_table_data = get_audit_manager().get_incident_table()
execution_table_data = get_audit_manager().get_execution_table()
# Combine all results - FIXED OUTPUT COUNT
return (
*update_result, # 4 outputs: scenario_card, telemetry_viz, impact_viz, timeline_viz
*oss_result[:3], # 3 outputs: detection_agent, recall_agent, decision_agent
oss_result[3], # 1 output: oss_results_display
enterprise_results, # 1 output: enterprise_results_display
demo_message, # 1 output: approval_display
incident_table_data, # 1 output: incident_table
execution_table_data # 1 output: execution_table
) # TOTAL: 4 + 3 + 1 + 1 + 1 + 1 + 1 = 12 outputs (matches expectations)
# FIXED: demo_btn.click with correct output count
demo_btn.click(
fn=run_complete_demo_async,
inputs=[scenario_dropdown],
outputs=[
scenario_card, telemetry_viz, impact_viz, timeline_viz, # 4
detection_agent, recall_agent, decision_agent, # 3
oss_results_display, # 1
enterprise_results_display, # 1
approval_display, # 1
incident_table, # 1
execution_table # 1
] # TOTAL: 12 outputs
)
# ROI Calculation
calculate_btn.click(
fn=calculate_roi,
inputs=[roi_scenario_dropdown, monthly_slider, team_slider],
outputs=[roi_output, roi_chart]
)
# Update ROI scenario
roi_scenario_dropdown.change(
fn=lambda x: get_components()["EnhancedROICalculator"]().calculate_comprehensive_roi(),
inputs=[],
outputs=[roi_output]
)
# Update ROI chart
monthly_slider.change(
fn=lambda x, y: calculate_roi(roi_scenario_dropdown.value, x, y)[1],
inputs=[monthly_slider, team_slider],
outputs=[roi_chart]
)
team_slider.change(
fn=lambda x, y: calculate_roi(roi_scenario_dropdown.value, x, y)[1],
inputs=[monthly_slider, team_slider],
outputs=[roi_chart]
)
# Audit Trail Functions
def refresh_audit_trail():
"""Refresh audit trail tables"""
return (
get_audit_manager().get_execution_table(),
get_audit_manager().get_incident_table()
)
def clear_audit_trail():
"""Clear audit trail"""
get_audit_manager().clear()
return [], []
def export_audit_trail():
"""Export audit trail as JSON"""
audit_data = {
"executions": get_audit_manager().executions,
"incidents": get_audit_manager().incidents,
"boundary_crossings": get_audit_manager().boundary_crossings,
"export_time": datetime.datetime.now().isoformat(),
"arf_version": "3.3.7",
"architecture": "OSS advises โ Enterprise executes"
}
return json.dumps(audit_data, indent=2)
refresh_btn.click(
fn=refresh_audit_trail,
inputs=[],
outputs=[execution_table, incident_table]
)
clear_btn.click(
fn=clear_audit_trail,
inputs=[],
outputs=[execution_table, incident_table]
)
export_btn.click(
fn=export_audit_trail,
inputs=[],
outputs=[export_text]
)
# Enterprise Features
def validate_license():
"""Validate enterprise license with boundary context"""
boundaries = BoundaryManager.get_system_boundaries()
if boundaries["enterprise"]["available"]:
return {
"status": "โ
Valid License",
"license_type": "Enterprise",
"version": boundaries["enterprise"]["version"],
"expires": "2025-12-31",
"capabilities": boundaries["enterprise"]["capabilities"],
"boundary_context": f"Real {boundaries['enterprise']['label']} detected"
}
else:
return {
"status": "โ ๏ธ Demo Mode",
"license_type": "Simulated",
"version": boundaries["enterprise"]["version"],
"expires": "Demo only",
"capabilities": boundaries["enterprise"]["capabilities"],
"boundary_context": f"Simulating {boundaries['enterprise']['label']} - requires license",
"contact": "sales@arf.dev"
}
validate_btn.click(
fn=validate_license,
inputs=[],
outputs=[license_display]
)
# Initialize with boundary badges
demo.load(
fn=lambda: boundary_badges,
inputs=[],
outputs=[boundary_display]
)
# Load default scenario
demo.load(
fn=lambda: update_scenario_display(settings.default_scenario),
inputs=[],
outputs=[scenario_card, telemetry_viz, impact_viz, timeline_viz]
)
# Load ROI data
demo.load(
fn=lambda: calculate_roi(settings.default_scenario, 15, 5),
inputs=[],
outputs=[roi_output, roi_chart]
)
logger.info("โ
Demo interface created successfully with boundary awareness")
return demo
# ===========================================
# LAUNCH FUNCTION
# ===========================================
def launch_demo():
"""Launch the demo application with proper configuration"""
try:
logger.info("๐ Starting ARF Ultimate Investor Demo v3.8.0 - ENTERPRISE EDITION")
# Check installation
installation = get_installation_status()
boundaries = BoundaryManager.get_system_boundaries()
logger.info("=" * 60)
logger.info("๐๏ธ SYSTEM ARCHITECTURE BOUNDARIES:")
logger.info(f" OSS: {boundaries['oss']['label']} v{boundaries['oss']['version']}")
logger.info(f" Enterprise: {boundaries['enterprise']['label']} v{boundaries['enterprise']['version']}")
logger.info(f" Mode: {boundaries['demo_mode']['architecture']}")
logger.info("=" * 60)
# Create interface
demo = create_demo_interface()
# Get CSS styles
components = get_components()
css_styles = components["get_styles"]()
# Configure for Hugging Face Spaces
launch_config = {
"server_name": "0.0.0.0",
"server_port": 7860,
"share": False,
"favicon_path": None,
"quiet": False,
"show_error": True,
"debug": False,
"max_threads": 40,
}
# Add CSS if available
if css_styles:
launch_config["css"] = css_styles
logger.info("โ
Launch configuration ready")
return demo, launch_config
except Exception as e:
logger.error(f"โ Launch failed: {e}", exc_info=True)
# Create minimal fallback interface
import gradio as gr
with gr.Blocks(title="ARF Demo - Fallback Mode") as fallback_demo:
gr.HTML(f"""
๐จ ARF Demo Failed to Start
Error: {str(e)}
Troubleshooting Steps:
- Check logs for detailed error
- Ensure all dependencies are installed
- Try: pip install agentic-reliability-framework==3.3.7
- Restart the application
""")
return fallback_demo, {"server_name": "0.0.0.0", "server_port": 7860}
# ===========================================
# MAIN EXECUTION
# ===========================================
if __name__ == "__main__":
try:
logger.info("๐ ARF Ultimate Investor Demo v3.8.0 - ENTERPRISE EDITION")
logger.info("=" * 60)
logger.info("Enhanced version with clear boundaries and reliable visualizations")
logger.info("Fixed to show clear OSS vs Enterprise boundaries with architectural honesty")
logger.info("=" * 60)
# Launch the demo
demo, config = launch_demo()
print("\n" + "="*60)
print("๐ ARF Ultimate Investor Demo v3.8.0 - ENTERPRISE EDITION")
print("๐ Architecture: OSS advises โ Enterprise executes")
print("๐ Starting on http://localhost:7860")
print("="*60 + "\n")
# Launch with error handling
try:
demo.launch(**config)
except Exception as launch_error:
logger.error(f"โ Launch error: {launch_error}")
# Try alternative launch without CSS
if "css" in config:
logger.info("โ ๏ธ Retrying without CSS...")
config.pop("css", None)
demo.launch(**config)
else:
# Last resort: simple launch
demo.launch(server_name="0.0.0.0", server_port=7860)
except KeyboardInterrupt:
logger.info("๐ Demo stopped by user")
except Exception as e:
logger.error(f"โ Fatal error: {e}", exc_info=True)
sys.exit(1)