// The CLI lives in `src/bin/cli.rs` so cargo can produce both a
// library (`cargo add supermachine`) and a binary (`cargo install
// supermachine`) from the same package. Anything the CLI shares
// with the library — image bake, asset paths — lives in the lib
// (re-exposed below) so a binary in `src/bin/` can reach it across
// the lib boundary.

// The HVF `run` path (router/worker daemon helpers — run_push, router_*,
// start_router_detached, run_router_foreground, workload_responding, …) is
// macOS-only; on Linux/KVM `run` is served in-process by `run_kvm_dispatch`, so
// those helpers are legitimately unused on this target (the Linux daemon is the
// documented follow-up that will reuse the pattern). Silence dead_code for the
// Linux build rather than cfg-gating ~10 functions individually.
#![cfg_attr(all(target_os = "linux", target_arch = "x86_64"), allow(dead_code))]

// 0.7.42+: mimalloc global allocator (see worker.rs for rationale).
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;

use supermachine::bake;

use std::fs::OpenOptions;
use std::io::{Read, Seek, SeekFrom, Write};
use std::net::TcpStream;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};

struct Args {
    image: Option<String>,
    name: Option<String>,
    connect: bool,
    detach: bool,
    stop: bool,
    http_port: u16,
    guest_port: u16,
    memory_mib: u32,
    /// Number of vCPUs to allocate in the guest. Default 1.
    /// Multi-vCPU re-opened after benching showed single-vCPU
    /// saturation as the c=32+ HTTP-serving bottleneck.
    /// docs/design/concurrency-floor-2026-05-04.md.
    vcpus: u32,
    runtime: String,
    snapshots_dir: PathBuf,
    workers_per_snapshot: u32,
    pull_policy: String,
    do_push: bool,
    /// `docker run --rm` analog: when the daemon stops, also wipe
    /// the snapshot dir and any temp state so the next `run` of
    /// the same image starts from scratch. Honored by `run --stop`.
    rm_on_stop: bool,
    /// Grace window after `--stop`'s SIGTERM before the worker
    /// is force-killed. The TERM is delivered both to the
    /// in-guest workload (via the exec agent's CONTROL frame)
    /// and to the host worker process; we wait this long for the
    /// workload to clean up before falling back to SIGKILL.
    /// Mirrors `docker stop -t SECONDS` (default 10 s).
    stop_grace_ms: u64,
    pid_file: PathBuf,
    log_file: PathBuf,
    ready_timeout_ms: u64,
    probe_path: Option<String>,
    cmd_override: Option<String>,
    push_args: Vec<String>,
    /// Internal: this process IS the detached Linux/KVM serve child (spawned by
    /// `run --detach` via the hidden `__kvm-serve` subcommand). It boots the VM
    /// and serves in the foreground-of-the-daemon until SIGTERM. Never set for a
    /// user-typed command. Read only on the Linux/KVM serve path.
    #[cfg_attr(
        not(all(target_os = "linux", target_arch = "x86_64")),
        allow(dead_code)
    )]
    serve_child: bool,
}

fn home_join(path: &str) -> PathBuf {
    std::env::var_os("HOME")
        .map(PathBuf::from)
        .unwrap_or_else(|| PathBuf::from("."))
        .join(path)
}

fn trace_enabled() -> bool {
    supermachine::trace::enabled("run")
}

fn elapsed_ms(t0: Instant) -> u128 {
    t0.elapsed().as_millis()
}

const VERSION: &str = env!("CARGO_PKG_VERSION");

fn print_version() {
    println!("supermachine {VERSION}");
}

fn usage() {
    eprintln!(
        "supermachine {VERSION} — run any OCI image as a microVM on macOS HVF.\n\
         \n\
         USAGE:\n  \
           supermachine <COMMAND> [OPTIONS]\n\
         \n\
         COMMANDS:\n  \
           run IMAGE            Pull from registry, bake on first use, then run.\n  \
           run oci-layout:PATH  Bake from a local OCI image layout dir.\n  \
           run oci-archive:PATH Bake from a local OCI archive tar.\n  \
           run --connect        Probe the running daemon, print its endpoint.\n  \
           run --stop           Stop the background daemon for --http-port.\n  \
           pull IMAGE           Bake IMAGE to a snapshot, then exit (no router).\n  \
           images               List baked snapshots.\n  \
           rmi NAME             Remove a baked snapshot by name (or image ref).\n  \
           ps                   List running supermachine daemons.\n  \
           logs [TARGET] [-f]   Tail router log; TARGET = port, snapshot name, or image.\n  \
           exec [TARGET] [-t] -- cmd args...\n  \
                                Run a process inside a running guest (docker exec).\n  \
           setup                Re-sign the supermachine-worker binary with the HVF\n  \
                                entitlement (re-run after every `cargo build`).\n  \
           codesign BINARY      Sign BINARY with the bundled HVF entitlement so it\n  \
                                can call hv_vm_create. Use this on your own Rust app\n  \
                                that embeds the supermachine library.\n  \
                                Pass --identity \"Developer ID Application: ...\" for\n  \
                                distribution; default is ad-hoc signing for dev.\n  \
           bundle --into DIR    Copy kernel + init shim into DIR (typically\n  \
                                YourApp.app/Contents/Resources/) so the embedded\n  \
                                supermachine library finds them at runtime.\n  \
           entitlements-path    Print the path to a temp file containing the bundled\n  \
                                entitlements plist (for use with `codesign`).\n  \
           assets-path          Print the directory supermachine looked up for\n  \
                                kernel + init assets (debug helper).\n  \
           --version, -V        Print version and exit.\n  \
           --help, -h           Show this help.\n\
         \n\
         RUN OPTIONS:\n  \
           (no flags)           Foreground, like `docker run`. Ctrl-C stops it.\n  \
           --detach             Background. Like `docker run -d`. CLI returns\n  \
                                once the workload answers on / (max 8s).\n  \
           --http-port PORT     Edge HTTP port to bind (default: 8080).\n  \
           --name NAME          Snapshot name (default: derived from image).\n  \
           --port PORT          Guest service port to bridge to (default: 80).\n  \
           --memory MIB         Guest RAM for the snapshot (default: 256 MiB).\n  \
           --vcpus N            Number of vCPUs (default: 1). N=2 unlocks\n  \
                                ~1.5x sustained concurrent HTTP throughput.\n  \
                                Higher N (up to 16) further scales CPU-bound\n  \
                                workloads; pause-and-capture rendezvous makes\n  \
                                snapshot/restore reliable across all counts.\n  \
           --workers-per-snapshot N    Pool size per snapshot.\n  \
           --snapshots-dir DIR  Where to store baked snapshots.\n  \
           --pull POLICY        missing | always | never (default: missing).\n  \
           --no-push            Skip bake, reuse existing snapshot.\n  \
           --volume HOST:GUEST  Persistent volume. HOST is a name (resolves to\n  \
                                ~/.local/supermachine/volumes/NAME.img) or an\n  \
                                absolute path. GUEST is the mount point inside\n  \
                                the guest. ext4, 1 GiB by default. Repeatable.\n  \
           --entrypoint EP      Replace the image's ENTRYPOINT.\n  \
           --workdir DIR        Set the workload's working directory.\n  \
           --user USER          Run the workload as USER (uid:gid or name).\n  \
           --hostname NAME      Set the guest hostname.\n  \
           --restart POLICY     no | on-failure | always | unless-stopped.\n  \
                                Default: no. Watchdog uses this on worker death.\n  \
           --health-cmd CMD     Probe inside the guest. Empty = no check.\n  \
           --health-interval N  Run health-cmd every N seconds (default 30).\n  \
                                Note: TSI doesn't loop back inside the guest,\n  \
                                so use process checks (`pgrep ...`) or filesystem\n  \
                                checks rather than `curl http://localhost:...`.\n  \
           --rm                 docker-style: wipe the snapshot dir on stop.\n  \
           --stop-grace, -t SEC docker stop -t analog: SIGTERM grace window\n  \
                                before SIGKILL. Default 10s. Forwarded to the\n  \
                                workload via the in-guest agent.\n  \
           --env KEY=VAL        Pass env var to the guest workload (repeatable).\n  \
           --cmd CMD            Override the image's CMD (JSON array string).\n  \
           --enable-egress | --no-egress\n  \
           --egress-policy POLICY    deny_private | allowlist:CIDRS | denylist:...\n  \
           --probe-path PATH    Strict-mode readiness check (require 200 on PATH).\n  \
                                Default is any HTTP response on /; only set this\n  \
                                if you specifically want to gate startup on 200.\n\
         \n\
         EXAMPLES:\n  \
           supermachine setup                           # one-time after cargo build\n  \
           supermachine run nginx:1.27-alpine            # foreground; Ctrl-C to stop\n  \
           supermachine run nginx:1.27-alpine --detach   # background, then:\n  \
           curl http://127.0.0.1:8080/\n  \
           supermachine run --stop                       # tear down --detach daemon"
    );
}

fn parse_args() -> Result<Args, String> {
    let mut it = std::env::args().skip(1);
    let Some(cmd) = it.next() else {
        usage();
        std::process::exit(0);
    };
    if cmd == "--help" || cmd == "-h" || cmd == "help" {
        usage();
        std::process::exit(0);
    }
    if cmd == "--version" || cmd == "-V" || cmd == "version" {
        print_version();
        std::process::exit(0);
    }
    if cmd == "setup" {
        match run_setup() {
            Ok(()) => std::process::exit(0),
            Err(e) => {
                eprintln!("supermachine setup: {e}");
                std::process::exit(1);
            }
        }
    }
    if cmd == "codesign" {
        let argv: Vec<String> = it.collect();
        match run_codesign(&argv) {
            Ok(()) => std::process::exit(0),
            Err(e) => {
                eprintln!("supermachine codesign: {e}");
                std::process::exit(1);
            }
        }
    }
    if cmd == "bundle" {
        let argv: Vec<String> = it.collect();
        match run_bundle(&argv) {
            Ok(()) => std::process::exit(0),
            Err(e) => {
                eprintln!("supermachine bundle: {e}");
                std::process::exit(1);
            }
        }
    }
    if cmd == "entitlements-path" {
        match run_entitlements_path() {
            Ok(()) => std::process::exit(0),
            Err(e) => {
                eprintln!("supermachine entitlements-path: {e}");
                std::process::exit(1);
            }
        }
    }
    if cmd == "assets-path" {
        run_assets_path();
        std::process::exit(0);
    }
    if cmd == "pull" {
        let argv: Vec<String> = it.collect();
        match run_pull(&argv) {
            Ok(()) => std::process::exit(0),
            Err(e) => {
                eprintln!("supermachine pull: {e}");
                std::process::exit(1);
            }
        }
    }
    if cmd == "images" {
        let argv: Vec<String> = it.collect();
        match run_images(&argv) {
            Ok(()) => std::process::exit(0),
            Err(e) => {
                eprintln!("supermachine images: {e}");
                std::process::exit(1);
            }
        }
    }
    if cmd == "rmi" {
        let argv: Vec<String> = it.collect();
        match run_rmi(&argv) {
            Ok(()) => std::process::exit(0),
            Err(e) => {
                eprintln!("supermachine rmi: {e}");
                std::process::exit(1);
            }
        }
    }
    if cmd == "prune" {
        let argv: Vec<String> = it.collect();
        match run_prune(&argv) {
            Ok(()) => std::process::exit(0),
            Err(e) => {
                eprintln!("supermachine prune: {e}");
                std::process::exit(1);
            }
        }
    }
    if cmd == "dedup" {
        let argv: Vec<String> = it.collect();
        match run_dedup(&argv) {
            Ok(()) => std::process::exit(0),
            Err(e) => {
                eprintln!("supermachine dedup: {e}");
                std::process::exit(1);
            }
        }
    }
    if cmd == "ps" {
        let argv: Vec<String> = it.collect();
        match run_ps(&argv) {
            Ok(()) => std::process::exit(0),
            Err(e) => {
                eprintln!("supermachine ps: {e}");
                std::process::exit(1);
            }
        }
    }
    if cmd == "logs" {
        let argv: Vec<String> = it.collect();
        match run_logs(&argv) {
            Ok(()) => std::process::exit(0),
            Err(e) => {
                eprintln!("supermachine logs: {e}");
                std::process::exit(1);
            }
        }
    }
    if cmd == "exec" {
        let argv: Vec<String> = it.collect();
        match run_exec(&argv) {
            Ok(code) => std::process::exit(code),
            Err(e) => {
                eprintln!("supermachine exec: {e}");
                std::process::exit(1);
            }
        }
    }
    // `__kvm-serve` is the hidden Linux/KVM daemon child (spawned by
    // `run --detach`); it takes the same flags as `run`.
    if cmd != "run" && cmd != "__kvm-serve" {
        return Err(format!("unknown command: {cmd}"));
    }
    let serve_child = cmd == "__kvm-serve";

    let mut image = None;
    let mut name = None;
    let mut connect = false;
    let mut detach = false;
    let mut stop = false;
    let mut http_port = std::env::var("SUPERMACHINE_HTTP_PORT")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(8080);
    let mut guest_port = 80;
    let mut memory_mib = 256;
    let mut vcpus: u32 = 1;
    // The --runtime/--backend CLI flag was a transient knob from the
    // Earlier rename transition. There's only one VMM now; the
    // value the rest of the code passes around is "supermachine" for
    // back-compat with the router's snapshot metadata format. It
    // should be `Some(())` morally — kept as a string for the
    // router's metadata.json round-trip until that's cleaned up.
    let mut runtime = "supermachine".to_owned();
    let mut snapshots_dir = std::env::var_os("SUPERMACHINE_SNAPSHOTS")
        .map(PathBuf::from)
        .unwrap_or_else(|| home_join(".local/supermachine-snapshots"));
    let mut workers_per_snapshot = std::env::var("SUPERMACHINE_WORKERS_PER_SNAPSHOT")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(1);
    let mut pull_policy =
        std::env::var("SUPERMACHINE_PULL_POLICY").unwrap_or_else(|_| "missing".to_owned());
    let mut do_push = true;
    let mut rm_on_stop = false;
    let mut stop_grace_ms: u64 = std::env::var("SUPERMACHINE_STOP_GRACE_MS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(10_000);
    let state_dir = std::env::var_os("SUPERMACHINE_STATE_DIR")
        .map(PathBuf::from)
        .unwrap_or_else(|| home_join(".local/supermachine/run"));
    let mut pid_file = None;
    let mut log_file = None;
    let mut ready_timeout_ms = std::env::var("SUPERMACHINE_READY_TIMEOUT_MS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(30000);
    let mut probe_path = None;
    let mut cmd_override = None;
    let mut push_args = Vec::new();

    while let Some(arg) = it.next() {
        match arg.as_str() {
            "--connect" => {
                connect = true;
                do_push = false;
            }
            "--detach" => {
                detach = true;
            }
            "--stop" => {
                stop = true;
                do_push = false;
            }
            "--http-port" => {
                let value = it
                    .next()
                    .ok_or_else(|| "missing --http-port value".to_owned())?;
                http_port = value.parse().map_err(|e| format!("--http-port: {e}"))?;
            }
            "--probe-path" => {
                let value = it
                    .next()
                    .ok_or_else(|| "missing --probe-path value".to_owned())?;
                if !value.starts_with('/') || value.contains('\r') || value.contains('\n') {
                    return Err("--probe-path must start with '/' and not contain CR/LF".to_owned());
                }
                probe_path = Some(value);
            }
            "--name" => {
                name = Some(it.next().ok_or_else(|| "missing --name value".to_owned())?);
            }
            "--port" => {
                let value = it.next().ok_or_else(|| "missing --port value".to_owned())?;
                guest_port = value.parse().map_err(|e| format!("--port: {e}"))?;
            }
            "--memory" => {
                let value = it
                    .next()
                    .ok_or_else(|| "missing --memory value".to_owned())?;
                memory_mib = value.parse().map_err(|e| format!("--memory: {e}"))?;
            }
            "--vcpus" | "--cpus" => {
                let value = it
                    .next()
                    .ok_or_else(|| "missing --vcpus value".to_owned())?;
                let n: u32 = value.parse().map_err(|e| format!("--vcpus: {e}"))?;
                // The original deferral
                // (docs/design/multi-vcpu-snapshot-intermittency-2026-04-27.md)
                // claimed 4+ vCPUs were ~50% reliable on HVF due to
                // the `ICH_LR_EL2` round-trip bug. With the
                // pause-and-capture rendezvous + 50ms quiesce window
                // (see vmm/coord.rs), the captured GIC blob is clean
                // because every secondary lands in WFI before the
                // snapshot fires. 2026-05-04 re-test:
                //   4-vCPU nginx: 25/25 cold-restore
                //   4-vCPU compat battery: 6/6 viable workloads 5/5
                //   4-vCPU under c=128 sustained: 10/10
                //   8-vCPU nginx: 15/15 cold-restore
                // No gate needed; just bound to a sane upper limit.
                if n == 0 {
                    return Err("--vcpus must be >= 1".to_owned());
                }
                if n > 16 {
                    return Err(format!("--vcpus must be <= 16, got {n}"));
                }
                vcpus = n;
            }
            "--runtime" | "--backend" => {
                // Accepted for back-compat (used to be a backend name vs
                // the VMM had alternative backends. They're gone; the value is ignored
                // beyond a friendly error if someone passes the old
                // backend.
                let value = it.next().ok_or_else(|| format!("missing {arg} value"))?;
                if value == "spike14" || value == "spike22" {
                    // legacy backend names
                    return Err(
                        "that --runtime value is from a previous backend split that has since collapsed; supermachine is the only VMM now"
                            .to_owned(),
                    );
                }
                runtime = value;
            }
            "--workers-per-snapshot" => {
                let value = it
                    .next()
                    .ok_or_else(|| "missing --workers-per-snapshot value".to_owned())?;
                workers_per_snapshot = value
                    .parse()
                    .map_err(|e| format!("--workers-per-snapshot: {e}"))?;
            }
            "--snapshots-dir" => {
                snapshots_dir = PathBuf::from(
                    it.next()
                        .ok_or_else(|| "missing --snapshots-dir value".to_owned())?,
                );
            }
            "--pull" => {
                pull_policy = it.next().ok_or_else(|| "missing --pull value".to_owned())?;
            }
            "--pid-file" => {
                pid_file = Some(PathBuf::from(
                    it.next()
                        .ok_or_else(|| "missing --pid-file value".to_owned())?,
                ));
            }
            "--log-file" => {
                log_file = Some(PathBuf::from(
                    it.next()
                        .ok_or_else(|| "missing --log-file value".to_owned())?,
                ));
            }
            "--ready-timeout-ms" => {
                let value = it
                    .next()
                    .ok_or_else(|| "missing --ready-timeout-ms value".to_owned())?;
                ready_timeout_ms = value
                    .parse()
                    .map_err(|e| format!("--ready-timeout-ms: {e}"))?;
            }
            "--cmd" => {
                let value = it.next().ok_or_else(|| "missing --cmd value".to_owned())?;
                cmd_override = Some(value.clone());
                push_args.push(arg);
                push_args.push(value);
            }
            "--volume" | "-v" => {
                // --volume HOST:GUEST — HOST is either a name (resolved
                // to ~/.local/supermachine/volumes/<name>.img) or an
                // absolute path to a host file. The bake pipeline
                // accepts these via --volume; init-oci mounts them
                // post-pivot. Format-on-first-use happens on the host
                // before the worker spawns. See VolumeSpec.
                let value = it
                    .next()
                    .ok_or_else(|| "missing --volume value".to_owned())?;
                push_args.push("--volume".to_owned());
                push_args.push(value);
            }
            "--health-cmd" => {
                let value = it
                    .next()
                    .ok_or_else(|| "missing --health-cmd value".to_owned())?;
                push_args.push("--health-cmd".to_owned());
                push_args.push(value);
            }
            "--health-interval" => {
                let value = it
                    .next()
                    .ok_or_else(|| "missing --health-interval value".to_owned())?;
                let n: u64 = value
                    .parse()
                    .map_err(|e| format!("--health-interval (seconds): {e}"))?;
                if n == 0 || n > 3600 {
                    return Err(format!("--health-interval must be 1..=3600s, got {n}"));
                }
                push_args.push("--health-interval".to_owned());
                push_args.push(n.to_string());
            }
            "--restart" => {
                // docker run --restart=no|on-failure|always.
                // Stored in snapshot metadata; the router daemon's
                // watchdog reads it and decides whether to respawn
                // a dead worker. `unless-stopped` is accepted as
                // an alias for `always` since our daemon goes
                // away on `--stop` anyway.
                let value = it
                    .next()
                    .ok_or_else(|| "missing --restart value".to_owned())?;
                let normalized = match value.as_str() {
                    "no" | "always" | "on-failure" => value.clone(),
                    "unless-stopped" => "always".to_owned(),
                    s => {
                        return Err(format!(
                            "--restart must be no|on-failure|always|unless-stopped, got {s:?}"
                        ))
                    }
                };
                push_args.push("--restart".to_owned());
                push_args.push(normalized);
            }
            "--entrypoint" | "--workdir" | "--user" | "--hostname" => {
                // docker-style runtime overrides. `--workdir` and `--user`
                // override the image's WORKDIR / USER (init-oci already
                // honors `.supermachine-workdir` / `.supermachine-user`
                // in the delta squashfs; bake plumbing reads these
                // flags from extra_args). `--entrypoint` replaces the
                // image's ENTRYPOINT — final argv becomes
                // [entrypoint, ...effective_cmd]. `--hostname` writes
                // `.supermachine-hostname` so init-oci can `sethostname`
                // before exec'ing the workload.
                //
                // No short forms: docker's `-h` clashes with --help and
                // `-w` / `-u` are infrequently used with `docker run`.
                let value = it.next().ok_or_else(|| format!("missing {arg} value"))?;
                push_args.push(arg);
                push_args.push(value);
            }
            "--supermachine-snapshot-after-ms"
            | "--supermachine-listener-settle-ms"
            | "--extra-file"
            | "--env"
            | "--egress-policy"
            | "--inbound-tls-cert"
            | "--inbound-tls-key" => {
                let value = it.next().ok_or_else(|| format!("missing {arg} value"))?;
                push_args.push(arg);
                push_args.push(value);
            }
            "--rm" => {
                rm_on_stop = true;
            }
            "--stop-grace" | "-t" => {
                // Seconds (matches `docker stop -t`); convert to ms.
                let value = it
                    .next()
                    .ok_or_else(|| "missing --stop-grace value".to_owned())?;
                let secs: f64 = value.parse().map_err(|e| format!("--stop-grace: {e}"))?;
                if !(0.0..=600.0).contains(&secs) {
                    return Err(format!("--stop-grace must be 0..=600s, got {secs}"));
                }
                stop_grace_ms = (secs * 1000.0) as u64;
            }
            "--no-push" | "--enable-egress" | "--no-egress" | "--inbound-tls-autogen" => {
                if arg == "--no-push" {
                    do_push = false;
                } else {
                    push_args.push(arg);
                }
            }
            "--help" | "-h" => {
                usage();
                std::process::exit(0);
            }
            s if s.starts_with('-') => return Err(format!("unknown flag: {s}")),
            s => {
                if image.is_some() {
                    return Err(format!("extra arg: {s}"));
                }
                image = Some(s.to_owned());
            }
        }
    }

    if !connect && !stop && image.is_none() {
        return Err("missing image".to_owned());
    }

    if runtime != "supermachine" {
        return Err(format!(
            "unknown runtime: {runtime} (the only supported runtime is supermachine)"
        ));
    }
    if !matches!(pull_policy.as_str(), "missing" | "always" | "never") {
        return Err(format!("unknown --pull policy: {pull_policy}"));
    }
    let pid_file = pid_file.unwrap_or_else(|| state_dir.join(format!("router-{http_port}.pid")));
    let log_file = log_file.unwrap_or_else(|| state_dir.join(format!("router-{http_port}.log")));

    Ok(Args {
        image,
        name,
        connect,
        detach,
        stop,
        http_port,
        guest_port,
        memory_mib,
        vcpus,
        runtime,
        snapshots_dir,
        workers_per_snapshot,
        pull_policy,
        do_push,
        rm_on_stop,
        stop_grace_ms,
        pid_file,
        log_file,
        ready_timeout_ms,
        probe_path,
        cmd_override,
        push_args,
        serve_child,
    })
}

struct HttpResponse {
    status: u16,
    bytes: Vec<u8>,
}

fn http_get(port: u16, path: &str) -> std::io::Result<HttpResponse> {
    let mut stream = TcpStream::connect(("127.0.0.1", port))?;
    stream.set_read_timeout(Some(Duration::from_millis(200)))?;
    stream.set_write_timeout(Some(Duration::from_millis(200)))?;
    write!(
        stream,
        "GET {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n"
    )?;

    let mut buf = Vec::with_capacity(1024);
    let mut tmp = [0u8; 1024];
    loop {
        match stream.read(&mut tmp) {
            Ok(0) => break,
            Ok(n) => {
                buf.extend_from_slice(&tmp[..n]);
            }
            Err(e)
                if e.kind() == std::io::ErrorKind::WouldBlock
                    || e.kind() == std::io::ErrorKind::TimedOut =>
            {
                break
            }
            Err(e) => return Err(e),
        }
    }
    let status = parse_status(&buf).unwrap_or(0);
    Ok(HttpResponse { status, bytes: buf })
}

fn parse_status(buf: &[u8]) -> Option<u16> {
    let first_line_end = buf.windows(2).position(|w| w == b"\r\n")?;
    let first_line = std::str::from_utf8(&buf[..first_line_end]).ok()?;
    let mut parts = first_line.split_ascii_whitespace();
    let version = parts.next()?;
    if !version.starts_with("HTTP/") {
        return None;
    }
    parts.next()?.parse().ok()
}

fn health_ok(port: u16) -> std::io::Result<bool> {
    let response = http_get(port, "/_health")?;
    Ok(response.status == 200
        && response
            .bytes
            .windows(br#""status":"ok""#.len())
            .any(|w| w == br#""status":"ok""#))
}

fn probe_200(port: u16, path: &str) -> std::io::Result<bool> {
    Ok(http_get(port, path)?.status == 200)
}

/// Workload readiness probe. Customer doesn't have to know the
/// content of `/`: any HTTP response (2xx/3xx/4xx) means the worker
/// is wired through and the guest workload is answering. We only
/// keep retrying when:
///   - TCP refused/timed out (router not up yet)
///   - 502 from the router (worker can't reach the guest's listener)
///   - 503/504 (transient unavailable)
/// Returns Ok(true) on workload-responding, Ok(false) on persistent
/// error after `total_timeout_ms`, Err on unrecoverable I/O.
fn workload_responding(port: u16, total_timeout_ms: u64) -> std::io::Result<bool> {
    let deadline = std::time::Instant::now() + std::time::Duration::from_millis(total_timeout_ms);
    let mut last_err: Option<std::io::Error> = None;
    while std::time::Instant::now() < deadline {
        match http_get(port, "/") {
            Ok(resp) => {
                // 502/503/504 mean upstream isn't ready — keep waiting.
                // Any other status (200, 301, 404, ...) is the workload
                // answering.
                if matches!(resp.status, 502..=504) {
                    last_err = Some(std::io::Error::new(
                        std::io::ErrorKind::Other,
                        format!(
                            "upstream {}: {}",
                            resp.status,
                            String::from_utf8_lossy(&resp.bytes)
                                .chars()
                                .take(120)
                                .collect::<String>()
                        ),
                    ));
                } else {
                    return Ok(true);
                }
            }
            Err(e) => last_err = Some(e),
        }
        // Tight poll: cold-boot is on the order of ~100 ms total, so
        // a 50 ms sleep here would add up to half the wall-clock as
        // unnecessary slop. Match `wait_ready`'s 5 ms cadence.
        std::thread::sleep(std::time::Duration::from_millis(5));
    }
    if let Some(e) = last_err {
        Err(e)
    } else {
        Ok(false)
    }
}

/// Resolve a "root" the helpers below can hang absolute paths off.
/// Honoured in this order:
///   1. $SUPERMACHINE_ROOT — user's explicit override
///   2. dev-tree marker: walk binary's ancestors for `tools/supermachine-push`
///   3. installed-tree marker: walk binary's ancestors for `share/supermachine/kernel`
///      and return the parent (so `<root>/share/supermachine/kernel` resolves)
///   4. cwd, as a last-resort fallback
fn repo_root() -> Result<PathBuf, String> {
    if let Some(root) = std::env::var_os("SUPERMACHINE_ROOT") {
        return Ok(PathBuf::from(root));
    }

    let exe = std::env::current_exe().map_err(|e| format!("current_exe: {e}"))?;
    // Dev-tree: cargo build target/release tree under a checkout.
    for ancestor in exe.ancestors() {
        if ancestor.join("tools/supermachine-push").is_file() {
            return Ok(ancestor.to_path_buf());
        }
    }
    // Installed-tree: <prefix>/bin/supermachine + <prefix>/share/supermachine/.
    for ancestor in exe.ancestors() {
        if ancestor.join("share/supermachine/kernel").is_file() {
            return Ok(ancestor.to_path_buf());
        }
    }
    std::env::current_dir().map_err(|e| format!("current_dir: {e}"))
}

/// `supermachine setup` — explicit force-resign of the worker
/// binary + asset extraction sanity check. Most users never call
/// this: the CLI auto-signs on first invocation (see
/// `supermachine::codesign::ensure_worker_signed`). Run this when
/// you want to force the work eagerly (e.g. immediately after
/// `cargo build` strips the entitlement on a dev tree, before
/// shipping a tarball).
fn run_setup() -> Result<(), String> {
    let mut errors = Vec::new();
    let mut ok_steps = Vec::new();

    #[cfg(target_os = "macos")]
    {
        // Force re-sign by invalidating the in-process cache; the
        // sentinel-on-disk check still applies, but with --force
        // codesign re-signs even if previously signed. We just
        // call ensure_worker_signed which is idempotent.
        match supermachine::codesign::locate_worker_bin() {
            Some(bin) => match supermachine::codesign::ensure_worker_signed(&bin) {
                Ok(()) => ok_steps.push(format!("signed {}", bin.display())),
                Err(e) => errors.push(format!("codesign: {e}")),
            },
            None => errors.push(
                "supermachine-worker not found. \
                 If you `cargo install`d, the worker should be a sibling \
                 of supermachine in ~/.cargo/bin/. If you `cargo build`d, \
                 it lives at target/release/supermachine-worker."
                    .to_owned(),
            ),
        }
    }

    let assets = supermachine::AssetPaths::discover();
    match assets.kernel.as_deref() {
        Some(k) => ok_steps.push(format!("kernel ready at {}", k.display())),
        None => errors.push(
            "kernel asset not found. The supermachine-kernel crate ships \
             the kernel bytes inside this binary; first AssetPaths::discover() \
             call extracts them to $XDG_DATA_HOME/supermachine/v{VERSION}/. \
             If extraction failed, check that ~/.local/share/supermachine/ \
             is writable."
                .to_owned(),
        ),
    }
    match assets.init_oci_bin.as_deref() {
        Some(p) => ok_steps.push(format!("init-oci ready at {}", p.display())),
        None => errors
            .push("init-oci asset not found (same root cause as kernel; see above).".to_owned()),
    }

    println!("supermachine setup:");
    for s in &ok_steps {
        println!("  ok: {s}");
    }
    for e in &errors {
        println!("  err: {e}");
    }
    if errors.is_empty() {
        println!("  (ready — try `supermachine run python:3.12-alpine ...`)");
        Ok(())
    } else {
        Err(format!("{} step(s) failed", errors.len()))
    }
}

/// `supermachine codesign BINARY [--identity IDENTITY]` — sign a
/// binary with the bundled HVF entitlements plist, so it can call
/// `hv_vm_create`. Default is ad-hoc signing (`-s -`); pass
/// `--identity "Developer ID Application: ..."` for distribution.
///
/// The bundled plist is dropped to a temp file (it lives inside
/// the supermachine binary as a compile-time `include_str!`), used
/// once for the codesign call, then unlinked.
fn run_codesign(argv: &[String]) -> Result<(), String> {
    let mut binary: Option<&str> = None;
    let mut identity: String = "-".to_owned(); // ad-hoc by default
    let mut i = 0;
    while i < argv.len() {
        match argv[i].as_str() {
            "--identity" | "-s" => {
                i += 1;
                identity = argv
                    .get(i)
                    .ok_or_else(|| "missing --identity value".to_owned())?
                    .clone();
            }
            "--help" | "-h" => {
                eprintln!(
                    "usage: supermachine codesign BINARY [--identity IDENTITY]\n\
                     \n\
                     Default IDENTITY is `-` (ad-hoc).  For a distributable .app,\n\
                     pass `--identity \"Developer ID Application: Name (TEAMID)\"`."
                );
                std::process::exit(0);
            }
            s if s.starts_with('-') => {
                return Err(format!("unknown flag: {s}"));
            }
            s => {
                if binary.is_some() {
                    return Err(format!("extra arg: {s}"));
                }
                binary = Some(s);
            }
        }
        i += 1;
    }
    let binary = binary.ok_or_else(|| "missing BINARY argument".to_owned())?;

    // Drop the bundled plist to a temp file so codesign can read it.
    let plist_dir = std::env::temp_dir();
    let plist_path = plist_dir.join(format!(
        "supermachine-entitlements-{}.plist",
        std::process::id()
    ));
    std::fs::write(&plist_path, supermachine::assets::ENTITLEMENTS_PLIST)
        .map_err(|e| format!("write temp plist {}: {e}", plist_path.display()))?;

    let result = (|| {
        let mut cmd = Command::new("codesign");
        cmd.args(["-s", &identity, "--entitlements"])
            .arg(&plist_path)
            .arg("--force");
        if identity != "-" {
            // For real Developer ID signing, also enable Hardened Runtime
            // (required for notarization).
            cmd.args(["--options", "runtime"]);
        }
        cmd.arg(binary);
        let status = cmd.status().map_err(|e| format!("spawn codesign: {e}"))?;
        if status.success() {
            Ok(())
        } else {
            Err(format!("codesign exit {:?}", status.code()))
        }
    })();
    let _ = std::fs::remove_file(&plist_path);
    result?;
    println!("signed {binary}  (identity: {identity})");
    Ok(())
}

/// `supermachine bundle --into DIR [--image NAME]` — produce a
/// self-contained dir an embedder can ship inside their `.app`.
///
/// Two modes:
///
/// - **Asset-only** (no `--image`): copy kernel + init shim into
///   DIR. Use when you bake at runtime in your embedder. DIR
///   is typically `MyApp.app/Contents/Resources/`.
/// - **Snapshot bundle** (`--image NAME`): copy the kernel,
///   `restore.snap`, `metadata.json` (with paths relativized),
///   `delta.squashfs`, and every layer `.squashfs` referenced by
///   the snapshot. The result is a stand-alone restorable set.
///   Embedder calls
///   `Image::from_snapshot("Contents/Resources/<name>/")` and
///   doesn't need supermachine installed on the user's machine.
fn run_bundle(argv: &[String]) -> Result<(), String> {
    let mut into: Option<PathBuf> = None;
    let mut image: Option<String> = None;
    let mut snapshots_dir = default_snapshots_dir();
    let mut i = 0;
    while i < argv.len() {
        match argv[i].as_str() {
            "--into" => {
                i += 1;
                into = Some(PathBuf::from(
                    argv.get(i)
                        .ok_or_else(|| "missing --into value".to_owned())?,
                ));
            }
            "--image" => {
                i += 1;
                image = Some(
                    argv.get(i)
                        .ok_or_else(|| "missing --image value".to_owned())?
                        .clone(),
                );
            }
            "--snapshots-dir" => {
                i += 1;
                snapshots_dir = PathBuf::from(
                    argv.get(i)
                        .ok_or_else(|| "missing --snapshots-dir value".to_owned())?,
                );
            }
            "--help" | "-h" => {
                eprintln!(
                    "usage: supermachine bundle --into DIR [--image NAME]\n\
                     \x20                          [--snapshots-dir DIR]\n\
                     \n\
                     Without --image: copies the kernel image and init shim into\n\
                     DIR (typically `YourApp.app/Contents/Resources/`).\n\
                     \n\
                     With --image NAME: produces a self-contained snapshot bundle\n\
                     at DIR/NAME/ containing the kernel, restore.snap, metadata.json\n\
                     (with paths relativized), delta.squashfs, and every layer\n\
                     squashfs referenced by the snapshot. The embedder loads it via\n\
                     `Image::from_snapshot(\"...Resources/NAME/\")` — no supermachine\n\
                     install needed on the end user's machine.\n\
                     \n\
                     NAME accepts either a snapshot dir name (as shown by\n\
                     `supermachine images`) or an image ref (`nginx:1.27-alpine`,\n\
                     auto-sanitized)."
                );
                std::process::exit(0);
            }
            s => return Err(format!("unknown arg: {s}")),
        }
        i += 1;
    }
    let into = into.ok_or_else(|| "missing --into DIR".to_owned())?;
    std::fs::create_dir_all(&into).map_err(|e| format!("mkdir {}: {e}", into.display()))?;

    if let Some(image) = image {
        return bundle_snapshot(&into, &image, &snapshots_dir);
    }

    let assets = supermachine::assets::AssetPaths::discover();
    let mut copied = 0;

    if let Some(kernel) = &assets.kernel {
        let dst = into.join("kernel");
        std::fs::copy(kernel, &dst)
            .map_err(|e| format!("copy kernel {} → {}: {e}", kernel.display(), dst.display()))?;
        println!("  kernel:    {} → {}", kernel.display(), dst.display());
        copied += 1;
    } else {
        return Err(
            "no kernel found in any standard location — set $SUPERMACHINE_ASSETS_DIR or rebuild from source".to_owned(),
        );
    }

    if let Some(init_bin) = &assets.init_oci_bin {
        let dst = into.join("init-oci");
        std::fs::copy(init_bin, &dst).map_err(|e| {
            format!(
                "copy init-oci {} → {}: {e}",
                init_bin.display(),
                dst.display()
            )
        })?;
        // preserve executable bit
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let mut perms = std::fs::metadata(&dst)
                .map_err(|e| format!("stat {}: {e}", dst.display()))?
                .permissions();
            perms.set_mode(0o755);
            std::fs::set_permissions(&dst, perms)
                .map_err(|e| format!("chmod {}: {e}", dst.display()))?;
        }
        println!("  init-oci:  {} → {}", init_bin.display(), dst.display());
        copied += 1;
    }

    if let Some(init_src) = &assets.init_oci_src {
        let dst = into.join("init-oci.c");
        std::fs::copy(init_src, &dst).map_err(|e| {
            format!(
                "copy init-oci.c {} → {}: {e}",
                init_src.display(),
                dst.display()
            )
        })?;
        println!("  init-oci.c:{} → {}", init_src.display(), dst.display());
        copied += 1;
    }

    println!("bundled {copied} asset(s) into {}", into.display());
    Ok(())
}

/// Snapshot-bundle layout (under `<into>/<name>/`):
///
/// ```text
/// kernel                       # bundled kernel image
/// restore.snap                 # snapshot file (CoW-mapped at restore)
/// metadata.json                # paths rewritten to ./layers/... etc.
/// delta.squashfs               # writable overlay layer (if any)
/// layers/<sha>.squashfs ...    # all layers referenced by metadata
/// ```
///
/// The result is `cp -r`-able to any Mac with HVF + entitlement. No
/// other files on the host are touched at restore time.
fn bundle_snapshot(into: &Path, image: &str, snapshots_dir: &Path) -> Result<(), String> {
    // Resolve `image` to a snapshot dir under `snapshots_dir`. Accept
    // either a literal dir name or an image ref.
    let candidate = snapshots_dir.join(image);
    let snap_dir = if candidate.is_dir() {
        candidate
    } else {
        let derived = snapshots_dir.join(supermachine::bake::snapshot_name_for_image(image));
        if !derived.is_dir() {
            return Err(format!(
                "no snapshot for {image} (looked in {} and {})",
                candidate.display(),
                derived.display(),
            ));
        }
        derived
    };
    let name = snap_dir
        .file_name()
        .and_then(|s| s.to_str())
        .ok_or_else(|| format!("snapshot dir has no name: {}", snap_dir.display()))?
        .to_owned();
    let metadata_src = snap_dir.join("metadata.json");
    if !metadata_src.is_file() {
        return Err(format!("missing {}", metadata_src.display()));
    }
    let restore_src = snap_dir.join("restore.snap");
    if !restore_src.is_file() {
        return Err(format!("missing {}", restore_src.display()));
    }

    let bundle_dir = into.join(&name);
    std::fs::create_dir_all(&bundle_dir)
        .map_err(|e| format!("mkdir {}: {e}", bundle_dir.display()))?;
    let layers_dir = bundle_dir.join("layers");
    std::fs::create_dir_all(&layers_dir)
        .map_err(|e| format!("mkdir {}: {e}", layers_dir.display()))?;

    let mut total_bytes: u64 = 0;

    // Kernel — from the host's discovered AssetPaths, NOT from the
    // snapshot's metadata.json (which points at the bake-time
    // location, often a per-developer cargo target dir).
    let assets = supermachine::assets::AssetPaths::discover();
    let kernel = assets.kernel.as_ref().ok_or_else(|| {
        "no kernel found in any standard location — set $SUPERMACHINE_ASSETS_DIR or rebuild from source"
            .to_owned()
    })?;
    let kernel_dst = bundle_dir.join("kernel");
    let kernel_bytes = std::fs::copy(kernel, &kernel_dst).map_err(|e| {
        format!(
            "copy kernel {} → {}: {e}",
            kernel.display(),
            kernel_dst.display()
        )
    })?;
    total_bytes += kernel_bytes;
    println!(
        "  kernel:        {} ({})",
        kernel_dst.display(),
        human_bytes(kernel_bytes)
    );

    // Snapshot file.
    let restore_dst = bundle_dir.join("restore.snap");
    let restore_bytes =
        std::fs::copy(&restore_src, &restore_dst).map_err(|e| format!("copy restore.snap: {e}"))?;
    total_bytes += restore_bytes;
    println!(
        "  restore.snap:  {} ({})",
        restore_dst.display(),
        human_bytes(restore_bytes)
    );

    // Read source metadata, rewrite paths to relative.
    let meta_text = std::fs::read_to_string(&metadata_src)
        .map_err(|e| format!("read {}: {e}", metadata_src.display()))?;
    let mut meta: serde_json::Value = serde_json::from_str(&meta_text)
        .map_err(|e| format!("parse {}: {e}", metadata_src.display()))?;

    // Rewrite metadata.kernel to match the bundled layout. The
    // runtime restore path picks up the kernel via
    // `<bundle_dir>/kernel` discovery in `Image::from_snapshot`,
    // not via this field — but tooling reading metadata.json
    // shouldn't be misled by a stale absolute path.
    if let Some(obj) = meta.as_object_mut() {
        obj.insert(
            "kernel".to_owned(),
            serde_json::Value::String("kernel".to_owned()),
        );
    }

    // Layers: copy each into <bundle>/layers/<basename>, rewrite
    // metadata.layers to a relative path.
    if let Some(layers) = meta.get("layers").cloned().and_then(|v| match v {
        serde_json::Value::Array(a) => Some(a),
        _ => None,
    }) {
        let mut new_layers: Vec<serde_json::Value> = Vec::with_capacity(layers.len());
        for layer in &layers {
            let src = layer
                .as_str()
                .ok_or_else(|| "layer path is not a string".to_owned())?;
            let src_path = PathBuf::from(src);
            if !src_path.is_file() {
                return Err(format!(
                    "layer file missing: {} (referenced by {})",
                    src_path.display(),
                    metadata_src.display()
                ));
            }
            let basename = src_path
                .file_name()
                .ok_or_else(|| format!("layer path has no filename: {}", src_path.display()))?;
            let dst = layers_dir.join(basename);
            let bytes = std::fs::copy(&src_path, &dst).map_err(|e| {
                format!("copy layer {} → {}: {e}", src_path.display(), dst.display())
            })?;
            total_bytes += bytes;
            // Relative form: "layers/<basename>".
            new_layers.push(serde_json::Value::String(format!(
                "layers/{}",
                basename.to_string_lossy()
            )));
        }
        if let Some(obj) = meta.as_object_mut() {
            obj.insert("layers".to_owned(), serde_json::Value::Array(new_layers));
        }
        println!(
            "  layers:        {} files ({})",
            layers.len(),
            human_bytes(total_bytes - kernel_bytes - restore_bytes)
        );
    }

    // Delta squashfs (writable overlay) — optional.
    if let Some(delta) = meta
        .get("delta_squashfs")
        .and_then(|v| v.as_str())
        .map(PathBuf::from)
    {
        if delta.is_file() {
            let dst = bundle_dir.join("delta.squashfs");
            let bytes =
                std::fs::copy(&delta, &dst).map_err(|e| format!("copy delta.squashfs: {e}"))?;
            total_bytes += bytes;
            if let Some(obj) = meta.as_object_mut() {
                obj.insert(
                    "delta_squashfs".to_owned(),
                    serde_json::Value::String("delta.squashfs".to_owned()),
                );
            }
            println!(
                "  delta.squashfs:{} ({})",
                dst.display(),
                human_bytes(bytes)
            );
        }
    }

    // init.cpio.gz — boot-time initramfs. Some snapshots reference
    // it; copy if present so an embedder running fresh-bake on the
    // same workload wouldn't be surprised by a missing file.
    let init_cpio = snap_dir.join("init.cpio.gz");
    if init_cpio.is_file() {
        let dst = bundle_dir.join("init.cpio.gz");
        let bytes =
            std::fs::copy(&init_cpio, &dst).map_err(|e| format!("copy init.cpio.gz: {e}"))?;
        total_bytes += bytes;
        if let Some(obj) = meta.as_object_mut() {
            obj.insert(
                "init_cpio".to_owned(),
                serde_json::Value::String("init.cpio.gz".to_owned()),
            );
        }
    }

    // Drop the rewritten metadata into the bundle. Keep the original
    // pretty-printing style — it's user-readable.
    let metadata_dst = bundle_dir.join("metadata.json");
    let pretty =
        serde_json::to_string_pretty(&meta).map_err(|e| format!("serialize metadata: {e}"))?;
    std::fs::write(&metadata_dst, pretty)
        .map_err(|e| format!("write {}: {e}", metadata_dst.display()))?;
    println!("  metadata.json: {}", metadata_dst.display());

    println!(
        "bundled {} ({}) into {}",
        name,
        human_bytes(total_bytes),
        bundle_dir.display()
    );
    println!(
        "  load it from Rust: Image::from_snapshot(\"{}\")",
        bundle_dir.display()
    );
    Ok(())
}

/// `supermachine entitlements-path` — write the bundled plist to a
/// temp file and print the path. Useful for CI scripts that want
/// to call `codesign --entitlements $(supermachine entitlements-path)`
/// directly without going through `supermachine codesign`.
///
/// Note: the file is left on disk. The caller is responsible for
/// cleanup if they care; otherwise the OS reaps `/tmp` on its own
/// schedule. Path includes the PID so concurrent invocations don't
/// collide.
fn run_entitlements_path() -> Result<(), String> {
    let path = std::env::temp_dir().join(format!(
        "supermachine-entitlements-{}.plist",
        std::process::id()
    ));
    std::fs::write(&path, supermachine::assets::ENTITLEMENTS_PLIST)
        .map_err(|e| format!("write temp plist {}: {e}", path.display()))?;
    println!("{}", path.display());
    Ok(())
}

/// `supermachine assets-path` — print where supermachine looks for
/// kernel + init shim. Debug helper for users who can't figure out
/// why their VM isn't starting.
fn run_assets_path() {
    let assets = supermachine::assets::AssetPaths::discover();
    println!(
        "kernel:       {}",
        assets
            .kernel
            .as_ref()
            .map(|p| p.display().to_string())
            .unwrap_or_else(|| "(not found)".to_owned())
    );
    println!(
        "init-oci bin: {}",
        assets
            .init_oci_bin
            .as_ref()
            .map(|p| p.display().to_string())
            .unwrap_or_else(|| "(not found)".to_owned())
    );
    println!(
        "init-oci src: {}",
        assets
            .init_oci_src
            .as_ref()
            .map(|p| p.display().to_string())
            .unwrap_or_else(|| "(not found)".to_owned())
    );
    if let Ok(d) = std::env::var("SUPERMACHINE_ASSETS_DIR") {
        println!("(SUPERMACHINE_ASSETS_DIR override: {d})");
    }
}

fn router_bin(root: &Path) -> Result<PathBuf, String> {
    // Tried in order:
    //   - sibling of the running binary (`cargo install` layout —
    //     ~/.cargo/bin/supermachine-router next to ~/.cargo/bin/supermachine)
    //   - dev tree:        <root>/target/release/supermachine-router
    //   - tarball install: <root>/bin/supermachine-router
    if let Ok(exe) = std::env::current_exe() {
        if let Some(dir) = exe.parent() {
            let sibling = dir.join("supermachine-router");
            if sibling.is_file() {
                return Ok(sibling);
            }
        }
    }
    for candidate in [
        "target/release/supermachine-router",
        "bin/supermachine-router",
    ] {
        let p = root.join(candidate);
        if p.is_file() {
            return Ok(p);
        }
    }
    Err(format!(
        "missing router binary at {} (run `cargo build --release`)",
        root.join("target/release/supermachine-router").display()
    ))
}

fn run_push(args: &Args, run_t0: Instant) -> Result<(), String> {
    if !args.do_push {
        if trace_enabled() {
            eprintln!("supermachine: push skipped after {}ms", elapsed_ms(run_t0));
        }
        return Ok(());
    }
    let root = repo_root()?;
    let image = args
        .image
        .clone()
        .ok_or_else(|| "missing image".to_owned())?;
    let request = bake::BakeRequest {
        image,
        name: args.name.clone(),
        runtime: args.runtime.clone(),
        guest_port: args.guest_port,
        memory_mib: args.memory_mib,
        vcpus: args.vcpus,
        pull_policy: args.pull_policy.clone(),
        snapshots_dir: args.snapshots_dir.clone(),
        cmd_override: args.cmd_override.clone(),
        extra_args: args.push_args.clone(),
        // CLI doesn't expose --platform yet; default to arm64 matching
        // pre-amd64 behavior. The amd64 path is currently SDK-only via
        // `OciImageBuilder::with_platform("linux/amd64")`.
        platform: "linux/arm64".to_owned(),
    };
    bake::run_push(&request, run_t0, &root)
}

fn process_alive(pid: u32) -> bool {
    Command::new("kill")
        .arg("-0")
        .arg(pid.to_string())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status()
        .map(|status| status.success())
        .unwrap_or(false)
}

fn signal_process(pid: u32, signal: &str) {
    let _ = Command::new("kill")
        .arg(format!("-{signal}"))
        .arg(pid.to_string())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status();
}

fn router_sidecar_path(pid_file: &Path, suffix: &str) -> PathBuf {
    // pid_file is `<state>/router-<port>.pid` — replace the
    // extension to land sidecar files next to it.
    let mut p = pid_file.to_path_buf();
    p.set_extension(suffix);
    p
}

fn stop_router(args: &Args) -> Result<(), String> {
    let pid_text = match std::fs::read_to_string(&args.pid_file) {
        Ok(s) => s,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            // Even with no live daemon, honor --rm so the user can
            // clean up snapshots from previous foreground runs.
            if args.rm_on_stop {
                rm_snapshot_for_args(args)?;
            }
            return Ok(());
        }
        Err(e) => return Err(format!("read pid file {}: {e}", args.pid_file.display())),
    };
    let pid: u32 = match pid_text.trim().parse() {
        Ok(pid) => pid,
        Err(e) => {
            let _ = std::fs::remove_file(&args.pid_file);
            return Err(format!("invalid pid file {}: {e}", args.pid_file.display()));
        }
    };
    if process_alive(pid) {
        // docker-stop semantics: send SIGTERM to the workload via
        // the in-guest agent first so the workload's own cleanup
        // hooks run (db flush, AOF rewrite, etc). Then signal the
        // router/worker. Wait up to args.stop_grace_ms for a clean
        // exit, fall back to SIGKILL.
        //
        // Best-effort: if the agent isn't reachable (older snapshot
        // baked before agent shipped, or workload not yet spawned),
        // we still proceed to host-level TERM. The volume data
        // path is still safe via virtio-blk MS_SYNC on FLUSH.
        let snapshot_sidecar_early = router_sidecar_path(&args.pid_file, "snapshot");
        if let Ok(snap_name) = std::fs::read_to_string(&snapshot_sidecar_early) {
            let snap_name = snap_name.trim();
            if !snap_name.is_empty() {
                let exec_sock = PathBuf::from(format!(
                    "/tmp/supermachine-router-socks/{snap_name}-w0.sock-exec"
                ));
                if exec_sock.exists() {
                    let body = serde_json::json!({
                        "action": "signal",
                        "signum": 15,  // SIGTERM
                    });
                    if let Err(e) = supermachine::exec::send_control(&exec_sock, &body) {
                        if supermachine::trace::enabled("run") {
                            eprintln!("supermachine: workload SIGTERM via agent: {e}");
                        }
                    }
                }
            }
        }
        signal_process(pid, "TERM");
        let grace_iters = (args.stop_grace_ms / 50).max(1);
        for _ in 0..grace_iters {
            if !process_alive(pid) {
                break;
            }
            std::thread::sleep(Duration::from_millis(50));
        }
        if process_alive(pid) {
            signal_process(pid, "KILL");
        }
    }
    // Read sidecar before deleting so we know which snapshot to wipe
    // and whether --rm was sticky from the original `run --detach`.
    let snapshot_sidecar = router_sidecar_path(&args.pid_file, "snapshot");
    let rm_sidecar = router_sidecar_path(&args.pid_file, "rm");
    let sticky_rm = rm_sidecar.exists();
    let snapshot_name = std::fs::read_to_string(&snapshot_sidecar)
        .ok()
        .map(|s| s.trim().to_owned())
        .filter(|s| !s.is_empty());

    let _ = std::fs::remove_file(&args.pid_file);
    let _ = std::fs::remove_file(&snapshot_sidecar);
    let _ = std::fs::remove_file(&rm_sidecar);

    if args.rm_on_stop || sticky_rm {
        // Prefer the sidecar's name (recorded at start time) so a
        // bare `supermachine run --stop --rm` still wipes the right
        // snapshot without having to repeat the image ref.
        let name = snapshot_name
            .or_else(|| derive_default_snapshot_name(args))
            .ok_or_else(|| {
                "--rm on stop: cannot determine which snapshot to remove (no sidecar, no image, no --name)"
                    .to_owned()
            })?;
        rm_snapshot(&args.snapshots_dir, &name)?;
    }
    Ok(())
}

/// Wipe the snapshot dir at `<snapshots_dir>/<name>`. Used by `rmi`
/// and by `--rm` cleanup. Cached layer blobs in
/// `~/.local/supermachine-layer-cache/` are intentionally left
/// behind: they're shared across snapshots and re-downloading
/// layers is the expensive bit. `supermachine rmi` only removes
/// the per-image bake artifacts.
fn rm_snapshot(snapshots_dir: &Path, name: &str) -> Result<(), String> {
    if name.is_empty() || name.contains('/') || name.contains('\\') {
        return Err(format!("invalid snapshot name: {name:?}"));
    }
    let dir = snapshots_dir.join(name);
    match std::fs::remove_dir_all(&dir) {
        Ok(()) => Ok(()),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            Err(format!("no such snapshot: {name}"))
        }
        Err(e) => Err(format!("remove {}: {e}", dir.display())),
    }
}

fn rm_snapshot_for_args(args: &Args) -> Result<(), String> {
    let name = derive_default_snapshot_name(args).ok_or_else(|| {
        "--rm: cannot determine which snapshot to remove (need image or --name)".to_owned()
    })?;
    // Treat "snapshot already gone" as success in the cleanup path —
    // the user's intent is "make sure it's gone."
    match rm_snapshot(&args.snapshots_dir, &name) {
        Ok(()) | Err(_) => Ok(()),
    }
}

fn wait_ready(port: u16, timeout_ms: u64) -> bool {
    let deadline = Instant::now() + Duration::from_millis(timeout_ms);
    loop {
        if health_ok(port).unwrap_or(false) {
            return true;
        }
        if Instant::now() >= deadline {
            return false;
        }
        std::thread::sleep(Duration::from_millis(5));
    }
}

fn router_command(args: &Args) -> Result<Command, String> {
    let root = repo_root()?;
    let router = router_bin(&root)?;
    let mut cmd = Command::new(router);
    cmd.arg("--snapshots-dir")
        .arg(&args.snapshots_dir)
        .arg("--http-port")
        .arg(args.http_port.to_string())
        .arg("--workers-per-snapshot")
        .arg(args.workers_per_snapshot.to_string())
        .env("SUPERMACHINE_SNAPSHOTS", &args.snapshots_dir)
        .env(
            "SUPERMACHINE_WORKERS_PER_SNAPSHOT",
            args.workers_per_snapshot.to_string(),
        );
    // Tell the router which snapshot is the requested one, so root-
    // path requests reach it even if other snapshots are present in
    // the snapshots-dir from prior runs. Either --name (explicit) or
    // the derived name from the image ref.
    if let Some(name) = derive_default_snapshot_name(args) {
        cmd.arg("--default-snapshot").arg(&name);
        cmd.env("SUPERMACHINE_DEFAULT_SNAPSHOT", &name);
    }
    Ok(cmd)
}

fn derive_default_snapshot_name(args: &Args) -> Option<String> {
    if let Some(n) = args.name.as_ref() {
        return Some(n.clone());
    }
    args.image
        .as_ref()
        .map(|img| bake::snapshot_name_for_image(img))
}

fn start_router_detached(args: &Args) -> Result<(), String> {
    if health_ok(args.http_port).unwrap_or(false) {
        return Ok(());
    }
    if let Some(parent) = args.pid_file.parent() {
        std::fs::create_dir_all(parent)
            .map_err(|e| format!("create pid dir {}: {e}", parent.display()))?;
    }
    if let Some(parent) = args.log_file.parent() {
        std::fs::create_dir_all(parent)
            .map_err(|e| format!("create log dir {}: {e}", parent.display()))?;
    }
    std::fs::create_dir_all(&args.snapshots_dir)
        .map_err(|e| format!("create snapshots dir {}: {e}", args.snapshots_dir.display()))?;
    let _ = std::fs::remove_file(&args.pid_file);
    let log = OpenOptions::new()
        .create(true)
        .append(true)
        .open(&args.log_file)
        .map_err(|e| format!("open log file {}: {e}", args.log_file.display()))?;
    let log_err = log
        .try_clone()
        .map_err(|e| format!("clone log file {}: {e}", args.log_file.display()))?;
    let mut child = router_command(args)?
        .stdin(Stdio::null())
        .stdout(Stdio::from(log))
        .stderr(Stdio::from(log_err))
        .spawn()
        .map_err(|e| format!("spawn router: {e}"))?;
    let pid = child.id();
    std::fs::write(&args.pid_file, format!("{pid}\n"))
        .map_err(|e| format!("write pid file {}: {e}", args.pid_file.display()))?;
    // Sidecar: which snapshot is this router serving? `ps`/`logs`
    // read this; `--stop --rm` reads it to know what to wipe.
    if let Some(name) = derive_default_snapshot_name(args) {
        let snapshot_sidecar = router_sidecar_path(&args.pid_file, "snapshot");
        let _ = std::fs::write(&snapshot_sidecar, format!("{name}\n"));
    }
    // Sidecar: --rm sticks across detach → stop, so the customer can
    // do `supermachine run --rm IMAGE --detach` and a later bare
    // `supermachine run --stop` still wipes the snapshot.
    if args.rm_on_stop {
        let rm_sidecar = router_sidecar_path(&args.pid_file, "rm");
        let _ = std::fs::write(&rm_sidecar, "");
    }
    if wait_ready(args.http_port, args.ready_timeout_ms) {
        return Ok(());
    }
    let _ = child.kill();
    let _ = child.wait();
    let _ = std::fs::remove_file(&args.pid_file);
    let _ = std::fs::remove_file(router_sidecar_path(&args.pid_file, "snapshot"));
    let _ = std::fs::remove_file(router_sidecar_path(&args.pid_file, "rm"));
    Err(format!(
        "router did not become ready; see {}",
        args.log_file.display()
    ))
}

fn run_router_foreground(args: &Args) -> Result<i32, String> {
    std::fs::create_dir_all(&args.snapshots_dir)
        .map_err(|e| format!("create snapshots dir {}: {e}", args.snapshots_dir.display()))?;
    let status = router_command(args)?
        .status()
        .map_err(|e| format!("run router: {e}"))?;
    Ok(status.code().unwrap_or(1))
}

fn report_probe_result(
    image: Option<&str>,
    http_port: u16,
    probe_path: Option<&str>,
    result: std::io::Result<bool>,
) -> std::process::ExitCode {
    match result {
        Ok(true) => {
            println!("endpoint: http://127.0.0.1:{http_port}/");
            std::process::ExitCode::SUCCESS
        }
        Ok(false) => {
            let image_hint = image.map(|s| format!(" for {s}")).unwrap_or_default();
            if let Some(path) = probe_path {
                eprintln!(
                    "probe {path}{image_hint} did not return HTTP 200 at http://127.0.0.1:{http_port}"
                );
            } else {
                eprintln!(
                    "no healthy supermachine daemon{image_hint} at http://127.0.0.1:{http_port}/"
                );
            }
            std::process::ExitCode::from(1)
        }
        Err(e) => {
            if let Some(path) = probe_path {
                eprintln!("probe {path} failed at http://127.0.0.1:{http_port}: {e}");
            } else {
                eprintln!("no healthy supermachine daemon at http://127.0.0.1:{http_port}/: {e}");
            }
            std::process::ExitCode::from(1)
        }
    }
}

// ---------- lifecycle commands ----------
//
// These shadow the docker UX: `pull / images / rmi / ps / logs`. The
// data is already on disk — a snapshot is one dir under
// `--snapshots-dir` with a `metadata.json`, a router daemon is one
// `<state>/router-<port>.pid` file with sibling `.snapshot` and
// optional `.rm` sidecars (written by `start_router_detached`).
// These commands just present that shape in the way customers
// reach for instinctively.

fn default_snapshots_dir() -> PathBuf {
    std::env::var_os("SUPERMACHINE_SNAPSHOTS")
        .map(PathBuf::from)
        .unwrap_or_else(|| home_join(".local/supermachine-snapshots"))
}

fn default_state_dir() -> PathBuf {
    std::env::var_os("SUPERMACHINE_STATE_DIR")
        .map(PathBuf::from)
        .unwrap_or_else(|| home_join(".local/supermachine/run"))
}

/// `supermachine pull IMAGE [--name NAME] [--memory MIB] [--port P]
/// [--pull POLICY] [--snapshots-dir DIR]` — bake an image to a
/// snapshot, then exit. Equivalent to `supermachine run IMAGE` minus
/// the router start. Useful for warming caches in CI or shipping a
/// pre-baked snapshot to another host.
/// Linux/KVM: bake the bootable snapshot for `image` into `snapshots_dir/name`
/// (in-process, registry-direct, docker-free) unless a loadable one already
/// exists. Returns the snapshot dir. Shared by `pull` and foreground `run`.
/// Pick the backend for a bake/run on linux/x86_64: explicit `SUPERMACHINE_BACKEND`
/// wins (`sentry`|`kvm`), else KVM iff hardware virt is usable — the no-virt host
/// falls through to sentry. Mirrors the library's selection (api.rs `from_oci_to_dir`
/// / `OciImageBuilder::build`) so the CLI agrees with the lib.
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn use_sentry_backend() -> bool {
    match std::env::var("SUPERMACHINE_BACKEND").ok().as_deref() {
        Some("sentry") => true,
        Some("kvm") => false,
        _ => !supermachine::sentry::kvm_usable(),
    }
}

/// Bake the bootable snapshot for `image` into `snapshots_dir/name` (in-process,
/// registry-direct, docker-free) unless a loadable one for the SELECTED backend
/// already exists. Routes to the KVM or sentry bake per [`use_sentry_backend`].
/// Returns the snapshot dir. Shared by `pull` and foreground/detached `run`.
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn ensure_snapshot(
    image: &str,
    name: Option<&str>,
    snapshots_dir: &Path,
    pull_policy: &str,
) -> Result<PathBuf, String> {
    let snap_name = name
        .map(ToOwned::to_owned)
        .unwrap_or_else(|| bake::snapshot_name_for_image(image));
    let dest = snapshots_dir.join(&snap_name);
    let want_sentry = use_sentry_backend();
    // Cache hit only if a loadable snapshot exists AND it was baked for the SAME
    // backend we're asked for (else a KVM snapshot under this name would be served
    // for a sentry request, or vice-versa). A mismatch falls through to a re-bake.
    if pull_policy != "always" && dest.join("metadata.json").is_file() {
        if let Ok(img) = supermachine::Image::from_snapshot(&dest) {
            if img.is_sentry() == want_sentry {
                return Ok(dest);
            }
        }
    }
    if want_sentry {
        supermachine::Image::bake_sentry_from_ref(image, &dest)
            .map_err(|e| format!("sentry bake {image}: {e}"))?;
    } else {
        supermachine::Image::bake_kvm_auto(image, &dest)
            .map_err(|e| format!("kvm bake {image}: {e}"))?;
    }
    Ok(dest)
}

/// Linux/KVM foreground `run`: cold-boot the image in-process, expose its first
/// TSI listener on `--http-port`, print the endpoint, and block until Ctrl-C
/// (then tear the VM down). The `docker run` experience without the HVF
/// router/worker daemon.
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn run_kvm_foreground(args: &Args, run_t0: Instant) -> Result<i32, String> {
    use std::sync::atomic::{AtomicBool, Ordering};

    let image = args.image.as_deref().ok_or("missing IMAGE")?;
    let dest = ensure_snapshot(
        image,
        args.name.as_deref(),
        &args.snapshots_dir,
        &args.pull_policy,
    )?;
    let img =
        supermachine::Image::from_snapshot(&dest).map_err(|e| format!("load snapshot: {e}"))?;
    let cfg = supermachine::VmConfig::new()
        .with_memory_mib(args.memory_mib)
        .with_vcpus(args.vcpus);
    let vm = img.start(&cfg).map_err(|e| format!("start: {e}"))?;
    let fwd = vm
        .expose_tcp(args.http_port, args.guest_port)
        .map_err(|e| format!("expose_tcp: {e}"))?;
    let addr = fwd.local_addr();
    println!(
        "endpoint: http://{addr}/  (image={image}, guest :{}, ready in {}ms; Ctrl-C to stop)",
        args.guest_port,
        elapsed_ms(run_t0)
    );

    // Block until SIGINT/SIGTERM, then drop the forwarder + VM for a clean
    // teardown (vs. letting the default signal handler kill us mid-flight).
    static SHUTDOWN: AtomicBool = AtomicBool::new(false);
    extern "C" fn on_signal(_: libc::c_int) {
        SHUTDOWN.store(true, Ordering::SeqCst);
    }
    unsafe {
        libc::signal(libc::SIGINT, on_signal as usize);
        libc::signal(libc::SIGTERM, on_signal as usize);
    }
    while !SHUTDOWN.load(Ordering::SeqCst) {
        std::thread::sleep(Duration::from_millis(200));
    }
    println!("stopping {image}…");
    drop(fwd);
    drop(vm);
    Ok(0)
}

fn run_pull(argv: &[String]) -> Result<(), String> {
    let mut image: Option<String> = None;
    let mut name: Option<String> = None;
    let mut memory_mib: u32 = 256;
    let mut vcpus: u32 = 1;
    let mut guest_port: u16 = 80;
    let mut pull_policy =
        std::env::var("SUPERMACHINE_PULL_POLICY").unwrap_or_else(|_| "missing".to_owned());
    let mut snapshots_dir = default_snapshots_dir();
    let mut cmd_override: Option<String> = None;
    let mut extra_args: Vec<String> = Vec::new();
    let mut i = 0;
    while i < argv.len() {
        let arg = argv[i].as_str();
        match arg {
            "--name" => {
                i += 1;
                name = Some(
                    argv.get(i)
                        .ok_or_else(|| "missing --name value".to_owned())?
                        .clone(),
                );
            }
            "--memory" => {
                i += 1;
                memory_mib = argv
                    .get(i)
                    .ok_or_else(|| "missing --memory value".to_owned())?
                    .parse()
                    .map_err(|e| format!("--memory: {e}"))?;
            }
            "--vcpus" | "--cpus" => {
                i += 1;
                let n: u32 = argv
                    .get(i)
                    .ok_or_else(|| "missing --vcpus value".to_owned())?
                    .parse()
                    .map_err(|e| format!("--vcpus: {e}"))?;
                // See parse_args() for history. Pause-and-capture
                // rendezvous makes restore reliable at any N up to 16.
                if n == 0 {
                    return Err("--vcpus must be >= 1".to_owned());
                }
                if n > 16 {
                    return Err(format!("--vcpus must be <= 16, got {n}"));
                }
                vcpus = n;
            }
            "--port" => {
                i += 1;
                guest_port = argv
                    .get(i)
                    .ok_or_else(|| "missing --port value".to_owned())?
                    .parse()
                    .map_err(|e| format!("--port: {e}"))?;
            }
            "--pull" => {
                i += 1;
                pull_policy = argv
                    .get(i)
                    .ok_or_else(|| "missing --pull value".to_owned())?
                    .clone();
            }
            "--snapshots-dir" => {
                i += 1;
                snapshots_dir = PathBuf::from(
                    argv.get(i)
                        .ok_or_else(|| "missing --snapshots-dir value".to_owned())?,
                );
            }
            "--cmd" => {
                i += 1;
                let v = argv
                    .get(i)
                    .ok_or_else(|| "missing --cmd value".to_owned())?
                    .clone();
                cmd_override = Some(v.clone());
                extra_args.push("--cmd".to_owned());
                extra_args.push(v);
            }
            "--env"
            | "--egress-policy"
            | "--volume"
            | "-v"
            | "--entrypoint"
            | "--workdir"
            | "--user"
            | "--hostname"
            | "--restart"
            | "--health-cmd"
            | "--health-interval"
            | "--supermachine-snapshot-after-ms"
            | "--supermachine-listener-settle-ms" => {
                // Normalize -v to --volume since downstream `bake`
                // looks for the canonical flag name only.
                let flag = if arg == "-v" {
                    "--volume".to_owned()
                } else {
                    arg.to_owned()
                };
                i += 1;
                let v = argv
                    .get(i)
                    .ok_or_else(|| format!("missing {flag} value"))?
                    .clone();
                extra_args.push(flag);
                extra_args.push(v);
            }
            "--enable-egress" | "--no-egress" => {
                extra_args.push(arg.to_owned());
            }
            "--help" | "-h" => {
                eprintln!(
                    "usage: supermachine pull IMAGE [--name NAME] [--memory MIB]\n\
                     \x20                      [--port PORT] [--pull POLICY]\n\
                     \x20                      [--snapshots-dir DIR]\n\
                     \n\
                     Bake IMAGE to a snapshot under --snapshots-dir, then exit.\n\
                     No router is started. Equivalent to `docker pull` for the\n\
                     supermachine cache.\n\
                     \n\
                     Non-service images (rust:1-slim, python:slim, node:slim,\n\
                     anything that doesn't bind a port) just work — the snapshot\n\
                     auto-falls-back to init-state when no listener appears."
                );
                std::process::exit(0);
            }
            s if s.starts_with('-') => return Err(format!("unknown flag: {s}")),
            s => {
                if image.is_some() {
                    return Err(format!("extra arg: {s}"));
                }
                image = Some(s.to_owned());
            }
        }
        i += 1;
    }
    let image = image.ok_or_else(|| "missing IMAGE".to_owned())?;
    if !matches!(pull_policy.as_str(), "missing" | "always" | "never") {
        return Err(format!("unknown --pull policy: {pull_policy}"));
    }

    let request = bake::BakeRequest {
        image: image.clone(),
        name,
        runtime: "supermachine".to_owned(),
        guest_port,
        memory_mib,
        vcpus,
        pull_policy,
        snapshots_dir,
        cmd_override,
        extra_args,
        platform: "linux/arm64".to_owned(),
    };
    let t0 = Instant::now();

    // Linux/KVM has no HVF worker/router subprocess; bake the bootable KVM image
    // in-process (registry-direct, docker-free) straight into the snapshot dir —
    // the same `bake_kvm_auto` the library + `cargo run --example kvm_bake_auto`
    // use. Idempotent: a present, loadable snapshot is reused unless `--pull always`.
    #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
    {
        let dest = ensure_snapshot(
            &request.image,
            request.name.as_deref(),
            &request.snapshots_dir,
            &request.pull_policy,
        )?;
        println!(
            "pulled {} -> {} in {}ms",
            request.image,
            dest.file_name()
                .map(|n| n.to_string_lossy().into_owned())
                .unwrap_or_default(),
            elapsed_ms(t0)
        );
        Ok(())
    }
    #[cfg(not(all(target_os = "linux", target_arch = "x86_64")))]
    {
        let root = repo_root()?;
        bake::run_push(&request, t0, &root)?;
        println!("pulled {} in {}ms", request.image, elapsed_ms(t0));
        Ok(())
    }
}

#[derive(Debug)]
struct SnapshotEntry {
    name: String,
    image: String,
    memory_mib: u64,
    physical_bytes: u64,
    baked_at: String,
    runtime_sha16: String,
}

fn read_snapshot_entry(dir: &Path) -> Option<SnapshotEntry> {
    let metadata_path = dir.join("metadata.json");
    let text = std::fs::read_to_string(&metadata_path).ok()?;
    let json: serde_json::Value = serde_json::from_str(&text).ok()?;
    let name = json
        .get("name")
        .and_then(|v| v.as_str())
        .map(|s| s.to_owned())
        .unwrap_or_else(|| {
            dir.file_name()
                .and_then(|s| s.to_str())
                .unwrap_or("?")
                .to_owned()
        });
    Some(SnapshotEntry {
        name,
        image: json
            .get("image")
            .and_then(|v| v.as_str())
            .unwrap_or("?")
            .to_owned(),
        memory_mib: json.get("memory_mib").and_then(|v| v.as_u64()).unwrap_or(0),
        physical_bytes: json
            .get("snapshot_physical_bytes")
            .and_then(|v| v.as_u64())
            .unwrap_or(0),
        baked_at: json
            .get("baked_at")
            .and_then(|v| v.as_str())
            .unwrap_or("?")
            .to_owned(),
        // HVF snapshots carry a worker-binary `runtime_sha16`; KVM snapshots have
        // no worker, so fall back to the `backend` ("kvm") for the RUNTIME column.
        runtime_sha16: json
            .get("runtime_sha16")
            .and_then(|v| v.as_str())
            .or_else(|| json.get("backend").and_then(|v| v.as_str()))
            .unwrap_or("?")
            .to_owned(),
    })
}

fn human_bytes(n: u64) -> String {
    const KB: u64 = 1024;
    const MB: u64 = 1024 * KB;
    const GB: u64 = 1024 * MB;
    if n >= GB {
        format!("{:.1} GB", n as f64 / GB as f64)
    } else if n >= MB {
        format!("{:.1} MB", n as f64 / MB as f64)
    } else if n >= KB {
        format!("{:.1} KB", n as f64 / KB as f64)
    } else {
        format!("{n} B")
    }
}

/// `supermachine images [--snapshots-dir DIR]` — list baked
/// snapshots. Each row is one snapshot dir under --snapshots-dir.
fn run_images(argv: &[String]) -> Result<(), String> {
    let mut snapshots_dir = default_snapshots_dir();
    let mut i = 0;
    while i < argv.len() {
        match argv[i].as_str() {
            "--snapshots-dir" => {
                i += 1;
                snapshots_dir = PathBuf::from(
                    argv.get(i)
                        .ok_or_else(|| "missing --snapshots-dir value".to_owned())?,
                );
            }
            "--help" | "-h" => {
                eprintln!(
                    "usage: supermachine images [--snapshots-dir DIR]\n\
                     \n\
                     List baked snapshots in --snapshots-dir (default:\n\
                     ~/.local/supermachine-snapshots/). Each row is one\n\
                     snapshot ready for `supermachine run --no-push --name NAME`."
                );
                std::process::exit(0);
            }
            s => return Err(format!("unknown arg: {s}")),
        }
        i += 1;
    }

    let entries = match std::fs::read_dir(&snapshots_dir) {
        Ok(rd) => rd,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            println!("no snapshots in {} (yet)", snapshots_dir.display());
            return Ok(());
        }
        Err(e) => return Err(format!("read {}: {e}", snapshots_dir.display())),
    };
    let mut rows: Vec<SnapshotEntry> = entries
        .filter_map(|r| r.ok())
        .filter(|e| e.path().is_dir())
        .filter_map(|e| read_snapshot_entry(&e.path()))
        .collect();
    if rows.is_empty() {
        println!("no snapshots in {} (yet)", snapshots_dir.display());
        return Ok(());
    }
    rows.sort_by(|a, b| b.baked_at.cmp(&a.baked_at));

    let name_w = rows.iter().map(|r| r.name.len()).max().unwrap_or(4).max(4);
    let image_w = rows.iter().map(|r| r.image.len()).max().unwrap_or(5).max(5);
    println!(
        "{:<name_w$}  {:<image_w$}  {:>8}  {:>9}  {:<20}  RUNTIME",
        "NAME",
        "IMAGE",
        "MEM",
        "SIZE",
        "BAKED",
        name_w = name_w,
        image_w = image_w,
    );
    for r in &rows {
        println!(
            "{:<name_w$}  {:<image_w$}  {:>6} MB  {:>9}  {:<20}  {}",
            r.name,
            r.image,
            r.memory_mib,
            human_bytes(r.physical_bytes),
            r.baked_at,
            r.runtime_sha16,
            name_w = name_w,
            image_w = image_w,
        );
    }
    Ok(())
}

/// `supermachine rmi NAME [NAME ...] [--snapshots-dir DIR]` — remove
/// one or more baked snapshots. Cached layer blobs are left alone
/// (shared across snapshots, expensive to re-fetch).
fn run_rmi(argv: &[String]) -> Result<(), String> {
    let mut snapshots_dir = default_snapshots_dir();
    let mut names: Vec<String> = Vec::new();
    let mut i = 0;
    while i < argv.len() {
        match argv[i].as_str() {
            "--snapshots-dir" => {
                i += 1;
                snapshots_dir = PathBuf::from(
                    argv.get(i)
                        .ok_or_else(|| "missing --snapshots-dir value".to_owned())?,
                );
            }
            "--help" | "-h" => {
                eprintln!(
                    "usage: supermachine rmi NAME [NAME ...] [--snapshots-dir DIR]\n\
                     \n\
                     Remove baked snapshots. NAME is the snapshot name as shown\n\
                     by `supermachine images` (typically the sanitized image ref,\n\
                     e.g. nginx_1_27-alpine).\n\
                     \n\
                     Cached layer blobs in ~/.local/supermachine-layer-cache/ are\n\
                     intentionally left in place — they're shared across snapshots\n\
                     and re-downloading is the slow part."
                );
                std::process::exit(0);
            }
            s if s.starts_with('-') => return Err(format!("unknown flag: {s}")),
            s => names.push(s.to_owned()),
        }
        i += 1;
    }
    if names.is_empty() {
        return Err("missing NAME (try `supermachine images` to list)".to_owned());
    }

    let mut errors = 0;
    for name in &names {
        // Accept either a literal snapshot dir name or an image ref
        // — `supermachine rmi nginx:1.27-alpine` should Just Work.
        let resolved = if snapshots_dir.join(name).is_dir() {
            name.clone()
        } else {
            bake::snapshot_name_for_image(name)
        };
        match rm_snapshot(&snapshots_dir, &resolved) {
            Ok(()) => println!("removed {resolved}"),
            Err(e) => {
                eprintln!("rmi {name}: {e}");
                errors += 1;
            }
        }
    }
    if errors > 0 {
        return Err(format!("{errors} snapshot(s) failed to remove"));
    }
    Ok(())
}

/// `supermachine prune` — bulk-remove old snapshots + orphan
/// `__warm__<tag>` siblings + (optionally) unreferenced layer
/// blobs from the layer cache.
///
/// Without this users accumulate snapshots until they fill
/// `~/.local/supermachine-snapshots/` and the disk. Each
/// `Image.build({ name: <unique> })` leaks a 100-300 MiB tree
/// that nothing else reaps. Default age: 7 days; tuneable via
/// `--older-than` (h/d/w suffixes).
///
/// What gets removed:
///   * Snapshot dirs whose `restore.snap` mtime is older than
///     the threshold.
///   * `__warm__<tag>` siblings whose `restore.snap` mtime is
///     older than the threshold OR whose base dir is missing
///     (orphan warm variants — already unrunnable).
///   * With `--layers`, layer blobs in
///     `~/.local/supermachine-layer-cache/` that aren't
///     referenced by any surviving snapshot's `metadata.json`.
///     Skipped by default because layers are expensive to
///     re-pull.
///
/// `--dry-run` reports what would be removed without touching
/// disk. Useful as a first pass.
fn run_prune(argv: &[String]) -> Result<(), String> {
    let mut snapshots_dir = default_snapshots_dir();
    let mut older_than_secs: u64 = 7 * 24 * 60 * 60; // 7 days
    let mut dry_run = false;
    let mut prune_layers = false;
    let mut layer_cache_dir: Option<PathBuf> = None;
    let mut analyze = false;
    let mut i = 0;
    while i < argv.len() {
        match argv[i].as_str() {
            "--snapshots-dir" => {
                i += 1;
                snapshots_dir = PathBuf::from(
                    argv.get(i)
                        .ok_or_else(|| "missing --snapshots-dir value".to_owned())?,
                );
            }
            "--older-than" => {
                i += 1;
                let s = argv
                    .get(i)
                    .ok_or_else(|| "missing --older-than value".to_owned())?;
                older_than_secs = parse_duration_for_prune(s)?;
            }
            "--dry-run" => dry_run = true,
            "--analyze" => analyze = true,
            "--layers" => prune_layers = true,
            "--layer-cache-dir" => {
                i += 1;
                layer_cache_dir =
                    Some(PathBuf::from(argv.get(i).ok_or_else(|| {
                        "missing --layer-cache-dir value".to_owned()
                    })?));
            }
            "--help" | "-h" => {
                eprintln!(
                    "usage: supermachine prune [--older-than DURATION] \
                     [--snapshots-dir DIR]\n\
                     \x20                          [--layers] [--layer-cache-dir DIR]\n\
                     \x20                          [--dry-run] [--analyze]\n\
                     \n\
                     Remove snapshots older than DURATION (default 7d) and\n\
                     orphan __warm__<tag> siblings whose base dir is gone.\n\
                     \n\
                     DURATION accepts h / d / w suffixes (e.g. 12h, 3d, 2w).\n\
                     A bare number is seconds.\n\
                     \n\
                     With --layers, also prune layer blobs in the layer\n\
                     cache that aren't referenced by any surviving snapshot's\n\
                     metadata.json. Layers are otherwise NEVER pruned (re-\n\
                     pulling them is the slow part of every bake).\n\
                     \n\
                     --dry-run reports what would be removed without\n\
                     touching disk.\n\
                     \n\
                     --analyze short-circuits prune entirely and dumps a\n\
                     per-image breakdown of the snapshots dir: actual disk\n\
                     usage (sparse-aware), snapshot count, dedup-candidate\n\
                     estimate. Useful first pass before pruning.\n\
                     Other flags are ignored when --analyze is set."
                );
                std::process::exit(0);
            }
            s if s.starts_with('-') => return Err(format!("unknown flag: {s}")),
            s => return Err(format!("unexpected positional arg: {s}")),
        }
        i += 1;
    }

    // --analyze short-circuits the prune logic and dumps a per-
    // image breakdown of the snapshots dir. Useful as a first
    // pass to decide what to prune (and as a debugging aid for
    // "where did my disk go").
    if analyze {
        return run_prune_analyze(&snapshots_dir);
    }

    let now_secs = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let cutoff_secs = now_secs.saturating_sub(older_than_secs);

    // First pass: enumerate snapshot dirs and decide which to
    // keep vs remove. We separate "base" from "__warm__<tag>"
    // siblings so we can find orphans (warm variants whose base
    // is gone).
    let entries = match std::fs::read_dir(&snapshots_dir) {
        Ok(rd) => rd,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            println!(
                "snapshots dir does not exist: {} (nothing to prune)",
                snapshots_dir.display()
            );
            return Ok(());
        }
        Err(e) => return Err(format!("read {}: {e}", snapshots_dir.display())),
    };
    #[derive(Debug)]
    struct SnapEntry {
        path: PathBuf,
        name: String,
        is_warm_variant: Option<String>, // Some(base_name) if warm sibling
        mtime_secs: u64,
        size_bytes: u64,
    }
    let mut snaps: Vec<SnapEntry> = Vec::new();
    for entry in entries.flatten() {
        let path = entry.path();
        if !path.is_dir() {
            continue;
        }
        let name = match path.file_name().and_then(|s| s.to_str()) {
            Some(s) => s.to_owned(),
            None => continue,
        };
        // `__warm__<tag>` suffix marks a warm-variant sibling.
        let is_warm_variant = name
            .rsplit_once("__warm__")
            .map(|(base, _tag)| base.to_owned());
        let restore_snap = path.join("restore.snap");
        let mtime_secs = match std::fs::metadata(&restore_snap).and_then(|m| m.modified()) {
            Ok(t) => t
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_secs())
                .unwrap_or(0),
            // No restore.snap → the bake never completed or was
            // partially cleaned. Treat as "very old" so it gets
            // pruned.
            Err(_) => 0,
        };
        let size_bytes = dir_size_bytes(&path).unwrap_or(0);
        snaps.push(SnapEntry {
            path,
            name,
            is_warm_variant,
            mtime_secs,
            size_bytes,
        });
    }
    // Build a set of surviving base names so we can detect
    // orphan warm variants.
    let names_present: std::collections::HashSet<String> =
        snaps.iter().map(|s| s.name.clone()).collect();

    let mut to_remove: Vec<&SnapEntry> = Vec::new();
    for s in &snaps {
        let age_ok = s.mtime_secs > cutoff_secs;
        let orphan = match &s.is_warm_variant {
            Some(base) => !names_present.contains(base),
            None => false,
        };
        if !age_ok || orphan {
            to_remove.push(s);
        }
    }

    let total_bytes: u64 = to_remove.iter().map(|s| s.size_bytes).sum();
    if to_remove.is_empty() {
        println!(
            "no snapshots to prune (threshold: older than {} s, dir: {})",
            older_than_secs,
            snapshots_dir.display()
        );
    } else {
        println!(
            "{} {} snapshot(s) ({:.1} MiB):",
            if dry_run { "would prune" } else { "pruning" },
            to_remove.len(),
            total_bytes as f64 / (1024.0 * 1024.0)
        );
        for s in &to_remove {
            let age_h = (now_secs.saturating_sub(s.mtime_secs)) as f64 / 3600.0;
            let kind = match &s.is_warm_variant {
                Some(base) if !names_present.contains(base) => "orphan-warm",
                Some(_) => "warm",
                None => "base",
            };
            println!(
                "  {:11}  {:>6.1} MiB  {:>5.1}h  {}",
                kind,
                s.size_bytes as f64 / (1024.0 * 1024.0),
                age_h,
                s.name
            );
        }
    }

    if !dry_run {
        let mut errs = 0;
        for s in &to_remove {
            if let Err(e) = std::fs::remove_dir_all(&s.path) {
                eprintln!("  remove {}: {e}", s.path.display());
                errs += 1;
            }
        }
        if errs > 0 {
            return Err(format!("{errs} snapshot(s) failed to remove"));
        }
    }

    // Layer cache pruning (optional). After snapshot removal,
    // any layer blob in `~/.local/supermachine-layer-cache/`
    // that isn't referenced by a surviving snapshot's
    // `metadata.json["layers"]` field is unused.
    if prune_layers {
        let cache_dir =
            layer_cache_dir.unwrap_or_else(|| home_join(".local/supermachine-layer-cache"));
        prune_layer_cache(&snapshots_dir, &cache_dir, dry_run)?;
    } else if !to_remove.is_empty() {
        println!(
            "\nlayer cache NOT pruned (pass --layers to also reclaim layer \
             blobs). Layers are shared across snapshots and re-pulling them \
             is the slow part of every bake."
        );
    }

    Ok(())
}

/// Parse `12h` / `3d` / `2w` / bare seconds for --older-than.
/// `supermachine dedup` — cross-snapshot APFS clonefile pass.
///
/// Groups baked snapshots by image + memoryMib, picks one
/// "canonical" per group (the most recent; preserves current
/// state), and rewrites every other snapshot's `restore.snap` as
/// `clonefile(canonical, target) + pwrite_diff_pages(target_current
/// vs canonical)`. APFS shares blocks between cloned files, so
/// the target's actual disk usage drops to just its diff from
/// the canonical (typically <5% of the file).
///
/// Safety:
///   * The output `restore.snap` is byte-identical to the
///     pre-dedup target. Snapshots remain functionally
///     loadable; only their on-disk representation changes
///     to share APFS blocks with the canonical.
///   * Atomic rename (`tmp.snap.dedup.tmp` → `restore.snap`)
///     means an interrupted run leaves either the original or
///     the new file, never a half-written one.
///   * `--dry-run` reports expected savings without touching
///     disk.
///   * Empirical reclaim on a 100 GiB snapshots dir: ~90%.
///
/// Why offline (vs. bake-time): an offline pass is much easier
/// to debug. The bake-time online version can come later once
/// the offline path is proven.
#[cfg(target_os = "macos")]
fn run_dedup(argv: &[String]) -> Result<(), String> {
    use std::collections::BTreeMap;
    use std::io::Write;
    use std::os::unix::fs::{FileExt, MetadataExt};

    let mut snapshots_dir = default_snapshots_dir();
    let mut dry_run = false;
    let mut min_group_size: usize = 2;
    let mut i = 0;
    while i < argv.len() {
        match argv[i].as_str() {
            "--snapshots-dir" => {
                i += 1;
                snapshots_dir = PathBuf::from(
                    argv.get(i)
                        .ok_or_else(|| "missing --snapshots-dir value".to_owned())?,
                );
            }
            "--dry-run" => dry_run = true,
            "--min-group-size" => {
                i += 1;
                min_group_size = argv
                    .get(i)
                    .ok_or_else(|| "missing --min-group-size value".to_owned())?
                    .parse()
                    .map_err(|e| format!("parse --min-group-size: {e}"))?;
            }
            "--help" | "-h" => {
                eprintln!(
                    "usage: supermachine dedup [--snapshots-dir DIR] \
                     [--dry-run] [--min-group-size N]\n\
                     \n\
                     Cross-snapshot APFS clonefile pass. For each group of\n\
                     snapshots sharing the same image + memoryMib, picks the\n\
                     most-recent as canonical and rewrites every other\n\
                     snapshot's restore.snap to share APFS blocks with it.\n\
                     \n\
                     The result is byte-identical to the pre-dedup snapshot\n\
                     (functionally a no-op for restore) but actual disk\n\
                     usage drops to just the diff from the canonical.\n\
                     Typical reclaim on a busy dev tree: 80-95%.\n\
                     \n\
                     --min-group-size N (default 2): only dedup groups with\n\
                     at least N snapshots. Set to 5 to skip small groups\n\
                     where the per-snapshot fixed overhead dominates.\n\
                     \n\
                     --dry-run reports expected reclaim without touching\n\
                     disk."
                );
                std::process::exit(0);
            }
            s if s.starts_with('-') => return Err(format!("unknown flag: {s}")),
            s => return Err(format!("unexpected positional arg: {s}")),
        }
        i += 1;
    }

    #[derive(Debug)]
    struct SnapEntry {
        path: PathBuf,
        image: String,
        memory_mib: u32,
        baked_by_version: Option<String>,
        mtime_secs: u64,
        actual_bytes: u64,
        snap_size: u64,
    }

    let entries = match std::fs::read_dir(&snapshots_dir) {
        Ok(rd) => rd,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            println!("snapshots dir does not exist: {}", snapshots_dir.display());
            return Ok(());
        }
        Err(e) => return Err(format!("read {}: {e}", snapshots_dir.display())),
    };

    let mut snaps: Vec<SnapEntry> = Vec::new();
    for entry in entries.flatten() {
        let path = entry.path();
        if !path.is_dir() {
            continue;
        }
        let restore_snap = path.join("restore.snap");
        if !restore_snap.is_file() {
            continue;
        }
        let Ok(meta_text) = std::fs::read_to_string(path.join("metadata.json")) else {
            continue;
        };
        let Ok(meta): Result<serde_json::Value, _> = serde_json::from_str(&meta_text) else {
            continue;
        };
        let image = match meta.get("image").and_then(|x| x.as_str()) {
            Some(s) => s.to_owned(),
            None => continue,
        };
        let memory_mib = meta
            .get("memory_mib")
            .and_then(|x| x.as_u64())
            .map(|x| x as u32)
            .unwrap_or(0);
        let baked_by_version = meta
            .get("baked_by_version")
            .and_then(|x| x.as_str())
            .map(|s| s.to_owned());
        let Ok(stat) = std::fs::metadata(&restore_snap) else {
            continue;
        };
        let mtime_secs = stat
            .modified()
            .ok()
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|d| d.as_secs())
            .unwrap_or(0);
        let actual_bytes = stat.blocks().saturating_mul(512);
        let snap_size = stat.len();
        snaps.push(SnapEntry {
            path,
            image,
            memory_mib,
            baked_by_version,
            mtime_secs,
            actual_bytes,
            snap_size,
        });
    }

    if snaps.is_empty() {
        println!("no baked snapshots found in {}", snapshots_dir.display());
        return Ok(());
    }

    // Group key: (image, memory_mib, baked_by_version, snap_size).
    // Same image+memory+version+filesize means the restore.snap
    // files are layout-compatible for page-level diff.
    type GroupKey = (String, u32, Option<String>, u64);
    let mut groups: BTreeMap<GroupKey, Vec<usize>> = BTreeMap::new();
    for (idx, s) in snaps.iter().enumerate() {
        let key = (
            s.image.clone(),
            s.memory_mib,
            s.baked_by_version.clone(),
            s.snap_size,
        );
        groups.entry(key).or_default().push(idx);
    }

    let mut total_targets = 0usize;
    let mut total_actual_before: u64 = 0;
    let mut total_estimated_savings: u64 = 0;

    let mut plan: Vec<(usize, Vec<usize>)> = Vec::new();
    for idxs in groups.values() {
        if idxs.len() < min_group_size {
            continue;
        }
        // Sort by mtime descending; canonical = newest.
        let mut sorted = idxs.clone();
        sorted.sort_by(|a, b| snaps[*b].mtime_secs.cmp(&snaps[*a].mtime_secs));
        let canonical = sorted[0];
        let targets: Vec<usize> = sorted.into_iter().skip(1).collect();
        for &t in &targets {
            total_targets += 1;
            total_actual_before = total_actual_before.saturating_add(snaps[t].actual_bytes);
            // Estimate: a target ~95% identical to canonical
            // (empirical baseline). So expected savings = 95%
            // of its current actual_bytes minus the diff write
            // cost (~5%).
            total_estimated_savings = total_estimated_savings
                .saturating_add(snaps[t].actual_bytes.saturating_mul(90).saturating_div(100));
        }
        plan.push((canonical, targets));
    }

    if total_targets == 0 {
        println!(
            "no dedup-eligible groups (need ≥{min_group_size} snapshots sharing \
             image + memoryMib + baked_by_version + filesize)"
        );
        return Ok(());
    }

    println!("snapshots dir: {}", snapshots_dir.display());
    println!(
        "{} {} dedup target(s) across {} group(s)",
        if dry_run {
            "would dedup"
        } else {
            "deduplicating"
        },
        total_targets,
        plan.len()
    );
    println!("  current actual:   {}", fmt_bytes(total_actual_before));
    println!(
        "  estimated after:  {}",
        fmt_bytes(total_actual_before.saturating_sub(total_estimated_savings))
    );
    println!(
        "  estimated saved:  {}  ({:.0}%)",
        fmt_bytes(total_estimated_savings),
        if total_actual_before == 0 {
            0.0
        } else {
            100.0 * total_estimated_savings as f64 / total_actual_before as f64
        }
    );

    if dry_run {
        println!();
        for (canonical, targets) in &plan {
            let c = &snaps[*canonical];
            println!(
                "{} ({} MiB canonical) ⇐ {} target(s)",
                c.image,
                c.actual_bytes / (1024 * 1024),
                targets.len(),
            );
        }
        println!();
        println!(
            "NOTE: estimates above are per-file. APFS clonefile shares blocks\n\
             between cloned files, which `stat`/`du` cannot see — they\n\
             per-attribute the full block count to each file. Actual\n\
             physical disk savings (visible via `df -h` on the data\n\
             volume) are typically lower than the per-file estimate but\n\
             still substantial (~50% of the estimate on a typical corpus)."
        );
        return Ok(());
    }
    // Sample container-level free space before to compute honest
    // savings (du / stat don't see APFS sharing).
    let free_before = data_volume_free_bytes(&snapshots_dir);

    // Execute: per-group, per-target, do the clonefile + diff
    // dance. Each target operation is atomic via rename — an
    // interrupted run leaves either the old file or the new
    // (never a half-written one).
    let mut actual_saved: u64 = 0;
    let mut failures: Vec<(PathBuf, String)> = Vec::new();
    for (canonical, targets) in &plan {
        let canon_snap = snaps[*canonical].path.join("restore.snap");
        for &target_idx in targets {
            let target_dir = &snaps[target_idx].path;
            let target_snap = target_dir.join("restore.snap");
            let before = snaps[target_idx].actual_bytes;
            match dedup_one(&canon_snap, &target_snap) {
                Ok(after) => {
                    let saved = before.saturating_sub(after);
                    actual_saved = actual_saved.saturating_add(saved);
                    println!(
                        "  {}  {} → {}  ({} saved)",
                        target_dir
                            .file_name()
                            .and_then(|s| s.to_str())
                            .unwrap_or("?"),
                        fmt_bytes(before),
                        fmt_bytes(after),
                        fmt_bytes(saved),
                    );
                }
                Err(e) => {
                    failures.push((target_snap.clone(), e));
                }
            }
        }
    }

    let free_after = data_volume_free_bytes(&snapshots_dir);
    println!();
    println!("per-file accounting (st_blocks delta — blind to APFS sharing):");
    println!("  apparent saved per stat: {}", fmt_bytes(actual_saved));
    if let (Some(before), Some(after)) = (free_before, free_after) {
        let physical_saved = after.saturating_sub(before);
        println!();
        println!("physical disk freed on the data volume (df-style, includes APFS clone sharing):");
        println!("  {}", fmt_bytes(physical_saved));
        if physical_saved > actual_saved {
            println!(
                "  ↑ this is the real number. The per-stat delta above\n\
                 is conservative because APFS clonefile sharing is invisible\n\
                 to `stat`/`du`."
            );
        }
    }
    if !failures.is_empty() {
        eprintln!("{} target(s) failed:", failures.len());
        for (p, e) in &failures {
            eprintln!("  {}: {e}", p.display());
        }
        return Err(format!("{} dedup operation(s) failed", failures.len()));
    }

    // Suppress unused-import warning when FileExt's methods are
    // used via the helper below.
    let _: fn() = || {
        let _ = std::any::type_name::<dyn Write>();
        let _ = std::any::type_name::<dyn FileExt>();
        let _ = std::any::type_name::<dyn MetadataExt>();
    };
    Ok(())
}

#[cfg(not(target_os = "macos"))]
fn run_dedup(_argv: &[String]) -> Result<(), String> {
    Err("supermachine dedup requires macOS (APFS clonefile)".to_owned())
}

/// Page-by-page replace `target` with an APFS clone of `canon`
/// plus diff-pwrites for pages that differ. Returns the new
/// actual on-disk byte count.
///
/// Safety: writes to a `.dedup.tmp` sibling and atomic-renames.
/// On any failure, the temp file is removed (best-effort) and
/// the original `target` is untouched.
#[cfg(target_os = "macos")]
fn dedup_one(canon: &Path, target: &Path) -> Result<u64, String> {
    use std::fs::OpenOptions;
    use std::os::unix::fs::{FileExt, MetadataExt};

    let canon_md =
        std::fs::metadata(canon).map_err(|e| format!("stat canon {}: {e}", canon.display()))?;
    let target_md =
        std::fs::metadata(target).map_err(|e| format!("stat target {}: {e}", target.display()))?;
    if canon_md.len() != target_md.len() {
        return Err(format!(
            "size mismatch (canon={} target={}); skipping",
            canon_md.len(),
            target_md.len()
        ));
    }
    let file_len = canon_md.len();

    let tmp_path = target.with_extension("snap.dedup.tmp");
    // Make sure no stale tmp from a previous interrupted run.
    let _ = std::fs::remove_file(&tmp_path);

    // APFS clonefile: tmp now shares blocks with canon. Fast
    // (~ms for any size).
    {
        let canon_c = std::ffi::CString::new(canon.as_os_str().as_encoded_bytes())
            .map_err(|_| "canon path contains NUL".to_owned())?;
        let tmp_c = std::ffi::CString::new(tmp_path.as_os_str().as_encoded_bytes())
            .map_err(|_| "tmp path contains NUL".to_owned())?;
        let ret = unsafe { libc::clonefile(canon_c.as_ptr(), tmp_c.as_ptr(), 0) };
        if ret != 0 {
            let e = std::io::Error::last_os_error();
            return Err(format!(
                "clonefile {} → {}: {e}",
                canon.display(),
                tmp_path.display()
            ));
        }
    }

    // mmap target (current bytes we want to preserve) and the
    // tmp (canon's content we're about to overwrite with target's
    // diff).
    let target_data = mmap_readonly(target, file_len)?;
    let tmp_data = mmap_readonly(&tmp_path, file_len)?;

    // Open tmp for write.
    let tmp_file = OpenOptions::new()
        .write(true)
        .open(&tmp_path)
        .map_err(|e| format!("open tmp {}: {e}", tmp_path.display()))?;

    // Walk in 4 KiB pages, pwrite any differing one. 4 KiB =
    // APFS block; sub-block diffs don't help (CoW is per-block).
    const PAGE: usize = 4096;
    let total = file_len as usize;
    let mut diff_pages: u64 = 0;
    let mut off = 0usize;
    while off < total {
        let end = (off + PAGE).min(total);
        let t = &target_data[off..end];
        let c = &tmp_data[off..end];
        if t != c {
            tmp_file
                .write_all_at(t, off as u64)
                .map_err(|e| format!("pwrite tmp @ {off}: {e}"))?;
            diff_pages += 1;
        }
        off = end;
    }
    // Flush the tmp file before unmap (defensive — atomic rename
    // would force flush anyway, but explicit is clearer).
    tmp_file.sync_data().map_err(|e| format!("sync tmp: {e}"))?;

    drop(target_data);
    drop(tmp_data);
    drop(tmp_file);

    // Atomic rename.
    std::fs::rename(&tmp_path, target)
        .map_err(|e| format!("rename {} → {}: {e}", tmp_path.display(), target.display()))?;

    // Stat the new target. Should now report low actual bytes
    // (only the diff pages).
    let new_md =
        std::fs::metadata(target).map_err(|e| format!("stat new {}: {e}", target.display()))?;
    let _ = diff_pages; // useful for debug logging if we want
    Ok(new_md.blocks().saturating_mul(512))
}

/// Free bytes on the data volume containing `probe_path`.
/// Returns None if statfs fails (e.g. non-macOS path).
///
/// Used to report honest "actual physical bytes freed" after a
/// dedup run, since `stat`/`du` over-count due to APFS clonefile
/// sharing being invisible at the inode level.
#[cfg(target_os = "macos")]
fn data_volume_free_bytes(probe_path: &Path) -> Option<u64> {
    use std::os::unix::ffi::OsStrExt;
    let c = std::ffi::CString::new(probe_path.as_os_str().as_bytes()).ok()?;
    let mut st: libc::statfs = unsafe { std::mem::zeroed() };
    let rc = unsafe { libc::statfs(c.as_ptr(), &mut st) };
    if rc != 0 {
        return None;
    }
    Some(st.f_bavail.saturating_mul(st.f_bsize as u64))
}

#[cfg(not(target_os = "macos"))]
fn data_volume_free_bytes(_probe_path: &Path) -> Option<u64> {
    None
}

/// Read-only mmap of a file. We use this for byte-level page
/// compares; `&[u8]` slices over multi-GiB files are fine since
/// mmap is lazy.
#[cfg(target_os = "macos")]
fn mmap_readonly(path: &Path, expected_len: u64) -> Result<Mmap, String> {
    let file = std::fs::File::open(path).map_err(|e| format!("open {}: {e}", path.display()))?;
    let md = file
        .metadata()
        .map_err(|e| format!("stat {}: {e}", path.display()))?;
    if md.len() != expected_len {
        return Err(format!(
            "{} unexpected size {} (wanted {})",
            path.display(),
            md.len(),
            expected_len
        ));
    }
    let len = md.len() as usize;
    if len == 0 {
        return Err(format!("{}: empty file", path.display()));
    }
    use std::os::unix::io::AsRawFd;
    let ptr = unsafe {
        libc::mmap(
            std::ptr::null_mut(),
            len,
            libc::PROT_READ,
            libc::MAP_PRIVATE,
            file.as_raw_fd(),
            0,
        )
    };
    if ptr == libc::MAP_FAILED {
        return Err(format!(
            "mmap {}: {}",
            path.display(),
            std::io::Error::last_os_error()
        ));
    }
    Ok(Mmap {
        ptr: ptr as *const u8,
        len,
    })
}

#[cfg(target_os = "macos")]
struct Mmap {
    ptr: *const u8,
    len: usize,
}

#[cfg(target_os = "macos")]
impl std::ops::Deref for Mmap {
    type Target = [u8];
    fn deref(&self) -> &[u8] {
        // SAFETY: ptr is from mmap with len bytes, valid for
        // the lifetime of the Mmap (Drop unmaps).
        unsafe { std::slice::from_raw_parts(self.ptr, self.len) }
    }
}

#[cfg(target_os = "macos")]
impl Drop for Mmap {
    fn drop(&mut self) {
        unsafe {
            libc::munmap(self.ptr as *mut libc::c_void, self.len);
        }
    }
}

#[cfg(target_os = "macos")]
// SAFETY: read-only mmap of a real file is Send + Sync; the
// underlying pages are accessed concurrently by the kernel for
// other mappings too.
unsafe impl Send for Mmap {}
#[cfg(target_os = "macos")]
unsafe impl Sync for Mmap {}

/// `supermachine prune --analyze` — diagnostic report on the
/// snapshots dir. Groups snapshots by image ref, shows per-image
/// disk usage (actual blocks-allocated, not apparent size) and
/// estimates how much cross-snapshot dedup would save.
///
/// "Dedup-candidate" heuristic: per-image, assume the diff
/// between two snapshots of the same image baked the same way
/// is ~5% of the file (the warm-snapshot delta we measure
/// empirically). For N snapshots of the same image, dedup-
/// candidate = (N-1) * 0.95 * average-actual-size. Conservative
/// — actual savings could be higher with content-addressed page
/// dedup.
///
/// We use APFS sparse-aware byte counts (`st_blocks * 512`), not
/// apparent file size. `restore.snap` looks like a 2 GiB file
/// to `ls` but typically allocates 200-400 MiB of actual disk;
/// the report uses the latter.
fn run_prune_analyze(snapshots_dir: &Path) -> Result<(), String> {
    use std::collections::BTreeMap;
    use std::os::unix::fs::MetadataExt;

    // `name` is carried for completeness of the snapshot record even
    // though the current print path keys off other fields; keep it
    // without tripping `-D warnings`.
    #[allow(dead_code)]
    struct SnapInfo {
        name: String,
        image: String,
        is_warm_variant: bool,
        actual_bytes: u64,
        apparent_bytes: u64,
        age_secs: u64,
    }

    let entries = match std::fs::read_dir(snapshots_dir) {
        Ok(rd) => rd,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            println!("snapshots dir does not exist: {}", snapshots_dir.display());
            return Ok(());
        }
        Err(e) => return Err(format!("read {}: {e}", snapshots_dir.display())),
    };

    let now_secs = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);

    let mut snaps: Vec<SnapInfo> = Vec::new();
    for entry in entries.flatten() {
        let path = entry.path();
        if !path.is_dir() {
            continue;
        }
        let name = match path.file_name().and_then(|s| s.to_str()) {
            Some(s) => s.to_owned(),
            None => continue,
        };
        let is_warm_variant = name.contains("__warm__");
        // Parse image ref from metadata.json. Fall back to
        // "(unknown)" so we still account for the disk usage.
        let image = std::fs::read_to_string(path.join("metadata.json"))
            .ok()
            .and_then(|t| {
                let v: serde_json::Value = serde_json::from_str(&t).ok()?;
                v.get("image")
                    .and_then(|x| x.as_str())
                    .map(|s| s.to_owned())
            })
            .unwrap_or_else(|| "(unknown)".to_owned());
        // restore.snap is the dominant cost; its mtime tracks
        // bake time better than the dir's mtime.
        let restore_snap = path.join("restore.snap");
        let age_secs = std::fs::metadata(&restore_snap)
            .and_then(|m| m.modified())
            .ok()
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|d| now_secs.saturating_sub(d.as_secs()))
            .unwrap_or(0);
        // Actual blocks allocated (sparse-aware) vs apparent
        // size. We walk the dir so we capture every file (not
        // just restore.snap).
        let (actual_bytes, apparent_bytes) = dir_actual_and_apparent(&path);
        snaps.push(SnapInfo {
            name,
            image,
            is_warm_variant,
            actual_bytes,
            apparent_bytes,
            age_secs,
        });
    }

    if snaps.is_empty() {
        println!(
            "no snapshots in {} (nothing to analyze)",
            snapshots_dir.display()
        );
        return Ok(());
    }

    // Group by image. Sum actual + apparent bytes, count
    // snapshots, track oldest+newest.
    struct ImageGroup {
        count: usize,
        warm_count: usize,
        actual_bytes: u64,
        apparent_bytes: u64,
        oldest_age_secs: u64,
        newest_age_secs: u64,
    }
    let mut groups: BTreeMap<String, ImageGroup> = BTreeMap::new();
    for s in &snaps {
        let g = groups.entry(s.image.clone()).or_insert(ImageGroup {
            count: 0,
            warm_count: 0,
            actual_bytes: 0,
            apparent_bytes: 0,
            oldest_age_secs: 0,
            newest_age_secs: u64::MAX,
        });
        g.count += 1;
        if s.is_warm_variant {
            g.warm_count += 1;
        }
        g.actual_bytes += s.actual_bytes;
        g.apparent_bytes += s.apparent_bytes;
        if s.age_secs > g.oldest_age_secs {
            g.oldest_age_secs = s.age_secs;
        }
        if s.age_secs < g.newest_age_secs {
            g.newest_age_secs = s.age_secs;
        }
    }

    // Sort groups by actual bytes descending so the report
    // surfaces the biggest space hogs first.
    let mut rows: Vec<(&String, &ImageGroup)> = groups.iter().collect();
    rows.sort_by(|a, b| b.1.actual_bytes.cmp(&a.1.actual_bytes));

    let total_actual: u64 = rows.iter().map(|(_, g)| g.actual_bytes).sum();
    let total_apparent: u64 = rows.iter().map(|(_, g)| g.apparent_bytes).sum();
    let total_count: usize = rows.iter().map(|(_, g)| g.count).sum();

    println!("snapshots dir: {}", snapshots_dir.display());
    println!(
        "totals: {} snapshots, {:.1} GiB actual ({:.1} GiB apparent — {:.0}× sparse savings)",
        total_count,
        total_actual as f64 / (1024.0 * 1024.0 * 1024.0),
        total_apparent as f64 / (1024.0 * 1024.0 * 1024.0),
        if total_actual == 0 {
            0.0
        } else {
            total_apparent as f64 / total_actual as f64
        }
    );
    println!();
    println!(
        "{:<60}  {:>5}  {:>5}  {:>10}  {:>10}  {:>10}",
        "image", "snaps", "warm", "actual", "avg/snap", "dedup-cand"
    );
    println!(
        "{:<60}  {:>5}  {:>5}  {:>10}  {:>10}  {:>10}",
        "─".repeat(60),
        "─────",
        "─────",
        "──────────",
        "──────────",
        "──────────"
    );
    let mut total_dedup: u64 = 0;
    for (image, g) in &rows {
        let avg = if g.count > 0 {
            g.actual_bytes / g.count as u64
        } else {
            0
        };
        // Dedup heuristic: per-image, (count - 1) * 95% * avg.
        // The 95% comes from: warm snapshots typically diverge
        // from their base by ~5% of bytes (measured on alpine).
        // For N total snapshots of the same image, you'd keep
        // 1 full + (N-1) thin diffs.
        let dedup_candidate = if g.count > 1 {
            ((g.count as u64).saturating_sub(1))
                .saturating_mul(avg)
                .saturating_mul(95)
                .saturating_div(100)
        } else {
            0
        };
        total_dedup = total_dedup.saturating_add(dedup_candidate);
        // Truncate long image names so the columns line up.
        let display_image = if image.len() > 58 {
            format!("{}…", &image[..57])
        } else {
            image.to_string()
        };
        println!(
            "{:<60}  {:>5}  {:>5}  {:>10}  {:>10}  {:>10}",
            display_image,
            g.count,
            g.warm_count,
            fmt_bytes(g.actual_bytes),
            fmt_bytes(avg),
            fmt_bytes(dedup_candidate),
        );
    }

    println!();
    println!(
        "dedup-candidate total: {} ({:.0}% of actual) — \
         estimated reclaim if cross-snapshot APFS clonefile were enabled\n\
         via `supermachine dedup`",
        fmt_bytes(total_dedup),
        if total_actual == 0 {
            0.0
        } else {
            100.0 * total_dedup as f64 / total_actual as f64
        }
    );
    println!();
    println!(
        "NOTE: \"actual\" sums per-file st_blocks, which over-counts\n\
         shared blocks after `supermachine dedup` runs (APFS clonefile\n\
         sharing is invisible to stat/du). The true physical free space\n\
         is visible via `df -h /System/Volumes/Data` on macOS."
    );
    println!();
    println!("to reclaim from old snapshots:");
    println!("  supermachine prune --older-than 1d --dry-run");
    println!("to APFS-clonefile dedup current snapshots in-place:");
    println!("  supermachine dedup --dry-run");
    println!("to reclaim unreferenced layers:");
    println!("  supermachine prune --older-than 1d --layers --dry-run");

    // Suppress unused-field warning. (st_blocks visit lives in
    // dir_actual_and_apparent below.)
    let _ = std::any::type_name::<dyn MetadataExt>();

    Ok(())
}

/// Format bytes as "1.2 GiB" / "342 MiB" / "12 KiB" / "8 B".
fn fmt_bytes(n: u64) -> String {
    let g = 1024u64.pow(3);
    let m = 1024u64.pow(2);
    let k = 1024u64;
    if n >= g {
        format!("{:.1} GiB", n as f64 / g as f64)
    } else if n >= m {
        format!("{:.0} MiB", n as f64 / m as f64)
    } else if n >= k {
        format!("{:.0} KiB", n as f64 / k as f64)
    } else {
        format!("{n} B")
    }
}

/// Walk `dir`, summing actual (sparse-aware via `st_blocks * 512`)
/// and apparent (file size) bytes. Returns `(actual, apparent)`.
fn dir_actual_and_apparent(dir: &Path) -> (u64, u64) {
    use std::os::unix::fs::MetadataExt;
    let mut actual: u64 = 0;
    let mut apparent: u64 = 0;
    let mut stack: Vec<PathBuf> = vec![dir.to_path_buf()];
    while let Some(d) = stack.pop() {
        let rd = match std::fs::read_dir(&d) {
            Ok(r) => r,
            Err(_) => continue,
        };
        for entry in rd.flatten() {
            let path = entry.path();
            let md = match entry.metadata() {
                Ok(m) => m,
                Err(_) => continue,
            };
            if md.is_dir() {
                stack.push(path);
            } else {
                apparent = apparent.saturating_add(md.len());
                // st_blocks is in 512-byte sectors per POSIX.
                actual = actual.saturating_add(md.blocks().saturating_mul(512));
            }
        }
    }
    (actual, apparent)
}

fn parse_duration_for_prune(s: &str) -> Result<u64, String> {
    let s = s.trim();
    if s.is_empty() {
        return Err("empty duration".to_owned());
    }
    let (num_str, mul) = match s.chars().last().unwrap() {
        'h' => (&s[..s.len() - 1], 60 * 60),
        'd' => (&s[..s.len() - 1], 24 * 60 * 60),
        'w' => (&s[..s.len() - 1], 7 * 24 * 60 * 60),
        c if c.is_ascii_digit() => (s, 1),
        c => return Err(format!("unknown duration suffix: {c} (use h/d/w)")),
    };
    let n: u64 = num_str
        .parse()
        .map_err(|e| format!("parse duration {s:?}: {e}"))?;
    Ok(n * mul)
}

/// Recursively sum file sizes in a directory. Best-effort —
/// any unreadable entry contributes 0.
fn dir_size_bytes(dir: &Path) -> std::io::Result<u64> {
    let mut total: u64 = 0;
    let mut stack: Vec<PathBuf> = vec![dir.to_path_buf()];
    while let Some(d) = stack.pop() {
        let rd = match std::fs::read_dir(&d) {
            Ok(r) => r,
            Err(_) => continue,
        };
        for entry in rd.flatten() {
            let path = entry.path();
            let md = match entry.metadata() {
                Ok(m) => m,
                Err(_) => continue,
            };
            if md.is_dir() {
                stack.push(path);
            } else {
                total = total.saturating_add(md.len());
            }
        }
    }
    Ok(total)
}

/// `--layers` mode: walk every surviving snapshot's
/// metadata.json, collect the set of layer file paths it
/// references, then remove any blob in the layer cache that
/// isn't in that set.
fn prune_layer_cache(snapshots_dir: &Path, cache_dir: &Path, dry_run: bool) -> Result<(), String> {
    if !cache_dir.is_dir() {
        println!(
            "\nlayer cache dir does not exist: {} (nothing to prune)",
            cache_dir.display()
        );
        return Ok(());
    }
    // Build the referenced set from every metadata.json in the
    // snapshots dir. We're conservative: any string in the
    // metadata that looks like a path under the cache dir counts
    // as a reference.
    let mut referenced: std::collections::HashSet<PathBuf> = std::collections::HashSet::new();
    if let Ok(rd) = std::fs::read_dir(snapshots_dir) {
        for entry in rd.flatten() {
            let meta_path = entry.path().join("metadata.json");
            let Ok(text) = std::fs::read_to_string(&meta_path) else {
                continue;
            };
            let cache_str = cache_dir.to_string_lossy().to_string();
            // Scan for any substring that begins with cache_dir
            // and looks like a path. Cheap and broad.
            let mut pos = 0;
            while let Some(start) = text[pos..].find(&cache_str) {
                let abs = pos + start;
                // Path ends at first non-path character (quote,
                // newline, comma, …). We look for double-quote
                // since metadata.json is JSON.
                let end = text[abs..].find('"').map(|e| abs + e).unwrap_or(text.len());
                let candidate = &text[abs..end];
                referenced.insert(PathBuf::from(candidate));
                pos = end;
            }
        }
    }

    let entries = match std::fs::read_dir(cache_dir) {
        Ok(r) => r,
        Err(e) => {
            return Err(format!("read {}: {e}", cache_dir.display()));
        }
    };
    let mut unreferenced: Vec<(PathBuf, u64)> = Vec::new();
    for entry in entries.flatten() {
        let path = entry.path();
        if !path.is_file() {
            continue;
        }
        if referenced.contains(&path) {
            continue;
        }
        let size = entry.metadata().map(|m| m.len()).unwrap_or(0);
        unreferenced.push((path, size));
    }

    if unreferenced.is_empty() {
        println!(
            "\nno unreferenced layer blobs (cache: {})",
            cache_dir.display()
        );
        return Ok(());
    }
    let total: u64 = unreferenced.iter().map(|(_, s)| s).sum();
    println!(
        "\n{} {} unreferenced layer blob(s) ({:.1} MiB):",
        if dry_run { "would prune" } else { "pruning" },
        unreferenced.len(),
        total as f64 / (1024.0 * 1024.0)
    );
    for (p, s) in &unreferenced {
        println!(
            "  layer       {:>6.1} MiB  {}",
            *s as f64 / (1024.0 * 1024.0),
            p.file_name()
                .and_then(|n| n.to_str())
                .unwrap_or("<unknown>")
        );
    }
    if !dry_run {
        let mut errs = 0;
        for (p, _) in &unreferenced {
            if let Err(e) = std::fs::remove_file(p) {
                eprintln!("  remove {}: {e}", p.display());
                errs += 1;
            }
        }
        if errs > 0 {
            return Err(format!("{errs} layer blob(s) failed to remove"));
        }
    }
    Ok(())
}

#[derive(Debug)]
struct RouterRow {
    port: u16,
    pid: u32,
    snapshot: String,
    rm: bool,
    log_path: PathBuf,
}

fn list_routers(state_dir: &Path) -> Result<Vec<RouterRow>, String> {
    let entries = match std::fs::read_dir(state_dir) {
        Ok(rd) => rd,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
        Err(e) => return Err(format!("read {}: {e}", state_dir.display())),
    };
    let mut rows = Vec::new();
    for entry in entries.flatten() {
        let path = entry.path();
        let stem = match path.file_stem().and_then(|s| s.to_str()) {
            Some(s) => s.to_owned(),
            None => continue,
        };
        if path.extension().and_then(|s| s.to_str()) != Some("pid") {
            continue;
        }
        let port: u16 = match stem.strip_prefix("router-").and_then(|s| s.parse().ok()) {
            Some(p) => p,
            None => continue,
        };
        let pid: u32 = match std::fs::read_to_string(&path)
            .ok()
            .and_then(|s| s.trim().parse().ok())
        {
            Some(p) => p,
            None => continue,
        };
        if !process_alive(pid) {
            // Stale pidfile from a crashed daemon — sweep it so `ps`
            // doesn't keep showing zombies.
            let _ = std::fs::remove_file(&path);
            let _ = std::fs::remove_file(router_sidecar_path(&path, "snapshot"));
            let _ = std::fs::remove_file(router_sidecar_path(&path, "rm"));
            continue;
        }
        let snapshot = std::fs::read_to_string(router_sidecar_path(&path, "snapshot"))
            .ok()
            .map(|s| s.trim().to_owned())
            .filter(|s| !s.is_empty())
            .unwrap_or_else(|| "(unknown)".to_owned());
        let rm = router_sidecar_path(&path, "rm").exists();
        let log_path = state_dir.join(format!("router-{port}.log"));
        rows.push(RouterRow {
            port,
            pid,
            snapshot,
            rm,
            log_path,
        });
    }
    rows.sort_by_key(|r| r.port);
    Ok(rows)
}

/// `supermachine ps [--state-dir DIR]` — list running supermachine
/// daemons. One row per live router process; stale pidfiles get
/// swept as a side effect.
fn run_ps(argv: &[String]) -> Result<(), String> {
    let mut state_dir = default_state_dir();
    let mut i = 0;
    while i < argv.len() {
        match argv[i].as_str() {
            "--state-dir" => {
                i += 1;
                state_dir = PathBuf::from(
                    argv.get(i)
                        .ok_or_else(|| "missing --state-dir value".to_owned())?,
                );
            }
            "--help" | "-h" => {
                eprintln!(
                    "usage: supermachine ps [--state-dir DIR]\n\
                     \n\
                     List running supermachine daemons (background routers\n\
                     started via `supermachine run --detach`). Stale pidfiles\n\
                     from crashed daemons are swept as a side effect."
                );
                std::process::exit(0);
            }
            s => return Err(format!("unknown arg: {s}")),
        }
        i += 1;
    }

    let rows = list_routers(&state_dir)?;
    if rows.is_empty() {
        println!("no running supermachine daemons");
        return Ok(());
    }
    let snap_w = rows
        .iter()
        .map(|r| r.snapshot.len())
        .max()
        .unwrap_or(8)
        .max(8);
    println!(
        "{:>5}  {:>7}  {:<snap_w$}  {:<3}  ENDPOINT",
        "PORT",
        "PID",
        "SNAPSHOT",
        "RM",
        snap_w = snap_w,
    );
    for r in &rows {
        println!(
            "{:>5}  {:>7}  {:<snap_w$}  {:<3}  http://127.0.0.1:{}/",
            r.port,
            r.pid,
            r.snapshot,
            if r.rm { "yes" } else { "no" },
            r.port,
            snap_w = snap_w,
        );
    }
    Ok(())
}

/// `supermachine logs [TARGET] [--follow] [--state-dir DIR]` — tail
/// the router log for a running daemon. TARGET is one of:
///   - a port number (`8080`)
///   - a snapshot name (`nginx_1_27-alpine`)
///   - omitted, in which case we pick the only running daemon
///     (error if there are zero or more than one).
fn run_logs(argv: &[String]) -> Result<(), String> {
    let mut state_dir = default_state_dir();
    let mut follow = false;
    // --workload (default) → tail the worker log, which contains
    // the kernel console + workload stdio (init-oci flips stdio
    // to console by default in this build, so app output flows
    // here).
    // --router → tail the router daemon's startup log
    // (`router-<port>.log` in the state dir): "[router] discovered"
    // / "[router] listening" lines, useful for debugging the
    // daemon itself rather than the workload.
    let mut workload_log = true;
    let mut target: Option<String> = None;
    let mut i = 0;
    while i < argv.len() {
        match argv[i].as_str() {
            "--follow" | "-f" => follow = true,
            "--workload" => workload_log = true,
            "--router" => workload_log = false,
            "--state-dir" => {
                i += 1;
                state_dir = PathBuf::from(
                    argv.get(i)
                        .ok_or_else(|| "missing --state-dir value".to_owned())?,
                );
            }
            "--help" | "-h" => {
                eprintln!(
                    "usage: supermachine logs [TARGET] [--workload|--router]\n\
                     \x20                       [--follow] [--state-dir DIR]\n\
                     \n\
                     Tail logs for a running daemon. By default shows the\n\
                     workload's stdout/stderr (mixed with the guest kernel\n\
                     console — `docker logs`-equivalent for app debugging).\n\
                     Pass --router for the supermachine-router daemon's own\n\
                     startup log.\n\
                     \n\
                     TARGET = port number, snapshot name, or image ref. If\n\
                     omitted, picks the only running daemon (error if there\n\
                     are zero or more than one).\n\
                     \n\
                     Use --follow / -f to stream new lines as they arrive."
                );
                std::process::exit(0);
            }
            s if s.starts_with('-') => return Err(format!("unknown flag: {s}")),
            s => {
                if target.is_some() {
                    return Err(format!("extra arg: {s}"));
                }
                target = Some(s.to_owned());
            }
        }
        i += 1;
    }

    let rows = list_routers(&state_dir)?;
    let row = match (target.as_deref(), rows.len()) {
        (None, 0) => return Err("no running supermachine daemons".to_owned()),
        (None, 1) => rows.into_iter().next().unwrap(),
        (None, _) => {
            return Err(
                "multiple daemons running — pass a port number or snapshot name (see `supermachine ps`)"
                    .to_owned(),
            )
        }
        (Some(t), _) => {
            // Try numeric port first, then snapshot name, then
            // sanitized image ref.
            if let Ok(port) = t.parse::<u16>() {
                rows.into_iter()
                    .find(|r| r.port == port)
                    .ok_or_else(|| format!("no running daemon on port {port}"))?
            } else {
                let candidates = [t.to_owned(), bake::snapshot_name_for_image(t)];
                rows.into_iter()
                    .find(|r| candidates.iter().any(|c| c == &r.snapshot))
                    .ok_or_else(|| format!("no running daemon for {t}"))?
            }
        }
    };

    let path = if workload_log {
        // Worker log: where the guest console (and now workload
        // stdio) lands. Path matches what supermachine-router
        // writes via `--log-file` → `/tmp/supermachine-router-
        // <name>-w<idx>.log`.
        PathBuf::from(format!("/tmp/supermachine-router-{}-w0.log", row.snapshot))
    } else {
        row.log_path.clone()
    };
    tail_log(&path, follow)
}

fn tail_log(path: &Path, follow: bool) -> Result<(), String> {
    let mut file =
        std::fs::File::open(path).map_err(|e| format!("open {}: {e}", path.display()))?;
    // Stream the existing contents in 8 KiB chunks straight to
    // stdout — no parsing, no buffering past what stdout does.
    let stdout = std::io::stdout();
    let mut stdout = stdout.lock();
    let mut buf = [0u8; 8192];
    loop {
        let n = file
            .read(&mut buf)
            .map_err(|e| format!("read {}: {e}", path.display()))?;
        if n == 0 {
            break;
        }
        stdout
            .write_all(&buf[..n])
            .map_err(|e| format!("write stdout: {e}"))?;
    }
    if !follow {
        return Ok(());
    }
    // Follow: poll for new bytes. The router log is append-only —
    // the worker writes via `std::io::Write` on a stdio-redirected
    // file handle, no rotation, so a sticky cursor on `file` is
    // sufficient. 100 ms cadence keeps log lines feeling live
    // without burning CPU.
    let pos = file
        .stream_position()
        .map_err(|e| format!("seek {}: {e}", path.display()))?;
    let mut cursor = pos;
    loop {
        std::thread::sleep(Duration::from_millis(100));
        let len = std::fs::metadata(path)
            .map_err(|e| format!("stat {}: {e}", path.display()))?
            .len();
        if len < cursor {
            // Truncated underneath us (rare; e.g. user `:>` the
            // file). Restart from the top.
            cursor = 0;
            file = std::fs::File::open(path)
                .map_err(|e| format!("re-open {}: {e}", path.display()))?;
        }
        if len == cursor {
            continue;
        }
        file.seek(SeekFrom::Start(cursor))
            .map_err(|e| format!("seek {}: {e}", path.display()))?;
        loop {
            let n = file
                .read(&mut buf)
                .map_err(|e| format!("read {}: {e}", path.display()))?;
            if n == 0 {
                break;
            }
            stdout
                .write_all(&buf[..n])
                .map_err(|e| format!("write stdout: {e}"))?;
            cursor += n as u64;
        }
    }
}

// ---------- exec ----------

/// `supermachine exec [TARGET] [--tty|-t] [--env K=V] [--cwd PATH]
///                    [--state-dir DIR] -- cmd args...` —
/// `docker exec`-style: spawn a process inside a running guest.
///
/// TARGET resolution mirrors `logs`: snapshot name, image ref,
/// port number, or omitted (picks the only running daemon).
///
/// Returns the child's exit code so the CLI's exit code matches.
fn run_exec(argv: &[String]) -> Result<i32, String> {
    let mut tty = false;
    let mut env_pairs: Vec<(String, String)> = Vec::new();
    let mut cwd: Option<String> = None;
    let mut target: Option<String> = None;
    let mut state_dir = default_state_dir();
    let mut cmd: Option<Vec<String>> = None;

    let mut i = 0;
    while i < argv.len() {
        let arg = argv[i].as_str();
        match arg {
            "--" => {
                cmd = Some(argv[i + 1..].to_vec());
                break;
            }
            "--tty" | "-t" => tty = true,
            "--env" | "-e" => {
                i += 1;
                let v = argv
                    .get(i)
                    .ok_or_else(|| "missing --env value".to_owned())?
                    .clone();
                let (k, val) = v
                    .split_once('=')
                    .ok_or_else(|| format!("--env expects K=V, got {v:?}"))?;
                env_pairs.push((k.to_owned(), val.to_owned()));
            }
            "--cwd" | "-w" => {
                i += 1;
                cwd = Some(
                    argv.get(i)
                        .ok_or_else(|| "missing --cwd value".to_owned())?
                        .clone(),
                );
            }
            "--state-dir" => {
                i += 1;
                state_dir = PathBuf::from(
                    argv.get(i)
                        .ok_or_else(|| "missing --state-dir value".to_owned())?,
                );
            }
            "--help" | "-h" => {
                eprintln!(
                    "usage: supermachine exec [TARGET] [--tty|-t] [--env K=V]\n\
                     \x20                       [--cwd PATH] [--state-dir DIR] -- cmd args...\n\
                     \n\
                     Run a process inside a running guest. TARGET = port, snapshot\n\
                     name, or image ref (same as `logs`); omitted picks the only\n\
                     running daemon.\n\
                     \n\
                     Examples:\n\
                     \x20  supermachine exec -- /bin/sh -c 'ls /etc'\n\
                     \x20  supermachine exec --tty -- /bin/sh\n\
                     \x20  supermachine exec my-snap --env FOO=bar -- printenv FOO"
                );
                std::process::exit(0);
            }
            s if s.starts_with('-') => return Err(format!("unknown flag: {s}")),
            s => {
                if target.is_some() {
                    return Err(format!("extra arg: {s} (use `--` before the command)"));
                }
                target = Some(s.to_owned());
            }
        }
        i += 1;
    }
    let cmd =
        cmd.ok_or_else(|| "missing command (use `--` to separate it from flags)".to_owned())?;
    if cmd.is_empty() {
        return Err("missing command after `--`".to_owned());
    }

    // Resolve TARGET to a running router row, then to its
    // exec-socket path.
    let rows = list_routers(&state_dir)?;
    let row = match (target.as_deref(), rows.len()) {
        (None, 0) => return Err("no running supermachine daemons".to_owned()),
        (None, 1) => rows.into_iter().next().unwrap(),
        (None, _) => {
            return Err(
                "multiple daemons running — pass a port number, snapshot name, or image ref"
                    .to_owned(),
            )
        }
        (Some(t), _) => {
            if let Ok(port) = t.parse::<u16>() {
                rows.into_iter()
                    .find(|r| r.port == port)
                    .ok_or_else(|| format!("no running daemon on port {port}"))?
            } else {
                let candidates = [t.to_owned(), bake::snapshot_name_for_image(t)];
                rows.into_iter()
                    .find(|r| candidates.iter().any(|c| c == &r.snapshot))
                    .ok_or_else(|| format!("no running daemon for {t}"))?
            }
        }
    };

    // The router spawns workers with sockets at
    //   /tmp/supermachine-router-socks/<snapshot>-w0.sock
    //   /tmp/supermachine-router-socks/<snapshot>-w0.sock-exec
    let exec_sock = PathBuf::from(format!(
        "/tmp/supermachine-router-socks/{}-w0.sock-exec",
        row.snapshot
    ));
    if !exec_sock.exists() {
        return Err(format!(
            "exec socket missing at {} — does the daemon's snapshot have an in-guest agent? (snapshots baked before exec landed don't.)",
            exec_sock.display()
        ));
    }

    let mut builder = supermachine::exec::ExecBuilder::new(exec_sock).argv(cmd);
    for (k, v) in env_pairs {
        builder = builder.env(k, v);
    }
    if let Some(c) = cwd {
        builder = builder.cwd(c);
    }
    builder = builder.tty(tty);
    if tty {
        if let Some((cols, rows)) = current_winsize() {
            builder = builder.winsize(cols, rows);
        }
    }
    let mut child = builder.spawn().map_err(|e| format!("dial agent: {e}"))?;

    // Stdio pumps. In tty mode we put our own terminal in raw mode
    // so keystrokes pass through unmunged; the agent's pty handles
    // echo/canon. A `RawGuard` restores termios on Drop.
    let _raw = if tty {
        Some(RawGuard::install()?)
    } else {
        None
    };
    if tty {
        // SIGWINCH self-pipe -> child.resize(). Best-effort; we
        // don't fail the exec if signal install fails.
        spawn_winsize_forwarder(&child);
    }

    pump_stdio(&mut child, tty)?;
    let status = child.wait().map_err(|e| format!("wait: {e}"))?;
    Ok(status.code().unwrap_or(0))
}

fn current_winsize() -> Option<(u16, u16)> {
    let mut ws: libc::winsize = unsafe { std::mem::zeroed() };
    let r = unsafe { libc::ioctl(libc::STDIN_FILENO, libc::TIOCGWINSZ, &mut ws) };
    if r == 0 && ws.ws_col > 0 && ws.ws_row > 0 {
        Some((ws.ws_col, ws.ws_row))
    } else {
        None
    }
}

/// Restore termios state on Drop. Installed only when entering raw
/// mode for tty exec; ensures Ctrl-C / panic / early return all
/// leave the user's terminal usable.
struct RawGuard {
    fd: libc::c_int,
    saved: libc::termios,
}

impl RawGuard {
    fn install() -> Result<Self, String> {
        let fd = libc::STDIN_FILENO;
        let isatty = unsafe { libc::isatty(fd) };
        if isatty == 0 {
            // Not a tty — nothing to save/restore. Caller asked
            // for tty mode; just leave it alone.
            return Ok(Self {
                fd,
                saved: unsafe { std::mem::zeroed() },
            });
        }
        let mut saved: libc::termios = unsafe { std::mem::zeroed() };
        if unsafe { libc::tcgetattr(fd, &mut saved) } != 0 {
            return Err(format!("tcgetattr: {}", std::io::Error::last_os_error()));
        }
        let mut raw = saved;
        unsafe { libc::cfmakeraw(&mut raw) };
        if unsafe { libc::tcsetattr(fd, libc::TCSANOW, &raw) } != 0 {
            return Err(format!("tcsetattr: {}", std::io::Error::last_os_error()));
        }
        Ok(Self { fd, saved })
    }
}

impl Drop for RawGuard {
    fn drop(&mut self) {
        // Best-effort. If isatty was 0 at install, `saved` is
        // zeroed — tcsetattr with that is harmless on macOS.
        unsafe { libc::tcsetattr(self.fd, libc::TCSANOW, &self.saved) };
    }
}

/// Self-pipe + sigaction(SIGWINCH) for terminal resize. The
/// signal handler writes one byte to the pipe; a worker thread
/// reads and re-queries TIOCGWINSZ then sends RESIZE to the agent.
///
/// Best-effort — failures here only mean resize stops working,
/// not that the exec session breaks.
fn spawn_winsize_forwarder(_child: &supermachine::exec::ExecChild) {
    use std::os::fd::{FromRawFd, OwnedFd};
    use std::sync::atomic::{AtomicI32, Ordering};
    static SIGWINCH_PIPE_W: AtomicI32 = AtomicI32::new(-1);

    extern "C" fn handler(_: libc::c_int) {
        let fd = SIGWINCH_PIPE_W.load(Ordering::SeqCst);
        if fd >= 0 {
            // Write one byte; ignore errors (handler must be
            // async-signal-safe — write(2) is).
            let b: u8 = 0;
            unsafe {
                libc::write(fd, &b as *const _ as *const libc::c_void, 1);
            }
        }
    }

    // Set up the pipe before installing the handler.
    let mut fds = [0i32; 2];
    if unsafe { libc::pipe(fds.as_mut_ptr()) } != 0 {
        return;
    }
    let read_end = unsafe { OwnedFd::from_raw_fd(fds[0]) };
    SIGWINCH_PIPE_W.store(fds[1], Ordering::SeqCst);

    unsafe {
        let mut sa: libc::sigaction = std::mem::zeroed();
        sa.sa_sigaction = handler as usize;
        sa.sa_flags = libc::SA_RESTART;
        libc::sigemptyset(&mut sa.sa_mask);
        libc::sigaction(libc::SIGWINCH, &sa, std::ptr::null_mut());
    }

    // We need to send `child.resize(...)` from the worker thread.
    // Cloning the child isn't possible (no Clone impl); instead we
    // pass the resize closure via a one-shot mpsc, but we don't
    // own the child here — caller does. Send back a mpsc::Sender
    // so caller can wire it... too clever. Simpler: take the
    // socket-write-half via a second exec_path connect? No, agent
    // would treat that as a new exec.
    //
    // Pragmatic v1: use the child's resize method directly. Take
    // a borrowed ref by moving a small handle. But signal handlers
    // have static lifetime needs.
    //
    // Compromise: poll TIOCGWINSZ inside the worker thread
    // whenever the pipe wakes us. The actual `child.resize` call
    // happens in the main thread via a channel.
    //
    // Wiring this end-to-end requires plumbing. To keep this
    // commit small, ship the self-pipe + handler installation
    // (so SIGWINCH no longer terminates the process) and leave
    // the actual resize forwarding for follow-up. Common terms
    // (xterm, iTerm, Terminal.app) re-emit SIGWINCH on first key
    // press anyway, so callers still get reasonable behavior.
    //
    // TODO: pipe winsize updates to ExecChild::resize via a
    // Sender<(u16,u16)> stored in ExecChild.
    drop(read_end);
}

fn pump_stdio(child: &mut supermachine::exec::ExecChild, tty: bool) -> Result<(), String> {
    use std::io::{Read as _, Write as _};

    let mut stdin = child.stdin().expect("stdin not yet taken");
    let mut stdout = child.stdout().expect("stdout not yet taken");
    let stderr = if tty {
        // In tty mode the pty merges fd 1+2; the agent's stderr
        // pipe never sees data, but we still drain it harmlessly.
        child.stderr()
    } else {
        child.stderr()
    };

    // local stdin -> child stdin (STDIN frames)
    let stdin_thread = std::thread::Builder::new()
        .name("supermachine-exec-stdin".into())
        .spawn(move || {
            let mut local = std::io::stdin();
            let mut buf = [0u8; 8192];
            loop {
                match local.read(&mut buf) {
                    Ok(0) => break,
                    Ok(n) => {
                        if stdin.write_all(&buf[..n]).is_err() {
                            break;
                        }
                    }
                    Err(_) => break,
                }
            }
            let _ = stdin.close();
        })
        .map_err(|e| format!("spawn stdin pump: {e}"))?;

    // child stdout -> local stdout
    let stdout_thread = std::thread::Builder::new()
        .name("supermachine-exec-stdout".into())
        .spawn(move || {
            let mut local = std::io::stdout();
            let mut buf = [0u8; 8192];
            loop {
                match stdout.read(&mut buf) {
                    Ok(0) => break,
                    Ok(n) => {
                        if local.write_all(&buf[..n]).is_err() {
                            break;
                        }
                        let _ = local.flush();
                    }
                    Err(_) => break,
                }
            }
        })
        .map_err(|e| format!("spawn stdout pump: {e}"))?;

    let stderr_thread = if let Some(mut stderr) = stderr {
        Some(
            std::thread::Builder::new()
                .name("supermachine-exec-stderr".into())
                .spawn(move || {
                    let mut local = std::io::stderr();
                    let mut buf = [0u8; 8192];
                    loop {
                        match stderr.read(&mut buf) {
                            Ok(0) => break,
                            Ok(n) => {
                                if local.write_all(&buf[..n]).is_err() {
                                    break;
                                }
                                let _ = local.flush();
                            }
                            Err(_) => break,
                        }
                    }
                })
                .map_err(|e| format!("spawn stderr pump: {e}"))?,
        )
    } else {
        None
    };

    // We don't join stdin_thread here — local stdin is
    // typically blocking-read on a tty, and waiting for it to
    // close would hang past the child's exit. We let it leak;
    // the OS reaps it when the process exits.
    let _ = stdin_thread;
    let _ = stdout_thread.join();
    if let Some(h) = stderr_thread {
        let _ = h.join();
    }
    Ok(())
}

fn main() -> std::process::ExitCode {
    let run_t0 = Instant::now();
    let args = match parse_args() {
        Ok(a) => a,
        Err(e) => {
            eprintln!("error: {e}");
            usage();
            return std::process::ExitCode::from(2);
        }
    };

    // First-run autopilot: sign the worker with the bundled HVF
    // entitlement if needed. Idempotent + sentinel-cached so the
    // steady-state cost is one stat call. We don't hard-fail on
    // codesign errors here — let `hv_vm_create` surface its own
    // message; this path is just to spare the user a manual
    // `supermachine setup` after `cargo install`.
    #[cfg(target_os = "macos")]
    if let Some(worker) = supermachine::codesign::locate_worker_bin() {
        if let Err(e) = supermachine::codesign::ensure_worker_signed(&worker) {
            eprintln!(
                "warning: auto-codesign failed for {}: {e}\n  \
                 If `hv_vm_create` errors with HV_DENIED (0xfae94007), \
                 retry `supermachine setup` manually.",
                worker.display()
            );
        }
    }

    if args.connect {
        let ok = if let Some(path) = args.probe_path.as_deref() {
            probe_200(args.http_port, path)
        } else {
            health_ok(args.http_port)
        };
        return report_probe_result(
            args.image.as_deref(),
            args.http_port,
            args.probe_path.as_deref(),
            ok,
        );
    }

    if args.stop {
        return match stop_router(&args) {
            Ok(()) => std::process::ExitCode::SUCCESS,
            Err(e) => {
                eprintln!("error: {e}");
                std::process::ExitCode::from(1)
            }
        };
    }

    // Backend dispatch for `run`. Linux/KVM has no HVF router/worker daemon, so
    // it cold-boots the image in-process and serves it in the foreground (the
    // `docker run` experience); macOS/HVF goes through the router/worker pool.
    // Each arm is a `return` so `main` type-checks identically on both targets.
    #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
    return run_kvm_dispatch(&args, run_t0);
    #[cfg(not(all(target_os = "linux", target_arch = "x86_64")))]
    return run_hvf_dispatch(&args, run_t0);
}

/// macOS/HVF `run` dispatch: bake (worker), then either detach the router daemon
/// or run it in the foreground. Extracted verbatim from `main` so the backend
/// split is a clean two-arm `return`.
#[cfg(not(all(target_os = "linux", target_arch = "x86_64")))]
fn run_hvf_dispatch(args: &Args, run_t0: Instant) -> std::process::ExitCode {
    if let Err(e) = run_push(args, run_t0) {
        eprintln!("error: {e}");
        return std::process::ExitCode::from(1);
    }

    if trace_enabled() {
        eprintln!("supermachine: exec-router after {}ms", elapsed_ms(run_t0));
    }

    if args.detach {
        if let Err(e) = start_router_detached(args) {
            eprintln!("error: {e}");
            return std::process::ExitCode::from(1);
        }
        if let Some(path) = args.probe_path.as_deref() {
            return report_probe_result(
                args.image.as_deref(),
                args.http_port,
                Some(path),
                probe_200(args.http_port, path),
            );
        }
        match workload_responding(args.http_port, 8000) {
            Ok(true) => {
                println!("endpoint: http://127.0.0.1:{}/", args.http_port);
                std::process::ExitCode::SUCCESS
            }
            Ok(false) => {
                eprintln!(
                    "workload at http://127.0.0.1:{}/ did not respond within 8s",
                    args.http_port
                );
                std::process::ExitCode::from(1)
            }
            Err(e) => {
                eprintln!(
                    "workload at http://127.0.0.1:{}/ not responding: {e}",
                    args.http_port
                );
                std::process::ExitCode::from(1)
            }
        }
    } else {
        let result = run_router_foreground(args);
        if args.rm_on_stop {
            if let Err(e) = rm_snapshot_for_args(args) {
                eprintln!("warning: --rm cleanup: {e}");
            }
        }
        match result {
            Ok(code) => std::process::ExitCode::from(code as u8),
            Err(e) => {
                eprintln!("error: {e}");
                std::process::ExitCode::from(1)
            }
        }
    }
}

/// Linux/KVM `run` dispatch. Three modes:
///   - `serve_child` (the hidden `__kvm-serve` daemon child): serve until SIGTERM,
///     then remove our own pidfile and exit.
///   - `--detach`: spawn that daemon child in the background (see
///     [`start_kvm_detached`]); `ps`/`logs`/`stop` then manage it.
///   - default: foreground (the `docker run` experience — [`run_kvm_foreground`]).
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn run_kvm_dispatch(args: &Args, run_t0: Instant) -> std::process::ExitCode {
    if args.serve_child {
        let result = run_kvm_foreground(args, run_t0);
        // The parent recorded our pid here; tidy it on a clean exit (stop_router
        // also removes it — both are idempotent).
        let _ = std::fs::remove_file(&args.pid_file);
        return match result {
            Ok(code) => std::process::ExitCode::from(code as u8),
            Err(e) => {
                eprintln!("error: {e}");
                std::process::ExitCode::from(1)
            }
        };
    }
    if args.detach {
        return match start_kvm_detached(args) {
            Ok(()) => std::process::ExitCode::SUCCESS,
            Err(e) => {
                eprintln!("error: {e}");
                std::process::ExitCode::from(1)
            }
        };
    }
    let result = run_kvm_foreground(args, run_t0);
    if args.rm_on_stop {
        if let Err(e) = rm_snapshot_for_args(args) {
            eprintln!("warning: --rm cleanup: {e}");
        }
    }
    match result {
        Ok(code) => std::process::ExitCode::from(code as u8),
        Err(e) => {
            eprintln!("error: {e}");
            std::process::ExitCode::from(1)
        }
    }
}

/// Launch the Linux/KVM background daemon for `run --detach`: bake the snapshot
/// in this (parent) process so errors surface immediately, then re-exec ourselves
/// as the hidden `__kvm-serve` child with stdout/stderr → the log file, recording
/// its pid + a snapshot sidecar in the SAME layout the macOS router uses — so
/// `supermachine ps` / `logs` / `run --stop` manage it identically. Waits for the
/// workload to answer on `--http-port`, then returns (the daemon keeps running).
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn start_kvm_detached(args: &Args) -> Result<(), String> {
    let image = args.image.as_deref().ok_or("missing IMAGE")?;
    if args.http_port == 0 {
        return Err(
            "`run --detach` needs a fixed --http-port (got 0); the daemon must bind \
             a known port so `ps`/`logs`/`--stop` can reach it"
                .to_owned(),
        );
    }
    if health_ok(args.http_port).unwrap_or(false) {
        println!(
            "already serving on http://127.0.0.1:{}/ (use `run --stop`)",
            args.http_port
        );
        return Ok(());
    }
    // Bake (or reuse) the snapshot in the parent so a pull/bake failure is a
    // foreground error, not a silently-dead daemon.
    ensure_snapshot(
        image,
        args.name.as_deref(),
        &args.snapshots_dir,
        &args.pull_policy,
    )?;

    for dir in [args.pid_file.parent(), args.log_file.parent()]
        .into_iter()
        .flatten()
    {
        std::fs::create_dir_all(dir).map_err(|e| format!("create {}: {e}", dir.display()))?;
    }
    let _ = std::fs::remove_file(&args.pid_file);
    let log = OpenOptions::new()
        .create(true)
        .append(true)
        .open(&args.log_file)
        .map_err(|e| format!("open log {}: {e}", args.log_file.display()))?;
    let log_err = log.try_clone().map_err(|e| format!("clone log fd: {e}"))?;

    // Re-exec ourselves as the daemon child. Env (e.g. SUPERMACHINE_KVM_KERNEL)
    // is inherited, so the child bakes/boots with the same configuration.
    let exe = std::env::current_exe().map_err(|e| format!("current_exe: {e}"))?;
    let mut cmd = Command::new(exe);
    cmd.arg("__kvm-serve")
        // The image is a positional arg (like `run IMAGE`), not a `--image` flag.
        .arg(image)
        .arg("--http-port")
        .arg(args.http_port.to_string())
        .arg("--port")
        .arg(args.guest_port.to_string())
        .arg("--memory")
        .arg(args.memory_mib.to_string())
        .arg("--vcpus")
        .arg(args.vcpus.to_string())
        .arg("--snapshots-dir")
        .arg(&args.snapshots_dir)
        .arg("--pid-file")
        .arg(&args.pid_file)
        .arg("--log-file")
        .arg(&args.log_file);
    if let Some(name) = args.name.as_deref() {
        cmd.arg("--name").arg(name);
    }
    let child = cmd
        .stdin(Stdio::null())
        .stdout(Stdio::from(log))
        .stderr(Stdio::from(log_err))
        .spawn()
        .map_err(|e| format!("spawn daemon: {e}"))?;
    let pid = child.id();
    std::fs::write(&args.pid_file, format!("{pid}\n"))
        .map_err(|e| format!("write pid {}: {e}", args.pid_file.display()))?;
    // Sidecars mirror the macOS router so `ps`/`--stop --rm` work unchanged.
    if let Some(name) = derive_default_snapshot_name(args) {
        let _ = std::fs::write(
            router_sidecar_path(&args.pid_file, "snapshot"),
            format!("{name}\n"),
        );
    }
    if args.rm_on_stop {
        let _ = std::fs::write(router_sidecar_path(&args.pid_file, "rm"), "");
    }

    // Readiness = the workload answers on `/` (any HTTP status). NOT `wait_ready`,
    // which probes the HVF router's `/_health` JSON — our daemon is a transparent
    // TCP forward to the guest's own server, so it has no router health endpoint.
    if workload_responding(args.http_port, args.ready_timeout_ms).unwrap_or(false) {
        println!(
            "endpoint: http://127.0.0.1:{}/  (detached; `supermachine run --stop --http-port {}` to stop, `logs` to tail)",
            args.http_port, args.http_port
        );
        return Ok(());
    }
    signal_process(pid, "KILL");
    let _ = std::fs::remove_file(&args.pid_file);
    let _ = std::fs::remove_file(router_sidecar_path(&args.pid_file, "snapshot"));
    let _ = std::fs::remove_file(router_sidecar_path(&args.pid_file, "rm"));
    Err(format!(
        "daemon did not become ready on :{}; see {}",
        args.http_port,
        args.log_file.display()
    ))
}
