ai/workflow_troubleshooting/troubleshooting_engine.py

import asyncio
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
import json
import logging
import re
import traceback
from typing import Any, Dict, List, Optional, Set, Tuple
import uuid

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class IssueSeverity(Enum):
    """Severity levels for workflow automation issues"""

    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "info"


class IssueCategory(Enum):
    """Categories of workflow automation issues"""

    CONFIGURATION = "configuration"
    CONNECTIVITY = "connectivity"
    PERMISSIONS = "permissions"
    PERFORMANCE = "performance"
    DATA = "data"
    LOGIC = "logic"
    EXTERNAL_SERVICE = "external_service"
    TIMEOUT = "timeout"
    RESOURCE = "resource"


class TroubleshootingStep(Enum):
    """Steps in the troubleshooting process"""

    IDENTIFICATION = "identification"
    ANALYSIS = "analysis"
    DIAGNOSIS = "diagnosis"
    RESOLUTION = "resolution"
    VERIFICATION = "verification"


@dataclass
class WorkflowIssue:
    """Represents a detected workflow automation issue"""

    issue_id: str
    workflow_id: str
    category: IssueCategory
    severity: IssueSeverity
    description: str
    symptoms: List[str]
    root_cause: Optional[str] = None
    detection_time: datetime = None
    affected_components: List[str] = None
    metrics_impact: Dict[str, Any] = None

    def __post_init__(self):
        if self.detection_time is None:
            self.detection_time = datetime.now()
        if self.affected_components is None:
            self.affected_components = []
        if self.metrics_impact is None:
            self.metrics_impact = {}


@dataclass
class TroubleshootingSession:
    """Represents a troubleshooting session for workflow automation"""

    session_id: str
    workflow_id: str
    issues: List[WorkflowIssue]
    steps_completed: List[TroubleshootingStep]
    current_step: TroubleshootingStep
    recommendations: List[str]
    resolution_status: str = "in_progress"
    start_time: datetime = None
    end_time: Optional[datetime] = None

    def __post_init__(self):
        if self.start_time is None:
            self.start_time = datetime.now()


class WorkflowTroubleshootingEngine:
    """
    AI-Powered Workflow Automation Troubleshooting Engine
    Provides intelligent diagnosis and resolution for workflow automation issues
    """

    def __init__(self):
        self.sessions: Dict[str, TroubleshootingSession] = {}
        self.issue_patterns = self._initialize_issue_patterns()
        self.resolution_strategies = self._initialize_resolution_strategies()
        self.metrics_history: Dict[str, List[Dict[str, Any]]] = {}

    def _initialize_issue_patterns(self) -> Dict[str, Dict[str, Any]]:
        """Initialize patterns for detecting common workflow automation issues"""
        return {
            "connection_timeout": {
                "category": IssueCategory.CONNECTIVITY,
                "severity": IssueSeverity.HIGH,
                "patterns": [
                    r"timeout.*connection",
                    r"connection.*timed out",
                    r"failed to connect",
                    r"network.*unreachable",
                    r"database.*connection.*timeout",
                ],
                "symptoms": [
                    "Slow response times",
                    "Failed API calls",
                    "Network errors",
                ],
            },
            "authentication_failure": {
                "category": IssueCategory.PERMISSIONS,
                "severity": IssueSeverity.CRITICAL,
                "patterns": [
                    r"authentication.*failed",
                    r"unauthorized",
                    r"invalid.*token",
                    r"permission.*denied",
                ],
                "symptoms": [
                    "Access denied errors",
                    "Token expiration",
                    "Credential issues",
                ],
            },
            "data_validation_error": {
                "category": IssueCategory.DATA,
                "severity": IssueSeverity.MEDIUM,
                "patterns": [
                    r"invalid.*data",
                    r"validation.*error",
                    r"malformed.*request",
                    r"missing.*required",
                    r"invalid.*response.*format",
                ],
                "symptoms": [
                    "Data format errors",
                    "Missing required fields",
                    "Schema violations",
                ],
            },
            "performance_degradation": {
                "category": IssueCategory.PERFORMANCE,
                "severity": IssueSeverity.MEDIUM,
                "patterns": [
                    r"slow.*performance",
                    r"high.*latency",
                    r"response.*time.*high",
                    r"throughput.*low",
                ],
                "symptoms": [
                    "Increased response times",
                    "Reduced throughput",
                    "Resource exhaustion",
                ],
            },
            "workflow_logic_error": {
                "category": IssueCategory.LOGIC,
                "severity": IssueSeverity.HIGH,
                "patterns": [
                    r"logic.*error",
                    r"incorrect.*condition",
                    r"workflow.*stuck",
                    r"infinite.*loop",
                ],
                "symptoms": [
                    "Workflow hangs",
                    "Incorrect branching",
                    "Unexpected results",
                ],
            },
            "external_service_unavailable": {
                "category": IssueCategory.EXTERNAL_SERVICE,
                "severity": IssueSeverity.HIGH,
                "patterns": [
                    r"service.*unavailable",
                    r"api.*down",
                    r"external.*service.*error",
                    r"third.*party.*failure",
                    r"api.*call.*failed.*status.*500",
                ],
                "symptoms": [
                    "External API failures",
                    "Service outages",
                    "Dependency issues",
                ],
            },
        }

    def _initialize_resolution_strategies(self) -> Dict[str, List[str]]:
        """Initialize resolution strategies for different issue types"""
        return {
            "connection_timeout": [
                "Check network connectivity and firewall settings",
                "Verify API endpoint URLs and availability",
                "Increase timeout configurations if appropriate",
                "Implement retry mechanisms with exponential backoff",
                "Monitor network latency and bandwidth",
            ],
            "authentication_failure": [
                "Verify API keys, tokens, and credentials",
                "Check token expiration and refresh mechanisms",
                "Validate OAuth configurations and scopes",
                "Review permission settings and access controls",
                "Test authentication flows with valid credentials",
            ],
            "data_validation_error": [
                "Validate input data formats and schemas",
                "Implement comprehensive data sanitization",
                "Add missing required fields with default values",
                "Review data transformation logic",
                "Enhance error handling for malformed data",
            ],
            "performance_degradation": [
                "Analyze workflow execution metrics and bottlenecks",
                "Optimize database queries and API calls",
                "Implement caching strategies for repeated operations",
                "Scale resources based on workload patterns",
                "Review and optimize workflow logic",
            ],
            "workflow_logic_error": [
                "Review workflow conditions and branching logic",
                "Add comprehensive logging and debugging",
                "Test edge cases and boundary conditions",
                "Implement timeout mechanisms for long-running operations",
                "Validate workflow state transitions",
            ],
            "external_service_unavailable": [
                "Implement circuit breaker patterns for external services",
                "Add fallback mechanisms and alternative workflows",
                "Monitor external service health and status",
                "Cache responses to reduce dependency on external services",
                "Implement graceful degradation strategies",
            ],
        }

    def start_troubleshooting_session(
        self, workflow_id: str, error_logs: List[str]
    ) -> TroubleshootingSession:
        """Start a new troubleshooting session for a workflow"""
        session_id = str(uuid.uuid4())

        # Analyze error logs to identify issues
        issues = self._analyze_error_logs(workflow_id, error_logs)

        session = TroubleshootingSession(
            session_id=session_id,
            workflow_id=workflow_id,
            issues=issues,
            steps_completed=[TroubleshootingStep.IDENTIFICATION],
            current_step=TroubleshootingStep.ANALYSIS,
            recommendations=[],
        )

        self.sessions[session_id] = session
        logger.info(
            f"Started troubleshooting session {session_id} for workflow {workflow_id}"
        )

        return session

    def _analyze_error_logs(
        self, workflow_id: str, error_logs: List[str]
    ) -> List[WorkflowIssue]:
        """Analyze error logs to identify workflow automation issues"""
        issues = []

        for log_entry in error_logs:
            for issue_type, pattern_info in self.issue_patterns.items():
                for pattern in pattern_info["patterns"]:
                    if re.search(pattern, log_entry, re.IGNORECASE):
                        issue = WorkflowIssue(
                            issue_id=str(uuid.uuid4()),
                            workflow_id=workflow_id,
                            category=pattern_info["category"],
                            severity=pattern_info["severity"],
                            description=f"Detected {issue_type} issue in workflow {workflow_id}",
                            symptoms=pattern_info["symptoms"],
                            affected_components=["Workflow Engine", "API Connectors"],
                        )
                        issues.append(issue)
                        break

        # Remove duplicates based on description
        unique_issues = []
        seen_descriptions = set()
        for issue in issues:
            if issue.description not in seen_descriptions:
                unique_issues.append(issue)
                seen_descriptions.add(issue.description)

        return unique_issues

    async def analyze_workflow_metrics(
        self, session_id: str, metrics: Dict[str, Any]
    ) -> List[WorkflowIssue]:
        """Analyze workflow metrics to identify performance and operational issues"""
        session = self.sessions.get(session_id)
        if not session:
            raise ValueError(f"Session {session_id} not found")

        issues = []

        # Store metrics for historical analysis
        if session.workflow_id not in self.metrics_history:
            self.metrics_history[session.workflow_id] = []
        self.metrics_history[session.workflow_id].append(
            {"timestamp": datetime.now(), "metrics": metrics}
        )

        # Analyze performance metrics
        if metrics.get("avg_response_time", 0) > 5.0:  # seconds
            issue = WorkflowIssue(
                issue_id=str(uuid.uuid4()),
                workflow_id=session.workflow_id,
                category=IssueCategory.PERFORMANCE,
                severity=IssueSeverity.MEDIUM,
                description="High response times detected in workflow execution",
                symptoms=["Slow performance", "Increased latency"],
                metrics_impact={"avg_response_time": metrics["avg_response_time"]},
            )
            issues.append(issue)

        # Analyze error rates
        if metrics.get("error_rate", 0) > 0.1:  # 10% error rate
            issue = WorkflowIssue(
                issue_id=str(uuid.uuid4()),
                workflow_id=session.workflow_id,
                category=IssueCategory.LOGIC,
                severity=IssueSeverity.HIGH,
                description="High error rate detected in workflow execution",
                symptoms=["Frequent failures", "Unreliable execution"],
                metrics_impact={"error_rate": metrics["error_rate"]},
            )
            issues.append(issue)

        # Analyze completion rates
        if metrics.get("completion_rate", 1.0) < 0.8:  # 80% completion rate
            issue = WorkflowIssue(
                issue_id=str(uuid.uuid4()),
                workflow_id=session.workflow_id,
                category=IssueCategory.LOGIC,
                severity=IssueSeverity.HIGH,
                description="Low completion rate detected in workflow execution",
                symptoms=["Workflow interruptions", "Incomplete executions"],
                metrics_impact={"completion_rate": metrics["completion_rate"]},
            )
            issues.append(issue)

        # Add new issues to session
        session.issues.extend(issues)

        # Update session step
        if TroubleshootingStep.ANALYSIS not in session.steps_completed:
            session.steps_completed.append(TroubleshootingStep.ANALYSIS)
            session.current_step = TroubleshootingStep.DIAGNOSIS

        return issues

    def diagnose_root_causes(self, session_id: str) -> List[str]:
        """Diagnose root causes for identified issues"""
        session = self.sessions.get(session_id)
        if not session:
            raise ValueError(f"Session {session_id} not found")

        root_causes = []

        for issue in session.issues:
            # Generate root cause analysis based on issue type and patterns
            if issue.category == IssueCategory.CONNECTIVITY:
                issue.root_cause = (
                    "Network connectivity issues or service unavailability"
                )
                root_causes.append(f"Connectivity issue: {issue.root_cause}")

            elif issue.category == IssueCategory.PERMISSIONS:
                issue.root_cause = (
                    "Authentication or authorization configuration problems"
                )
                root_causes.append(f"Permission issue: {issue.root_cause}")

            elif issue.category == IssueCategory.PERFORMANCE:
                issue.root_cause = "Resource constraints or inefficient workflow design"
                root_causes.append(f"Performance issue: {issue.root_cause}")

            elif issue.category == IssueCategory.DATA:
                issue.root_cause = "Data format, validation, or transformation issues"
                root_causes.append(f"Data issue: {issue.root_cause}")

            elif issue.category == IssueCategory.LOGIC:
                issue.root_cause = (
                    "Workflow logic errors or conditional branching issues"
                )
                root_causes.append(f"Logic issue: {issue.root_cause}")

            elif issue.category == IssueCategory.EXTERNAL_SERVICE:
                issue.root_cause = (
                    "Dependency on external services with availability issues"
                )
                root_causes.append(f"External service issue: {issue.root_cause}")

        # Update session step
        if TroubleshootingStep.DIAGNOSIS not in session.steps_completed:
            session.steps_completed.append(TroubleshootingStep.DIAGNOSIS)
            session.current_step = TroubleshootingStep.RESOLUTION

        return root_causes

    def generate_recommendations(self, session_id: str) -> List[str]:
        """Generate resolution recommendations for identified issues"""
        session = self.sessions.get(session_id)
        if not session:
            raise ValueError(f"Session {session_id} not found")

        recommendations = []

        for issue in session.issues:
            # Map issue patterns to resolution strategies
            for issue_type, strategies in self.resolution_strategies.items():
                if any(
                    pattern in issue.description.lower()
                    for pattern in self.issue_patterns[issue_type]["patterns"]
                ):
                    recommendations.extend(strategies)
                    break

        # Add general recommendations
        general_recommendations = [
            "Implement comprehensive logging and monitoring",
            "Add automated health checks for all workflow components",
            "Create backup and recovery procedures",
            "Establish alerting mechanisms for critical issues",
            "Document troubleshooting procedures for common problems",
        ]

        recommendations.extend(general_recommendations)

        # Update session recommendations
        session.recommendations = list(set(recommendations))  # Remove duplicates

        # Update session step
        if TroubleshootingStep.RESOLUTION not in session.steps_completed:
            session.steps_completed.append(TroubleshootingStep.RESOLUTION)
            session.current_step = TroubleshootingStep.VERIFICATION

        return session.recommendations

    def verify_resolution(
        self, session_id: str, test_results: Dict[str, bool]
    ) -> Dict[str, Any]:
        """Verify that issues have been resolved"""
        session = self.sessions.get(session_id)
        if not session:
            raise ValueError(f"Session {session_id} not found")

        verification_results = {
            "session_id": session_id,
            "workflow_id": session.workflow_id,
            "verification_time": datetime.now(),
            "tests_passed": [],
            "tests_failed": [],
            "overall_status": "pending",
        }

        # Check test results
        for test_name, test_passed in test_results.items():
            if test_passed:
                verification_results["tests_passed"].append(test_name)
            else:
                verification_results["tests_failed"].append(test_name)

        # Determine overall status
        if not verification_results["tests_failed"]:
            verification_results["overall_status"] = "resolved"
            session.resolution_status = "resolved"
        else:
            verification_results["overall_status"] = "partial"
            session.resolution_status = "partial"

        # Update session
        session.end_time = datetime.now()
        if TroubleshootingStep.VERIFICATION not in session.steps_completed:
            session.steps_completed.append(TroubleshootingStep.VERIFICATION)

        return verification_results

    def get_session_summary(self, session_id: str) -> Dict[str, Any]:
        """Get comprehensive summary of troubleshooting session"""
        session = self.sessions.get(session_id)
        if not session:
            raise ValueError(f"Session {session_id} not found")

        return {
            "session_id": session_id,
            "workflow_id": session.workflow_id,
            "start_time": session.start_time.isoformat(),
            "end_time": session.end_time.isoformat() if session.end_time else None,
            "duration": (session.end_time - session.start_time).total_seconds()
            if session.end_time
            else None,
            "issues_found": len(session.issues),
            "issues_by_severity": self._count_issues_by_severity(session.issues),
            "issues_by_category": self._count_issues_by_category(session.issues),
            "steps_completed": [step.value for step in session.steps_completed],
            "current_step": session.current_step.value,
            "resolution_status": session.resolution_status,
            "recommendations_count": len(session.recommendations),
        }

    def _count_issues_by_severity(self, issues: List[WorkflowIssue]) -> Dict[str, int]:
        """Count issues by severity level"""
        counts = {}
        for severity in IssueSeverity:
            counts[severity.value] = len(
                [issue for issue in issues if issue.severity == severity]
            )
        return counts

    def _count_issues_by_category(self, issues: List[WorkflowIssue]) -> Dict[str, int]:
        """Count issues by category"""
        counts = {}
        for category in IssueCategory:
            counts[category.value] = len(
                [issue for issue in issues if issue.category == category]
            )
        return counts

    def get_workflow_health_score(self, workflow_id: str) -> Dict[str, Any]:
        """Calculate health score for a workflow based on historical metrics"""
        if workflow_id not in self.metrics_history:
            return {
                "health_score": 100,
                "status": "unknown",
                "reason": "No metrics available",
            }

        metrics_history = self.metrics_history[workflow_id]
        if not metrics_history:
            return {
                "health_score": 100,
                "status": "unknown",
                "reason": "No metrics available",
            }

        # Calculate health score based on recent metrics