mothership 0.0.100

//! Ship - A single process in the fleet
//!
//! All ships are equal peers in the fleet.

use std::collections::{HashMap, VecDeque};
use std::process::Stdio;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};

use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::{Child, Command};
use tokio::sync::Mutex;
use tracing::{debug, info, warn};

use breaker_machines::CircuitBreaker;
use state_machines::state_machine;

use crate::charter::{Bind, RouteConfig};

/// Maximum log lines to retain per ship
const MAX_LOG_LINES: usize = 1000;
/// Maximum log bytes to retain per ship (1MB)
const MAX_LOG_BYTES: usize = 1024 * 1024;
/// Maximum length per log line (truncate longer lines)
const MAX_LINE_LENGTH: usize = 4096;

/// Restart circuit defaults (crash loop protection)
const RESTART_FAILURE_THRESHOLD: usize = 5;
const RESTART_FAILURE_WINDOW_SECS: f64 = 60.0;
const RESTART_BACKOFF_SECS: f64 = 30.0;
const RESTART_SUCCESS_THRESHOLD: usize = 1;

/// Grace period after launch before health checks trigger restarts (seconds)
const STARTUP_GRACE_SECS: u64 = 10;

/// Spawn a task to forward process output to logging and buffer
fn spawn_log_forwarder<R>(
    stream: Option<R>,
    stream_name: &'static str,
    ship_name: &str,
    log_buffer: Arc<Mutex<LogBuffer>>,
) where
    R: tokio::io::AsyncRead + Unpin + Send + 'static,
{
    if let Some(reader) = stream {
        let ship_name = ship_name.to_string();
        tokio::spawn(async move {
            let reader = BufReader::new(reader);
            let mut lines = reader.lines();
            while let Ok(Some(line)) = lines.next_line().await {
                if !SUPPRESS_STDOUT.load(Ordering::SeqCst) {
                    info!(ship = %ship_name, stream = stream_name, "{}", line);
                }
                log_buffer.lock().await.push(stream_name, line);
            }
        });
    }
}

/// Graceful shutdown timeout before SIGKILL (seconds)
const TERMINATE_TIMEOUT_SECS: u64 = 5;

/// Global flag to suppress stdout logging (for TUI mode)
static SUPPRESS_STDOUT: AtomicBool = AtomicBool::new(false);

/// Set whether to suppress stdout logging
pub fn set_suppress_stdout(suppress: bool) {
    SUPPRESS_STDOUT.store(suppress, Ordering::SeqCst);
}

/// Check if stdout logging is suppressed
pub fn is_stdout_suppressed() -> bool {
    SUPPRESS_STDOUT.load(Ordering::SeqCst)
}

/// Log entry from a ship
#[derive(Debug, Clone)]
pub struct LogEntry {
    pub stream: &'static str,
    pub message: String,
}

/// Bounded log buffer with byte and line limits
#[derive(Debug)]
struct LogBuffer {
    entries: VecDeque<LogEntry>,
    total_bytes: usize,
}

impl LogBuffer {
    fn new() -> Self {
        Self {
            entries: VecDeque::with_capacity(MAX_LOG_LINES),
            total_bytes: 0,
        }
    }

    fn push(&mut self, stream: &'static str, mut message: String) {
        // Truncate long lines
        if message.len() > MAX_LINE_LENGTH {
            message.truncate(MAX_LINE_LENGTH - 3);
            message.push_str("...");
        }

        let msg_len = message.len();

        // Evict old entries if we'd exceed byte limit
        while self.total_bytes + msg_len > MAX_LOG_BYTES && !self.entries.is_empty() {
            if let Some(old) = self.entries.pop_front() {
                self.total_bytes = self.total_bytes.saturating_sub(old.message.len());
            }
        }

        // Evict if we'd exceed line limit
        while self.entries.len() >= MAX_LOG_LINES {
            if let Some(old) = self.entries.pop_front() {
                self.total_bytes = self.total_bytes.saturating_sub(old.message.len());
            }
        }

        self.total_bytes += msg_len;
        self.entries.push_back(LogEntry { stream, message });
    }

    fn recent(&self, limit: usize) -> Vec<LogEntry> {
        self.entries.iter().rev().take(limit).cloned().collect()
    }
}

/// Current status of a ship
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum ShipStatus {
    /// Not yet started
    #[default]
    Pending,
    /// Starting up, waiting for health check
    Starting,
    /// Running and healthy
    Running,
    /// Running but health check failed
    Unhealthy,
    /// Restart backoff (circuit open)
    Backoff,
    /// Process exited cleanly
    Stopped,
    /// Process crashed
    Failed,
}

state_machine! {
    name: ShipLifecycle,
    context: (),
    dynamic: true,

    initial: Pending,
    states: [
        Pending,
        Starting,
        Running,
        Unhealthy,
        Backoff,
        Stopped,
        Failed,
    ],
    events {
        start {
            transition: { from: [Pending, Stopped, Failed, Backoff], to: Starting }
        }
        healthy {
            transition: { from: [Starting, Unhealthy], to: Running }
        }
        unhealthy {
            transition: { from: [Starting, Running], to: Unhealthy }
        }
        crash {
            transition: { from: [Starting, Running, Unhealthy], to: Failed }
        }
        backoff {
            transition: { from: [Failed, Starting, Running, Unhealthy], to: Backoff }
        }
        stop {
            transition: { from: [Pending, Starting, Running, Unhealthy, Backoff, Failed], to: Stopped }
        }
    }
}

#[derive(Debug)]
struct ShipState {
    machine: DynamicShipLifecycle,
    status: ShipStatus,
    backoff_until: Option<Instant>,
}

impl ShipState {
    fn new() -> Self {
        let machine = DynamicShipLifecycle::new(());
        let status = status_from_state(machine.current_state());
        Self {
            machine,
            status,
            backoff_until: None,
        }
    }

    fn refresh_status(&mut self) {
        self.status = status_from_state(self.machine.current_state());
    }
}

fn status_from_state(state: &str) -> ShipStatus {
    match state {
        "Pending" => ShipStatus::Pending,
        "Starting" => ShipStatus::Starting,
        "Running" => ShipStatus::Running,
        "Unhealthy" => ShipStatus::Unhealthy,
        "Backoff" => ShipStatus::Backoff,
        "Stopped" => ShipStatus::Stopped,
        "Failed" => ShipStatus::Failed,
        _ => ShipStatus::Pending,
    }
}

/// Snapshot of ship state for sync access (e.g., TUI)
#[derive(Debug, Clone)]
pub struct ShipSnapshot {
    name: String,
    group: String,
    command: String,
    status: ShipStatus,
    pid: Option<u32>,
    healthcheck: Option<String>,
    restart_count: u32,
    critical: bool,
    oneshot: bool,
    routes: Vec<String>,
}

impl ShipSnapshot {
    /// Get ship name
    pub fn name(&self) -> &str {
        &self.name
    }

    /// Get ship group
    pub fn group(&self) -> &str {
        &self.group
    }

    /// Get command
    pub fn command(&self) -> &str {
        &self.command
    }

    /// Get status
    pub fn status(&self) -> ShipStatus {
        self.status
    }

    /// Get PID if running
    pub fn pid(&self) -> Option<u32> {
        self.pid
    }

    /// Get healthcheck URL
    pub fn healthcheck(&self) -> Option<&str> {
        self.healthcheck.as_deref()
    }

    /// Get restart count
    pub fn restart_count(&self) -> u32 {
        self.restart_count
    }

    /// Is this ship critical (crash kills fleet)?
    pub fn is_critical(&self) -> bool {
        self.critical
    }

    /// Is this a oneshot job?
    pub fn is_oneshot(&self) -> bool {
        self.oneshot
    }

    /// Get routes (as display strings)
    pub fn routes(&self) -> &[String] {
        &self.routes
    }
}

/// Builder for Ship
#[derive(Default)]
pub struct ShipBuilder {
    name: String,
    group: String,
    command: String,
    args: Vec<String>,
    env: HashMap<String, String>,
    bind: Option<Bind>,
    healthcheck: Option<String>,
    depends_on: Vec<String>,
    routes: Vec<RouteConfig>,
    critical: bool,
    oneshot: bool,
}

impl ShipBuilder {
    /// Create a new ship builder
    pub fn new(name: impl Into<String>, command: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            command: command.into(),
            critical: true, // Default: all ships are critical
            ..Default::default()
        }
    }

    /// Set the fleet group
    pub fn group(mut self, group: impl Into<String>) -> Self {
        self.group = group.into();
        self
    }

    /// Set command arguments
    pub fn args(mut self, args: Vec<String>) -> Self {
        self.args = args;
        self
    }

    /// Set environment variables
    pub fn env(mut self, env: HashMap<String, String>) -> Self {
        self.env = env;
        self
    }

    /// Set bind address
    pub fn bind(mut self, bind: Option<Bind>) -> Self {
        self.bind = bind;
        self
    }

    /// Set healthcheck endpoint
    pub fn healthcheck(mut self, healthcheck: Option<String>) -> Self {
        self.healthcheck = healthcheck;
        self
    }

    /// Set dependencies
    pub fn depends_on(mut self, depends_on: Vec<String>) -> Self {
        self.depends_on = depends_on;
        self
    }

    /// Set HTTP routes
    pub fn routes(mut self, routes: Vec<RouteConfig>) -> Self {
        self.routes = routes;
        self
    }

    /// Set whether this ship is critical (crash kills fleet)
    pub fn critical(mut self, critical: bool) -> Self {
        self.critical = critical;
        self
    }

    /// Set whether this is a oneshot job
    pub fn oneshot(mut self, oneshot: bool) -> Self {
        self.oneshot = oneshot;
        self
    }

    /// Build the Ship
    pub fn build(self) -> Ship {
        let backoff_delay = Duration::from_secs_f64(RESTART_BACKOFF_SECS);
        let name = self.name;
        let breaker = CircuitBreaker::builder(format!("ship:{}", name))
            .failure_threshold(RESTART_FAILURE_THRESHOLD)
            .failure_window_secs(RESTART_FAILURE_WINDOW_SECS)
            .half_open_timeout_secs(RESTART_BACKOFF_SECS)
            .success_threshold(RESTART_SUCCESS_THRESHOLD)
            .build();

        Ship {
            name,
            group: self.group,
            command: self.command,
            args: self.args,
            env: self.env,
            bind: self.bind,
            healthcheck: self.healthcheck,
            depends_on: self.depends_on,
            routes: self.routes,
            critical: self.critical,
            oneshot: self.oneshot,
            state: Arc::new(Mutex::new(ShipState::new())),
            child: Arc::new(Mutex::new(None)),
            process_meta: Arc::new(Mutex::new(ProcessMeta::default())),
            last_health_check: Arc::new(Mutex::new(None)),
            restart_count: Arc::new(Mutex::new(0)),
            log_buffer: Arc::new(Mutex::new(LogBuffer::new())),
            breaker: Arc::new(Mutex::new(breaker)),
            backoff_delay,
            memory_sampler: Arc::new(Mutex::new(None)),
            memory_metrics: Arc::new(Mutex::new(None)),
        }
    }
}

/// Process metadata for proper cleanup
#[derive(Debug, Default)]
struct ProcessMeta {
    /// Process group ID (same as child PID when using setpgid(0,0))
    pgid: Option<i32>,
    /// Time of last launch (for grace period)
    launched_at: Option<Instant>,
}

/// A ship in the fleet (process wrapper)
pub struct Ship {
    /// Ship name (unique identifier)
    pub name: String,
    /// Fleet group this ship belongs to
    pub group: String,
    /// Command to run
    pub command: String,
    /// Command arguments
    pub args: Vec<String>,
    /// Environment variables
    pub env: HashMap<String, String>,
    /// Bind address (for health check inference)
    pub bind: Option<Bind>,
    /// Health check endpoint
    pub healthcheck: Option<String>,
    /// Dependencies (other ship names)
    pub depends_on: Vec<String>,
    /// HTTP routes (bind name -> pattern)
    pub routes: Vec<RouteConfig>,
    /// Critical ship - crash kills fleet (default: true)
    pub critical: bool,
    /// Oneshot job - runs once then exits (default: false)
    pub oneshot: bool,
    /// Current lifecycle state
    state: Arc<Mutex<ShipState>>,
    /// Child process handle
    child: Arc<Mutex<Option<Child>>>,
    /// Process metadata (pgid, launch time)
    process_meta: Arc<Mutex<ProcessMeta>>,
    /// Last health check time
    last_health_check: Arc<Mutex<Option<Instant>>>,
    /// Restart count
    restart_count: Arc<Mutex<u32>>,
    /// Log buffer for TUI display (bounded by bytes and lines)
    log_buffer: Arc<Mutex<LogBuffer>>,
    /// Crash loop circuit breaker
    breaker: Arc<Mutex<CircuitBreaker>>,
    /// Backoff delay when circuit opens
    backoff_delay: Duration,
    /// Memory sampler task handle
    memory_sampler: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>,
    /// Latest memory sample
    memory_metrics: Arc<Mutex<Option<crate::monitoring::MemorySample>>>,
}

impl Ship {
    /// Get display name
    pub fn display_name(&self) -> &str {
        &self.name
    }

    async fn transition(&self, event: ShipLifecycleEvent) {
        let mut state = self.state.lock().await;
        if state.machine.handle(event).is_ok() {
            state.refresh_status();
        }
    }

    /// Get current status
    pub async fn status(&self) -> ShipStatus {
        self.state.lock().await.status
    }

    /// Get restart count
    pub async fn restart_count(&self) -> u32 {
        *self.restart_count.lock().await
    }

    /// Increment restart count
    pub async fn increment_restart(&self) {
        let mut count = self.restart_count.lock().await;
        *count += 1;
    }

    /// Launch the ship (spawn process)
    pub async fn launch(&self) -> anyhow::Result<()> {
        let name = self.display_name();

        // Guard against double launch
        if self.is_running().await {
            warn!(ship = name, "Ship already running, skipping launch");
            return Ok(());
        }

        info!(ship = name, command = %self.command, "Launching ship");

        self.transition(ShipLifecycleEvent::Start).await;

        let mut cmd = Command::new(&self.command);

        // Inject Mothership protocol env vars for ships that need them
        let mothership_pid = std::process::id();
        let socket_dir = std::env::var("MS_SOCKET_DIR").unwrap_or_else(|_| {
            std::env::var("XDG_RUNTIME_DIR")
                .unwrap_or_else(|_| std::env::temp_dir().to_string_lossy().to_string())
                + "/mothership"
        });

        cmd.args(&self.args)
            .envs(&self.env)
            .env("NO_COLOR", "1")
            .env("MS_PID", mothership_pid.to_string())
            .env("MS_SHIP", &self.name)
            .env("MS_SOCKET_DIR", socket_dir)
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);

        // Set up process group and death signal
        // SAFETY: pre_exec runs after fork, before exec. We only call
        // async-signal-safe functions (setpgid, prctl/kqueue, getpid, _exit).
        let parent_pid = std::process::id() as i32;
        unsafe {
            cmd.pre_exec(move || {
                // Create new process group with this process as leader
                // This allows us to kill all children with killpg()
                if libc::setpgid(0, 0) != 0 {
                    // Non-fatal: continue even if setpgid fails
                    // (can happen if already process group leader)
                }

                // Set up parent death notification (cross-platform)
                // On Linux: uses prctl(PR_SET_PDEATHSIG)
                // On macOS/FreeBSD: spawns kqueue watcher process
                let pgid = libc::getpid(); // After setpgid(0,0), our PID is our PGID
                super::parent_death::setup_parent_death_signal_preexec(
                    parent_pid,
                    pgid,
                    libc::SIGTERM,
                );

                Ok(())
            });
        }

        let mut child = cmd.spawn()?;
        let pid = child.id();

        // Spawn log forwarders - format as JSON via tracing
        spawn_log_forwarder(child.stdout.take(), "stdout", name, self.log_buffer.clone());
        spawn_log_forwarder(child.stderr.take(), "stderr", name, self.log_buffer.clone());

        // Store child and metadata
        *self.child.lock().await = Some(child);
        {
            let mut meta = self.process_meta.lock().await;
            meta.pgid = pid.map(|p| p as i32);
            meta.launched_at = Some(Instant::now());
        }

        if let Some(p) = pid {
            info!(ship = name, pid = p, "Ship launched");
        } else {
            info!(ship = name, "Ship launched");
        }

        // Spawn memory sampler if we have a PID
        if let Some(p) = pid {
            let sampler = self.spawn_memory_sampler(p);
            *self.memory_sampler.lock().await = Some(sampler);
        }

        Ok(())
    }

    /// Check if ship is within startup grace period
    pub async fn in_grace_period(&self) -> bool {
        let meta = self.process_meta.lock().await;
        match meta.launched_at {
            Some(launched_at) => launched_at.elapsed() < Duration::from_secs(STARTUP_GRACE_SECS),
            None => false,
        }
    }

    /// Check if process is still running
    pub async fn is_running(&self) -> bool {
        let mut child_guard = self.child.lock().await;
        if let Some(child) = child_guard.as_mut() {
            match child.try_wait() {
                Ok(Some(_)) => false,
                Ok(None) => true,
                Err(_) => false,
            }
        } else {
            false
        }
    }

    /// Wait for process to exit
    pub async fn wait(&self) -> anyhow::Result<i32> {
        let mut child_guard = self.child.lock().await;
        if let Some(child) = child_guard.as_mut() {
            let status = child.wait().await?;
            let code = status.code().unwrap_or(-1);

            if code == 0 {
                self.transition(ShipLifecycleEvent::Stop).await;
            } else {
                self.transition(ShipLifecycleEvent::Crash).await;
            }

            Ok(code)
        } else {
            Ok(-1)
        }
    }

    /// Terminate the process and its entire process group
    pub async fn terminate(&self) -> anyhow::Result<()> {
        let name = self.display_name();
        info!(ship = name, "Terminating ship");

        let pgid = self.process_meta.lock().await.pgid;
        let mut child_guard = self.child.lock().await;

        if let Some(child) = child_guard.as_mut() {
            use nix::sys::signal::{self, Signal};
            use nix::unistd::Pid;

            // Try graceful shutdown first (SIGTERM to process group)
            if let Some(pgid) = pgid {
                debug!(ship = name, pgid = pgid, "Sending SIGTERM to process group");
                if signal::killpg(Pid::from_raw(pgid), Signal::SIGTERM).is_err()
                    && let Some(pid) = child.id()
                {
                    let _ = signal::kill(Pid::from_raw(pid as i32), Signal::SIGTERM);
                }
            } else if let Some(pid) = child.id() {
                let _ = signal::kill(Pid::from_raw(pid as i32), Signal::SIGTERM);
            }

            // Wait briefly for graceful shutdown
            let timeout = Duration::from_secs(TERMINATE_TIMEOUT_SECS);
            tokio::select! {
                _ = tokio::time::sleep(timeout) => {
                    // Force kill process group if still running
                    if let Some(pgid) = pgid {
                        debug!(ship = name, pgid = pgid, "Sending SIGKILL to process group");
                        if signal::killpg(Pid::from_raw(pgid), Signal::SIGKILL).is_err()
                            && let Some(pid) = child.id() {
                                let _ = signal::kill(Pid::from_raw(pid as i32), Signal::SIGKILL);
                            }
                    }
                    let _ = child.kill().await;
                }
                _ = child.wait() => {
                    debug!(ship = name, "Ship exited gracefully");
                }
            }
        }

        // Clear process metadata
        {
            let mut meta = self.process_meta.lock().await;
            meta.pgid = None;
            meta.launched_at = None;
        }

        // Stop memory sampler
        if let Some(handle) = self.memory_sampler.lock().await.take() {
            handle.abort();
        }
        *self.memory_metrics.lock().await = None;

        self.transition(ShipLifecycleEvent::Stop).await;
        Ok(())
    }

    /// Get PID if running
    pub async fn pid(&self) -> Option<u32> {
        let child_guard = self.child.lock().await;
        if let Some(child) = child_guard.as_ref() {
            child.id()
        } else {
            None
        }
    }

    /// Spawn memory sampling task for this ship
    fn spawn_memory_sampler(&self, pid: u32) -> tokio::task::JoinHandle<()> {
        let metrics = self.memory_metrics.clone();
        let name = self.name.clone();

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(10));

            loop {
                interval.tick().await;

                match crate::monitoring::sample_process_memory(pid) {
                    Ok(sample) => {
                        *metrics.lock().await = Some(sample);
                    }
                    Err(crate::monitoring::MemoryError::ProcessNotFound(_)) => {
                        // Process died, stop sampling
                        break;
                    }
                    Err(e) => {
                        warn!(ship = %name, pid = pid, error = %e, "Memory sampling failed");
                    }
                }
            }
        })
    }

    /// Get latest memory sample
    pub async fn memory_sample(&self) -> Option<crate::monitoring::MemorySample> {
        self.memory_metrics.lock().await.clone()
    }

    /// Get healthcheck URL if configured
    pub fn healthcheck_url(&self) -> Option<String> {
        let endpoint = self.healthcheck.as_ref()?;

        // If endpoint is already a full URL, use it directly
        if endpoint.starts_with("http://") || endpoint.starts_with("https://") {
            return Some(endpoint.clone());
        }

        let base = self.bind.as_ref()?.healthcheck_base();
        Some(format!("{}{}", base, endpoint))
    }

    /// Mark ship as healthy
    pub async fn mark_healthy(&self) {
        {
            let mut state = self.state.lock().await;
            if state.machine.handle(ShipLifecycleEvent::Healthy).is_ok() {
                state.refresh_status();
            }
        }
        *self.last_health_check.lock().await = Some(Instant::now());
    }

    /// Mark ship as unhealthy
    pub async fn mark_unhealthy(&self) {
        self.transition(ShipLifecycleEvent::Unhealthy).await;
    }

    /// Mark ship as failed (crash)
    pub async fn mark_failed(&self) {
        self.transition(ShipLifecycleEvent::Crash).await;
    }

    /// Record a healthy check (for rate-based thresholds)
    pub async fn record_health_success(&self) {
        let breaker = self.breaker.lock().await;
        breaker.record_success(0.0);
    }

    /// Record an unhealthy check, returns true if circuit is open
    pub async fn record_health_failure(&self) -> bool {
        let mut breaker = self.breaker.lock().await;
        breaker.record_failure_and_maybe_trip(0.0);
        breaker.is_open()
    }

    /// Record a crash against the restart circuit, returns true if circuit is open
    pub async fn record_crash(&self) -> bool {
        let mut breaker = self.breaker.lock().await;
        breaker.record_failure_and_maybe_trip(0.0);
        breaker.is_open()
    }

    /// Reset the restart circuit (clears failure history)
    pub async fn reset_breaker(&self) {
        let mut breaker = self.breaker.lock().await;
        breaker.reset();
    }

    /// Enter restart backoff state
    pub async fn enter_backoff(&self) {
        let until = Instant::now() + self.backoff_delay;
        let mut state = self.state.lock().await;
        if state.machine.handle(ShipLifecycleEvent::Backoff).is_ok() {
            state.refresh_status();
        }
        state.backoff_until = Some(until);
    }

    /// Clear restart backoff state
    pub async fn clear_backoff(&self) {
        let mut state = self.state.lock().await;
        state.backoff_until = None;
    }

    /// Check if ship is currently backing off
    pub async fn is_backing_off(&self) -> bool {
        self.state.lock().await.status == ShipStatus::Backoff
    }

    /// Check if backoff delay has elapsed
    pub async fn backoff_expired(&self) -> bool {
        let state = self.state.lock().await;
        match state.backoff_until {
            Some(until) => Instant::now() >= until,
            None => true,
        }
    }

    /// Get recent logs
    pub async fn logs(&self, limit: usize) -> Vec<LogEntry> {
        self.log_buffer.lock().await.recent(limit)
    }

    /// Take a snapshot of current state
    pub async fn snapshot(&self) -> ShipSnapshot {
        ShipSnapshot {
            name: self.name.clone(),
            group: self.group.clone(),
            command: self.command.clone(),
            status: self.status().await,
            pid: self.pid().await,
            healthcheck: self.healthcheck_url(),
            restart_count: *self.restart_count.lock().await,
            critical: self.critical,
            oneshot: self.oneshot,
            routes: self.routes.iter().map(|r| r.to_string()).collect(),
        }
    }
}

#[cfg(all(test, unix))]
mod tests {
    use std::{collections::HashMap, fs, path::Path, time::Duration};

    use tempfile::tempdir;
    use tokio::time::{Instant, sleep};

    use super::ShipBuilder;

    async fn wait_for_pid_file(path: &Path) -> i32 {
        let deadline = Instant::now() + Duration::from_secs(2);
        loop {
            if let Ok(contents) = fs::read_to_string(path)
                && let Ok(pid) = contents.trim().parse::<i32>()
                && pid > 0
            {
                return pid;
            }
            if Instant::now() >= deadline {
                panic!("timed out waiting for pid file: {}", path.display());
            }
            sleep(Duration::from_millis(20)).await;
        }
    }

    fn process_exists(pid: i32) -> bool {
        let result = unsafe { libc::kill(pid, 0) };
        if result == 0 {
            return true;
        }
        std::io::Error::last_os_error().raw_os_error() != Some(libc::ESRCH)
    }

    async fn wait_for_exit(pid: i32) {
        let deadline = Instant::now() + Duration::from_secs(2);
        loop {
            if !process_exists(pid) {
                return;
            }
            if Instant::now() >= deadline {
                panic!("process {} still running after shutdown", pid);
            }
            sleep(Duration::from_millis(20)).await;
        }
    }

    #[tokio::test]
    async fn terminate_kills_process_group() {
        let temp = tempdir().expect("temp dir");
        let pid_path = temp.path().join("ship-child.pid");

        let mut env = HashMap::new();
        env.insert(
            "PID_FILE".to_string(),
            pid_path.to_string_lossy().to_string(),
        );

        let script = "sleep 1000 & echo $! > \"$PID_FILE\"; wait";
        let ship = ShipBuilder::new("test-ship", "sh")
            .args(vec!["-c".to_string(), script.to_string()])
            .env(env)
            .critical(false)
            .build();

        ship.launch().await.expect("launch ship");

        let child_pid = wait_for_pid_file(&pid_path).await;
        ship.terminate().await.expect("terminate ship");

        assert!(!ship.is_running().await, "ship still running");
        wait_for_exit(child_pid).await;
    }
}