"""
ARF 3.3.9 Demo - Psychology-Optimized Version
Fixed with proper error handling
"""
import gradio as gr
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import json
import time
import uuid
import hashlib
import random
import traceback
import sys
from typing import Dict, List, Any, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
import re
print("=" * 70)
print("š§ ARF 3.3.9 DEMO - Psychology-Optimized Edition")
print("=" * 70)
# ============================================================================
# ENUMS AND DATA CLASSES
# ============================================================================
class RiskLevel(str, Enum):
CRITICAL = "CRITICAL"
HIGH = "HIGH"
MEDIUM = "MEDIUM"
LOW = "LOW"
SAFE = "SAFE"
class EnforcementLevel(str, Enum):
ADVISORY = "ADVISORY"
HUMAN_APPROVAL = "HUMAN_APPROVAL"
AUTONOMOUS = "AUTONOMOUS"
FULL_MECHANICAL = "FULL_MECHANICAL"
class LicenseTier(str, Enum):
TRIAL = "TRIAL"
STARTER = "STARTER"
PROFESSIONAL = "PROFESSIONAL"
ENTERPRISE = "ENTERPRISE"
@dataclass
class RiskAssessment:
score: float
level: RiskLevel
confidence: float
factors: Dict[str, float]
@property
def color(self) -> str:
return {
RiskLevel.CRITICAL: "#D32F2F",
RiskLevel.HIGH: "#F44336",
RiskLevel.MEDIUM: "#FF9800",
RiskLevel.LOW: "#4CAF50",
RiskLevel.SAFE: "#2E7D32"
}[self.level]
@property
def icon(self) -> str:
return {
RiskLevel.CRITICAL: "š„",
RiskLevel.HIGH: "ā ļø",
RiskLevel.MEDIUM: "š¶",
RiskLevel.LOW: "ā
",
RiskLevel.SAFE: "š”ļø"
}[self.level]
@dataclass
class GateResult:
name: str
passed: bool
message: str
required: bool = True
weight: float = 1.0
@property
def color(self) -> str:
return "#4CAF50" if self.passed else "#F44336"
@property
def icon(self) -> str:
return "ā
" if self.passed else "ā"
# ============================================================================
# SIMPLIFIED BUT ROBUST ENGINE
# ============================================================================
class ARFEngine:
"""Simplified but robust ARF engine"""
def __init__(self):
self.stats = {
"total_processed": 0,
"blocked_actions": 0,
"autonomous_executions": 0,
"avg_processing_time": 0.0
}
print("ā
ARF Engine initialized")
def assess_action(self, action: str, context: Dict, license_key: str = None) -> Dict:
"""Assess an action with realistic scoring"""
start_time = time.time()
# 1. Risk Assessment
risk = self._assess_risk(action, context)
# 2. Policy Evaluation
policy_result = self._evaluate_policies(action, context, risk.score)
# 3. License Validation
license_info = self._validate_license(license_key)
# 4. Gate Evaluation
gate_results = self._evaluate_gates(risk, policy_result, license_info)
# 5. Generate results
processing_time = time.time() - start_time
# Update stats
self._update_stats(policy_result, gate_results, processing_time)
return {
"timestamp": datetime.now().isoformat(),
"action": action,
"risk_assessment": risk,
"policy_result": policy_result,
"license_info": license_info,
"gate_results": gate_results,
"processing_time": round(processing_time, 4),
"recommendation": self._generate_recommendation(policy_result, gate_results, license_info)
}
def _assess_risk(self, action: str, context: Dict) -> RiskAssessment:
"""Assess risk with realistic scoring"""
action_lower = action.lower()
# Calculate risk factors
destructiveness = 0.3
if "drop" in action_lower and "database" in action_lower:
destructiveness = 0.95
elif "delete" in action_lower:
destructiveness = 0.85
elif "truncate" in action_lower:
destructiveness = 0.80
complexity = min(0.8, len(action.split()) * 0.05)
env = context.get("environment", "development")
environment_risk = {
"production": 0.8,
"staging": 0.5,
"testing": 0.3,
"development": 0.2
}.get(env, 0.5)
# Weighted risk score
risk_score = (
destructiveness * 0.4 +
complexity * 0.2 +
environment_risk * 0.4
)
# Determine risk level
if risk_score >= 0.8:
level = RiskLevel.CRITICAL
elif risk_score >= 0.6:
level = RiskLevel.HIGH
elif risk_score >= 0.4:
level = RiskLevel.MEDIUM
elif risk_score >= 0.2:
level = RiskLevel.LOW
else:
level = RiskLevel.SAFE
# Confidence (higher for clear patterns)
confidence = 0.8
if "drop database" in action_lower:
confidence = 0.95
elif "delete from" in action_lower:
confidence = 0.90
return RiskAssessment(
score=round(risk_score, 3),
level=level,
confidence=round(confidence, 3),
factors={
"destructiveness": destructiveness,
"complexity": complexity,
"environment": environment_risk
}
)
def _evaluate_policies(self, action: str, context: Dict, risk_score: float) -> Dict:
"""Evaluate against safety policies"""
violations = []
# Policy 1: No destructive operations in production without approval
if ("drop" in action.lower() or "delete" in action.lower()) and context.get("environment") == "production":
violations.append({
"policy": "Destructive Operation Prevention",
"severity": "CRITICAL",
"action": "BLOCK"
})
# Policy 2: High risk actions need review
if risk_score > 0.7 and context.get("environment") == "production":
violations.append({
"policy": "High Risk Review Required",
"severity": "HIGH",
"action": "REQUIRE_APPROVAL"
})
return {
"violations": violations,
"blocked": any(v["action"] == "BLOCK" for v in violations),
"requires_approval": any(v["action"] == "REQUIRE_APPROVAL" for v in violations),
"total_violations": len(violations)
}
def _validate_license(self, license_key: str = None) -> Dict:
"""Validate license key"""
if not license_key:
return {
"tier": None,
"valid": False,
"name": "OSS Edition",
"enforcement": "ADVISORY",
"message": "šµ Using ARF OSS (Open Source)"
}
if license_key.startswith("ARF-"):
if "TRIAL" in license_key:
return {
"tier": "TRIAL",
"valid": True,
"name": "Trial",
"enforcement": "ADVISORY",
"message": "š Trial License Active (14 days)"
}
elif "PRO" in license_key:
return {
"tier": "PROFESSIONAL",
"valid": True,
"name": "Professional",
"enforcement": "AUTONOMOUS",
"message": "ā
Professional License Active"
}
return {
"tier": None,
"valid": False,
"name": "Invalid",
"enforcement": "ADVISORY",
"message": "ā Invalid License Key"
}
def _evaluate_gates(self, risk: RiskAssessment, policy: Dict, license: Dict) -> List[GateResult]:
"""Evaluate mechanical gates"""
gates = []
# Gate 1: License Validation
gates.append(GateResult(
name="License Validation",
passed=license.get("valid", False),
message=license.get("message", "No license"),
required=True
))
# Gate 2: Risk Threshold
risk_threshold = 0.8 if license.get("tier") in ["PROFESSIONAL", "ENTERPRISE"] else 0.7
gates.append(GateResult(
name="Risk Assessment",
passed=risk.score <= risk_threshold,
message=f"Risk {risk.score:.1%} ā¤ {risk_threshold:.0%} threshold",
required=True
))
# Gate 3: Confidence Threshold
gates.append(GateResult(
name="Confidence Threshold",
passed=risk.confidence >= 0.7,
message=f"Confidence {risk.confidence:.1%} ā„ 70%",
required=True
))
# Gate 4: Policy Compliance
gates.append(GateResult(
name="Policy Compliance",
passed=not policy.get("blocked", False),
message=f"{policy.get('total_violations', 0)} policy violations",
required=True
))
return gates
def _generate_recommendation(self, policy: Dict, gates: List[GateResult], license: Dict) -> str:
"""Generate recommendation"""
if policy.get("blocked"):
return "š« BLOCKED: Action violates safety policies"
all_gates_passed = all(g.passed for g in gates if g.required)
if not license.get("valid"):
return "šµ OSS ADVISORY: Human review recommended"
elif all_gates_passed and license.get("tier") in ["PROFESSIONAL", "ENTERPRISE"]:
return "š” ENTERPRISE APPROVED: Autonomous execution permitted"
elif all_gates_passed and license.get("tier") == "STARTER":
return "š¤ HUMAN APPROVAL: Gates passed, awaiting confirmation"
else:
return "ā ļø REVIEW REQUIRED: Additional validation needed"
def _update_stats(self, policy: Dict, gates: List[GateResult], processing_time: float):
"""Update statistics"""
self.stats["total_processed"] += 1
if policy.get("blocked"):
self.stats["blocked_actions"] += 1
if all(g.passed for g in gates if g.required):
self.stats["autonomous_executions"] += 1
# Update average processing time
self.stats["avg_processing_time"] = (
self.stats["avg_processing_time"] * 0.9 + processing_time * 0.1
)
def get_stats(self) -> Dict:
"""Get statistics with safe defaults"""
stats = self.stats.copy()
# Calculate derived stats with error handling
total = stats.get("total_processed", 0)
blocked = stats.get("blocked_actions", 0)
autonomous = stats.get("autonomous_executions", 0)
stats["blocked_percentage"] = round(blocked / total * 100, 1) if total > 0 else 0.0
stats["autonomous_percentage"] = round(autonomous / total * 100, 1) if total > 0 else 0.0
stats["processing_speed"] = f"{stats.get('avg_processing_time', 0)*1000:.0f}ms"
# Add default values for any missing keys
defaults = {
"risk_distribution": {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0, "SAFE": 0},
"system_health": "ā
Optimal",
"license_distribution": {"OSS": 0, "TRIAL": 0, "ENTERPRISE": 0}
}
for key, value in defaults.items():
if key not in stats:
stats[key] = value
return stats
def generate_trial_license(self, email: str) -> Dict:
"""Generate trial license"""
if not email or "@" not in email:
return {
"success": False,
"message": "Please enter a valid email address"
}
license_key = f"ARF-TRIAL-{hashlib.sha256(email.encode()).hexdigest()[:8].upper()}"
return {
"success": True,
"license_key": license_key,
"message": "š 14-Day Trial License Generated!",
"expires": (datetime.now() + timedelta(days=14)).strftime("%Y-%m-%d"),
"features": [
"Mechanical gate evaluation",
"Enterprise dashboard",
"Basic audit trail"
]
}
# ============================================================================
# DEMO DATA
# ============================================================================
DEMO_SCENARIOS = [
{
"id": "high_risk",
"name": "š„ High-Risk Database Operation",
"action": "DROP DATABASE production_users CASCADE",
"context": {"environment": "production", "criticality": "critical"}
},
{
"id": "safe_deploy",
"name": "ā
Safe Service Deployment",
"action": "deploy_service payment_api:v2.3.1 to staging with 25% canary",
"context": {"environment": "staging", "service": "payment_api", "canary": 25}
},
{
"id": "config_change",
"name": "š§ Configuration Update",
"action": "UPDATE config SET timeout_ms=30000 WHERE service='api'",
"context": {"environment": "production", "service": "api"}
}
]
# ============================================================================
# GRADIO INTERFACE
# ============================================================================
def create_demo_interface():
"""Create the demo interface"""
# Initialize engine
arf_engine = ARFEngine()
# Get initial stats with error handling
try:
stats = arf_engine.get_stats()
except Exception as e:
print(f"Error getting stats: {e}")
stats = {
"total_processed": 0,
"blocked_percentage": 0.0,
"autonomous_percentage": 0.0,
"processing_speed": "0ms"
}
# Safely get values with defaults
total_processed = stats.get("total_processed", 0)
blocked_percentage = stats.get("blocked_percentage", 0.0)
autonomous_percentage = stats.get("autonomous_percentage", 0.0)
processing_speed = stats.get("processing_speed", "0ms")
with gr.Blocks(
title="ARF 3.3.9 - OSS vs Enterprise Demo",
theme=gr.themes.Soft(),
css="""
.gradio-container {
max-width: 1400px;
margin: 0 auto;
font-family: -apple-system, BlinkMacSystemFont, sans-serif;
}
.demo-card {
border-radius: 10px;
padding: 20px;
margin-bottom: 20px;
box-shadow: 0 2px 8px rgba(0,0,0,0.1);
}
.oss-card {
border: 2px solid #1E88E5;
background: linear-gradient(135deg, #E3F2FD 0%, #BBDEFB 100%);
}
.enterprise-card {
border: 2px solid #FFB300;
background: linear-gradient(135deg, #FFF8E1 0%, #FFECB3 100%);
}
.stat-card {
background: white;
padding: 15px;
border-radius: 8px;
text-align: center;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
.gate-passed {
background: #E8F5E9;
border-left: 4px solid #4CAF50;
padding: 10px;
margin: 5px 0;
border-radius: 4px;
}
.gate-failed {
background: #FFEBEE;
border-left: 4px solid #F44336;
padding: 10px;
margin: 5px 0;
border-radius: 4px;
}
"""
) as demo:
# Header
gr.Markdown("""
# š¤ Agentic Reliability Framework (ARF) 3.3.9
### From Advisory to Mechanical Enforcement
""")
# Statistics
gr.Markdown(f"""
Total Assessments
{total_processed}
Risks Blocked
{blocked_percentage}%
Autonomous Rate
{autonomous_percentage}%
Processing Speed
{processing_speed}
""")
# Main controls
with gr.Row():
with gr.Column(scale=2):
# Scenario selection
scenario_select = gr.Dropdown(
choices=[s["name"] for s in DEMO_SCENARIOS],
label="Select Demo Scenario",
value=DEMO_SCENARIOS[0]["name"]
)
# Action input
action_input = gr.Textbox(
label="Action to Evaluate",
value=DEMO_SCENARIOS[0]["action"],
lines=2
)
# Context input
context_input = gr.Textbox(
label="Context (JSON)",
value=json.dumps(DEMO_SCENARIOS[0]["context"], indent=2),
lines=3
)
# License input
license_input = gr.Textbox(
label="Enterprise License Key (Optional)",
placeholder="Enter ARF-TRIAL-XXXXXXX or leave blank for OSS",
value=""
)
# Buttons
with gr.Row():
process_btn = gr.Button("š Process Action", variant="primary", size="lg")
clear_btn = gr.Button("š Clear", variant="secondary")
with gr.Column(scale=1):
# Quick examples
gr.Markdown("### Quick Examples")
quick_examples = gr.Radio(
choices=["High Risk", "Medium Risk", "Low Risk"],
label="Risk Level",
value="High Risk"
)
# License status
gr.Markdown("### License Status")
license_status = gr.HTML("""
šµ ARF OSS Edition
Advisory Mode - No license required
""")
# Results
with gr.Row():
with gr.Column(scale=1):
oss_output = gr.HTML(label="šµ ARF OSS Results")
with gr.Column(scale=1):
enterprise_output = gr.HTML(label="š” ARF Enterprise Results")
# Comparison
comparison_output = gr.HTML(label="š Side-by-Side Comparison")
# Trial section
with gr.Accordion("š Get 14-Day Trial License", open=False):
with gr.Row():
with gr.Column(scale=1):
email_input = gr.Textbox(
label="Work Email",
placeholder="you@company.com"
)
trial_btn = gr.Button("Get Trial License", variant="primary")
with gr.Column(scale=2):
trial_output = gr.JSON(label="Your Trial License")
# Functions
def process_action(scenario_name, action_text, context_text, license_key, quick_example):
"""Process an action"""
try:
# Update based on quick example
if quick_example == "High Risk":
action_text = "DELETE FROM users WHERE created_at < '2023-01-01'"
context_text = '{"environment": "production"}'
elif quick_example == "Medium Risk":
action_text = "ALTER TABLE payments ADD COLUMN metadata JSONB"
context_text = '{"environment": "staging"}'
elif quick_example == "Low Risk":
action_text = "SELECT COUNT(*) FROM logs"
context_text = '{"environment": "development"}'
# Use scenario if selected
for scenario in DEMO_SCENARIOS:
if scenario["name"] == scenario_name:
action_text = scenario["action"]
context_text = json.dumps(scenario["context"])
break
# Parse context
try:
context = json.loads(context_text) if context_text.strip() else {}
except:
context = {"environment": "production"}
# Process action
result = arf_engine.assess_action(action_text, context, license_key)
# Create displays
oss_html = create_oss_display(result)
enterprise_html = create_enterprise_display(result)
comparison_html = create_comparison_display(result)
# Update license status
license_info = result["license_info"]
if license_info.get("valid"):
status_html = f"""
ā
{license_info['name']} License
{license_info['message']}
"""
else:
status_html = """
šµ ARF OSS Edition
Advisory Mode - No license required
"""
return oss_html, enterprise_html, comparison_html, status_html
except Exception as e:
error_html = f"""
ā Error Processing Action
{str(e)}
"""
return error_html, error_html, error_html, ""
def create_oss_display(result):
"""Create OSS display"""
risk = result["risk_assessment"]
policy = result["policy_result"]
return f"""
šµ ARF OSS 3.3.9
Reliability
{(1-risk.score):.1%}
Risk Level
{risk.level.value}
Confidence
{risk.confidence:.1%}
š” Recommendation
{result['recommendation']}
ā ļø OSS Limitations
- Human decision required for all actions
- No mechanical enforcement
- {policy['total_violations']} policy violation(s) detected
- Processing time: {result['processing_time']}s
"""
def create_enterprise_display(result):
"""Create Enterprise display"""
risk = result["risk_assessment"]
gates = result["gate_results"]
license_info = result["license_info"]
# Calculate gate progress
required_gates = [g for g in gates if g.required]
passed_required = sum(1 for g in required_gates if g.passed)
# Create gates HTML
gates_html = ""
for gate in gates:
gate_class = "gate-passed" if gate.passed else "gate-failed"
gates_html += f"""
{gate.icon}
{gate.name}
{gate.message}
"""
return f"""
š” ARF Enterprise 3.3.9
{license_info.get('name', 'OSS')}
License Tier
{license_info.get('name', 'OSS')}
Gates Passed
{passed_required}/{len(required_gates)}
Enforcement
{license_info.get('enforcement', 'ADVISORY').replace('_', ' ').title()}
Mechanical Gates
{gates_html}
š Enterprise Benefits
- Mechanical enforcement with automated gates
- Processing time: {result['processing_time']}s
- {license_info.get('message', 'Enterprise-grade protection')}
"""
def create_comparison_display(result):
"""Create comparison display"""
policy = result["policy_result"]
gates = result["gate_results"]
license_info = result["license_info"]
oss_can_execute = not policy.get("blocked", False)
all_gates_passed = all(g.passed for g in gates if g.required)
if all_gates_passed and license_info.get("valid"):
enterprise_auth = "AUTONOMOUS_EXECUTION"
elif license_info.get("valid"):
enterprise_auth = "ADVISORY_ONLY"
else:
enterprise_auth = "OSS_ONLY"
return f"""
š Side-by-Side Comparison
šµ OSS 3.3.9
{'ā ļø' if not oss_can_execute else 'ā
'}
{'Advisory Only' if not oss_can_execute else 'Can Execute'}
š” Enterprise
{'ā
' if enterprise_auth == 'AUTONOMOUS_EXECUTION' else 'š¤' if enterprise_auth == 'ADVISORY_ONLY' else 'šµ'}
{enterprise_auth.replace('_', ' ').title()}
šÆ Enterprise Value Proposition
"""
def generate_trial(email):
"""Generate trial license"""
return arf_engine.generate_trial_license(email)
def clear_inputs():
"""Clear all inputs"""
return DEMO_SCENARIOS[0]["action"], json.dumps(DEMO_SCENARIOS[0]["context"], indent=2), "", "High Risk", "", "", "", ""
def update_scenario(scenario_name):
"""Update inputs based on scenario"""
for scenario in DEMO_SCENARIOS:
if scenario["name"] == scenario_name:
return scenario["action"], json.dumps(scenario["context"], indent=2)
return "", "{}"
# Connect events
process_btn.click(
process_action,
[scenario_select, action_input, context_input, license_input, quick_examples],
[oss_output, enterprise_output, comparison_output, license_status]
)
clear_btn.click(
clear_inputs,
outputs=[action_input, context_input, license_input, quick_examples, oss_output, enterprise_output, comparison_output, license_status]
)
scenario_select.change(
update_scenario,
[scenario_select],
[action_input, context_input]
)
trial_btn.click(
generate_trial,
[email_input],
[trial_output]
)
# Initial load
demo.load(
lambda: process_action(
DEMO_SCENARIOS[0]["name"],
DEMO_SCENARIOS[0]["action"],
json.dumps(DEMO_SCENARIOS[0]["context"]),
"",
"High Risk"
),
outputs=[oss_output, enterprise_output, comparison_output, license_status]
)
return demo
# ============================================================================
# MAIN EXECUTION
# ============================================================================
if __name__ == "__main__":
print("\nš Starting ARF 3.3.9 Demo")
try:
demo = create_demo_interface()
demo.launch(
server_name="0.0.0.0",
server_port=7860,
share=False,
debug=False,
show_error=True
)
except Exception as e:
print(f"ā Error launching demo: {e}")
traceback.print_exc()
# Fallback minimal interface
with gr.Blocks() as fallback:
gr.Markdown("""
# š¤ ARF 3.3.9 Demo
The main demo is experiencing technical difficulties.
**Key Value Proposition:**
- **OSS**: Advisory recommendations only
- **Enterprise**: Mechanical enforcement with automated gates
**Visit [arf.dev](https://arf.dev) for the full demo.**
""")
fallback.launch()