""" ARF 3.3.9 Demo - Final Clean Version Fixes all display issues """ import gradio as gr import json import time import uuid import hashlib import random from datetime import datetime, timedelta print("=" * 70) print("🚀 ARF 3.3.9 Demo - Final Version") print("=" * 70) class ARFDemo: def __init__(self): self.stats = { "total_processed": 0, "blocked_actions": 0, "autonomous_executions": 0, "avg_processing_time": 0.0 } print("✅ ARF Demo initialized") def assess_action(self, action, context, license_key=None): """Assess an action""" start_time = time.time() # Parse context if isinstance(context, str): try: context = json.loads(context) if context.strip() else {} except: context = {"environment": "production"} # 1. Calculate risk risk_score, confidence = self._calculate_risk(action, context) # 2. Check policies policy_result = self._check_policies(action, context, risk_score) # 3. Validate license license_info = self._validate_license(license_key) # 4. Evaluate gates gate_results = self._evaluate_gates(risk_score, confidence, policy_result, license_info) # 5. Generate recommendation recommendation = self._generate_recommendation(policy_result, gate_results, license_info) processing_time = time.time() - start_time # Update stats self._update_stats(policy_result, gate_results, processing_time, license_info) return { "action": action, "risk_score": risk_score, "confidence": confidence, "policy_result": policy_result, "license_info": license_info, "gate_results": gate_results, "recommendation": recommendation, "processing_time": processing_time, "stats": self.get_stats() } def _calculate_risk(self, action, context): """Calculate risk score""" action_lower = action.lower() # Base risk risk = 0.3 # Adjust for destructive operations if "drop database" in action_lower: risk = 0.95 elif "delete from" in action_lower: risk = 0.85 elif "truncate" in action_lower: risk = 0.80 # Adjust for environment if context.get("environment") == "production": risk = min(1.0, risk + 0.2) # Confidence confidence = 0.95 if "drop" in action_lower else 0.85 return round(risk, 3), round(confidence, 3) def _check_policies(self, action, context, risk_score): """Check against policies""" violations = [] 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" }) if risk_score > 0.7: 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): """Validate license key""" if not license_key: return { "tier": "oss", "valid": False, "name": "OSS Edition", "enforcement": "advisory", "color": "#1E88E5", "icon": "🔵", "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", "color": "#FFB300", "icon": "🟡", "message": "Trial License Active (14 days)" } elif "PRO" in license_key: return { "tier": "professional", "valid": True, "name": "Professional", "enforcement": "autonomous", "color": "#FF9800", "icon": "🟠", "message": "Professional License Active" } elif "ENTERPRISE" in license_key: return { "tier": "enterprise", "valid": True, "name": "Enterprise", "enforcement": "full_mechanical", "color": "#FF6F00", "icon": "🔶", "message": "Enterprise License Active" } return { "tier": "invalid", "valid": False, "name": "Invalid", "enforcement": "advisory", "color": "#9E9E9E", "icon": "❌", "message": "Invalid License Key" } def _evaluate_gates(self, risk_score, confidence, policy_result, license_info): """Evaluate mechanical gates""" gates = [] # Gate 1: License Validation gates.append({ "name": "License Validation", "passed": license_info["valid"], "message": license_info["message"], "required": True, "weight": 0.3 }) # Gate 2: Risk Assessment risk_threshold = 0.8 if license_info["tier"] in ["professional", "enterprise"] else 0.7 gates.append({ "name": "Risk Assessment", "passed": risk_score <= risk_threshold, "message": f"Risk {risk_score:.1%} ≤ {risk_threshold:.0%}", "required": True, "weight": 0.3 }) # Gate 3: Confidence Threshold gates.append({ "name": "Confidence Threshold", "passed": confidence >= 0.7, "message": f"Confidence {confidence:.1%} ≥ 70%", "required": True, "weight": 0.2 }) # Gate 4: Policy Compliance gates.append({ "name": "Policy Compliance", "passed": not policy_result["blocked"], "message": f"{policy_result['total_violations']} policy violation(s)", "required": True, "weight": 0.2 }) return gates def _generate_recommendation(self, policy_result, gate_results, license_info): """Generate recommendation""" if policy_result["blocked"]: return "🚫 BLOCKED: Action violates safety policies" all_gates_passed = all(g["passed"] for g in gate_results if g["required"]) if not license_info["valid"]: return "🔵 OSS ADVISORY: Human review recommended" elif all_gates_passed and license_info["tier"] in ["professional", "enterprise"]: return "🟡 ENTERPRISE APPROVED: Autonomous execution permitted" elif all_gates_passed and license_info["tier"] == "trial": return "🎁 TRIAL: Gates passed (advisory only)" else: return "⚠️ REVIEW REQUIRED: Additional validation needed" def _update_stats(self, policy_result, gate_results, processing_time, license_info): """Update statistics""" self.stats["total_processed"] += 1 if policy_result["blocked"]: self.stats["blocked_actions"] += 1 if all(g["passed"] for g in gate_results if g["required"]) and license_info["tier"] != "oss": 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): """Get statistics""" stats = self.stats.copy() total = stats["total_processed"] blocked = stats["blocked_actions"] autonomous = stats["autonomous_executions"] 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['avg_processing_time']*1000:.0f}ms" return stats def generate_trial_license(self, email): """Generate trial license""" if not email or "@" not in email: return {"success": False, "message": "Please enter a valid email"} 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") } # Demo scenarios DEMO_SCENARIOS = [ { "name": "🔥 High-Risk Database Operation", "action": "DROP DATABASE production_users CASCADE", "context": {"environment": "production", "criticality": "critical"} }, { "name": "✅ Safe Service Deployment", "action": "deploy_service payment_api:v2.3.1 to staging", "context": {"environment": "staging", "service": "payment_api"} }, { "name": "🔧 Configuration Update", "action": "UPDATE config SET timeout_ms=30000", "context": {"environment": "production", "service": "api"} } ] # Initialize engine arf_demo = ARFDemo() # Create interface 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; } .stat-card { background: white; padding: 15px; border-radius: 10px; text-align: center; box-shadow: 0 2px 8px rgba(0,0,0,0.1); } .oss-panel { border: 2px solid #1E88E5; border-radius: 10px; padding: 20px; background: #E3F2FD; margin-bottom: 20px; } .enterprise-panel { border: 2px solid #FFB300; border-radius: 10px; padding: 20px; background: #FFF8E1; margin-bottom: 20px; } .gate-passed { background: #E8F5E9; padding: 10px; margin: 5px 0; border-radius: 5px; border-left: 4px solid #4CAF50; } .gate-failed { background: #FFEBEE; padding: 10px; margin: 5px 0; border-radius: 5px; border-left: 4px solid #F44336; } """ ) as demo: # Header gr.Markdown(""" # 🤖 Agentic Reliability Framework 3.3.9 ### From Advisory to Mechanical Enforcement """) # Statistics stats = arf_demo.get_stats() gr.Markdown(f"""
Total Assessments
{stats['total_processed']}
Risks Prevented
{stats['blocked_percentage']}%
Autonomous Rate
{stats['autonomous_percentage']}%
Processing Speed
{stats['processing_speed']}
""") # Main controls with gr.Row(): with gr.Column(scale=2): scenario = gr.Dropdown( choices=[s["name"] for s in DEMO_SCENARIOS], label="Select Demo Scenario", value=DEMO_SCENARIOS[0]["name"] ) action = gr.Textbox( label="Action to Evaluate", value=DEMO_SCENARIOS[0]["action"], lines=2 ) context = gr.Textbox( label="Context (JSON)", value=json.dumps(DEMO_SCENARIOS[0]["context"], indent=2), lines=3 ) license_key = gr.Textbox( label="License Key (leave blank for OSS, use ARF-TRIAL-XXX for trial)", value="" ) with gr.Row(): process_btn = gr.Button("🚀 Process Action", variant="primary") clear_btn = gr.Button("🔄 Clear") with gr.Column(scale=1): # License status license_status = gr.HTML("""
🔵 ARF OSS Edition
Advisory Mode
""") # Results with gr.Row(): oss_output = gr.HTML(label="🔵 ARF OSS Results") enterprise_output = gr.HTML(label="🟡 ARF Enterprise Results") # Trial section with gr.Accordion("Get Trial License", open=False): email = gr.Textbox(label="Email") trial_btn = gr.Button("Get Trial") trial_output = gr.JSON(label="Trial License") # Functions def process_action(scenario_name, action_text, context_text, license_text): """Process action""" try: # Use scenario if selected for s in DEMO_SCENARIOS: if s["name"] == scenario_name: action_text = s["action"] context_text = json.dumps(s["context"]) break # Process action result = arf_demo.assess_action(action_text, context_text, license_text) # Create displays oss_html = create_oss_display(result) enterprise_html = create_enterprise_display(result) # Update license status license_info = result["license_info"] status_html = f"""
{license_info['icon']} {license_info['name']}
{license_info['message']}
""" return oss_html, enterprise_html, status_html except Exception as e: error_html = f"
Error: {str(e)}
" return error_html, error_html, "" def create_oss_display(result): """Create OSS display""" risk_score = result["risk_score"] confidence = result["confidence"] policy = result["policy_result"] risk_color = "#F44336" if risk_score > 0.7 else "#FF9800" if risk_score > 0.4 else "#4CAF50" risk_level = "HIGH" if risk_score > 0.7 else "MEDIUM" if risk_score > 0.4 else "LOW" return f"""

🔵 ARF OSS 3.3.9

Reliability
{(1-risk_score):.1%}
Risk Level
{risk_level}
Confidence
{confidence:.1%}
💡 Recommendation
{result['recommendation']}
⚠️ OSS Limitations
""" def create_enterprise_display(result): """Create Enterprise display""" 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" icon = "✅" if gate["passed"] else "❌" gates_html += f"""
{icon}
{gate['name']}
{gate['message']}
""" return f"""

🟡 ARF Enterprise 3.3.9

{license_info['name'].upper()}
License
{license_info['name']}
Gates Passed
{passed_required}/{len(required_gates)}
Enforcement
{license_info['enforcement'].title()}
Mechanical Gates
{gates_html}
🚀 Enterprise Benefits
""" def generate_trial(email_text): """Generate trial license""" return arf_demo.generate_trial_license(email_text) # Connect events process_btn.click( process_action, [scenario, action, context, license_key], [oss_output, enterprise_output, license_status] ) trial_btn.click( generate_trial, [email], [trial_output] ) # Initial load demo.load( lambda: process_action( DEMO_SCENARIOS[0]["name"], DEMO_SCENARIOS[0]["action"], json.dumps(DEMO_SCENARIOS[0]["context"]), "" ), outputs=[oss_output, enterprise_output, license_status] ) # Launch if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860)