demo/mock_arf.py

"""
Mock ARF components for demo purposes
In production, these would use the real agentic-reliability-framework package
"""
import time
import json
from typing import Dict, Any, List
import random

def simulate_arf_analysis(scenario: Dict[str, Any]) -> Dict[str, Any]:
    """Simulate ARF analysis pipeline"""
    return {
        "analysis_complete": True,
        "anomaly_detected": True,
        "severity": "critical",
        "root_cause": scenario.get('root_cause', 'unknown'),
        "pattern_detected": True,
        "pattern_confidence": random.uniform(0.8, 0.95),
        "analysis_timestamp": time.time(),
        "processing_time_ms": random.randint(200, 500)
    }

def run_rag_similarity_search(scenario: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Simulate RAG similarity search"""
    component = scenario.get('component', 'redis_cache')
    
    # Mock similar incidents based on scenario
    similar_incidents = []
    
    # Generate 3-5 similar incidents
    for i in range(random.randint(3, 5)):
        similarity = random.uniform(0.7, 0.95)
        success = similarity > 0.8
        
        incident = {
            "incident_id": f"inc_{int(time.time())}_{i}",
            "component": component,
            "similarity_score": similarity,
            "success": success,
            "resolution": "scale_out" if component == "redis_cache" else "restart",
            "actions_taken": ["scale_out", "adjust_cache_ttl"] if component == "redis_cache" else ["restart_container"],
            "resolution_time_minutes": random.uniform(5, 15),
            "timestamp": time.time() - random.randint(86400, 2592000)  # 1-30 days ago
        }
        
        if success:
            incident["cost_savings"] = random.randint(1000, 10000)
        
        similar_incidents.append(incident)
    
    # Sort by similarity
    similar_incidents.sort(key=lambda x: x['similarity_score'], reverse=True)
    
    return similar_incidents

def calculate_pattern_confidence(scenario: Dict[str, Any], similar_incidents: List[Dict[str, Any]]) -> float:
    """Calculate pattern detection confidence"""
    if not similar_incidents:
        return 0.7
    
    # Base confidence
    base_confidence = 0.75
    
    # Boost based on number of similar incidents
    incident_boost = min(0.15, len(similar_incidents) * 0.03)
    
    # Boost based on average similarity
    avg_similarity = sum(i['similarity_score'] for i in similar_incidents) / len(similar_incidents)
    similarity_boost = avg_similarity * 0.1
    
    # Boost based on success rate
    success_rate = sum(1 for i in similar_incidents if i['success']) / len(similar_incidents)
    success_boost = success_rate * 0.1
    
    total_confidence = base_confidence + incident_boost + similarity_boost + success_boost
    
    return min(0.98, total_confidence)

def create_mock_healing_intent(scenario: Dict[str, Any], 
                               similar_incidents: List[Dict[str, Any]], 
                               confidence: float = 0.85) -> Dict[str, Any]:
    """Create a mock HealingIntent object"""
    
    # Determine action based on scenario
    component = scenario.get('component', 'redis_cache')
    if component == 'redis_cache':
        action = 'scale_out'
        parameters = {'scale_factor': 2, 'cache_ttl': 300}
        justification = "Scale Redis cluster and adjust cache TTL based on historical pattern"
    elif component == 'database':
        action = 'optimize_connections'
        parameters = {'max_connections': 200, 'connection_timeout': 30}
        justification = "Optimize database connection pool settings"
    else:
        action = 'restart_container'
        parameters = {}
        justification = "Restart container to resolve memory issues"
    
    # Calculate RAG similarity score
    rag_score = None
    if similar_incidents:
        rag_score = sum(i['similarity_score'] for i in similar_incidents[:3]) / min(3, len(similar_incidents))
    
    return {
        "action": action,
        "component": component,
        "parameters": parameters,
        "justification": justification,
        "confidence": confidence,
        "incident_id": scenario.get('incident_id', f"inc_{int(time.time())}"),
        "detected_at": time.time(),
        "similar_incidents": similar_incidents,
        "rag_similarity_score": rag_score,
        "source": "oss_analysis",
        "intent_id": f"intent_{int(time.time())}",
        "created_at": time.time(),
        "status": "created",
        "oss_edition": "community",
        "requires_enterprise": True,
        "execution_allowed": False,
        "deterministic_id": f"intent_{hash(json.dumps(parameters, sort_keys=True)) % 10000:04d}"
    }