enact_core/workflow/

health.rs

//! Run Health Watchdog - Monitor and detect stuck/abandoned runs
//!
//! Implements Antfarm-style medic checks:
//! - Stuck running steps
//! - Abandoned claims
//! - Dead runs with no state movement
//! - Orphan scheduler jobs

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// Health check configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthCheckConfig {
    /// Whether health checks are enabled
    #[serde(default = "default_true")]
    pub enabled: bool,
    /// Threshold for considering a step stuck (seconds)
    #[serde(default = "default_stuck_threshold")]
    pub stuck_threshold_seconds: i64,
    /// Threshold for considering a run abandoned (minutes)
    #[serde(default = "default_abandoned_threshold")]
    pub abandoned_threshold_minutes: i64,
    /// Check interval (seconds)
    #[serde(default = "default_check_interval")]
    pub check_interval_seconds: u64,
}

fn default_true() -> bool {
    true
}

fn default_stuck_threshold() -> i64 {
    300 // 5 minutes
}

fn default_abandoned_threshold() -> i64 {
    30 // 30 minutes
}

fn default_check_interval() -> u64 {
    60 // 1 minute
}

impl Default for HealthCheckConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            stuck_threshold_seconds: 300,
            abandoned_threshold_minutes: 30,
            check_interval_seconds: 60,
        }
    }
}

/// Run state for health monitoring
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MonitoredRunState {
    /// Run is pending/queued
    Pending,
    /// Run is currently executing
    Running,
    /// Run is paused/waiting
    Paused,
    /// Run completed successfully
    Completed,
    /// Run failed
    Failed,
    /// Run was cancelled
    Cancelled,
    /// Run timed out
    TimedOut,
}

/// Information about a running step
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StepExecution {
    /// Step ID
    pub step_id: String,
    /// Step name
    pub step_name: String,
    /// When step started
    pub started_at: DateTime<Utc>,
    /// Last activity timestamp
    pub last_activity: DateTime<Utc>,
    /// Step state
    pub state: StepState,
}

/// Step execution state
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StepState {
    /// Step is executing
    Executing,
    /// Step is waiting (e.g., for user input)
    Waiting,
    /// Step is in retry backoff
    Retrying,
    /// Step is blocked
    Blocked,
}

/// Monitored run information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MonitoredRun {
    /// Run ID
    pub run_id: String,
    /// Workflow name
    pub workflow_name: String,
    /// Current state
    pub state: MonitoredRunState,
    /// When run started
    pub started_at: DateTime<Utc>,
    /// Last state change
    pub last_state_change: DateTime<Utc>,
    /// Current step (if running)
    pub current_step: Option<StepExecution>,
    /// Number of state changes
    pub state_change_count: usize,
    /// Retry count
    pub retry_count: u32,
    /// Whether run is orphaned (no active process)
    #[serde(default)]
    pub is_orphaned: bool,
    /// Last health check
    pub last_health_check: Option<DateTime<Utc>>,
}

/// Health issue detected
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthIssue {
    /// Issue type
    pub issue_type: IssueType,
    /// Run ID affected
    pub run_id: String,
    /// Description
    pub description: String,
    /// When issue was detected
    pub detected_at: DateTime<Utc>,
    /// Severity
    pub severity: IssueSeverity,
    /// Recommended action
    pub recommended_action: RecommendedAction,
}

/// Types of health issues
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum IssueType {
    /// Step is stuck
    StuckStep,
    /// Run is abandoned
    AbandonedRun,
    /// Run has no state movement
    DeadRun,
    /// Run is orphaned
    OrphanedRun,
    /// Step retry exhaustion
    RetryExhaustion,
}

/// Issue severity
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, PartialOrd, Ord)]
#[serde(rename_all = "lowercase")]
pub enum IssueSeverity {
    Info,
    Warning,
    Error,
    Critical,
}

/// Recommended action for an issue
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum RecommendedAction {
    /// Alert operator
    Alert,
    /// Auto-retry
    AutoRetry,
    /// Terminate run
    Terminate,
    /// Escalate to human
    Escalate,
    /// Collect diagnostics
    CollectDiagnostics,
    /// No action needed
    None,
}

/// Health check result
#[derive(Debug, Clone)]
pub struct HealthCheckResult {
    /// Issues found
    pub issues: Vec<HealthIssue>,
    /// Runs checked
    pub runs_checked: usize,
    /// Timestamp
    pub checked_at: DateTime<Utc>,
}

/// Run health watchdog
pub struct RunHealthWatchdog {
    config: HealthCheckConfig,
    runs: HashMap<String, MonitoredRun>,
}

impl RunHealthWatchdog {
    /// Create a new watchdog
    pub fn new(config: HealthCheckConfig) -> Self {
        Self {
            config,
            runs: HashMap::new(),
        }
    }

    /// Register a run for monitoring
    pub fn register_run(&mut self, run_id: String, workflow_name: String) {
        let now = Utc::now();
        self.runs.insert(
            run_id.clone(),
            MonitoredRun {
                run_id,
                workflow_name,
                state: MonitoredRunState::Pending,
                started_at: now,
                last_state_change: now,
                current_step: None,
                state_change_count: 0,
                retry_count: 0,
                is_orphaned: false,
                last_health_check: None,
            },
        );
    }

    /// Update run state
    pub fn update_run_state(&mut self, run_id: &str, new_state: MonitoredRunState) {
        if let Some(run) = self.runs.get_mut(run_id) {
            run.state = new_state;
            run.last_state_change = Utc::now();
            run.state_change_count += 1;
        }
    }

    /// Update current step
    pub fn update_current_step(
        &mut self,
        run_id: &str,
        step_id: &str,
        step_name: &str,
        step_state: StepState,
    ) {
        let now = Utc::now();
        if let Some(run) = self.runs.get_mut(run_id) {
            run.current_step = Some(StepExecution {
                step_id: step_id.to_string(),
                step_name: step_name.to_string(),
                started_at: now,
                last_activity: now,
                state: step_state,
            });
            run.last_state_change = now;
        }
    }

    /// Record step activity
    pub fn record_step_activity(&mut self, run_id: &str) {
        if let Some(run) = self.runs.get_mut(run_id) {
            if let Some(step) = &mut run.current_step {
                step.last_activity = Utc::now();
            }
        }
    }

    /// Mark run as orphaned
    pub fn mark_orphaned(&mut self, run_id: &str) {
        if let Some(run) = self.runs.get_mut(run_id) {
            run.is_orphaned = true;
        }
    }

    /// Perform health checks on all runs
    pub fn check_health(&self) -> HealthCheckResult {
        let mut issues = Vec::new();
        let now = Utc::now();

        for run in self.runs.values() {
            // Skip completed/failed/cancelled runs
            match run.state {
                MonitoredRunState::Completed
                | MonitoredRunState::Failed
                | MonitoredRunState::Cancelled => continue,
                _ => {}
            }

            // Check for stuck steps
            if let Some(step) = &run.current_step {
                let step_duration = now.signed_duration_since(step.started_at);
                let inactive_duration = now.signed_duration_since(step.last_activity);

                // Check if step has been running too long
                if step_duration.num_seconds() > self.config.stuck_threshold_seconds {
                    issues.push(HealthIssue {
                        issue_type: IssueType::StuckStep,
                        run_id: run.run_id.clone(),
                        description: format!(
                            "Step '{}' has been executing for {} seconds (threshold: {}s)",
                            step.step_name,
                            step_duration.num_seconds(),
                            self.config.stuck_threshold_seconds
                        ),
                        detected_at: now,
                        severity: IssueSeverity::Error,
                        recommended_action: RecommendedAction::Escalate,
                    });
                }

                // Check for inactive steps (no activity)
                if inactive_duration.num_seconds() > self.config.stuck_threshold_seconds {
                    issues.push(HealthIssue {
                        issue_type: IssueType::DeadRun,
                        run_id: run.run_id.clone(),
                        description: format!(
                            "Step '{}' has had no activity for {} seconds",
                            step.step_name,
                            inactive_duration.num_seconds()
                        ),
                        detected_at: now,
                        severity: IssueSeverity::Warning,
                        recommended_action: RecommendedAction::CollectDiagnostics,
                    });
                }
            }

            // Check for abandoned runs
            let run_duration = now.signed_duration_since(run.started_at);
            if run_duration.num_minutes() > self.config.abandoned_threshold_minutes {
                issues.push(HealthIssue {
                    issue_type: IssueType::AbandonedRun,
                    run_id: run.run_id.clone(),
                    description: format!(
                        "Run has been active for {} minutes (threshold: {}m)",
                        run_duration.num_minutes(),
                        self.config.abandoned_threshold_minutes
                    ),
                    detected_at: now,
                    severity: IssueSeverity::Warning,
                    recommended_action: RecommendedAction::Alert,
                });
            }

            // Check for dead runs (no state changes)
            let since_last_change = now.signed_duration_since(run.last_state_change);
            if since_last_change.num_minutes() > self.config.abandoned_threshold_minutes {
                issues.push(HealthIssue {
                    issue_type: IssueType::DeadRun,
                    run_id: run.run_id.clone(),
                    description: format!(
                        "No state changes for {} minutes",
                        since_last_change.num_minutes()
                    ),
                    detected_at: now,
                    severity: IssueSeverity::Error,
                    recommended_action: RecommendedAction::Terminate,
                });
            }

            // Check for orphaned runs
            if run.is_orphaned {
                issues.push(HealthIssue {
                    issue_type: IssueType::OrphanedRun,
                    run_id: run.run_id.clone(),
                    description: "Run process no longer exists".to_string(),
                    detected_at: now,
                    severity: IssueSeverity::Critical,
                    recommended_action: RecommendedAction::Terminate,
                });
            }
        }

        HealthCheckResult {
            issues,
            runs_checked: self.runs.len(),
            checked_at: now,
        }
    }

    /// Get a specific run
    pub fn get_run(&self, run_id: &str) -> Option<&MonitoredRun> {
        self.runs.get(run_id)
    }

    /// Get all monitored runs
    pub fn get_all_runs(&self) -> &HashMap<String, MonitoredRun> {
        &self.runs
    }

    /// Remove a run from monitoring
    pub fn remove_run(&mut self, run_id: &str) {
        self.runs.remove(run_id);
    }

    /// Get runs by state
    pub fn get_runs_by_state(&self, state: MonitoredRunState) -> Vec<&MonitoredRun> {
        self.runs.values().filter(|r| r.state == state).collect()
    }

    /// Get runs that need attention (have issues)
    pub fn get_runs_needing_attention(&self) -> Vec<&MonitoredRun> {
        let health_result = self.check_health();
        let run_ids: std::collections::HashSet<_> =
            health_result.issues.iter().map(|i| &i.run_id).collect();

        self.runs
            .values()
            .filter(|r| run_ids.contains(&r.run_id))
            .collect()
    }

    /// Generate health report
    pub fn generate_report(&self) -> String {
        let result = self.check_health();
        let mut report = String::new();

        report.push_str(&format!(
            "# Health Check Report\n\n**Checked at:** {}\n\n",
            result.checked_at.format("%Y-%m-%d %H:%M:%S UTC")
        ));

        report.push_str(&format!("**Runs monitored:** {}\n\n", result.runs_checked));

        if result.issues.is_empty() {
            report.push_str("✅ All runs healthy\n");
        } else {
            report.push_str(&format!(
                "⚠️  **Issues found:** {}\n\n",
                result.issues.len()
            ));

            // Group by severity
            let mut critical = vec![];
            let mut errors = vec![];
            let mut warnings = vec![];
            let mut infos = vec![];

            for issue in &result.issues {
                match issue.severity {
                    IssueSeverity::Critical => critical.push(issue),
                    IssueSeverity::Error => errors.push(issue),
                    IssueSeverity::Warning => warnings.push(issue),
                    IssueSeverity::Info => infos.push(issue),
                }
            }

            for (severity, issues) in [
                ("🔴 Critical", critical),
                ("❌ Errors", errors),
                ("⚠️  Warnings", warnings),
                ("ℹ️  Info", infos),
            ] {
                if !issues.is_empty() {
                    report.push_str(&format!("## {}\n\n", severity));
                    for issue in issues {
                        report.push_str(&format!("### {}\n", issue.run_id));
                        report.push_str(&format!("**Type:** {:?}\n\n", issue.issue_type));
                        report.push_str(&format!("{}\n\n", issue.description));
                        report.push_str(&format!(
                            "**Recommended action:** {:?}\n\n",
                            issue.recommended_action
                        ));
                    }
                }
            }
        }

        // Summary by state
        report.push_str("\n## Run States\n\n");
        for state in [
            MonitoredRunState::Pending,
            MonitoredRunState::Running,
            MonitoredRunState::Paused,
            MonitoredRunState::Completed,
            MonitoredRunState::Failed,
            MonitoredRunState::Cancelled,
            MonitoredRunState::TimedOut,
        ] {
            let count = self.get_runs_by_state(state).len();
            report.push_str(&format!("- {:?}: {}\n", state, count));
        }

        report
    }
}

/// Background health check task
pub struct HealthCheckTask {
    watchdog: RunHealthWatchdog,
    last_check: Option<DateTime<Utc>>,
}

impl HealthCheckTask {
    /// Create a new health check task
    pub fn new(watchdog: RunHealthWatchdog) -> Self {
        Self {
            watchdog,
            last_check: None,
        }
    }

    /// Run a single health check
    pub fn check(&mut self) -> HealthCheckResult {
        let result = self.watchdog.check_health();
        self.last_check = Some(Utc::now());
        result
    }

    /// Get the watchdog
    pub fn watchdog(&self) -> &RunHealthWatchdog {
        &self.watchdog
    }

    /// Get mutable watchdog
    pub fn watchdog_mut(&mut self) -> &mut RunHealthWatchdog {
        &mut self.watchdog
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::Duration;

    #[test]
    fn test_watchdog_registration() {
        let config = HealthCheckConfig::default();
        let mut watchdog = RunHealthWatchdog::new(config);

        watchdog.register_run("run-1".to_string(), "feature-dev".to_string());

        let run = watchdog.get_run("run-1").unwrap();
        assert_eq!(run.run_id, "run-1");
        assert_eq!(run.workflow_name, "feature-dev");
        assert!(matches!(run.state, MonitoredRunState::Pending));
    }

    #[test]
    fn test_stuck_step_detection() {
        let config = HealthCheckConfig {
            enabled: true,
            stuck_threshold_seconds: 10,
            abandoned_threshold_minutes: 30,
            check_interval_seconds: 60,
        };

        let mut watchdog = RunHealthWatchdog::new(config);
        watchdog.register_run("run-1".to_string(), "feature-dev".to_string());

        // Simulate a step running for too long
        let long_ago = Utc::now() - Duration::seconds(20);
        watchdog.runs.get_mut("run-1").unwrap().current_step = Some(StepExecution {
            step_id: "step-1".to_string(),
            step_name: "Long Step".to_string(),
            started_at: long_ago,
            last_activity: long_ago,
            state: StepState::Executing,
        });

        let result = watchdog.check_health();
        assert!(!result.issues.is_empty());
        assert!(result
            .issues
            .iter()
            .any(|i| matches!(i.issue_type, IssueType::StuckStep)));
    }

    #[test]
    fn test_abandoned_run_detection() {
        let config = HealthCheckConfig {
            enabled: true,
            stuck_threshold_seconds: 300,
            abandoned_threshold_minutes: 1, // Very short for testing
            check_interval_seconds: 60,
        };

        let mut watchdog = RunHealthWatchdog::new(config);
        watchdog.register_run("run-1".to_string(), "feature-dev".to_string());

        // Set start time to long ago
        let long_ago = Utc::now() - Duration::minutes(5);
        watchdog.runs.get_mut("run-1").unwrap().started_at = long_ago;

        let result = watchdog.check_health();
        assert!(!result.issues.is_empty());
        assert!(result
            .issues
            .iter()
            .any(|i| matches!(i.issue_type, IssueType::AbandonedRun)));
    }

    #[test]
    fn test_orphaned_run_detection() {
        let config = HealthCheckConfig::default();
        let mut watchdog = RunHealthWatchdog::new(config);
        watchdog.register_run("run-1".to_string(), "feature-dev".to_string());
        watchdog.mark_orphaned("run-1");

        let result = watchdog.check_health();
        assert!(!result.issues.is_empty());
        assert!(result
            .issues
            .iter()
            .any(|i| matches!(i.issue_type, IssueType::OrphanedRun)));
    }

    #[test]
    fn test_health_report() {
        let config = HealthCheckConfig::default();
        let mut watchdog = RunHealthWatchdog::new(config);
        watchdog.register_run("run-1".to_string(), "feature-dev".to_string());
        watchdog.register_run("run-2".to_string(), "bug-fix".to_string());

        let report = watchdog.generate_report();
        assert!(report.contains("Health Check Report"));
        assert!(report.contains("Runs monitored:** 2"));
    }
}