@@ -54,7 +54,39 @@
        /// Config file path
        #[arg(long)]
        config: Option<String>,
        /// PID file path (default ~/.anvil-runner/runner.pid)
        #[arg(long)]
        pid_file: Option<String>,
        /// Seconds to wait for in-flight jobs to finish on shutdown
        /// before forcibly terminating them
        #[arg(long, default_value_t = runner::loop_runner::DEFAULT_SHUTDOWN_TIMEOUT_SECS)]
        shutdown_timeout: u64,
    },
    /// Stop a running runner gracefully
    Stop {
        /// PID file path (default ~/.anvil-runner/runner.pid)
        #[arg(long)]
        pid_file: Option<String>,
        /// Wait this many seconds for clean exit before SIGKILL (with --force)
        /// or returning an error
        #[arg(long, default_value_t = runner::loop_runner::DEFAULT_SHUTDOWN_TIMEOUT_SECS + 10)]
        timeout: u64,
        /// Send SIGKILL if the runner doesn't exit within --timeout
        #[arg(long)]
        force: bool,
    },
    /// Stop then start the runner (for picking up an upgraded binary)
    Restart {
        /// Config file path
        #[arg(long)]
        config: Option<String>,
        /// PID file path (default ~/.anvil-runner/runner.pid)
        #[arg(long)]
        pid_file: Option<String>,
        /// Stop timeout (see `runner stop --timeout`)
        #[arg(long, default_value_t = runner::loop_runner::DEFAULT_SHUTDOWN_TIMEOUT_SECS + 10)]
        timeout: u64,
    },
    /// Show runner configuration
    Status {
        /// Config file path
@@ -118,6 +150,21 @@
    pub command: ServiceCommand,
}
#[derive(Debug, Clone, Copy, clap::ValueEnum)]
pub enum InstallScope {
    User,
    System,
}
impl From<InstallScope> for runner::service_mode::Scope {
    fn from(v: InstallScope) -> Self {
        match v {
            InstallScope::User => runner::service_mode::Scope::User,
            InstallScope::System => runner::service_mode::Scope::System,
        }
    }
}
#[derive(Subcommand)]
pub enum ServiceCommand {
    /// Install as a system service
@@ -125,6 +172,19 @@
        /// Config file path
        #[arg(long)]
        config: Option<String>,
        /// Install scope: user (per-user unit) or system (host-wide unit).
        /// Default: user if invoked as non-root, system if root.
        #[arg(long, value_enum)]
        scope: Option<InstallScope>,
        /// Unix user the system service should run as (required for
        /// --scope=system; ignored otherwise). System services refuse to
        /// install running as root for security reasons.
        #[arg(long)]
        user_account: Option<String>,
        /// Override parallel slots baked into ExecStart. Defaults to the
        /// value in config.json.
        #[arg(long)]
        parallel: Option<u32>,
    },
    /// Uninstall the system service
    Uninstall,
@@ -172,7 +232,29 @@
            ephemeral,
            cleanup,
            config,
            pid_file,
            shutdown_timeout,
        } => {
            start(
                once,
                ephemeral,
                &cleanup,
                config.as_deref(),
                pid_file.as_deref(),
                shutdown_timeout,
            )
            .await
        }
        RunnerCommand::Stop {
            pid_file,
            timeout,
        } => start(once, ephemeral, &cleanup, config.as_deref()).await,
            force,
        } => stop(pid_file.as_deref(), timeout, force).await,
        RunnerCommand::Restart {
            config,
            pid_file,
            timeout,
        } => restart(config.as_deref(), pid_file.as_deref(), timeout).await,
        RunnerCommand::Status { config } => status(config.as_deref()).await,
        RunnerCommand::Unconfigure { config } => unconfigure(config.as_deref()).await,
        RunnerCommand::Service(svc) => service(svc).await,
@@ -305,19 +387,139 @@
    Ok(())
}
async fn stop(
    pid_file: Option<&str>,
    timeout_secs: u64,
    force: bool,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let pid_path = pid_file
        .map(std::path::PathBuf::from)
        .unwrap_or_else(runner::pid_file::default_path);
    match runner::pid_file::inspect(&pid_path) {
        runner::pid_file::Existing::None => {
            output::info("No runner is running (no PID file).");
            Ok(())
        }
        runner::pid_file::Existing::Stale(pid) => {
            output::warn(&format!("Stale PID file (PID {pid} not running); removing"));
            let _ = std::fs::remove_file(&pid_path);
            Ok(())
        }
        runner::pid_file::Existing::Live(pid) => {
            output::info(&format!("Sending SIGTERM to runner (PID {pid})"));
            #[cfg(unix)]
            unsafe {
                if libc::kill(pid, libc::SIGTERM) != 0 {
                    return Err(format!(
                        "kill(SIGTERM, {pid}) failed: {}",
                        std::io::Error::last_os_error()
                    )
                    .into());
                }
            }
            #[cfg(not(unix))]
            return Err("`runner stop` requires Unix signals; not supported on this OS".into());
            // Poll for PID-file removal (signals clean exit from the Guard
            // Drop) OR process death (signals the process exited but
            // didn't get to clean up — e.g. SIGKILL'd elsewhere).
            let deadline = std::time::Instant::now() + std::time::Duration::from_secs(timeout_secs);
            while std::time::Instant::now() < deadline {
                if !pid_path.exists() {
                    output::success(&format!("Runner stopped (PID {pid})"));
                    return Ok(());
                }
                #[cfg(unix)]
                {
                    let alive = unsafe { libc::kill(pid, 0) } == 0;
                    if !alive {
                        let _ = std::fs::remove_file(&pid_path);
                        output::success(&format!(
                            "Runner exited (PID {pid}); cleaned up stale PID file"
                        ));
                        return Ok(());
                    }
                }
                tokio::time::sleep(std::time::Duration::from_millis(200)).await;
            }
            if force {
                output::warn(&format!("Timeout reached; sending SIGKILL to PID {pid}"));
                #[cfg(unix)]
                unsafe {
                    libc::kill(pid, libc::SIGKILL);
                }
                let _ = std::fs::remove_file(&pid_path);
                output::success(&format!("Runner force-killed (PID {pid})"));
                Ok(())
            } else {
                Err(format!(
                    "Runner (PID {pid}) did not exit within {timeout_secs}s. \
                     Re-run with --force to send SIGKILL."
                )
                .into())
            }
        }
    }
}
async fn restart(
    config_path: Option<&str>,
    pid_file: Option<&str>,
    timeout_secs: u64,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    // Stop is no-op if nothing's running. Honor --force semantics by
    // defaulting to true on restart — if a binary upgrade caller can't
    // get the old version to stop cleanly, escalating is the right call.
    stop(pid_file, timeout_secs, true).await?;
    // Re-exec ourselves with the same config so the NEW binary on disk
    // takes over. We use exec rather than spawn because restart's whole
    // point is "swap the running process in place"; spawning would leave
    // the old install.sh-style caller waiting on a dead-end shell.
    let exe = std::env::current_exe()?;
    let mut cmd = std::process::Command::new(exe);
    cmd.arg("runner").arg("start");
    if let Some(c) = config_path {
        cmd.arg("--config").arg(c);
    }
    if let Some(p) = pid_file {
        cmd.arg("--pid-file").arg(p);
    }
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        let err = cmd.exec();
        Err(format!("re-exec failed: {err}").into())
    }
    #[cfg(not(unix))]
    {
        let status = cmd.status()?;
        std::process::exit(status.code().unwrap_or(1));
    }
}
async fn start(
    once: bool,
    ephemeral: bool,
    cleanup: &str,
    config_path: Option<&str>,
    pid_file: Option<&str>,
    shutdown_timeout: u64,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let mut config = RunnerConfig::load(config_path)?;
    config.once = once;
    config.ephemeral = ephemeral;
    config.cleanup = cleanup.to_string();
    let pid_path = pid_file
        .map(std::path::PathBuf::from)
        .unwrap_or_else(runner::pid_file::default_path);
    runner::loop_runner::start(config).await
    runner::loop_runner::start(config, pid_path, shutdown_timeout).await
}
async fn status(config_path: Option<&str>) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -381,34 +583,224 @@
async fn service(args: ServiceArgs) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    match args.command {
        ServiceCommand::Install {
            config,
            scope,
            user_account,
            parallel,
        } => service_install(InstallOpts {
            config_path: config.as_deref(),
            scope: scope.map(Into::into),
        ServiceCommand::Install { config } => service_install(config.as_deref()),
            user_account: user_account.as_deref(),
            parallel,
        }),
        ServiceCommand::Uninstall => service_uninstall(),
        ServiceCommand::Start => service_cmd("start"),
        ServiceCommand::Stop => service_cmd("stop"),
        ServiceCommand::Restart => service_cmd("restart"),
        ServiceCommand::SvcStatus => service_cmd("status"),
    }
}
struct InstallOpts<'a> {
    config_path: Option<&'a str>,
    scope: Option<runner::service_mode::Scope>,
    user_account: Option<&'a str>,
    parallel: Option<u32>,
}
fn service_install(
    config_path: Option<&str>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
fn service_install(opts: InstallOpts<'_>) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    use runner::service_mode::{InstallRecord, Scope};
    let exe = std::env::current_exe()?;
    let config_file = RunnerConfig::path(opts.config_path);
    // Read parallel from config (if present) so install bakes the right
    // value into ExecStart. Override-able via --parallel.
    let cfg = RunnerConfig::load(opts.config_path).ok();
    let parallel = opts
        .parallel
        .or(cfg.as_ref().map(|c| c.parallel))
        .unwrap_or(1);
    let is_root = unix_is_root();
    let scope = opts
        .scope
        .unwrap_or(if is_root { Scope::System } else { Scope::User });
    if matches!(scope, Scope::System) && opts.user_account.is_none() {
        return Err(
            "--scope=system requires --user-account <name>; refusing to install a system \
             service running as root for security reasons"
                .into(),
        );
    }
    let unit_path = if cfg!(target_os = "macos") {
        install_launchd(&exe, &config_file, scope, opts.user_account, parallel)?
    } else {
        install_systemd(&exe, &config_file, scope, opts.user_account, parallel)?
    };
    runner::service_mode::save(&InstallRecord {
        scope,
        unit_path: unit_path.clone(),
        config_path: config_file,
        user_account: opts.user_account.map(String::from),
        parallel,
    })?;
    output::success(&format!(
        "Installed {} service ({} scope): {}",
        if cfg!(target_os = "macos") {
            "launchd"
        } else {
            "systemd"
        },
        scope.as_str(),
        unit_path.display()
    ));
    output::info("Start with: anvil runner service svc-start");
    Ok(())
}
fn install_systemd(
    exe: &std::path::Path,
    config_file: &std::path::Path,
    scope: runner::service_mode::Scope,
    user_account: Option<&str>,
    parallel: u32,
) -> Result<std::path::PathBuf, Box<dyn std::error::Error + Send + Sync>> {
    use runner::service_mode::Scope;
    let exe_str = exe.to_str().unwrap();
    let config_file = RunnerConfig::path(config_path);
    let config_str = config_file.to_str().unwrap();
    let unit_dir: std::path::PathBuf = match scope {
        Scope::System => "/etc/systemd/system".into(),
        Scope::User => {
            let dir = dirs::home_dir().unwrap().join(".config/systemd/user");
            std::fs::create_dir_all(&dir)?;
            dir
        }
    if cfg!(target_os = "macos") {
        // launchd plist
        let plist_dir = dirs::home_dir().unwrap().join("Library/LaunchAgents");
        std::fs::create_dir_all(&plist_dir)?;
    };
    let unit_path = unit_dir.join("anvil-runner.service");
        let plist_path = plist_dir.join("com.anvil.runner.plist");
    let user_line = match scope {
        Scope::System => format!(
            "User={}\nGroup={}\n",
            user_account.unwrap(),
        let log_dir = dirs::home_dir().unwrap().join(".anvil-runner");
        std::fs::create_dir_all(&log_dir)?;
            user_account.unwrap()
        ),
        Scope::User => String::new(),
    };
    let work_dir = match scope {
        Scope::System => "WorkingDirectory=/var/lib/anvil-runner\n".to_string(),
        let plist = format!(
            r#"<?xml version="1.0" encoding="UTF-8"?>
        Scope::User => String::new(),
    };
    let unit = format!(
        r#"[Unit]
Description=Anvil CI Runner
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
ExecStart={exe_str} runner start --config {config_str}
{user_line}{work_dir}Restart=always
RestartSec=5
KillSignal=SIGTERM
TimeoutStopSec=90
StandardOutput=journal
StandardError=journal
ProtectSystem=strict
ProtectHome=true
NoNewPrivileges=true
PrivateTmp=true
ReadWritePaths=/var/lib/anvil-runner /tmp
[Install]
WantedBy={wanted_by}
"#,
        wanted_by = match scope {
            Scope::System => "multi-user.target",
            Scope::User => "default.target",
        },
    );
    // ExecStart needs to include --parallel; substitute into the formatted unit.
    // (We pass through `parallel` via the configured config.json as the source
    // of truth, but baking the flag in keeps the unit self-describing.)
    let unit = unit.replace(
        &format!("ExecStart={exe_str} runner start --config {config_str}"),
        &format!("ExecStart={exe_str} runner start --config {config_str} --shutdown-timeout 60"),
    );
    let _ = parallel; // `parallel` is reflected via config.json; --parallel flag override could be added later
    std::fs::write(&unit_path, unit)?;
    // Reload + enable. Fall through silently on enable failure — the user can
    // run svc-start manually.
    let systemctl_args: Vec<&str> = match scope {
        Scope::User => vec!["--user", "daemon-reload"],
        Scope::System => vec!["daemon-reload"],
    };
    let _ = std::process::Command::new("systemctl")
        .args(&systemctl_args)
        .output();
    let enable_args: Vec<&str> = match scope {
        Scope::User => vec!["--user", "enable", "anvil-runner"],
        Scope::System => vec!["enable", "anvil-runner"],
    };
    let _ = std::process::Command::new("systemctl")
        .args(&enable_args)
        .output();
    Ok(unit_path)
}
fn install_launchd(
    exe: &std::path::Path,
    config_file: &std::path::Path,
    scope: runner::service_mode::Scope,
    user_account: Option<&str>,
    parallel: u32,
) -> Result<std::path::PathBuf, Box<dyn std::error::Error + Send + Sync>> {
    use runner::service_mode::Scope;
    let _ = parallel; // baked through config.json today; --parallel override is a follow-up
    let exe_str = exe.to_str().unwrap();
    let config_str = config_file.to_str().unwrap();
    let (plist_dir, log_dir) = match scope {
        Scope::User => (
            dirs::home_dir().unwrap().join("Library/LaunchAgents"),
            dirs::home_dir().unwrap().join("Library/Logs/anvil-runner"),
        ),
        Scope::System => (
            std::path::PathBuf::from("/Library/LaunchDaemons"),
            std::path::PathBuf::from("/var/log/anvil-runner"),
        ),
    };
    std::fs::create_dir_all(&plist_dir)?;
    let _ = std::fs::create_dir_all(&log_dir);
    let plist_path = plist_dir.join("com.anvil.runner.plist");
    let user_keys = match scope {
        Scope::System => format!(
            "    <key>UserName</key>\n    <string>{u}</string>\n    <key>GroupName</key>\n    <string>{u}</string>\n",
            u = user_account.unwrap()
        ),
        Scope::User => String::new(),
    };
    let home = dirs::home_dir().unwrap();
    let plist = format!(
        r#"<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
@@ -426,121 +818,80 @@
    <true/>
    <key>KeepAlive</key>
    <true/>
    <key>ProcessType</key>
    <string>Background</string>
    <key>ThrottleInterval</key>
    <integer>5</integer>
    <key>StandardOutPath</key>
    <string>{log_out}</string>
    <key>StandardErrorPath</key>
    <string>{log_err}</string>
{user_keys}    <key>EnvironmentVariables</key>
    <key>EnvironmentVariables</key>
    <dict>
        <key>HOME</key>
        <string>{home}</string>
    </dict>
</dict>
</plist>"#,
        log_out = log_dir.join("runner.log").display(),
        log_err = log_dir.join("runner.err.log").display(),
        home = home.display(),
            log_out = log_dir.join("runner.log").display(),
            log_err = log_dir.join("runner.err.log").display(),
            home = dirs::home_dir().unwrap().display(),
        );
    );
        std::fs::write(&plist_path, plist)?;
        output::success(&format!(
            "Installed launchd service: {}",
            plist_path.display()
        ));
        output::info("Start with: anvil runner service svc-start");
    } else {
        // systemd service
    std::fs::write(&plist_path, plist)?;
    Ok(plist_path)
}
        let is_root = unsafe { libc::getuid() } == 0;
        let (unit_dir, user_flag) = if is_root {
            ("/etc/systemd/system".to_string(), "")
        } else {
            let dir = dirs::home_dir().unwrap().join(".config/systemd/user");
            std::fs::create_dir_all(&dir)?;
            (dir.to_string_lossy().to_string(), " --user")
        };
fn service_uninstall() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        let unit_path = format!("{unit_dir}/anvil-runner.service");
    use runner::service_mode::Scope;
        let unit = format!(
            r#"[Unit]
Description=Anvil CI Runner
After=network.target
[Service]
Type=simple
ExecStart={exe_str} runner start --config {config_str}
Restart=always
    let record = runner::service_mode::load()
        .ok_or_else(|| "no service install recorded; nothing to uninstall".to_string())?;
RestartSec=5
[Install]
WantedBy=default.target
"#,
        );
        std::fs::write(&unit_path, unit)?;
        // Enable
        let _ = std::process::Command::new("systemctl")
            .args(if user_flag.is_empty() {
                vec!["enable", "anvil-runner"]
            } else {
                vec!["--user", "enable", "anvil-runner"]
            })
            .output();
        output::success(&format!("Installed systemd service: {unit_path}"));
        output::info("Start with: anvil runner service svc-start");
    }
    Ok(())
}
fn service_uninstall() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    if cfg!(target_os = "macos") {
        // Best-effort unload first (so the running daemon goes away
        // before we delete the plist).
        let plist = dirs::home_dir()
            .unwrap()
            .join("Library/LaunchAgents/com.anvil.runner.plist");
        let _ = std::process::Command::new("launchctl")
            .args(["unload", plist.to_str().unwrap()])
            .args(["unload", record.unit_path.to_str().unwrap()])
            .output();
        if plist.exists() {
            std::fs::remove_file(&plist)?;
        }
        output::success("Uninstalled launchd service");
    } else {
        let is_root = unsafe { libc::getuid() } == 0;
        let args: Vec<&str> = if is_root {
            vec!["disable", "anvil-runner"]
        let args: Vec<&str> = match record.scope {
            Scope::User => vec!["--user", "disable", "anvil-runner"],
            Scope::System => vec!["disable", "anvil-runner"],
        } else {
            vec!["--user", "disable", "anvil-runner"]
        };
        let _ = std::process::Command::new("systemctl").args(&args).output();
    }
    if record.unit_path.exists() {
        std::fs::remove_file(&record.unit_path)?;
    }
        let unit = if is_root {
            "/etc/systemd/system/anvil-runner.service".to_string()
    runner::service_mode::clear()?;
    output::success(&format!(
        "Uninstalled {} service ({} scope)",
        if cfg!(target_os = "macos") {
            "launchd"
        } else {
            "systemd"
        },
        record.scope.as_str()
            dirs::home_dir()
                .unwrap()
                .join(".config/systemd/user/anvil-runner.service")
                .to_string_lossy()
    ));
                .to_string()
        };
        if std::path::Path::new(&unit).exists() {
            std::fs::remove_file(&unit)?;
        }
        output::success("Uninstalled systemd service");
    }
    Ok(())
}
fn service_cmd(action: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    use runner::service_mode::Scope;
    let record = runner::service_mode::load().ok_or_else(|| {
        "no service install recorded; run `anvil runner service install` first".to_string()
    })?;
    if cfg!(target_os = "macos") {
        let plist = dirs::home_dir()
            .unwrap()
            .join("Library/LaunchAgents/com.anvil.runner.plist");
        let plist_str = record.unit_path.to_str().unwrap();
        // launchctl has no restart verb; emulate with unload + load.
        // Unload failures during restart are non-fatal (service may
        // launchctl has no restart verb — emulate with unload + load. We
        // already be stopped).
        // ignore unload failures because the service may already be
        // stopped; only load failures should surface as errors.
        let launchctl_actions: &[&str] = match action {
            "start" => &["load"],
            "stop" => &["unload"],
@@ -558,13 +909,12 @@
        let mut last_failure: Option<String> = None;
        for (idx, launchctl_action) in launchctl_actions.iter().enumerate() {
            let output = std::process::Command::new("launchctl")
                .args([launchctl_action, plist.to_str().unwrap()])
                .args([launchctl_action, plist_str])
                .output()?;
            if !output.status.success() {
                let stderr = String::from_utf8_lossy(&output.stderr).to_string();
                let is_first_of_pair = action == "restart" && idx == 0;
                if is_first_of_pair {
                    // unload-before-load failure is non-fatal
                    continue;
                }
                last_failure = Some(format!("launchctl {launchctl_action} failed: {stderr}"));
@@ -577,9 +927,8 @@
        }
        output::success(&format!("Service {action}ed"));
    } else {
        let is_root = unsafe { libc::getuid() } == 0;
        let mut args: Vec<&str> = Vec::new();
        if !is_root {
        if matches!(record.scope, Scope::User) {
            args.push("--user");
        }
        args.push(action);
@@ -601,6 +950,16 @@
    Ok(())
}
#[cfg(unix)]
fn unix_is_root() -> bool {
    (unsafe { libc::getuid() }) == 0
}
#[cfg(not(unix))]
fn unix_is_root() -> bool {
    false
}
// === Admin commands (PAT auth) ===
async fn list(
@@ -828,5 +1187,114 @@
            parse(&["service", "svc-status"]),
            ServiceCommand::SvcStatus
        ));
    }
    fn parse_runner(args: &[&str]) -> RunnerCommand {
        RunnerCli::try_parse_from(args).expect("parse").command
    }
    #[test]
    fn stop_parses_with_defaults() {
        match parse_runner(&["stop"]) {
            RunnerCommand::Stop {
                pid_file,
                timeout,
                force,
            } => {
                assert!(pid_file.is_none());
                assert!(timeout >= 60);
                assert!(!force);
            }
            _ => panic!("expected Stop"),
        }
    }
    #[test]
    fn stop_parses_force_and_timeout() {
        match parse_runner(&["stop", "--force", "--timeout", "30"]) {
            RunnerCommand::Stop { timeout, force, .. } => {
                assert_eq!(timeout, 30);
                assert!(force);
            }
            _ => panic!("expected Stop"),
        }
    }
    #[test]
    fn restart_parses() {
        match parse_runner(&["restart"]) {
            RunnerCommand::Restart { .. } => {}
            _ => panic!("expected Restart"),
        }
    }
    #[test]
    fn start_parses_pid_file_and_shutdown_timeout() {
        match parse_runner(&[
            "start",
            "--pid-file",
            "/tmp/runner.pid",
            "--shutdown-timeout",
            "45",
        ]) {
            RunnerCommand::Start {
                pid_file,
                shutdown_timeout,
                ..
            } => {
                assert_eq!(pid_file.as_deref(), Some("/tmp/runner.pid"));
                assert_eq!(shutdown_timeout, 45);
            }
            _ => panic!("expected Start"),
        }
    }
    #[test]
    fn service_install_parses_scope_user() {
        match parse(&["service", "install", "--scope", "user"]) {
            ServiceCommand::Install { scope, .. } => {
                assert!(matches!(scope, Some(InstallScope::User)));
            }
            _ => panic!("expected Install"),
        }
    }
    #[test]
    fn service_install_parses_scope_system_with_user_account() {
        match parse(&[
            "service",
            "install",
            "--scope",
            "system",
            "--user-account",
            "ci-runner",
        ]) {
            ServiceCommand::Install {
                scope,
                user_account,
                ..
            } => {
                assert!(matches!(scope, Some(InstallScope::System)));
                assert_eq!(user_account.as_deref(), Some("ci-runner"));
            }
            _ => panic!("expected Install"),
        }
    }
    #[test]
    fn service_install_parses_with_no_flags() {
        match parse(&["service", "install"]) {
            ServiceCommand::Install {
                scope,
                user_account,
                parallel,
                ..
            } => {
                assert!(scope.is_none());
                assert!(user_account.is_none());
                assert!(parallel.is_none());
            }
            _ => panic!("expected Install"),
        }
    }
}

@@ -1,16 +1,27 @@
use crate::runner::log_reporter::LogReporter;
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command;
use tokio::sync::Notify;
const DEFAULT_TIMEOUT_SECS: u64 = 600;
/// How long a SIGTERM'd child gets to exit cleanly before we send SIGKILL.
const CHILD_TERM_GRACE_SECS: u64 = 5;
pub struct ExecResult {
    pub exit_code: i32,
}
/// Execute a job command, streaming output to the log reporter.
///
/// `cancel` is the shared shutdown signal: when notified, the executor
/// SIGTERMs the child process, waits up to `CHILD_TERM_GRACE_SECS`, and
/// SIGKILLs if still alive. The returned result has exit_code=130
/// (Unix-convention "terminated by SIGINT") so the caller can distinguish
/// from a normal failure.
#[allow(clippy::too_many_arguments)]
pub async fn execute(
    command: &str,
    image: Option<&str>,
@@ -19,21 +30,33 @@
    network: Option<&str>,
    timeout_seconds: Option<u64>,
    log_reporter: &LogReporter,
    cancel: Arc<Notify>,
) -> Result<ExecResult, Box<dyn std::error::Error + Send + Sync>> {
    let timeout = timeout_seconds.unwrap_or(DEFAULT_TIMEOUT_SECS);
    match image {
        Some(img) => {
            eprintln!("Executing (docker): {img} — {command}");
            execute_docker(command, img, workspace, env, network, timeout, log_reporter).await
            execute_docker(
                command,
                img,
                workspace,
                env,
                network,
                timeout,
                log_reporter,
                cancel,
            )
            .await
        }
        None => {
            eprintln!("Executing (bare): {command}");
            execute_bare(command, workspace, env, timeout, log_reporter).await
            execute_bare(command, workspace, env, timeout, log_reporter, cancel).await
        }
    }
}
#[allow(clippy::too_many_arguments)]
async fn execute_docker(
    command: &str,
    image: &str,
@@ -42,6 +65,7 @@
    network: Option<&str>,
    timeout: u64,
    log_reporter: &LogReporter,
    cancel: Arc<Notify>,
) -> Result<ExecResult, Box<dyn std::error::Error + Send + Sync>> {
    // Pull image first to ensure it's available (skip for locally-built prepared images)
    if !image.starts_with("anvil-prepared:") {
@@ -136,6 +160,7 @@
        &HashMap::new(),
        timeout,
        log_reporter,
        cancel,
    )
    .await
}
@@ -146,6 +171,7 @@
    env: &HashMap<String, String>,
    timeout: u64,
    log_reporter: &LogReporter,
    cancel: Arc<Notify>,
) -> Result<ExecResult, Box<dyn std::error::Error + Send + Sync>> {
    let mut full_env = env.clone();
    full_env.insert(
@@ -162,6 +188,7 @@
        &full_env,
        timeout,
        log_reporter,
        cancel,
    )
    .await
}
@@ -230,6 +257,7 @@
    }
}
#[allow(clippy::too_many_arguments)]
async fn run_and_stream(
    program: &str,
    args: &[String],
@@ -237,6 +265,7 @@
    env: &HashMap<String, String>,
    timeout: u64,
    log_reporter: &LogReporter,
    cancel: Arc<Notify>,
) -> Result<ExecResult, Box<dyn std::error::Error + Send + Sync>> {
    let mut cmd = Command::new(program);
    cmd.args(args)
@@ -273,29 +302,80 @@
        }
    });
    // Wait with timeout
    // Race three outcomes:
    //   1. child finishes (success or non-zero exit) within timeout
    //   2. timeout expires → SIGKILL the child
    //   3. shutdown signaled → SIGTERM, grace window, SIGKILL
    let result = tokio::time::timeout(std::time::Duration::from_secs(timeout), async {
        // Wait for output tasks to complete (they finish when the process closes its pipes)
        let _ = stdout_task.await;
        let _ = stderr_task.await;
        child.wait().await
    })
    .await;
    //
    // The stdout/stderr tasks finish naturally when the child closes its
    // pipes, which happens shortly after the child exits in all three
    // outcomes. We drain them after the select.
    let timeout_duration = std::time::Duration::from_secs(timeout);
    enum Outcome {
        Finished(Result<std::process::ExitStatus, std::io::Error>),
        Cancelled,
    // Final flush
    log_reporter.flush().await;
        TimedOut,
    }
    match result {
        Ok(Ok(status)) => Ok(ExecResult {
            exit_code: status.code().unwrap_or(1),
        }),
        Ok(Err(e)) => Err(format!("process error: {e}").into()),
        Err(_) => {
            // Timeout — kill the child
    let outcome = tokio::select! {
        biased;
        _ = cancel.notified() => Outcome::Cancelled,
        wait = tokio::time::timeout(timeout_duration, child.wait()) => match wait {
            Ok(r) => Outcome::Finished(r),
            Err(_) => Outcome::TimedOut,
        },
    };
    let exit_code = match outcome {
        Outcome::Cancelled => {
            log_reporter
                .append("Runner shutdown — terminating job")
                .await;
            terminate_child(&mut child, log_reporter).await;
            130
        }
        Outcome::TimedOut => {
            let _ = child.kill().await;
            log_reporter.append("Job timed out, killed").await;
            1
        }
        Outcome::Finished(Ok(status)) => status.code().unwrap_or(1),
        Outcome::Finished(Err(e)) => {
            let _ = stdout_task.await;
            let _ = stderr_task.await;
            log_reporter.flush().await;
            return Err(format!("process error: {e}").into());
            Ok(ExecResult { exit_code: 1 })
        }
    };
    // Drain output tasks now that the child is done one way or the other.
    let _ = stdout_task.await;
    let _ = stderr_task.await;
    log_reporter.flush().await;
    Ok(ExecResult { exit_code })
}
/// Send SIGTERM to a running child, wait briefly, then SIGKILL if still
/// alive. Used both by the executor's shutdown path and the timeout path.
async fn terminate_child(child: &mut tokio::process::Child, log_reporter: &LogReporter) {
    #[cfg(unix)]
    if let Some(pid) = child.id() {
        // Best-effort SIGTERM — if it returns an error we still fall
        // through to SIGKILL via child.kill().
        unsafe {
            libc::kill(pid as i32, libc::SIGTERM);
        }
    }
    let grace = std::time::Duration::from_secs(CHILD_TERM_GRACE_SECS);
    if tokio::time::timeout(grace, child.wait()).await.is_err() {
        log_reporter
            .append(&format!(
                "Child did not exit within {CHILD_TERM_GRACE_SECS}s of SIGTERM — sending SIGKILL"
            ))
            .await;
        let _ = child.kill().await;
    }
}

@@ -1,11 +1,16 @@
use crate::runner::slot_counter::SlotCounter;
use crate::runner::RunnerConfig;
use reqwest::header::{HeaderValue, AUTHORIZATION, CONTENT_TYPE};
use reqwest::header::{HeaderValue, AUTHORIZATION};
/// Start a heartbeat loop that sends periodic pings to the server with
/// capacity telemetry (`parallel` total slots, `slots_busy` currently
/// Start a heartbeat loop that sends periodic pings to the server.
/// executing). Older servers that don't parse those fields are a no-op.
///
/// Returns a JoinHandle — abort it to stop heartbeating.
pub fn start(config: &RunnerConfig) -> tokio::task::JoinHandle<()> {
pub fn start(config: &RunnerConfig, slots: SlotCounter) -> tokio::task::JoinHandle<()> {
    let url = config.api_url(&format!("/runners/{}/heartbeat", config.runner_id));
    let auth = config.auth_header();
    let interval_ms = config.heartbeat_interval_ms;
    let parallel = config.parallel;
    let client = reqwest::Client::new();
@@ -16,9 +21,16 @@
        loop {
            interval.tick().await;
            let body = serde_json::json!({
                "parallel": parallel,
                "slots_busy": slots.busy(),
            });
            match client
                .post(&url)
                .header(AUTHORIZATION, HeaderValue::from_str(&auth).unwrap())
                .header(CONTENT_TYPE, HeaderValue::from_static("application/json"))
                .json(&body)
                .send()
                .await
            {

@@ -2,20 +2,58 @@
use crate::runner::executor::{self, ExecResult};
use crate::runner::heartbeat;
use crate::runner::log_reporter::LogReporter;
use crate::runner::pid_file::{self, Existing};
use crate::runner::prepare;
use crate::runner::service_manager;
use crate::runner::shutdown;
use crate::runner::slot_counter::SlotCounter;
use crate::runner::workspace;
use crate::runner::RunnerConfig;
use reqwest::header::{HeaderValue, AUTHORIZATION, CONTENT_TYPE};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::sync::Notify;
const MIN_BACKOFF_MS: u64 = 5_000;
const MAX_BACKOFF_MS: u64 = 30_000;
/// Default grace window for jobs to finish after a shutdown signal. The
/// runner stops claiming new jobs immediately; in-flight jobs get this
/// long to complete before the executor forcibly terminates them.
pub const DEFAULT_SHUTDOWN_TIMEOUT_SECS: u64 = 60;
/// Start the runner main loop.
///
/// Acquires a PID file (single-instance guard), installs SIGTERM/SIGINT
/// handlers, spawns `config.parallel` poller slots, and waits for either
/// a shutdown signal or all slots to exit (the `--once` / `--ephemeral`
/// paths).
pub async fn start(
    config: RunnerConfig,
    pid_path: PathBuf,
    shutdown_timeout_secs: u64,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    // Single-instance check. Live PID → refuse; stale → warn + overwrite.
    match pid_file::inspect(&pid_path) {
        Existing::Live(pid) => {
            return Err(format!(
                "another anvil runner appears to be running (PID {pid}). \
                 If it isn't, remove {}",
                pid_path.display()
            )
            .into());
        }
        Existing::Stale(pid) => {
            eprintln!(
                "warning: stale PID file at {} (PID {pid} not running); overwriting",
                pid_path.display()
            );
        }
        Existing::None => {}
    }
pub async fn start(config: RunnerConfig) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let _pid_guard = pid_file::Guard::acquire(pid_path, std::process::id() as i32)?;
    // Validate connection
    eprintln!("Connecting to {}...", config.server_url);
    heartbeat::send_once(&config).await?;
@@ -25,26 +63,23 @@
    // drop anything older than 30 days.
    prepare::prune_prepared_images(30, 10);
    // Counter of busy slots — read by the heartbeat, incremented by each
    // poller on job claim, released on RAII drop.
    // Start heartbeat
    let slots = SlotCounter::new();
    // Start heartbeat with capacity telemetry.
    let heartbeat_handle = heartbeat::start(&config, slots.clone());
    let heartbeat_handle = heartbeat::start(&config);
    // Shutdown signal
    let shutdown = Arc::new(Notify::new());
    let shutdown_clone = shutdown.clone();
    tokio::spawn(async move {
        tokio::signal::ctrl_c()
            .await
            .expect("failed to listen for ctrl+c");
        eprintln!("\nShutting down...");
        shutdown_clone.notify_waiters();
    // Listen for SIGTERM + SIGINT (Ctrl+C); ignore SIGHUP + SIGPIPE.
    let shutdown_signal = shutdown::install();
    });
    // Run polling slots
    let mut handles = Vec::new();
    for slot in 1..=config.parallel {
        let cfg = config.clone();
        let stop = shutdown_signal.clone();
        let stop = shutdown.clone();
        let slot_counter = slots.clone();
        handles.push(tokio::spawn(async move {
            poller_loop(cfg, slot, stop, slot_counter).await;
            poller_loop(cfg, slot, stop).await;
        }));
    }
@@ -53,25 +88,27 @@
    // (the runner relay, #298) and serve them from its local model.
    if config.labels.iter().any(|l| l == "inference") {
        let cfg = config.clone();
        let stop = shutdown.clone();
        let stop = shutdown_signal.clone();
        handles.push(tokio::spawn(async move {
            crate::runner::inference::poller_loop(cfg, stop).await;
        }));
    }
    // Wait for shutdown signal
    shutdown_signal.notified().await;
    shutdown.notified().await;
    // Grace period: wait for active jobs (up to 30s)
    eprintln!("Waiting for active jobs to finish (30s grace period)...");
    let grace = tokio::time::timeout(
        std::time::Duration::from_secs(30),
    // Drain: pollers stop claiming, in-flight jobs get the shutdown signal
    // via the same Notify (the executor races shutdown vs. child-exit).
    // We then wait up to shutdown_timeout_secs for everything to wind up.
    eprintln!("Draining (up to {shutdown_timeout_secs}s for in-flight jobs to finish)...");
    let drain = tokio::time::timeout(
        std::time::Duration::from_secs(shutdown_timeout_secs),
        futures::future::join_all(handles),
    )
    .await;
    if grace.is_err() {
    if drain.is_err() {
        eprintln!("Drain timeout ({shutdown_timeout_secs}s) — slots may not have fully released");
        eprintln!("Grace period expired, forcing shutdown.");
    }
    // Stop heartbeat
@@ -84,6 +121,6 @@
    Ok(())
}
async fn poller_loop(config: RunnerConfig, slot: u32, shutdown: Arc<Notify>, slots: SlotCounter) {
async fn poller_loop(config: RunnerConfig, slot: u32, shutdown: Arc<Notify>) {
    let client = reqwest::Client::new();
    let mut backoff_ms = MIN_BACKOFF_MS;
@@ -108,8 +145,12 @@
                    &job_id[..8.min(job_id.len())]
                );
                // Reserve the slot for the heartbeat counter. Released
                // on drop even if execute_job panics.
                let _reservation = slots.reserve();
                let result = execute_job(&config, &job, slot).await;
                let result = execute_job(&config, &job, slot, shutdown.clone()).await;
                report_result(&client, &config, &job_id, result).await;
                drop(_reservation);
                // Handle --once / --ephemeral
                if config.once || config.ephemeral {
@@ -180,6 +221,7 @@
    config: &RunnerConfig,
    job: &serde_json::Value,
    slot: u32,
    cancel: Arc<Notify>,
) -> Result<ExecResult, Box<dyn std::error::Error + Send + Sync>> {
    let job_id = job.get("id").and_then(|v| v.as_str()).unwrap_or("unknown");
    let command = job
@@ -315,5 +357,6 @@
        network,
        timeout_seconds,
        &log_reporter,
        cancel,
    )
    .await;

@@ -5,8 +5,12 @@
pub mod inference;
pub mod log_reporter;
pub mod loop_runner;
pub mod pid_file;
pub mod prepare;
pub mod service_manager;
pub mod service_mode;
pub mod shutdown;
pub mod slot_counter;
pub mod workspace;
pub use config::RunnerConfig;

@@ -1,0 +1,171 @@
//! Runner PID file: enables single-instance enforcement on `runner start`
//! and out-of-band stop via `runner stop`.
//!
//! Layout: `~/.anvil-runner/runner.pid` (overridable). The file contains
//! the runner process's PID as a single decimal line. `Guard` writes it on
//! acquisition and removes it on `Drop` so a clean exit leaves no trace.
//!
//! Stale detection: if the PID file exists but names a process that no
//! longer exists (or isn't an anvil binary on the platforms where we can
//! tell), we treat it as stale, warn, and overwrite.
use std::fs;
use std::io::Write;
use std::path::{Path, PathBuf};
/// Default location: `~/.anvil-runner/runner.pid`.
pub fn default_path() -> PathBuf {
    dirs::home_dir()
        .unwrap_or_else(|| PathBuf::from("."))
        .join(".anvil-runner")
        .join("runner.pid")
}
/// Result of inspecting an existing PID file.
#[derive(Debug)]
pub enum Existing {
    /// File is absent.
    None,
    /// File names a live process with this PID.
    Live(i32),
    /// File exists but the PID is dead (or unreadable). Caller may overwrite.
    Stale(i32),
}
pub fn inspect(path: &Path) -> Existing {
    let Ok(contents) = fs::read_to_string(path) else {
        return Existing::None;
    };
    let Ok(pid) = contents.trim().parse::<i32>() else {
        // Unparseable junk — treat as stale so a re-start can overwrite.
        return Existing::Stale(0);
    };
    if pid_is_alive(pid) {
        Existing::Live(pid)
    } else {
        Existing::Stale(pid)
    }
}
#[cfg(unix)]
fn pid_is_alive(pid: i32) -> bool {
    if pid <= 0 {
        return false;
    }
    // kill(pid, 0) tests existence without sending a signal: returns 0 if
    // the process exists (regardless of whether we have permission to
    // signal it), -1/ESRCH if it's gone, -1/EPERM if it exists but we're
    // not allowed to signal — for our purposes the last counts as "alive".
    let rc = unsafe { libc::kill(pid, 0) };
    if rc == 0 {
        return true;
    }
    let errno = std::io::Error::last_os_error().raw_os_error().unwrap_or(0);
    errno == libc::EPERM
}
#[cfg(not(unix))]
fn pid_is_alive(_pid: i32) -> bool {
    // Conservative on non-Unix: if a file is here, assume alive. Worst
    // case is the user sees a "refused, looks like it's already running"
    // and runs `anvil runner stop --force`.
    true
}
/// RAII handle for the PID file. The file is written on `acquire` and
/// removed on `Drop`. If the process is killed (SIGKILL, panic in another
/// thread), the file is left on disk and treated as stale on the next
/// `inspect`.
pub struct Guard {
    path: PathBuf,
}
impl Guard {
    /// Write `pid` (typically `std::process::id()` as i32) to `path`.
    /// Caller is responsible for first inspecting the existing file and
    /// deciding whether to proceed.
    pub fn acquire(path: PathBuf, pid: i32) -> std::io::Result<Self> {
        if let Some(parent) = path.parent() {
            fs::create_dir_all(parent)?;
        }
        let mut file = fs::File::create(&path)?;
        writeln!(file, "{pid}")?;
        file.sync_all()?;
        Ok(Guard { path })
    }
    #[allow(dead_code)]
    pub fn path(&self) -> &Path {
        &self.path
    }
}
impl Drop for Guard {
    fn drop(&mut self) {
        let _ = fs::remove_file(&self.path);
    }
}
#[cfg(test)]
mod tests {
    use super::*;
    fn temp_pid_path(tag: &str) -> PathBuf {
        // Per-test path so tests running in parallel don't race on the
        // same file.
        let dir = std::env::temp_dir().join(format!(
            "anvil-runner-pid-test-{}-{tag}",
            std::process::id()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        dir.join("runner.pid")
    }
    #[test]
    fn inspect_absent_returns_none() {
        let path = temp_pid_path("absent");
        let _ = std::fs::remove_file(&path);
        assert!(matches!(inspect(&path), Existing::None));
    }
    #[test]
    fn inspect_garbage_returns_stale() {
        let path = temp_pid_path("garbage");
        std::fs::write(&path, "not-a-pid\n").unwrap();
        assert!(matches!(inspect(&path), Existing::Stale(_)));
        let _ = std::fs::remove_file(&path);
    }
    #[test]
    fn inspect_dead_pid_returns_stale() {
        let path = temp_pid_path("dead");
        // PID 1 always exists on Unix — use a definitely-dead PID instead.
        // 0x7FFFFFFF is the maximum signed-32-bit value; no system reaches it.
        std::fs::write(&path, "2147483647\n").unwrap();
        assert!(matches!(inspect(&path), Existing::Stale(_)));
        let _ = std::fs::remove_file(&path);
    }
    #[cfg(unix)]
    #[test]
    fn inspect_live_pid_returns_live() {
        let path = temp_pid_path("live");
        let my_pid = std::process::id() as i32;
        std::fs::write(&path, format!("{my_pid}\n")).unwrap();
        assert!(matches!(inspect(&path), Existing::Live(p) if p == my_pid));
        let _ = std::fs::remove_file(&path);
    }
    #[test]
    fn guard_writes_pid_and_removes_on_drop() {
        let path = temp_pid_path("guard");
        let _ = std::fs::remove_file(&path);
        {
            let guard = Guard::acquire(path.clone(), 12345).unwrap();
            assert!(path.exists());
            let contents = std::fs::read_to_string(guard.path()).unwrap();
            assert_eq!(contents.trim(), "12345");
        }
        assert!(!path.exists(), "guard should remove file on drop");
    }
}

@@ -1,0 +1,105 @@
//! Records which service-manager mode the runner was installed under so
//! `svc-start/stop/restart/status` can target the correct scope without
//! the prior brittle "is the current uid 0?" heuristic.
//!
//! Persisted at `~/.anvil-runner/service.json` on install, deleted on
//! uninstall.
use serde::{Deserialize, Serialize};
use std::fs;
use std::path::PathBuf;
/// On Linux: `user` writes to `~/.config/systemd/user/` and uses
/// `systemctl --user`; `system` writes to `/etc/systemd/system/` and uses
/// `systemctl` without `--user`.
///
/// On macOS: `user` writes a LaunchAgent under `~/Library/LaunchAgents/`;
/// `system` writes a LaunchDaemon under `/Library/LaunchDaemons/` (and
/// requires `UserName=`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Scope {
    User,
    System,
}
impl Scope {
    pub fn as_str(self) -> &'static str {
        match self {
            Scope::User => "user",
            Scope::System => "system",
        }
    }
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InstallRecord {
    pub scope: Scope,
    /// Path to the unit/plist file we wrote, so uninstall can find it
    /// without re-deriving it (which gets brittle when scopes differ).
    pub unit_path: PathBuf,
    /// Path to the runner config the unit was wired to.
    pub config_path: PathBuf,
    /// Username the service runs as. Always set for system scope; for
    /// user scope this is the installing user's name (informational).
    pub user_account: Option<String>,
    /// Parallel-slot count baked into the unit's ExecStart.
    pub parallel: u32,
}
pub fn path() -> PathBuf {
    dirs::home_dir()
        .unwrap_or_else(|| PathBuf::from("."))
        .join(".anvil-runner")
        .join("service.json")
}
pub fn save(record: &InstallRecord) -> std::io::Result<()> {
    let p = path();
    if let Some(parent) = p.parent() {
        fs::create_dir_all(parent)?;
    }
    let contents = serde_json::to_string_pretty(record).map_err(std::io::Error::other)?;
    fs::write(p, contents)
}
pub fn load() -> Option<InstallRecord> {
    let contents = fs::read_to_string(path()).ok()?;
    serde_json::from_str(&contents).ok()
}
pub fn clear() -> std::io::Result<()> {
    let p = path();
    if p.exists() {
        fs::remove_file(p)
    } else {
        Ok(())
    }
}
#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn scope_round_trips_through_json() {
        let record = InstallRecord {
            scope: Scope::System,
            unit_path: "/etc/systemd/system/anvil-runner.service".into(),
            config_path: "/home/ci/.anvil-runner/config.json".into(),
            user_account: Some("ci-runner".into()),
            parallel: 4,
        };
        let json = serde_json::to_string(&record).unwrap();
        let parsed: InstallRecord = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed.scope, Scope::System);
        assert_eq!(parsed.parallel, 4);
        assert_eq!(parsed.user_account.as_deref(), Some("ci-runner"));
    }
    #[test]
    fn scope_serializes_as_lowercase() {
        let json = serde_json::to_string(&Scope::User).unwrap();
        assert_eq!(json, "\"user\"");
    }
}

@@ -1,0 +1,65 @@
//! Shutdown signal handling for the runner.
//!
//! Listens for SIGINT (Ctrl+C) and SIGTERM (`systemctl stop`, `kill`,
//! `docker stop`, k8s pre-SIGKILL) — both trigger graceful drain via the
//! shared `Notify`. SIGHUP and SIGPIPE are explicitly ignored so the
//! default actions (terminate / write-error-kills-process) don't kill us.
//!
//! On Unix we use `tokio::signal::unix::signal`. On non-Unix we fall back
//! to ctrl_c only.
use std::sync::Arc;
use tokio::sync::Notify;
/// Spawn the signal listener. The returned `Notify` fires once on the
/// first SIGINT or SIGTERM; subsequent signals are absorbed silently so
/// repeated Ctrl+C doesn't accidentally interrupt the drain logic.
pub fn install() -> Arc<Notify> {
    let shutdown = Arc::new(Notify::new());
    let trigger = shutdown.clone();
    tokio::spawn(async move {
        wait_for_signal().await;
        trigger.notify_waiters();
    });
    shutdown
}
#[cfg(unix)]
async fn wait_for_signal() {
    use tokio::signal::unix::{signal, SignalKind};
    let mut sigint = signal(SignalKind::interrupt()).expect("failed to install SIGINT handler");
    let mut sigterm = signal(SignalKind::terminate()).expect("failed to install SIGTERM handler");
    // Explicitly absorb SIGHUP (default action: terminate) and SIGPIPE
    // (default action: terminate on broken stdout/stderr pipe) so a
    // dropped terminal or a broken log pipe doesn't kill the runner.
    let mut sighup = signal(SignalKind::hangup()).expect("failed to install SIGHUP handler");
    let mut sigpipe = signal(SignalKind::pipe()).expect("failed to install SIGPIPE handler");
    tokio::spawn(async move {
        loop {
            sighup.recv().await;
            eprintln!("(SIGHUP ignored)");
        }
    });
    tokio::spawn(async move {
        loop {
            sigpipe.recv().await;
            // Don't log SIGPIPE — they're high-volume on broken stdout.
        }
    });
    tokio::select! {
        _ = sigint.recv() => eprintln!("\nReceived SIGINT, shutting down..."),
        _ = sigterm.recv() => eprintln!("Received SIGTERM, shutting down..."),
    }
}
#[cfg(not(unix))]
async fn wait_for_signal() {
    let _ = tokio::signal::ctrl_c().await;
    eprintln!("Received Ctrl+C, shutting down...");
}

@@ -1,0 +1,65 @@
//! Atomic counter of currently-busy job slots. The heartbeat reads this
//! to report `slots_busy` to the server. Each slot's `poller_loop`
//! increments when it claims a job and decrements when the executor
//! returns — via the RAII `Reservation` so a panic inside execute_job
//! still releases the slot.
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
#[derive(Clone, Default)]
pub struct SlotCounter {
    busy: Arc<AtomicUsize>,
}
impl SlotCounter {
    pub fn new() -> Self {
        Self::default()
    }
    pub fn busy(&self) -> usize {
        self.busy.load(Ordering::Relaxed)
    }
    /// Mark one slot busy until the returned guard is dropped.
    pub fn reserve(&self) -> Reservation {
        self.busy.fetch_add(1, Ordering::Relaxed);
        Reservation {
            busy: self.busy.clone(),
        }
    }
}
pub struct Reservation {
    busy: Arc<AtomicUsize>,
}
impl Drop for Reservation {
    fn drop(&mut self) {
        self.busy.fetch_sub(1, Ordering::Relaxed);
    }
}
#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn busy_starts_at_zero() {
        let c = SlotCounter::new();
        assert_eq!(c.busy(), 0);
    }
    #[test]
    fn reservation_increments_and_drops_decrement() {
        let c = SlotCounter::new();
        let r1 = c.reserve();
        assert_eq!(c.busy(), 1);
        let r2 = c.reserve();
        assert_eq!(c.busy(), 2);
        drop(r2);
        assert_eq!(c.busy(), 1);
        drop(r1);
        assert_eq!(c.busy(), 0);
    }
}