supervisor/src/runner.rs

//! Runner implementation for actor process management.

// use sal_service_manager::{ProcessManagerError as ServiceProcessManagerError, ProcessStatus, ProcessConfig};

/// Simple process status enum to replace sal_service_manager dependency
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub enum ProcessStatus {
    NotStarted,
    Starting,
    Running,
    Stopping,
    Stopped,
    Failed,
    Error(String),
}

/// Simple process config to replace sal_service_manager dependency
#[derive(Debug, Clone)]
pub struct ProcessConfig {
    pub command: String,
    pub args: Vec<String>,
    pub working_dir: Option<String>,
    pub env_vars: Vec<(String, String)>,
}

impl ProcessConfig {
    pub fn new(command: String, args: Vec<String>, working_dir: Option<String>, env_vars: Vec<(String, String)>) -> Self {
        Self {
            command,
            args,
            working_dir,
            env_vars,
        }
    }
}

/// Simple process manager error to replace sal_service_manager dependency
#[derive(Debug, thiserror::Error)]
pub enum ProcessManagerError {
    #[error("Process execution failed: {0}")]
    ExecutionFailed(String),
    #[error("Process not found: {0}")]
    ProcessNotFound(String),
    #[error("IO error: {0}")]
    IoError(String),
}
use std::path::PathBuf;

/// Represents the current status of an actor/runner (alias for ProcessStatus)
pub type RunnerStatus = ProcessStatus;

/// Log information structure with serialization support
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct LogInfo {
    pub timestamp: String,
    pub level: String,
    pub message: String,
}

/// Runner configuration and state (merged from RunnerConfig)
#[derive(Debug, Clone)]
pub struct Runner {
    /// Unique identifier for the runner
    pub id: String,
    pub name: String,
    pub namespace: String,
    /// Path to the actor binary
    pub command: PathBuf, // Command to run runner by, used only if supervisor is used to run runners
    /// Redis URL for job queue
    pub redis_url: String,
    /// Additional command-line arguments
    pub extra_args: Vec<String>,
}

impl Runner {
    /// Create a new runner from configuration
    pub fn from_config(config: RunnerConfig) -> Self {
        Self {
            id: config.id,
            name: config.name,
            namespace: config.namespace,
            command: config.command,
            redis_url: config.redis_url,
            extra_args: config.extra_args,
        }
    }
    
    /// Create a new runner with extra arguments
    pub fn with_args(
        id: String,
        name: String,
        namespace: String,
        command: PathBuf,
        redis_url: String,
        extra_args: Vec<String>,
    ) -> Self {
        Self {
            id,
            name,
            namespace,
            command,
            redis_url,
            extra_args,
        }
    }

    /// Get the queue key for this runner with the given namespace
    pub fn get_queue(&self) -> String {
        if self.namespace == "" {
            format!("runner:{}", self.name)
        } else {
            format!("{}:runner:{}", self.namespace, self.name)
        }
    }
}

/// Result type for runner operations
pub type RunnerResult<T> = Result<T, RunnerError>;

/// Errors that can occur during runner operations
#[derive(Debug, thiserror::Error)]
pub enum RunnerError {
    #[error("Actor '{actor_id}' not found")]
    ActorNotFound { actor_id: String },
    
    #[error("Actor '{actor_id}' is already running")]
    ActorAlreadyRunning { actor_id: String },
    
    #[error("Actor '{actor_id}' is not running")]
    ActorNotRunning { actor_id: String },
    
    #[error("Failed to start actor '{actor_id}': {reason}")]
    StartupFailed { actor_id: String, reason: String },
    
    #[error("Failed to stop actor '{actor_id}': {reason}")]
    StopFailed { actor_id: String, reason: String },
    
    #[error("Timeout waiting for actor '{actor_id}' to start")]
    StartupTimeout { actor_id: String },
    
    #[error("Job queue error for actor '{actor_id}': {reason}")]
    QueueError { actor_id: String, reason: String },
    
    #[error("Process manager error: {source}")]
    ProcessManagerError {
        #[from]
        source: ProcessManagerError,
    },
    
    #[error("Configuration error: {reason}")]
    ConfigError { reason: String },
    
    #[error("Invalid secret: {0}")]
    InvalidSecret(String),
    
    #[error("IO error: {source}")]
    IoError {
        #[from]
        source: std::io::Error,
    },
    
    #[error("Redis error: {source}")]
    RedisError {
        #[from]
        source: redis::RedisError,
    },
    
    #[error("Job error: {source}")]
    JobError {
        #[from]
        source: hero_job::JobError,
    },
    
    #[error("Job client error: {source}")]
    JobClientError {
        #[from]
        source: hero_job_client::ClientError,
    },
    
    #[error("Job '{job_id}' not found")]
    JobNotFound { job_id: String },
    
    #[error("Authentication error: {message}")]
    AuthenticationError { message: String },
}

// Type alias for backward compatibility
pub type RunnerConfig = Runner;

/// Convert Runner to ProcessConfig
pub fn runner_to_process_config(config: &Runner) -> ProcessConfig {
    let mut args = vec![
        config.id.clone(),  // First positional argument is the runner ID
        "--redis-url".to_string(),
        config.redis_url.clone(),
    ];
    
    // Add extra arguments (e.g., context configurations)
    args.extend(config.extra_args.clone());
    
    ProcessConfig::new(
        config.command.to_string_lossy().to_string(),
        args,
        Some("/tmp".to_string()), // Default working directory since Runner doesn't have working_dir field
        vec![]
    )
}