swactor/crates/ci/src/local_runner.rs

//! LocalRunner actor: executes job commands directly on the host via shell.
//!
//! Short-lived actor, one per job. Spawned by LocalCoordinator when a job
//! is ready to execute.

use std::io::BufRead;
use std::process::{Command, Stdio};
use std::sync::{Arc, Mutex};

use swactor::actor::{ActorAddress, ActorInterface, Ctx};

use crate::local_coordinator::LocalCoordinatorMsg;
use crate::{JobComplete, JobFailure, JobProgress, JobSuccess, LocalStartJob};

/// Messages the LocalRunner can receive.
#[derive(Debug, Clone)]
pub enum LocalRunnerMsg {
    /// Begin executing the job (sent to self in on_start).
    Execute,
    /// Simulated: job completed (for testing without real shell).
    SimComplete(Result<(), String>),
}

/// LocalRunner actor state.
pub struct LocalRunner {
    coordinator_addr: ActorAddress,
    start_job: LocalStartJob,
}

impl LocalRunner {
    pub fn new(coordinator_addr: ActorAddress, start_job: LocalStartJob) -> Self {
        Self {
            coordinator_addr,
            start_job,
        }
    }

    /// Execute all commands in the job definition, streaming output back.
    fn execute(&self, ctx: &Ctx) {
        let job_id = &self.start_job.job_id;
        let work_dir = &self.start_job.work_dir;
        let timeout_secs = self.start_job.job_def.timeout_secs;

        for cmd_str in &self.start_job.job_def.run {
            // Send progress: command being run.
            let _ = ctx.send(
                self.coordinator_addr,
                LocalCoordinatorMsg::JobProgress(JobProgress {
                    job_id: job_id.clone(),
                    output_line: format!("$ {cmd_str}"),
                }),
            );

            let child_result = Command::new("sh")
                .arg("-c")
                .arg(cmd_str)
                .current_dir(work_dir)
                .envs(&self.start_job.env_overrides)
                .envs(&self.start_job.job_def.env)
                .stdout(Stdio::piped())
                .stderr(Stdio::piped())
                .spawn();

            let mut child = match child_result {
                Ok(c) => c,
                Err(e) => {
                    let _ = ctx.send(
                        self.coordinator_addr,
                        LocalCoordinatorMsg::JobComplete(JobComplete {
                            job_id: job_id.clone(),
                            result: Err(JobFailure::ExecError(e.to_string())),
                            artifacts: Vec::new(),
                        }),
                    );
                    ctx.stop_self();
                    return;
                }
            };

            // Timeout mechanism: share child handle, spawn thread that kills after timeout.
            let kill_flag = Arc::new(Mutex::new(false));
            let kill_flag_clone = Arc::clone(&kill_flag);
            // Use a pipe to signal the timeout thread when the command finishes.
            let (done_tx, done_rx) = std::sync::mpsc::channel::<()>();
            let timeout_handle = std::thread::spawn(move || {
                // Wait for either timeout or command completion.
                if done_rx.recv_timeout(std::time::Duration::from_secs(timeout_secs)).is_err() {
                    *kill_flag_clone.lock().unwrap() = true;
                }
            });

            // Read stdout line-by-line.
            let stdout = child.stdout.take();
            let stderr = child.stderr.take();

            let mut last_lines: Vec<String> = Vec::new();

            if let Some(stdout) = stdout {
                let reader = std::io::BufReader::new(stdout);
                for line in reader.lines() {
                    if let Ok(line) = line {
                        let _ = ctx.send(
                            self.coordinator_addr,
                            LocalCoordinatorMsg::JobProgress(JobProgress {
                                job_id: job_id.clone(),
                                output_line: line.clone(),
                            }),
                        );
                        last_lines.push(line);
                        if last_lines.len() > 50 {
                            last_lines.remove(0);
                        }
                    }
                }
            }

            if let Some(stderr) = stderr {
                let reader = std::io::BufReader::new(stderr);
                for line in reader.lines() {
                    if let Ok(line) = line {
                        let _ = ctx.send(
                            self.coordinator_addr,
                            LocalCoordinatorMsg::JobProgress(JobProgress {
                                job_id: job_id.clone(),
                                output_line: format!("[stderr] {line}"),
                            }),
                        );
                        last_lines.push(line);
                        if last_lines.len() > 50 {
                            last_lines.remove(0);
                        }
                    }
                }
            }

            let status = child.wait();

            // Signal timeout thread that command finished.
            let _ = done_tx.send(());
            let _ = timeout_handle.join();

            // Check if killed by timeout.
            if *kill_flag.lock().unwrap() {
                let _ = ctx.send(
                    self.coordinator_addr,
                    LocalCoordinatorMsg::JobComplete(JobComplete {
                        job_id: job_id.clone(),
                        result: Err(JobFailure::Timeout),
                        artifacts: Vec::new(),
                    }),
                );
                ctx.stop_self();
                return;
            }

            match status {
                Ok(exit) if exit.success() => {
                    // Command passed, continue to next.
                }
                Ok(exit) => {
                    let exit_code = exit.code().unwrap_or(-1);
                    let _ = ctx.send(
                        self.coordinator_addr,
                        LocalCoordinatorMsg::JobComplete(JobComplete {
                            job_id: job_id.clone(),
                            result: Err(JobFailure::CommandFailed {
                                exit_code,
                                last_lines,
                            }),
                            artifacts: Vec::new(),
                        }),
                    );
                    ctx.stop_self();
                    return;
                }
                Err(e) => {
                    let _ = ctx.send(
                        self.coordinator_addr,
                        LocalCoordinatorMsg::JobComplete(JobComplete {
                            job_id: job_id.clone(),
                            result: Err(JobFailure::ExecError(e.to_string())),
                            artifacts: Vec::new(),
                        }),
                    );
                    ctx.stop_self();
                    return;
                }
            }
        }

        // All commands passed.
        let _ = ctx.send(
            self.coordinator_addr,
            LocalCoordinatorMsg::JobComplete(JobComplete {
                job_id: job_id.clone(),
                result: Ok(JobSuccess),
                artifacts: Vec::new(),
            }),
        );
        ctx.stop_self();
    }
}

impl ActorInterface for LocalRunner {
    type Incoming = LocalRunnerMsg;
    type Response = ();

    fn on_start(&mut self, ctx: &Ctx) {
        let _ = ctx.send(ctx.self_addr(), LocalRunnerMsg::Execute);
    }

    fn handle(&mut self, ctx: &Ctx, msg: LocalRunnerMsg) {
        match msg {
            LocalRunnerMsg::Execute => {
                self.execute(ctx);
            }
            LocalRunnerMsg::SimComplete(result) => {
                let complete = JobComplete {
                    job_id: self.start_job.job_id.clone(),
                    result: match result {
                        Ok(()) => Ok(JobSuccess),
                        Err(msg) => Err(JobFailure::CommandFailed {
                            exit_code: 1,
                            last_lines: vec![msg],
                        }),
                    },
                    artifacts: Vec::new(),
                };
                let _ = ctx.send(
                    self.coordinator_addr,
                    LocalCoordinatorMsg::JobComplete(complete),
                );
                ctx.stop_self();
            }
        }
    }
}