supermachine/

exec.rs

//! Host-side framing for the in-guest exec agent.
//!
//! Mirrors the wire protocol implemented in
//! `crates/supermachine-guest-agent`. See
//! `docs/design/exec-2026-05-03.md` for the protocol overview.
//!
//! Public surface (re-exported from the crate root):
//!
//! - [`crate::Vm::exec`] / [`crate::Vm::exec_builder`] (in `api.rs`)
//! - [`ExecBuilder`] — chainable spawn config
//! - [`ExecChild`] — handle to a running guest process
//! - [`ExecStdin`] / [`ExecStdout`] / [`ExecStderr`] — std-style stdio
//!
//! Internals:
//!
//! - `spawn` dials the host-side `<vsock_mux>-exec.sock`, sends
//!   the REQUEST frame, then spawns a demux thread that reads
//!   frames and pumps STDOUT/STDERR bytes into pipe pairs (so the
//!   caller's [`std::io::Read`] just works), and posts the EXIT
//!   status onto a channel.

use std::collections::BTreeMap;
use std::io::{self, Read, Write};
use std::net::Shutdown;
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
use std::os::unix::ffi::OsStrExt;
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::UnixStream;
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
use std::os::unix::process::CommandExt;
use std::path::Path;
use std::path::PathBuf;
use std::process::ExitStatus;
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
use std::process::{ChildStdin, Command, Stdio};
use std::sync::atomic::AtomicBool;
use std::sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender, TryRecvError};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
use serde::Deserialize;
use serde::Serialize;

// Wire frame types — must match the agent's constants.
const FRAME_REQUEST: u8 = 0xff;
const FRAME_CONTROL: u8 = 0xfe;
const FRAME_STDIN: u8 = 0;
const FRAME_STDOUT: u8 = 1;
const FRAME_STDERR: u8 = 2;
const FRAME_RESIZE: u8 = 3;
const FRAME_SIGNAL: u8 = 4;
const FRAME_EXIT: u8 = 5;
const FRAME_ERROR: u8 = 6;

/// Configurable spawn for [`crate::Vm::exec`]. Built via
/// [`crate::Vm::exec_builder`] or constructed inline by `Vm::exec`.
///
/// ```no_run
/// # use supermachine::{Image, Vm, VmConfig};
/// let image = Image::from_snapshot("path/to/snapshot")?;
/// let vm = Vm::start(&image, &VmConfig::new())?;
/// let mut child = vm.exec_builder()
///     .argv(["sh", "-c", "echo hi && exit 7"])
///     .env("FOO", "bar")
///     .spawn()?;
/// let status = child.wait()?;
/// assert_eq!(status.code(), Some(7));
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// ```
#[derive(Clone)]
pub struct ExecBuilder {
    exec_path: PathBuf,
    argv: Vec<String>,
    env: BTreeMap<String, String>,
    cwd: Option<String>,
    tty: bool,
    cols: Option<u16>,
    rows: Option<u16>,
    timeout: Option<Duration>,
    /// Files to atomically write inside the guest before exec.
    /// Eliminates a separate `write_file` round-trip.
    stage_files: Vec<StageFile>,
    /// Additional argvs to run sequentially after `argv` succeeds
    /// (`&&` semantics). Stops on first non-zero exit. Removes the
    /// need for a `sh -c "X && Y"` wrapper and its extra fork.
    chain: Vec<Vec<String>>,
    /// No-virt sentry WARM-DAEMON route: when set (by [`crate::Vm::exec_builder`] on
    /// a warm-daemon-pool-backed `Vm`), [`ExecBuilder::output`] runs `argv` as a
    /// fresh CLIENT cell via `crate::sentry::Pool::exec_capture` — which reaches the
    /// detached warm daemon over the supervisor-global owned loopback — instead of
    /// dialing `exec_path` (a warm-daemon pooled `Vm` has no exec socket). `None` for
    /// every other backend / path.
    #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
    warm_pool: Option<std::sync::Arc<crate::sentry::Pool>>,
    /// External no-KVM runsc backend route. `Vm::exec_builder` sets this for
    /// runsc-backed VMs; `output()` executes via `runsc exec`, while `spawn()`
    /// bridges the same framed `ExecChild` API onto a live `runsc exec` child.
    #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
    runsc: Option<RunscExecTarget>,
    /// Native kboxlike backend route. `output()` and streaming `spawn()` execute
    /// by launching a fresh process under the backend rootfs.
    #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
    kboxlike: Option<KboxlikeExecTarget>,
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
#[derive(Clone, Debug, Serialize, Deserialize)]
#[doc(hidden)]
pub struct RunscExecTarget {
    pub runsc_bin: PathBuf,
    pub runsc_root: PathBuf,
    #[serde(default)]
    pub network: Option<String>,
    pub container_id: String,
    pub image_env: Vec<(String, String)>,
    #[serde(default)]
    pub agent_path: Option<String>,
    pub file_agent_rootfs: Option<PathBuf>,
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
#[derive(Clone, Debug, Serialize, Deserialize)]
#[doc(hidden)]
pub struct KboxlikeExecTarget {
    pub rootfs: PathBuf,
    pub image_env: Vec<(String, String)>,
    #[serde(default)]
    pub user: Option<(u32, u32)>,
    #[serde(default)]
    pub mounts: Vec<(PathBuf, String)>,
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub(crate) const RUNSC_EXEC_TOKEN_PREFIX: &str = "supermachine-runsc-exec-v1:";
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub(crate) const KBOXLIKE_EXEC_TOKEN_PREFIX: &str = "supermachine-kboxlike-exec-v1:";
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub(crate) const RUNSC_NETWORK_NONE: &str = "none";
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub(crate) const RUNSC_NETWORK_SANDBOX: &str = "sandbox";

#[derive(Clone)]
struct StageFile {
    path: String,
    data: Vec<u8>,
    mode: Option<u32>,
}

impl ExecBuilder {
    /// Construct a builder that dials a specific exec-socket
    /// path. Most embedders should use [`crate::Vm::exec_builder`]
    /// instead; this lower-level constructor is for tooling that
    /// already has a router daemon running and just wants to
    /// dial a known socket (e.g. the `supermachine exec` CLI).
    pub fn new(exec_path: PathBuf) -> Self {
        Self {
            exec_path,
            argv: Vec::new(),
            env: BTreeMap::new(),
            cwd: None,
            tty: false,
            cols: None,
            rows: None,
            timeout: None,
            stage_files: Vec::new(),
            chain: Vec::new(),
            #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
            warm_pool: None,
            #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
            runsc: None,
            #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
            kboxlike: None,
        }
    }

    /// Rebuild an exec builder from the opaque token returned by
    /// `Vm::exec_binding_token`. Plain socket-path tokens remain accepted for
    /// existing KVM/HVF/sentry callers; runsc tokens carry the host-control-plane
    /// target needed to execute via `runsc exec`.
    #[doc(hidden)]
    pub fn from_binding_token(token: impl AsRef<str>) -> io::Result<Self> {
        let token = token.as_ref();
        #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
        if let Some(raw) = token.strip_prefix(RUNSC_EXEC_TOKEN_PREFIX) {
            let bytes = crate::api::b64_decode(raw)
                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
            let target: RunscExecTarget = serde_json::from_slice(&bytes)
                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
            return Ok(Self::new(PathBuf::from("runsc-exec-token")).with_runsc_exec(target));
        }
        #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
        if let Some(raw) = token.strip_prefix(KBOXLIKE_EXEC_TOKEN_PREFIX) {
            let bytes = crate::api::b64_decode(raw)
                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
            let target: KboxlikeExecTarget = serde_json::from_slice(&bytes)
                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
            return Ok(Self::new(PathBuf::from("kboxlike-exec-token")).with_kboxlike_exec(target));
        }
        Ok(Self::new(PathBuf::from(token)))
    }

    /// Route this builder's [`output`](Self::output) through a no-virt sentry
    /// warm-daemon pool (`Pool::exec_capture`) instead of an exec socket. Set by
    /// [`crate::Vm::exec_builder`] for a warm-daemon-pool-backed `Vm`.
    #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
    pub(crate) fn with_warm_pool(mut self, pool: std::sync::Arc<crate::sentry::Pool>) -> Self {
        self.warm_pool = Some(pool);
        self
    }

    /// Route this builder's [`output`](Self::output) through `runsc exec` for the
    /// experimental external no-KVM backend.
    #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
    #[doc(hidden)]
    pub fn with_runsc_exec(mut self, target: RunscExecTarget) -> Self {
        self.runsc = Some(target);
        self
    }

    /// Route this builder's [`output`](Self::output) through the experimental
    /// native kboxlike backend.
    #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
    #[doc(hidden)]
    pub fn with_kboxlike_exec(mut self, target: KboxlikeExecTarget) -> Self {
        self.kboxlike = Some(target);
        self
    }

    /// The argv to execute in the guest. First element is the
    /// program (resolved via PATH inside the guest).
    pub fn argv<I, S>(mut self, argv: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.argv = argv.into_iter().map(Into::into).collect();
        self
    }

    /// Set an environment variable for the spawned process.
    /// Repeatable. The agent merges these on top of its own
    /// environment, so `PATH`, `HOME`, etc. are inherited unless
    /// you explicitly override them.
    pub fn env(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.env.insert(key.into(), value.into());
        self
    }

    /// Working directory for the spawned process inside the guest.
    pub fn cwd(mut self, path: impl Into<String>) -> Self {
        self.cwd = Some(path.into());
        self
    }

    /// Allocate a pseudo-terminal. stdin/stdout pass through the
    /// pty master; stderr is empty (the pty merges fd 1 + fd 2).
    /// Use this for interactive shells.
    pub fn tty(mut self, on: bool) -> Self {
        self.tty = on;
        self
    }

    /// Initial terminal size for tty mode. Has no effect without
    /// `.tty(true)`. Use [`ExecChild::resize`] to change later.
    pub fn winsize(mut self, cols: u16, rows: u16) -> Self {
        self.cols = Some(cols);
        self.rows = Some(rows);
        self
    }

    /// Hard wall-clock timeout. If the process is still running
    /// when the deadline fires, the watchdog sends SIGKILL and
    /// the resulting [`ExecOutcome`] has `timed_out = true`.
    /// Only honored by [`ExecBuilder::output`] (the streaming
    /// [`ExecBuilder::spawn`] path leaves cancellation to the
    /// caller, who already has [`ExecChild::signal`]).
    ///
    /// Reliable: `timed_out = true` is always set when the
    /// deadline fires.
    ///
    /// Process group: the agent runs `setpgid(0, 0)` in the
    /// child after fork, then forwards SIGNAL to the entire
    /// process group via `kill(-pid, sig)`. Forking children
    /// (e.g. `sh -c "rustc x.rs && /tmp/x"`) get killed
    /// transitively — no orphaned processes blocking the EXIT
    /// frame. Verified to take ~500 ms end-to-end on a 500 ms
    /// timeout across `sleep`, `sh -c sleep`, busy loops, and
    /// shell pipelines.
    pub fn timeout(mut self, d: Duration) -> Self {
        self.timeout = Some(d);
        self
    }

    /// Stage `bytes` at `path` inside the guest before exec runs.
    /// Atomic (write+rename). Folded into the same vsock RPC as
    /// the exec request — eliminates the separate `Vm::write_file`
    /// round-trip. Repeatable for multiple files.
    ///
    /// Use when your workload starts with a known-small input file
    /// (source code, a test config) — replace
    /// `vm.write_file(p, b)?; vm.exec_builder().argv([…]).output()?`
    /// with `vm.exec_builder().stage_file(p, b).argv([…]).output()?`.
    pub fn stage_file(mut self, path: impl Into<String>, bytes: impl Into<Vec<u8>>) -> Self {
        self.stage_files.push(StageFile {
            path: path.into(),
            data: bytes.into(),
            mode: None,
        });
        self
    }

    /// Like [`Self::stage_file`] but also sets the file mode.
    pub fn stage_file_mode(
        mut self,
        path: impl Into<String>,
        bytes: impl Into<Vec<u8>>,
        mode: u32,
    ) -> Self {
        self.stage_files.push(StageFile {
            path: path.into(),
            data: bytes.into(),
            mode: Some(mode),
        });
        self
    }

    /// Append a second (third, …) argv to run after the previous
    /// one succeeds (`&&` semantics). Stops on first non-zero
    /// exit. Removes the need for a `sh -c "X && Y"` wrapper —
    /// the agent forks each command directly in the guest.
    pub fn chain<I, S>(mut self, argv: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.chain.push(argv.into_iter().map(Into::into).collect());
        self
    }

    /// Dial the agent, send the REQUEST, and return the
    /// [`ExecChild`] handle. Use this when you want to stream
    /// stdin/stdout/stderr yourself; for a one-call "run, drain,
    /// collect" pattern use [`ExecBuilder::output`].
    pub fn spawn(self) -> io::Result<ExecChild> {
        if self.argv.is_empty() {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "exec: argv is empty",
            ));
        }
        #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
        if self.runsc.is_some() {
            return runsc_spawn(self);
        }
        #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
        if self.kboxlike.is_some() {
            return kboxlike_spawn(self);
        }
        spawn(self)
    }

    /// Run to completion, collecting stdout + stderr + exit
    /// status into one [`ExecOutcome`]. Blocks until the process
    /// exits (or the configured [`ExecBuilder::timeout`] fires
    /// and the process is killed).
    ///
    /// Mirrors [`std::process::Command::output`]. Use this for
    /// the common "exec a command, look at the result" case;
    /// reach for [`ExecBuilder::spawn`] only when you need to
    /// stream stdio.
    ///
    /// ```no_run
    /// # use std::time::Duration;
    /// # use supermachine::{Image, Vm, VmConfig};
    /// let image = Image::from_snapshot("path/to/snapshot")?;
    /// let vm = Vm::start(&image, &VmConfig::new())?;
    /// let out = vm.exec_builder()
    ///     .argv(["sh", "-c", "echo hi; echo err >&2; exit 7"])
    ///     .timeout(Duration::from_secs(30))
    ///     .output()?;
    /// assert_eq!(out.status.code(), Some(7));
    /// assert_eq!(out.stdout, b"hi\n");
    /// assert_eq!(out.stderr, b"err\n");
    /// assert!(!out.timed_out);
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    pub fn output(self) -> io::Result<ExecOutcome> {
        // No-virt sentry warm-daemon route: run `argv` as a fresh CLIENT cell in the
        // daemon's supervisor (it reaches the warm daemon over the owned loopback).
        // `Pool::exec_capture` is blocking and returns (exit_code, combined_output);
        // bridge it onto an ExecOutcome (no exec socket, no streaming — the
        // daemon-client model is request/response, so TTY/stdin/timeout are N/A here).
        #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
        if let Some(pool) = &self.warm_pool {
            let started = std::time::Instant::now();
            // Send the FULL exec request (argv + env + cwd + stage_files + chain),
            // same JSON shape as the exec-socket path, so the daemon supervisor stages
            // files + applies per-exec env/cwd + chains — full feature parity (vs the
            // old argv-only route). Combined stdout+stderr; tty/streaming/stdin N/A on
            // the request/response daemon path.
            let stage_payload: Vec<StageFilePayload> = self
                .stage_files
                .iter()
                .map(|s| StageFilePayload {
                    path: &s.path,
                    data_b64: crate::api::b64_encode(&s.data),
                    mode: s.mode,
                })
                .collect();
            let payload = RequestPayload {
                argv: &self.argv,
                env: &self.env,
                cwd: self.cwd.as_deref(),
                tty: false,
                cols: None,
                rows: None,
                stage_files: stage_payload,
                chain: &self.chain,
            };
            let body = serde_json::to_vec(&payload)
                .map_err(|e| io::Error::other(format!("sentry exec request encode: {e}")))?;
            let (code, out) = pool
                .exec_capture_full(&body)
                .map_err(|e| io::Error::other(format!("sentry warm-daemon exec: {e}")))?;
            return Ok(ExecOutcome {
                status: synthesize_exit(code as u32),
                stdout: out,
                stderr: Vec::new(),
                duration: started.elapsed(),
                peak_rss_kib: None,
                timed_out: false,
            });
        }
        #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
        if self.runsc.is_some() {
            return runsc_output(self);
        }
        #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
        if self.kboxlike.is_some() {
            return kboxlike_output(self);
        }
        // `output()` and `output_with_cancel()` share the same body;
        // the only difference is that `output()` passes a Receiver
        // that never fires. Constructing one is essentially free
        // (a single mpsc allocation that immediately gets dropped
        // when this function returns).
        let (never_tx, never_rx) = channel::<()>();
        // Keep the sender alive for the lifetime of this call so
        // the channel doesn't "fire" via Disconnected — that would
        // be ambiguous with "user cancelled."
        let result = self.output_with_cancel(never_rx);
        drop(never_tx);
        result
    }

    /// Like [`output`] but transparently retries a connection-SETUP
    /// failure — the agent closed before the workload produced ANY output
    /// or EXIT (see [`ExecEof::is_setup_failure`]) — up to `retries` times
    /// with a short linear backoff, re-dialing a fresh connection each
    /// time. ONLY that case is retried: a mid-stream drop (output already
    /// delivered) and a timeout/cancel are returned unchanged, because a
    /// retry there could double-execute a side-effecting command.
    ///
    /// Semantics are at-least-once — a workload whose EXIT frame was lost
    /// *after* it already ran (rare) could run twice — so use this only for
    /// idempotent workloads. `retries == 0` is exactly [`output`] (no
    /// clone, no retry, the default).
    ///
    /// [`output`]: Self::output
    pub fn output_resilient(self, retries: u8) -> io::Result<ExecOutcome> {
        if retries == 0 {
            return self.output();
        }
        let mut attempt: u8 = 0;
        loop {
            // Clone so each attempt dials a fresh connection; `self` seeds
            // the clones and drops when the loop returns.
            let result = self.clone().output();
            let setup_failure = result
                .as_ref()
                .err()
                .and_then(|e| e.get_ref())
                .and_then(|e| e.downcast_ref::<ExecEof>())
                .map(ExecEof::is_setup_failure)
                .unwrap_or(false);
            if setup_failure && attempt < retries {
                attempt += 1;
                eprintln!(
                    "[exec] connection-setup EOF (agent not ready / worker gone) — retry \
                     {attempt}/{retries}"
                );
                thread::sleep(Duration::from_millis(50 * attempt as u64));
                continue;
            }
            return result;
        }
    }

    /// Like [`Self::output`] but additionally interrupts the running
    /// process if `cancel` receives a value (or is dropped). The
    /// interrupt path is best-effort `SIGKILL` via the agent
    /// followed by a host-side [`ExecChild::abort`] — so cancel
    /// works even if the worker / agent has already died.
    ///
    /// `timeout` (set via [`Self::timeout`]) and `cancel` are
    /// orthogonal: BOTH fire SIGKILL when their trigger arrives;
    /// whichever wins kills the child. `ExecOutcome::timed_out`
    /// reports whether the timeout specifically fired; the cancel
    /// path doesn't surface a distinguishing flag (the caller
    /// already knows they cancelled).
    ///
    /// Typical use:
    ///
    /// ```ignore
    /// let (cancel_tx, cancel_rx) = std::sync::mpsc::channel();
    /// std::thread::spawn(move || {
    ///     wait_for_user_ctrl_c();
    ///     let _ = cancel_tx.send(());
    /// });
    /// let outcome = vm.exec_builder(["./run.sh"])
    ///     .timeout(Duration::from_secs(60))
    ///     .output_with_cancel(cancel_rx)?;
    /// ```
    pub fn output_with_cancel(self, cancel: Receiver<()>) -> io::Result<ExecOutcome> {
        let timeout = self.timeout;
        let t0 = Instant::now();
        let mut child = self.spawn()?;
        // Close stdin immediately so workloads that block on
        // `read(0)` (cat, anything reading stdin) see EOF and
        // exit. Drop sends an empty STDIN frame to the agent;
        // the agent's input thread observes that as "EOF for
        // current child" without exiting, so SIGNAL frames from
        // our watchdog still reach the agent later.
        drop(child.stdin());
        let mut stdout = child.stdout();
        let mut stderr = child.stderr();
        // Drain both pipes in parallel so a chatty stderr never
        // backpressures stdout (or vice versa) and deadlocks.
        let stdout_handle = stdout.take().map(|mut s| {
            thread::spawn(move || {
                let mut buf = Vec::new();
                let _ = s.read_to_end(&mut buf);
                buf
            })
        });
        let stderr_handle = stderr.take().map(|mut s| {
            thread::spawn(move || {
                let mut buf = Vec::new();
                let _ = s.read_to_end(&mut buf);
                buf
            })
        });
        // Unified watchdog: waits for whichever fires first —
        //   * a stop-tx send (cooperative cleanup once wait returns)
        //   * a timeout (if configured)
        //   * a value on the user's cancel channel
        // …and on timeout or cancel sends SIGKILL via the agent,
        // then aborts the host-side socket so even a dead agent
        // can't keep us blocked.
        let (stop_tx, stop_rx) = channel::<()>();
        let timed_out = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let signaler = child.signaler();
        let watchdog = {
            let timed_out = Arc::clone(&timed_out);
            Some(thread::spawn(move || {
                // Two-phase escalation on cancel / timeout:
                //   1. Send SIGKILL via the agent (polite — gives
                //      the agent a chance to reap the child
                //      cleanly and post an EXIT frame so the caller
                //      gets a proper exit status, e.g. 128+9=137).
                //   2. Wait up to ABORT_GRACE for the wait-side to
                //      finish (stop_rx); if not, force-close the
                //      host-side socket (signaler.abort()) so the
                //      demux thread unblocks regardless of agent
                //      liveness.
                //
                // Phase 1 alone covers the happy path (test:
                // timeoutMs kills a sleep, expects exitCode=137).
                // Phase 2 covers the integrator's "agent/worker
                // died, host wait won't return" case.
                const TICK: Duration = Duration::from_millis(10);
                const ABORT_GRACE: Duration = Duration::from_secs(2);
                let deadline = timeout.map(|d| Instant::now() + d);

                // Trigger-and-grace: try the polite kill, then
                // wait up to ABORT_GRACE for the wait-side to
                // complete. If it doesn't, force-close.
                let escalate = |stop_rx: &Receiver<()>| {
                    let _ = signaler.kill();
                    match stop_rx.recv_timeout(ABORT_GRACE) {
                        Ok(()) | Err(RecvTimeoutError::Disconnected) => {
                            // Agent reaped child + wait returned —
                            // clean path. Skip the force-close.
                        }
                        Err(RecvTimeoutError::Timeout) => {
                            // Agent didn't reap within grace
                            // (likely dead / unresponsive). Force-
                            // close so demux unblocks and the
                            // caller doesn't hang.
                            let _ = signaler.abort();
                        }
                    }
                };

                loop {
                    // Stop signal (wait completed) — clean exit.
                    match stop_rx.try_recv() {
                        Ok(()) | Err(TryRecvError::Disconnected) => return,
                        Err(TryRecvError::Empty) => {}
                    }
                    // User cancel — escalate.
                    match cancel.try_recv() {
                        Ok(()) => {
                            escalate(&stop_rx);
                            return;
                        }
                        Err(TryRecvError::Disconnected) => {
                            // Cancel-tx dropped without firing —
                            // user gave up on cancellation. Keep
                            // polling for the stop signal.
                        }
                        Err(TryRecvError::Empty) => {}
                    }
                    // Timeout — escalate + mark.
                    if let Some(dl) = deadline {
                        if Instant::now() >= dl {
                            timed_out.store(true, std::sync::atomic::Ordering::SeqCst);
                            if crate::trace::enabled("timeout") {
                                eprintln!("[exec.timeout] fired at +{:?}", timeout.unwrap());
                            }
                            escalate(&stop_rx);
                            return;
                        }
                    }
                    // Sleep for either TICK or "until deadline,
                    // whichever is sooner" — keeps timeout latency
                    // tight without busy-looping.
                    let sleep_for = deadline
                        .map(|dl| dl.saturating_duration_since(Instant::now()).min(TICK))
                        .unwrap_or(TICK);
                    if sleep_for.is_zero() {
                        continue;
                    }
                    // Use recv_timeout so a stop_tx send wakes us
                    // immediately on the happy path.
                    match stop_rx.recv_timeout(sleep_for) {
                        Ok(()) => return,
                        Err(RecvTimeoutError::Timeout) => continue,
                        Err(RecvTimeoutError::Disconnected) => return,
                    }
                }
            }))
        };
        let exit = child.wait_with_rss();
        // Stop the watchdog (idempotent if it already exited).
        let _ = stop_tx.send(());
        if let Some(h) = watchdog {
            let _ = h.join();
        }
        let stdout_bytes = stdout_handle
            .map(|h| h.join().unwrap_or_default())
            .unwrap_or_default();
        let stderr_bytes = stderr_handle
            .map(|h| h.join().unwrap_or_default())
            .unwrap_or_default();
        let (status, peak_rss_kib) = exit?;
        Ok(ExecOutcome {
            status,
            stdout: stdout_bytes,
            stderr: stderr_bytes,
            duration: t0.elapsed(),
            peak_rss_kib,
            timed_out: timed_out.load(std::sync::atomic::Ordering::SeqCst),
        })
    }
}

/// Result of [`ExecBuilder::output`]. Mirrors the shape of
/// [`std::process::Output`] with two additions: `duration` (host-
/// side wall-clock) and `timed_out` (true iff the wall-clock
/// timeout fired and the process was SIGKILL'd).
///
/// `peak_rss_kib` is `None` today; a future supermachine-agent
/// rev will populate it via `getrusage(RUSAGE_CHILDREN)`.
#[derive(Debug)]
pub struct ExecOutcome {
    /// Exit status, exactly as [`std::process::Output::status`].
    pub status: ExitStatus,
    /// Captured stdout bytes (verbatim — caller decodes if needed).
    pub stdout: Vec<u8>,
    /// Captured stderr bytes (verbatim).
    pub stderr: Vec<u8>,
    /// Host-side wall-clock duration from [`ExecBuilder::output`]
    /// being called to the process exiting (or being killed).
    pub duration: Duration,
    /// Peak resident set size of the process and its children, in
    /// KiB. `None` until the guest agent gains rusage support.
    pub peak_rss_kib: Option<u64>,
    /// `true` iff the wall-clock timeout from
    /// [`ExecBuilder::timeout`] fired and the process was killed
    /// with SIGKILL.
    pub timed_out: bool,
}

impl ExecOutcome {
    /// `true` when the exit status is success AND the process
    /// wasn't forcibly killed by the timeout.
    pub fn success(&self) -> bool {
        self.status.success() && !self.timed_out
    }
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn runsc_output(builder: ExecBuilder) -> io::Result<ExecOutcome> {
    let target = builder.runsc.clone().ok_or_else(|| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            "runsc_output called without a runsc target",
        )
    })?;
    if builder.argv.is_empty() {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "exec: argv is empty",
        ));
    }
    if builder.tty {
        if target.agent_path.is_some() {
            return runsc_agent_output(builder, target);
        }
        return Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "runsc backend does not support TTY exec without a bundled agent",
        ));
    }
    if builder.tty || builder.cols.is_some() || builder.rows.is_some() {
        return Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "runsc backend does not support winsize without TTY",
        ));
    }
    runsc_stage_files(&target, &builder.stage_files)?;

    let started = Instant::now();
    let mut stdout = Vec::new();
    let mut stderr = Vec::new();
    let mut status = synthesize_exit(0);
    let mut timed_out = false;
    let mut argvs = Vec::with_capacity(1 + builder.chain.len());
    argvs.push(builder.argv.clone());
    argvs.extend(builder.chain.iter().cloned());

    for argv in argvs {
        let out = if target.file_agent_rootfs.is_some() {
            runsc_file_agent_output_one(&target, &builder, &argv, started)?
        } else {
            runsc_output_one(&target, &builder, &argv, started)?
        };
        timed_out |= out.timed_out;
        stdout.extend(out.stdout);
        stderr.extend(out.stderr);
        status = out.status;
        if timed_out || !status.success() {
            break;
        }
    }

    Ok(ExecOutcome {
        status,
        stdout,
        stderr,
        duration: started.elapsed(),
        peak_rss_kib: None,
        timed_out,
    })
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn runsc_agent_output(builder: ExecBuilder, target: RunscExecTarget) -> io::Result<ExecOutcome> {
    let started = Instant::now();
    runsc_stage_files(&target, &builder.stage_files)?;
    let agent_path = target.agent_path.as_deref().ok_or_else(|| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            "runsc agent exec target missing agent path",
        )
    })?;
    let mut cmd = Command::new(&target.runsc_bin);
    add_runsc_global_flags_with_network(&mut cmd, &target.runsc_root, runsc_network(&target));
    cmd.arg("exec")
        .arg(&target.container_id)
        .arg(agent_path)
        .arg("--exec-stdio")
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    let mut child = cmd.spawn().map_err(|e| {
        io::Error::new(
            e.kind(),
            format!(
                "runsc exec {} {} --exec-stdio: {e}",
                target.container_id, agent_path
            ),
        )
    })?;
    let stage_payload: Vec<StageFilePayload> = builder
        .stage_files
        .iter()
        .map(|s| StageFilePayload {
            path: &s.path,
            data_b64: crate::api::b64_encode(&s.data),
            mode: s.mode,
        })
        .collect();
    let payload = RequestPayload {
        argv: &builder.argv,
        env: &builder.env,
        cwd: builder.cwd.as_deref(),
        tty: builder.tty,
        cols: builder.cols,
        rows: builder.rows,
        stage_files: stage_payload,
        chain: &builder.chain,
    };
    let json = serde_json::to_vec(&payload).map_err(|e| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            format!("runsc agent exec: encode REQUEST: {e}"),
        )
    })?;
    {
        let stdin = child.stdin.as_mut().ok_or_else(|| {
            io::Error::new(io::ErrorKind::Other, "runsc agent exec stdin pipe missing")
        })?;
        write_raw_frame(stdin, FRAME_REQUEST, &json)?;
        write_raw_frame(stdin, FRAME_STDIN, &[])?;
    }

    let mut child_stdout = child.stdout.take().ok_or_else(|| {
        io::Error::new(io::ErrorKind::Other, "runsc agent exec stdout pipe missing")
    })?;
    let mut child_stderr = child.stderr.take().ok_or_else(|| {
        io::Error::new(io::ErrorKind::Other, "runsc agent exec stderr pipe missing")
    })?;
    let stderr_handle = thread::Builder::new()
        .name("supermachine-runsc-agent-output-stderr".into())
        .spawn(move || {
            let mut buf = Vec::new();
            let _ = child_stderr.read_to_end(&mut buf);
            buf
        })
        .map_err(|e| {
            io::Error::new(
                io::ErrorKind::Other,
                format!("runsc agent output stderr thread: {e}"),
            )
        })?;

    let mut stdout = Vec::new();
    let mut stderr = Vec::new();
    let mut status = synthesize_exit(0);
    let done = Arc::new(AtomicBool::new(false));
    let timed_out = Arc::new(AtomicBool::new(false));
    let watchdog = builder.timeout.map(|timeout| {
        let done = Arc::clone(&done);
        let timed_out = Arc::clone(&timed_out);
        let child_pid = child.id() as i32;
        thread::spawn(move || {
            let elapsed = started.elapsed();
            if elapsed < timeout {
                thread::sleep(timeout - elapsed);
            }
            if !done.load(std::sync::atomic::Ordering::SeqCst) {
                timed_out.store(true, std::sync::atomic::Ordering::SeqCst);
                let _ = unsafe { libc::kill(child_pid, libc::SIGKILL) };
            }
        })
    });

    loop {
        let (kind, payload) = match read_frame(&mut child_stdout) {
            Ok(frame) => frame,
            Err(e) => {
                if timed_out.load(std::sync::atomic::Ordering::SeqCst) {
                    break;
                }
                let _ = child.kill();
                let _ = child.wait();
                done.store(true, std::sync::atomic::Ordering::SeqCst);
                if let Some(h) = watchdog {
                    let _ = h.join();
                }
                let stderr = stderr_handle.join().unwrap_or_default();
                if !stderr.is_empty() {
                    return Err(io::Error::new(
                        e.kind(),
                        format!(
                            "runsc agent exec frame read failed: {e}; runsc stderr={}",
                            String::from_utf8_lossy(&stderr)
                        ),
                    ));
                }
                return Err(e);
            }
        };
        match kind {
            FRAME_STDOUT => stdout.extend_from_slice(&payload),
            FRAME_STDERR => stderr.extend_from_slice(&payload),
            FRAME_EXIT => {
                match decode_exit_payload(&payload) {
                    DemuxExit::Status { code, .. } => status = synthesize_exit(code),
                    DemuxExit::Error(msg) => {
                        let _ = child.kill();
                        let _ = child.wait();
                        done.store(true, std::sync::atomic::Ordering::SeqCst);
                        if let Some(h) = watchdog {
                            let _ = h.join();
                        }
                        let mut runsc_stderr = stderr_handle.join().unwrap_or_default();
                        stderr.append(&mut runsc_stderr);
                        return Err(io::Error::new(io::ErrorKind::InvalidData, msg));
                    }
                    _ => {}
                }
                break;
            }
            FRAME_ERROR => {
                let msg = String::from_utf8_lossy(&payload).into_owned();
                let _ = child.kill();
                let _ = child.wait();
                done.store(true, std::sync::atomic::Ordering::SeqCst);
                if let Some(h) = watchdog {
                    let _ = h.join();
                }
                let mut runsc_stderr = stderr_handle.join().unwrap_or_default();
                stderr.append(&mut runsc_stderr);
                return Err(io::Error::new(io::ErrorKind::Other, msg));
            }
            _ => {}
        }
    }
    drop(child.stdin.take());
    let host_status = child.wait()?;
    done.store(true, std::sync::atomic::Ordering::SeqCst);
    if let Some(h) = watchdog {
        let _ = h.join();
    }
    let mut runsc_stderr = stderr_handle.join().unwrap_or_default();
    stderr.append(&mut runsc_stderr);
    if timed_out.load(std::sync::atomic::Ordering::SeqCst) {
        status = synthesize_exit(runsc_status_code(host_status));
    }
    Ok(ExecOutcome {
        status,
        stdout,
        stderr,
        duration: started.elapsed(),
        peak_rss_kib: None,
        timed_out: timed_out.load(std::sync::atomic::Ordering::SeqCst),
    })
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn runsc_file_agent_output_one(
    target: &RunscExecTarget,
    builder: &ExecBuilder,
    argv: &[String],
    started: Instant,
) -> io::Result<ExecOutcome> {
    let rootfs = target.file_agent_rootfs.as_ref().ok_or_else(|| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            "runsc file-agent exec target missing rootfs",
        )
    })?;
    let req_dir = rootfs
        .join(".supermachine")
        .join("runsc-exec")
        .join("requests")
        .join(format!(
            "req-{}-{}",
            std::process::id(),
            unique_suffix_nanos()
        ));
    std::fs::create_dir_all(&req_dir)?;

    let mut script = String::from("set -e\n");
    if let Some(cwd) = builder.cwd.as_deref() {
        script.push_str("cd ");
        script.push_str(&sh_quote(cwd));
        script.push('\n');
    }
    let mut env = BTreeMap::new();
    for (k, v) in &target.image_env {
        env.insert(k.clone(), v.clone());
    }
    for (k, v) in &builder.env {
        env.insert(k.clone(), v.clone());
    }
    for (k, v) in env {
        script.push_str("export ");
        script.push_str(&k);
        script.push('=');
        script.push_str(&sh_quote(&v));
        script.push('\n');
    }
    script.push_str("exec");
    for arg in argv {
        script.push(' ');
        script.push_str(&sh_quote(arg));
    }
    script.push('\n');

    std::fs::write(req_dir.join("script"), script)?;
    std::fs::write(req_dir.join("ready"), b"1\n")?;

    let mut timed_out = false;
    loop {
        if req_dir.join("done").is_file() {
            break;
        }
        if let Some(timeout) = builder.timeout {
            if started.elapsed() >= timeout {
                timed_out = true;
                break;
            }
        }
        thread::sleep(Duration::from_millis(20));
    }

    let stdout = std::fs::read(req_dir.join("stdout")).unwrap_or_default();
    let stderr = std::fs::read(req_dir.join("stderr")).unwrap_or_default();
    let code = std::fs::read_to_string(req_dir.join("status"))
        .ok()
        .and_then(|s| s.trim().parse::<u32>().ok())
        .unwrap_or(if timed_out { 124 } else { 1 });
    let _ = std::fs::remove_dir_all(&req_dir);
    Ok(ExecOutcome {
        status: synthesize_exit(code),
        stdout,
        stderr,
        duration: started.elapsed(),
        peak_rss_kib: None,
        timed_out,
    })
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn kboxlike_output(builder: ExecBuilder) -> io::Result<ExecOutcome> {
    let target = builder.kboxlike.clone().ok_or_else(|| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            "kboxlike_output called without a kboxlike target",
        )
    })?;
    if builder.argv.is_empty() {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "exec: argv is empty",
        ));
    }
    if builder.tty || builder.cols.is_some() || builder.rows.is_some() {
        return Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "kboxlike backend does not support TTY exec yet",
        ));
    }
    kboxlike_stage_files(&target, &builder.stage_files)?;

    let started = Instant::now();
    let mut stdout = Vec::new();
    let mut stderr = Vec::new();
    let mut status = synthesize_exit(0);
    let mut timed_out = false;
    let mut argvs = Vec::with_capacity(1 + builder.chain.len());
    argvs.push(builder.argv.clone());
    argvs.extend(builder.chain.iter().cloned());

    for argv in argvs {
        let out = kboxlike_output_one(&target, &builder, &argv, started)?;
        timed_out |= out.timed_out;
        stdout.extend(out.stdout);
        stderr.extend(out.stderr);
        status = out.status;
        if timed_out || !status.success() {
            break;
        }
    }

    Ok(ExecOutcome {
        status,
        stdout,
        stderr,
        duration: started.elapsed(),
        peak_rss_kib: None,
        timed_out,
    })
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn kboxlike_output_one(
    target: &KboxlikeExecTarget,
    builder: &ExecBuilder,
    argv: &[String],
    started: Instant,
) -> io::Result<ExecOutcome> {
    let mut cmd = kboxlike_exec_command(target, builder, argv)?;
    cmd.stdin(Stdio::null());
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());
    let mut child = cmd
        .spawn()
        .map_err(|e| io::Error::new(e.kind(), format!("kboxlike exec {:?}: {e}", argv)))?;
    let mut timed_out = false;
    loop {
        if child.try_wait()?.is_some() {
            break;
        }
        if let Some(timeout) = builder.timeout {
            if started.elapsed() >= timeout {
                timed_out = true;
                let _ = child.kill();
                break;
            }
        }
        thread::sleep(Duration::from_millis(10));
    }
    let out = child.wait_with_output()?;
    Ok(ExecOutcome {
        status: out.status,
        stdout: out.stdout,
        stderr: out.stderr,
        duration: started.elapsed(),
        peak_rss_kib: None,
        timed_out,
    })
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn kboxlike_spawn(builder: ExecBuilder) -> io::Result<ExecChild> {
    let target = builder.kboxlike.clone().ok_or_else(|| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            "kboxlike_spawn called without a kboxlike target",
        )
    })?;
    if builder.argv.is_empty() {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "exec: argv is empty",
        ));
    }
    if builder.tty || builder.cols.is_some() || builder.rows.is_some() {
        return Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "kboxlike backend does not support TTY exec yet",
        ));
    }
    kboxlike_stage_files(&target, &builder.stage_files)?;

    let argv = if builder.chain.is_empty() {
        builder.argv.clone()
    } else {
        runsc_spawn_argv(&builder)
    };
    let mut cmd = kboxlike_exec_command(&target, &builder, &argv)?;
    cmd.stdin(Stdio::piped());
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());
    let mut child = cmd
        .spawn()
        .map_err(|e| io::Error::new(e.kind(), format!("kboxlike exec {:?}: {e}", argv)))?;
    let child_pid = child.id() as i32;
    let child_stdin = Arc::new(Mutex::new(child.stdin.take()));
    let child_stdout = child
        .stdout
        .take()
        .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "kboxlike exec stdout pipe missing"))?;
    let child_stderr = child
        .stderr
        .take()
        .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "kboxlike exec stderr pipe missing"))?;

    let (host_sock, bridge_sock) = UnixStream::pair()?;
    let sock_r = host_sock.try_clone()?;
    let interrupt_fd = Arc::new(host_sock.try_clone()?);
    let sock_w = Arc::new(Mutex::new(host_sock));

    let (stdout_tx, stdout_r) = channel::<Vec<u8>>();
    let (stderr_tx, stderr_r) = channel::<Vec<u8>>();
    let (exit_tx, exit_rx) = channel::<DemuxExit>();
    let demux = thread::Builder::new()
        .name("supermachine-kboxlike-exec-demux".into())
        .spawn(move || {
            demux_loop(sock_r, stdout_tx, stderr_tx, exit_tx);
        })
        .map_err(|e| {
            io::Error::new(
                io::ErrorKind::Other,
                format!("kboxlike exec: demux thread: {e}"),
            )
        })?;

    let bridge_w = Arc::new(Mutex::new(bridge_sock.try_clone()?));
    let control_stdin = Arc::clone(&child_stdin);
    let process_done = Arc::new(AtomicBool::new(false));
    let control_done = Arc::clone(&process_done);
    let mut control_r = bridge_sock.try_clone()?;
    thread::Builder::new()
        .name("supermachine-kboxlike-exec-control".into())
        .spawn(move || {
            runsc_spawn_control_loop(&mut control_r, child_pid, control_stdin, control_done);
        })
        .map_err(|e| {
            io::Error::new(
                io::ErrorKind::Other,
                format!("kboxlike exec: control thread: {e}"),
            )
        })?;

    let stdout_pump = runsc_spawn_pump(
        "supermachine-kboxlike-exec-stdout",
        child_stdout,
        FRAME_STDOUT,
        Arc::clone(&bridge_w),
    )?;
    let stderr_pump = runsc_spawn_pump(
        "supermachine-kboxlike-exec-stderr",
        child_stderr,
        FRAME_STDERR,
        Arc::clone(&bridge_w),
    )?;
    thread::Builder::new()
        .name("supermachine-kboxlike-exec-wait".into())
        .spawn(move || {
            drop(child_stdin);
            let status = child.wait();
            let payload = match status {
                Ok(status) => runsc_status_code(status).to_be_bytes(),
                Err(_) => 1u32.to_be_bytes(),
            };
            process_done.store(true, std::sync::atomic::Ordering::SeqCst);
            let _ = send_frame(&bridge_w, FRAME_EXIT, &payload);
            if let Ok(sock) = bridge_w.lock() {
                let _ = sock.shutdown(Shutdown::Both);
            }
            let _ = stdout_pump.join();
            let _ = stderr_pump.join();
        })
        .map_err(|e| {
            io::Error::new(
                io::ErrorKind::Other,
                format!("kboxlike exec: wait thread: {e}"),
            )
        })?;

    let stdin = ExecStdin::new(sock_w.clone());

    Ok(ExecChild {
        sock_w,
        we_signaled: Arc::new(AtomicBool::new(false)),
        aborted: Arc::new(AtomicBool::new(false)),
        interrupt_fd,
        stdout_r: Some(stdout_r),
        stderr_r: Some(stderr_r),
        stdin: Some(stdin),
        demux: Some(demux),
        exit_rx,
    })
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn kboxlike_exec_command(
    target: &KboxlikeExecTarget,
    builder: &ExecBuilder,
    argv: &[String],
) -> io::Result<Command> {
    if argv.is_empty() {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "exec: argv is empty",
        ));
    }
    let rootfs = std::ffi::CString::new(target.rootfs.as_os_str().as_bytes()).map_err(|_| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            "kboxlike rootfs path contains NUL byte",
        )
    })?;
    let cwd = std::ffi::CString::new(builder.cwd.as_deref().unwrap_or("/")).map_err(|_| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            "kboxlike exec cwd contains NUL byte",
        )
    })?;
    let mounts = kboxlike_prepare_runtime_mounts(&target.rootfs, &target.mounts)?;
    let mut cmd = Command::new(&argv[0]);
    cmd.args(&argv[1..]);
    cmd.env_clear();
    let mut env = BTreeMap::new();
    for (key, value) in &target.image_env {
        env.insert(key.clone(), value.clone());
    }
    for (key, value) in &builder.env {
        env.insert(key.clone(), value.clone());
    }
    for (key, value) in env {
        cmd.env(key, value);
    }
    let user = target.user;
    unsafe {
        cmd.pre_exec(move || {
            kboxlike_setup_runtime_mounts(&mounts)?;
            if libc::chroot(rootfs.as_ptr()) < 0 {
                return Err(io::Error::last_os_error());
            }
            if libc::chdir(cwd.as_ptr()) < 0 {
                return Err(io::Error::last_os_error());
            }
            if let Some((uid, gid)) = user {
                kboxlike_drop_uid_gid(uid, gid)?;
            }
            Ok(())
        });
    }
    Ok(cmd)
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
struct KboxlikePreparedRuntimeMounts {
    binds: Vec<(std::ffi::CString, std::ffi::CString)>,
    proc_target: std::ffi::CString,
    shm_target: std::ffi::CString,
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn kboxlike_prepare_runtime_mounts(
    rootfs: &Path,
    mounts: &[(PathBuf, String)],
) -> io::Result<KboxlikePreparedRuntimeMounts> {
    kboxlike_create_nested_mount_targets(mounts)?;
    let mut prepared = Vec::with_capacity(mounts.len());
    for (host_path, guest_path) in mounts {
        let target = kboxlike_rootfs_join(rootfs, guest_path)?;
        std::fs::create_dir_all(&target)?;
        let host = std::ffi::CString::new(host_path.as_os_str().as_bytes()).map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                format!(
                    "kboxlike mount host path contains NUL byte: {}",
                    host_path.display()
                ),
            )
        })?;
        let target = std::ffi::CString::new(target.as_os_str().as_bytes()).map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("kboxlike mount target contains NUL byte: {guest_path}"),
            )
        })?;
        prepared.push((host, target));
    }
    kboxlike_append_standard_dev_bind_mounts(rootfs, &mut prepared)?;
    let proc_target = rootfs.join("proc");
    std::fs::create_dir_all(&proc_target)?;
    let shm_target = rootfs.join("dev").join("shm");
    std::fs::create_dir_all(&shm_target)?;
    let proc_target = std::ffi::CString::new(proc_target.as_os_str().as_bytes()).map_err(|_| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            format!(
                "kboxlike proc target path contains NUL byte: {}",
                proc_target.display()
            ),
        )
    })?;
    let shm_target = std::ffi::CString::new(shm_target.as_os_str().as_bytes()).map_err(|_| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            format!(
                "kboxlike shm target path contains NUL byte: {}",
                shm_target.display()
            ),
        )
    })?;
    Ok(KboxlikePreparedRuntimeMounts {
        binds: prepared,
        proc_target,
        shm_target,
    })
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn kboxlike_create_nested_mount_targets(mounts: &[(PathBuf, String)]) -> io::Result<()> {
    for (parent_host, parent_guest) in mounts {
        for (_, child_guest) in mounts {
            let Some(relative_child) = kboxlike_nested_mount_suffix(parent_guest, child_guest)
            else {
                continue;
            };
            std::fs::create_dir_all(parent_host.join(relative_child))?;
        }
    }
    Ok(())
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn kboxlike_nested_mount_suffix(parent_guest: &str, child_guest: &str) -> Option<PathBuf> {
    let parent = Path::new(parent_guest.strip_prefix('/').unwrap_or(parent_guest));
    let child = Path::new(child_guest.strip_prefix('/').unwrap_or(child_guest));
    if parent == child {
        return None;
    }
    child.strip_prefix(parent).ok().map(Path::to_path_buf)
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn kboxlike_append_standard_dev_bind_mounts(
    rootfs: &Path,
    prepared: &mut Vec<(std::ffi::CString, std::ffi::CString)>,
) -> io::Result<()> {
    let dev_dir = rootfs.join("dev");
    std::fs::create_dir_all(&dev_dir)?;
    for name in ["null", "zero", "random", "urandom"] {
        let host_path = PathBuf::from("/dev").join(name);
        if !host_path.exists() {
            continue;
        }
        let target = dev_dir.join(name);
        if !target.exists() {
            std::fs::File::create(&target)?;
        }
        let host = std::ffi::CString::new(host_path.as_os_str().as_bytes()).map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                format!(
                    "kboxlike device host path contains NUL byte: {}",
                    host_path.display()
                ),
            )
        })?;
        let target = std::ffi::CString::new(target.as_os_str().as_bytes()).map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                format!(
                    "kboxlike device target path contains NUL byte: {}",
                    target.display()
                ),
            )
        })?;
        prepared.push((host, target));
    }
    Ok(())
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn kboxlike_setup_runtime_mounts(mounts: &KboxlikePreparedRuntimeMounts) -> io::Result<()> {
    unsafe {
        if libc::unshare(libc::CLONE_NEWNS) < 0 {
            return Err(io::Error::last_os_error());
        }
        if libc::mount(
            std::ptr::null(),
            c"/".as_ptr(),
            std::ptr::null(),
            (libc::MS_REC | libc::MS_PRIVATE) as libc::c_ulong,
            std::ptr::null(),
        ) < 0
        {
            return Err(io::Error::last_os_error());
        }
        for (host, target) in &mounts.binds {
            if libc::mount(
                host.as_ptr(),
                target.as_ptr(),
                std::ptr::null(),
                (libc::MS_BIND | libc::MS_REC) as libc::c_ulong,
                std::ptr::null(),
            ) < 0
            {
                return Err(io::Error::last_os_error());
            }
        }
        if libc::mount(
            c"proc".as_ptr(),
            mounts.proc_target.as_ptr(),
            c"proc".as_ptr(),
            (libc::MS_NOSUID | libc::MS_NOEXEC | libc::MS_NODEV) as libc::c_ulong,
            std::ptr::null(),
        ) < 0
        {
            return Err(io::Error::last_os_error());
        }
        if libc::mount(
            c"tmpfs".as_ptr(),
            mounts.shm_target.as_ptr(),
            c"tmpfs".as_ptr(),
            (libc::MS_NOSUID | libc::MS_NODEV) as libc::c_ulong,
            c"mode=1777,size=64m".as_ptr().cast(),
        ) < 0
        {
            return Err(io::Error::last_os_error());
        }
    }
    Ok(())
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn kboxlike_drop_uid_gid(uid: u32, gid: u32) -> io::Result<()> {
    unsafe {
        if libc::setgroups(0, std::ptr::null()) < 0 {
            return Err(io::Error::last_os_error());
        }
        if libc::setgid(gid as libc::gid_t) < 0 {
            return Err(io::Error::last_os_error());
        }
        if libc::setuid(uid as libc::uid_t) < 0 {
            return Err(io::Error::last_os_error());
        }
    }
    Ok(())
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn kboxlike_stage_files(target: &KboxlikeExecTarget, files: &[StageFile]) -> io::Result<()> {
    for file in files {
        let target_path = kboxlike_rootfs_join(&target.rootfs, &file.path)?;
        let parent = target_path.parent().ok_or_else(|| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("stage file has no parent: {}", file.path),
            )
        })?;
        std::fs::create_dir_all(parent)?;
        let name = target_path
            .file_name()
            .and_then(|name| name.to_str())
            .unwrap_or("stage");
        let tmp = parent.join(format!(".supermachine-stage-{}-{name}", std::process::id()));
        std::fs::write(&tmp, &file.data)?;
        let mode = file.mode.unwrap_or(0o644);
        std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(mode))?;
        std::fs::rename(&tmp, &target_path)?;
    }
    Ok(())
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn kboxlike_rootfs_join(rootfs: &Path, guest_path: &str) -> io::Result<PathBuf> {
    let rel = guest_path.strip_prefix('/').unwrap_or(guest_path);
    let p = Path::new(rel);
    if rel.is_empty()
        || p.components().any(|component| {
            matches!(
                component,
                std::path::Component::ParentDir | std::path::Component::RootDir
            )
        })
    {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            format!("guest path escapes rootfs: {guest_path}"),
        ));
    }
    Ok(rootfs.join(p))
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn runsc_output_one(
    target: &RunscExecTarget,
    builder: &ExecBuilder,
    argv: &[String],
    started: Instant,
) -> io::Result<ExecOutcome> {
    let mut cmd = Command::new(&target.runsc_bin);
    add_runsc_global_flags_with_network(&mut cmd, &target.runsc_root, runsc_network(target));
    cmd.arg("exec");
    if let Some(cwd) = builder.cwd.as_deref() {
        cmd.arg("-cwd").arg(cwd);
    }
    let mut env = BTreeMap::new();
    for (k, v) in &target.image_env {
        env.insert(k.clone(), v.clone());
    }
    for (k, v) in &builder.env {
        env.insert(k.clone(), v.clone());
    }
    for (k, v) in env {
        cmd.arg("-env").arg(format!("{k}={v}"));
    }
    cmd.arg(&target.container_id);
    cmd.args(argv);
    cmd.stdin(Stdio::null());
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());

    let mut child = cmd.spawn().map_err(|e| {
        io::Error::new(e.kind(), format!("runsc exec {}: {e}", target.container_id))
    })?;
    let mut timed_out = false;
    loop {
        if child.try_wait()?.is_some() {
            break;
        }
        if let Some(timeout) = builder.timeout {
            if started.elapsed() >= timeout {
                timed_out = true;
                let _ = child.kill();
                break;
            }
        }
        thread::sleep(Duration::from_millis(10));
    }
    let out = child.wait_with_output()?;
    Ok(ExecOutcome {
        status: out.status,
        stdout: out.stdout,
        stderr: out.stderr,
        duration: started.elapsed(),
        peak_rss_kib: None,
        timed_out,
    })
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn runsc_spawn(builder: ExecBuilder) -> io::Result<ExecChild> {
    let target = builder.runsc.clone().ok_or_else(|| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            "runsc_spawn called without a runsc target",
        )
    })?;
    if builder.argv.is_empty() {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "exec: argv is empty",
        ));
    }
    if target.agent_path.is_some() {
        return runsc_agent_spawn(builder, target);
    }
    if builder.tty || builder.cols.is_some() || builder.rows.is_some() {
        return Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "runsc backend does not support TTY exec yet",
        ));
    }
    runsc_stage_files(&target, &builder.stage_files)?;

    let argv = runsc_spawn_argv(&builder);
    let mut cmd = runsc_exec_command(&target, &builder, &argv);
    cmd.stdin(Stdio::piped());
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());
    let mut child = cmd.spawn().map_err(|e| {
        io::Error::new(e.kind(), format!("runsc exec {}: {e}", target.container_id))
    })?;
    let child_pid = child.id() as i32;
    let child_stdin = Arc::new(Mutex::new(child.stdin.take()));
    let child_stdout = child
        .stdout
        .take()
        .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "runsc exec stdout pipe missing"))?;
    let child_stderr = child
        .stderr
        .take()
        .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "runsc exec stderr pipe missing"))?;

    let (host_sock, bridge_sock) = UnixStream::pair()?;
    let sock_r = host_sock.try_clone()?;
    let interrupt_fd = Arc::new(host_sock.try_clone()?);
    let sock_w = Arc::new(Mutex::new(host_sock));

    let (stdout_tx, stdout_r) = channel::<Vec<u8>>();
    let (stderr_tx, stderr_r) = channel::<Vec<u8>>();
    let (exit_tx, exit_rx) = channel::<DemuxExit>();
    let demux = thread::Builder::new()
        .name("supermachine-runsc-exec-demux".into())
        .spawn(move || {
            demux_loop(sock_r, stdout_tx, stderr_tx, exit_tx);
        })
        .map_err(|e| {
            io::Error::new(
                io::ErrorKind::Other,
                format!("runsc exec: demux thread: {e}"),
            )
        })?;

    let bridge_w = Arc::new(Mutex::new(bridge_sock.try_clone()?));
    let control_stdin = Arc::clone(&child_stdin);
    let process_done = Arc::new(AtomicBool::new(false));
    let control_done = Arc::clone(&process_done);
    let mut control_r = bridge_sock.try_clone()?;
    thread::Builder::new()
        .name("supermachine-runsc-exec-control".into())
        .spawn(move || {
            runsc_spawn_control_loop(&mut control_r, child_pid, control_stdin, control_done);
        })
        .map_err(|e| {
            io::Error::new(
                io::ErrorKind::Other,
                format!("runsc exec: control thread: {e}"),
            )
        })?;

    let stdout_pump = runsc_spawn_pump(
        "supermachine-runsc-exec-stdout",
        child_stdout,
        FRAME_STDOUT,
        Arc::clone(&bridge_w),
    )?;
    let stderr_pump = runsc_spawn_pump(
        "supermachine-runsc-exec-stderr",
        child_stderr,
        FRAME_STDERR,
        Arc::clone(&bridge_w),
    )?;
    thread::Builder::new()
        .name("supermachine-runsc-exec-wait".into())
        .spawn(move || {
            drop(child_stdin);
            let status = child.wait();
            let payload = match status {
                Ok(status) => runsc_status_code(status).to_be_bytes(),
                Err(_) => 1u32.to_be_bytes(),
            };
            process_done.store(true, std::sync::atomic::Ordering::SeqCst);
            let _ = send_frame(&bridge_w, FRAME_EXIT, &payload);
            if let Ok(sock) = bridge_w.lock() {
                let _ = sock.shutdown(Shutdown::Both);
            }
            let _ = stdout_pump.join();
            let _ = stderr_pump.join();
        })
        .map_err(|e| {
            io::Error::new(
                io::ErrorKind::Other,
                format!("runsc exec: wait thread: {e}"),
            )
        })?;

    let stdin = ExecStdin::new(sock_w.clone());

    Ok(ExecChild {
        sock_w,
        we_signaled: Arc::new(AtomicBool::new(false)),
        aborted: Arc::new(AtomicBool::new(false)),
        interrupt_fd,
        stdout_r: Some(stdout_r),
        stderr_r: Some(stderr_r),
        stdin: Some(stdin),
        demux: Some(demux),
        exit_rx,
    })
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn runsc_agent_spawn(builder: ExecBuilder, target: RunscExecTarget) -> io::Result<ExecChild> {
    let debug = std::env::var_os("SUPERMACHINE_RUNSC_EXECDBG").is_some();
    let agent_path = target.agent_path.as_deref().ok_or_else(|| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            "runsc agent exec target missing agent path",
        )
    })?;
    let mut cmd = Command::new(&target.runsc_bin);
    add_runsc_global_flags_with_network(&mut cmd, &target.runsc_root, runsc_network(&target));
    cmd.arg("exec")
        .arg(&target.container_id)
        .arg(agent_path)
        .arg("--exec-stdio");
    cmd.stdin(Stdio::piped());
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());
    if debug {
        eprintln!(
            "[runsc.execdbg] spawn agent container={} path={} argv0={}",
            target.container_id,
            agent_path,
            builder.argv.first().map(String::as_str).unwrap_or("")
        );
    }
    let mut child = cmd.spawn().map_err(|e| {
        io::Error::new(
            e.kind(),
            format!(
                "runsc exec {} {} --exec-stdio: {e}",
                target.container_id, agent_path
            ),
        )
    })?;
    let child_pid = child.id() as i32;
    let mut child_stdin = child.stdin.take().ok_or_else(|| {
        io::Error::new(io::ErrorKind::Other, "runsc agent exec stdin pipe missing")
    })?;
    let mut child_stdout = child.stdout.take().ok_or_else(|| {
        io::Error::new(io::ErrorKind::Other, "runsc agent exec stdout pipe missing")
    })?;
    let child_stderr = child.stderr.take().ok_or_else(|| {
        io::Error::new(io::ErrorKind::Other, "runsc agent exec stderr pipe missing")
    })?;

    let (host_sock, bridge_sock) = UnixStream::pair()?;
    let sock_r = host_sock.try_clone()?;
    let interrupt_fd = Arc::new(host_sock.try_clone()?);
    let sock_w = Arc::new(Mutex::new(host_sock));

    let (stdout_tx, stdout_r) = channel::<Vec<u8>>();
    let (stderr_tx, stderr_r) = channel::<Vec<u8>>();
    let (exit_tx, exit_rx) = channel::<DemuxExit>();
    let demux = thread::Builder::new()
        .name("supermachine-runsc-agent-demux".into())
        .spawn(move || {
            demux_loop(sock_r, stdout_tx, stderr_tx, exit_tx);
        })
        .map_err(|e| {
            io::Error::new(
                io::ErrorKind::Other,
                format!("runsc agent exec: demux thread: {e}"),
            )
        })?;

    let bridge_w = Arc::new(Mutex::new(bridge_sock.try_clone()?));
    let mut bridge_to_agent = bridge_sock.try_clone()?;
    let process_done = Arc::new(AtomicBool::new(false));
    let pump_done = Arc::clone(&process_done);
    let debug_c2g = debug;
    thread::Builder::new()
        .name("supermachine-runsc-agent-c2g".into())
        .spawn(move || {
            let mut copied: u64 = 0;
            let mut buf = [0u8; 8192];
            let copied_result: io::Result<u64> = loop {
                match bridge_to_agent.read(&mut buf) {
                    Ok(0) => break Ok(copied),
                    Ok(n) => {
                        copied += n as u64;
                        if debug_c2g {
                            eprintln!("[runsc.execdbg] c2g read n={n} total={copied}");
                        }
                        if let Err(e) = child_stdin.write_all(&buf[..n]) {
                            break Err(e);
                        }
                        if let Err(e) = child_stdin.flush() {
                            break Err(e);
                        }
                    }
                    Err(e) => break Err(e),
                }
            };
            if debug_c2g {
                eprintln!("[runsc.execdbg] c2g done copied={copied_result:?}");
            }
            let _ = child_stdin.flush();
            drop(child_stdin);
            if !pump_done.load(std::sync::atomic::Ordering::SeqCst) {
                if debug_c2g {
                    eprintln!("[runsc.execdbg] c2g killing host runsc exec pid={child_pid}");
                }
                let _ = unsafe { libc::kill(child_pid, libc::SIGKILL) };
            }
        })
        .map_err(|e| {
            io::Error::new(
                io::ErrorKind::Other,
                format!("runsc agent exec: stdin bridge thread: {e}"),
            )
        })?;

    let stdout_bridge_w = Arc::clone(&bridge_w);
    let debug_g2c = debug;
    let stdout_pump = thread::Builder::new()
        .name("supermachine-runsc-agent-g2c".into())
        .spawn(move || {
            let mut buf = [0u8; 8192];
            let mut total: u64 = 0;
            loop {
                match child_stdout.read(&mut buf) {
                    Ok(0) => {
                        if debug_g2c {
                            eprintln!("[runsc.execdbg] g2c eof total={total}");
                        }
                        return;
                    }
                    Ok(n) => {
                        total += n as u64;
                        if debug_g2c {
                            eprintln!("[runsc.execdbg] g2c read n={n} total={total}");
                        }
                        let mut guard = match stdout_bridge_w.lock() {
                            Ok(g) => g,
                            Err(_) => return,
                        };
                        if guard.write_all(&buf[..n]).is_err() {
                            return;
                        }
                    }
                    Err(e) => {
                        if debug_g2c {
                            eprintln!("[runsc.execdbg] g2c read error {e}");
                        }
                        return;
                    }
                }
            }
        })
        .map_err(|e| {
            io::Error::new(
                io::ErrorKind::Other,
                format!("runsc agent exec: stdout bridge thread: {e}"),
            )
        })?;
    let stderr_pump = runsc_spawn_pump(
        "supermachine-runsc-agent-stderr",
        child_stderr,
        FRAME_STDERR,
        Arc::clone(&bridge_w),
    )?;

    thread::Builder::new()
        .name("supermachine-runsc-agent-wait".into())
        .spawn(move || {
            let status = child.wait();
            if debug {
                eprintln!("[runsc.execdbg] wait status={status:?}");
            }
            process_done.store(true, std::sync::atomic::Ordering::SeqCst);
            if let Ok(sock) = bridge_w.lock() {
                let _ = sock.shutdown(Shutdown::Both);
            }
            let _ = stdout_pump.join();
            let _ = stderr_pump.join();
        })
        .map_err(|e| {
            io::Error::new(
                io::ErrorKind::Other,
                format!("runsc agent exec: wait thread: {e}"),
            )
        })?;

    let stage_payload: Vec<StageFilePayload> = builder
        .stage_files
        .iter()
        .map(|s| StageFilePayload {
            path: &s.path,
            data_b64: crate::api::b64_encode(&s.data),
            mode: s.mode,
        })
        .collect();
    let payload = RequestPayload {
        argv: &builder.argv,
        env: &builder.env,
        cwd: builder.cwd.as_deref(),
        tty: builder.tty,
        cols: builder.cols,
        rows: builder.rows,
        stage_files: stage_payload,
        chain: &builder.chain,
    };
    let json = serde_json::to_vec(&payload).map_err(|e| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            format!("runsc agent exec: encode REQUEST: {e}"),
        )
    })?;
    send_frame(&sock_w, FRAME_REQUEST, &json)?;

    let stdin = ExecStdin::new(sock_w.clone());

    Ok(ExecChild {
        sock_w,
        we_signaled: Arc::new(AtomicBool::new(false)),
        aborted: Arc::new(AtomicBool::new(false)),
        interrupt_fd,
        stdout_r: Some(stdout_r),
        stderr_r: Some(stderr_r),
        stdin: Some(stdin),
        demux: Some(demux),
        exit_rx,
    })
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn runsc_exec_command(target: &RunscExecTarget, builder: &ExecBuilder, argv: &[String]) -> Command {
    let mut cmd = Command::new(&target.runsc_bin);
    add_runsc_global_flags_with_network(&mut cmd, &target.runsc_root, runsc_network(target));
    cmd.arg("exec");
    if let Some(cwd) = builder.cwd.as_deref() {
        cmd.arg("-cwd").arg(cwd);
    }
    let mut env = BTreeMap::new();
    for (k, v) in &target.image_env {
        env.insert(k.clone(), v.clone());
    }
    for (k, v) in &builder.env {
        env.insert(k.clone(), v.clone());
    }
    for (k, v) in env {
        cmd.arg("-env").arg(format!("{k}={v}"));
    }
    cmd.arg(&target.container_id);
    cmd.args(argv);
    cmd
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn runsc_spawn_argv(builder: &ExecBuilder) -> Vec<String> {
    if builder.chain.is_empty() {
        return builder.argv.clone();
    }
    let mut commands = Vec::with_capacity(1 + builder.chain.len());
    commands.push(shell_join_argv(&builder.argv));
    for argv in &builder.chain {
        commands.push(shell_join_argv(argv));
    }
    vec!["/bin/sh".into(), "-c".into(), commands.join(" && ")]
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn shell_join_argv(argv: &[String]) -> String {
    argv.iter()
        .map(|arg| sh_quote(arg))
        .collect::<Vec<_>>()
        .join(" ")
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn runsc_stage_files(target: &RunscExecTarget, files: &[StageFile]) -> io::Result<()> {
    for file in files {
        let q_path = sh_quote(&file.path);
        let mode = file.mode.unwrap_or(0o644);
        let script = format!(
            "set -e\n\
             p={q_path}\n\
             d=$(dirname -- \"$p\")\n\
             mkdir -p -- \"$d\"\n\
             t=\"$d/.supermachine-stage-$$-$(basename -- \"$p\")\"\n\
             cat > \"$t\"\n\
             chmod {mode:o} \"$t\"\n\
             mv -f -- \"$t\" \"$p\"\n"
        );
        let mut cmd = Command::new(&target.runsc_bin);
        add_runsc_global_flags_with_network(&mut cmd, &target.runsc_root, runsc_network(target));
        cmd.arg("exec")
            .arg(&target.container_id)
            .arg("/bin/sh")
            .arg("-c")
            .arg(script);
        cmd.stdin(Stdio::piped());
        cmd.stdout(Stdio::piped());
        cmd.stderr(Stdio::piped());
        let mut child = cmd.spawn().map_err(|e| {
            io::Error::new(
                e.kind(),
                format!(
                    "runsc stage_file {} in {}: {e}",
                    file.path, target.container_id
                ),
            )
        })?;
        if let Some(mut stdin) = child.stdin.take() {
            stdin.write_all(&file.data)?;
        }
        let out = child.wait_with_output()?;
        if !out.status.success() {
            return Err(io::Error::other(format!(
                "runsc stage_file {} failed: status={} stdout={} stderr={}",
                file.path,
                out.status,
                String::from_utf8_lossy(&out.stdout),
                String::from_utf8_lossy(&out.stderr)
            )));
        }
    }
    Ok(())
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn runsc_spawn_control_loop(
    sock: &mut UnixStream,
    child_pid: i32,
    child_stdin: Arc<Mutex<Option<ChildStdin>>>,
    process_done: Arc<AtomicBool>,
) {
    loop {
        let (kind, payload) = match read_frame(sock) {
            Ok(frame) => frame,
            Err(_) => {
                if !process_done.load(std::sync::atomic::Ordering::SeqCst) {
                    let _ = unsafe { libc::kill(child_pid, libc::SIGKILL) };
                }
                return;
            }
        };
        match kind {
            FRAME_STDIN => {
                let mut guard = match child_stdin.lock() {
                    Ok(g) => g,
                    Err(_) => return,
                };
                if payload.is_empty() {
                    guard.take();
                } else if let Some(stdin) = guard.as_mut() {
                    let _ = stdin.write_all(&payload);
                    let _ = stdin.flush();
                }
            }
            FRAME_SIGNAL => {
                if let Some(signum) = payload.first().copied() {
                    let _ = unsafe { libc::kill(child_pid, signum as i32) };
                }
            }
            FRAME_RESIZE => {}
            _ => {}
        }
    }
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn runsc_spawn_pump<R>(
    name: &'static str,
    mut reader: R,
    kind: u8,
    bridge_w: Arc<Mutex<UnixStream>>,
) -> io::Result<JoinHandle<()>>
where
    R: Read + Send + 'static,
{
    thread::Builder::new()
        .name(name.into())
        .spawn(move || {
            let mut buf = [0u8; 8192];
            loop {
                match reader.read(&mut buf) {
                    Ok(0) => return,
                    Ok(n) => {
                        if send_frame(&bridge_w, kind, &buf[..n]).is_err() {
                            return;
                        }
                    }
                    Err(_) => return,
                }
            }
        })
        .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("runsc exec: {name}: {e}")))
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn write_raw_frame<W: Write>(writer: &mut W, kind: u8, payload: &[u8]) -> io::Result<()> {
    let mut header = [0u8; 5];
    header[0] = kind;
    header[1..5].copy_from_slice(&(payload.len() as u32).to_be_bytes());
    writer.write_all(&header)?;
    if !payload.is_empty() {
        writer.write_all(payload)?;
    }
    writer.flush()
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn runsc_status_code(status: ExitStatus) -> u32 {
    #[cfg(unix)]
    {
        use std::os::unix::process::ExitStatusExt;
        if let Some(code) = status.code() {
            return code as u32;
        }
        if let Some(signal) = status.signal() {
            return 128 + signal as u32;
        }
    }
    1
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn sh_quote(s: &str) -> String {
    let mut out = String::from("'");
    for ch in s.chars() {
        if ch == '\'' {
            out.push_str("'\\''");
        } else {
            out.push(ch);
        }
    }
    out.push('\'');
    out
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn unique_suffix_nanos() -> u128 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0)
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub(crate) fn add_runsc_global_flags(cmd: &mut Command, root: &Path) {
    add_runsc_global_flags_with_network(cmd, root, RUNSC_NETWORK_NONE);
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub(crate) fn add_runsc_global_flags_with_network(cmd: &mut Command, root: &Path, network: &str) {
    cmd.arg("--root")
        .arg(root)
        .arg("--platform=systrap")
        .arg(format!("--network={network}"))
        .arg("--overlay2=none")
        .arg("--file-access=shared");
    if network == RUNSC_NETWORK_SANDBOX {
        cmd.arg("--reproduce-nat").arg("--reproduce-nftables");
    }
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn runsc_network(target: &RunscExecTarget) -> &str {
    target.network.as_deref().unwrap_or(RUNSC_NETWORK_NONE)
}

/// JSON shape for the REQUEST frame. Fields match the agent's
/// `ExecRequest` struct.
#[derive(Serialize)]
struct RequestPayload<'a> {
    argv: &'a [String],
    env: &'a BTreeMap<String, String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    cwd: Option<&'a str>,
    tty: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    cols: Option<u16>,
    #[serde(skip_serializing_if = "Option::is_none")]
    rows: Option<u16>,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    stage_files: Vec<StageFilePayload<'a>>,
    #[serde(skip_serializing_if = "<[_]>::is_empty")]
    chain: &'a [Vec<String>],
}

#[derive(Serialize)]
struct StageFilePayload<'a> {
    path: &'a str,
    data_b64: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    mode: Option<u32>,
}

/// A running guest process. Hold this until you call [`wait`] or
/// drop it; dropping closes the agent connection, which kills the
/// child via the agent's `SIGHUP` propagation.
///
/// [`wait`]: ExecChild::wait
pub struct ExecChild {
    /// Writer half of the unix socket to the agent. Wrapped in a
    /// mutex so STDIN frames don't interleave with RESIZE/SIGNAL
    /// frames coming from other places.
    sock_w: Arc<Mutex<UnixStream>>,
    /// 0.7.44+ flips to true as soon as we send the first SIGNAL
    /// frame (from `signal()`/`signaler.signal()`/`signaler.kill()`).
    /// Used by `wait_with_rss` to synthesize a kill-by-signal
    /// ExitStatus when the agent closes the connection before an
    /// EXIT frame arrives — which happens reliably under heavy
    /// stdout flood when the single exec vsock channel is so
    /// congested with STDOUT bytes that the EXIT frame can't be
    /// queued in time before the watchdog's ABORT_GRACE elapses
    /// and force-closes the socket. The host KNOWS the workload
    /// was killed (we sent the kill); the missing EXIT is a wire
    /// artifact, not a semantic loss. See `synthesize_killed_by_signal`.
    we_signaled: Arc<AtomicBool>,
    /// Set to `true` the moment the host tears the connection down
    /// locally — via [`Self::abort`], [`ExecSignaler::abort`], or Drop.
    /// Lets the `EofBeforeExit` path distinguish "we closed it" (a clean
    /// local abort) from "the agent/VM went away unexpectedly", so the
    /// surfaced error is self-diagnosing instead of a generic EOF.
    aborted: Arc<AtomicBool>,
    /// A third `try_clone`d handle to the same underlying socket,
    /// retained ONLY for the host-side shutdown path. Calling
    /// `shutdown(Both)` on this fd wakes the demux thread's blocking
    /// read on its own fd (both fds share the kernel-side socket
    /// state). Used by [`Self::abort`] and Drop so an embedder that
    /// needs to interrupt a blocked [`wait`] / [`output`] from another
    /// thread can do so WITHOUT relying on the worker / agent being
    /// alive — a missing capability before 0.7.10 that forced
    /// embedders into thread-spawn-and-poll workarounds.
    ///
    /// [`wait`]: ExecChild::wait
    /// [`output`]: ExecBuilder::output
    interrupt_fd: Arc<UnixStream>,
    /// stdout chunk receiver (caller side). Pre-0.7.12 this was an
    /// `Option<OwnedFd>` wrapping a Unix pipe read end; we replaced
    /// that with an in-memory channel because the kernel pipe
    /// buffer (16 KiB on macOS) capped streaming throughput — any
    /// workload that produced >~48 KiB before the caller drained
    /// deadlocked the agent's writer mutex. The channel has no such
    /// ceiling; bytes are queued in memory until `ExecStdout::read`
    /// drains them.
    stdout_r: Option<Receiver<Vec<u8>>>,
    /// stderr chunk receiver. Same channel-based plumbing as stdout.
    stderr_r: Option<Receiver<Vec<u8>>>,
    /// stdin pipe write end (caller side). `None` if already closed
    /// via [`ExecStdin::close`].
    stdin: Option<ExecStdin>,
    /// Demux thread join handle.
    demux: Option<JoinHandle<()>>,
    /// EXIT status receiver, posted by the demux thread.
    exit_rx: Receiver<DemuxExit>,
}

/// Send + Sync + Clone handle to an [`ExecChild`]'s control channel.
/// Lets a thread that doesn't own the child send signals AND abort
/// the connection — the canonical Rust answer to "I want to cancel
/// an in-flight `output()` / `wait()` from a registry / watchdog /
/// shutdown handler."
///
/// Why this exists: `ExecChild` itself is `!Sync` (its `exit_rx` is
/// `mpsc::Receiver`, which is `Send` but `!Sync`), so you can't share
/// `&ExecChild` across threads. `ExecSignaler` is a `!exit_rx`
/// slice of the child that IS thread-safe — built on the same
/// `Arc<Mutex<UnixStream>>` the writer-side already uses internally.
///
/// Cost: each clone is one atomic refcount increment.
///
/// Idiomatic shape:
///
/// ```ignore
/// let child = vm.exec_builder(["sleep", "30"]).spawn()?;
/// let signaler = child.signaler();
/// std::thread::spawn(move || {
///     // user pressed Ctrl+C — interrupt the running process.
///     let _ = signaler.kill();
/// });
/// let exit = child.wait()?;   // unblocks when kill arrives
/// ```
///
/// For "interrupt even if the worker died" semantics, use
/// [`Self::abort`] — it closes the host-side socket which causes
/// the demux thread to surface EOF immediately, regardless of
/// whether the guest agent is still alive.
#[derive(Clone)]
pub struct ExecSignaler {
    sock_w: Arc<Mutex<UnixStream>>,
    interrupt_fd: Arc<UnixStream>,
    /// Shared with the parent [`ExecChild`] so a successful
    /// `signal()`/`kill()` lets `wait_with_rss` synthesize a
    /// SIGKILL exit when the agent closes the channel without
    /// sending EXIT (vsock congestion → EXIT can't queue). See
    /// `ExecChild::we_signaled` for the full rationale.
    we_signaled: Arc<AtomicBool>,
    /// Shared with the parent [`ExecChild`]; set when this signaler's
    /// [`Self::abort`] tears the connection down, so the child's
    /// `EofBeforeExit` path reports a local abort rather than an
    /// unexplained agent disconnect. See [`ExecChild::aborted`].
    aborted: Arc<AtomicBool>,
}

impl std::fmt::Debug for ExecSignaler {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ExecSignaler").finish_non_exhaustive()
    }
}

impl ExecSignaler {
    /// Send a signal to the in-guest process. The agent forwards it
    /// via `kill(child_pid, signum)`. Returns `Ok(())` if the frame
    /// reached the agent; the result of the actual kill is async
    /// (the agent sends an EXIT frame when the process dies).
    ///
    /// Common signals: `libc::SIGTERM` (graceful), `libc::SIGINT`
    /// (Ctrl+C-style), `libc::SIGKILL` (forced). See [`Self::kill`]
    /// for the SIGKILL shortcut.
    pub fn signal(&self, signum: i32) -> io::Result<()> {
        // Mark "we signalled" BEFORE the send. Even if the send
        // fails (vsock congested, agent dead, etc.), the host has
        // committed to wanting the child dead — a subsequent EOF
        // should be interpreted as kill-success, not "agent
        // disconnected unexpectedly".
        self.we_signaled
            .store(true, std::sync::atomic::Ordering::SeqCst);
        let payload = [signum as u8];
        send_frame(&self.sock_w, FRAME_SIGNAL, &payload)
    }

    /// Shortcut for `signal(libc::SIGKILL)`. Mirrors
    /// [`std::process::Child::kill`].
    pub fn kill(&self) -> io::Result<()> {
        self.signal(libc::SIGKILL)
    }

    /// Tear down the host-side connection IMMEDIATELY — does NOT
    /// require the agent to be alive. Closes the underlying Unix
    /// socket (via `shutdown(Both)`), which wakes the demux thread
    /// from its blocking read and surfaces `UnexpectedEof` to the
    /// caller's [`ExecChild::wait`].
    ///
    /// Use this when the SIGNAL path won't work — e.g. the worker
    /// process died (pool shutdown) but the host-side `output()` /
    /// `wait()` is still blocked because EOF hasn't propagated.
    ///
    /// Safe to call from multiple threads / multiple times — the
    /// shutdown(2) syscall is idempotent on an already-shutdown
    /// socket.
    pub fn abort(&self) -> io::Result<()> {
        // Mark the teardown as local BEFORE the shutdown so the demux
        // thread's resulting EOF is attributed to us, not the agent.
        self.aborted
            .store(true, std::sync::atomic::Ordering::SeqCst);
        // Ignore "not connected" if we've already shut down.
        match self.interrupt_fd.shutdown(Shutdown::Both) {
            Ok(()) => Ok(()),
            Err(e) if e.kind() == io::ErrorKind::NotConnected => Ok(()),
            Err(e) => Err(e),
        }
    }
}

/// Outcome reported by the demux thread.
enum DemuxExit {
    /// Agent sent EXIT frame. `peak_rss_kib` is `None` for older
    /// agents that only send the 4-byte exit_code payload, and
    /// `Some(kib)` for v0.2.x+ agents that include rusage.
    Status {
        code: u32,
        peak_rss_kib: Option<u64>,
    },
    /// Agent closed the socket without sending EXIT. Carries how much
    /// the agent had streamed before the close (STDOUT/STDERR frame
    /// count + byte count) so the caller can self-diagnose the cause:
    /// zero output usually means the agent never started / the VM went
    /// away, while output-then-EOF means a mid-stream drop (congestion).
    EofBeforeExit { frames: u64, bytes: u64 },
    /// Agent sent ERROR frame.
    Error(String),
    /// Demux I/O failed.
    Io(io::Error),
}

impl ExecChild {
    /// Take ownership of the stdin handle. Calling this twice
    /// returns `None` the second time.
    pub fn stdin(&mut self) -> Option<ExecStdin> {
        self.stdin.take()
    }

    /// Take ownership of stdout. Returns a `Read`er. Useful when
    /// you want to spawn your own pump thread.
    pub fn stdout(&mut self) -> Option<ExecStdout> {
        self.stdout_r.take().map(ExecStdout::from_rx)
    }

    /// Take ownership of stderr. Empty in tty mode.
    pub fn stderr(&mut self) -> Option<ExecStderr> {
        self.stderr_r.take().map(ExecStderr::from_rx)
    }

    /// Send a signal (SIGTERM/SIGINT/...) to the guest process.
    pub fn signal(&self, signum: i32) -> io::Result<()> {
        // Same we_signaled marking as ExecSignaler::signal — see
        // there for rationale.
        self.we_signaled
            .store(true, std::sync::atomic::Ordering::SeqCst);
        let payload = [signum as u8];
        send_frame(&self.sock_w, FRAME_SIGNAL, &payload)
    }

    /// Hand back a [`Send + Sync + Clone`](ExecSignaler) handle for
    /// signaling THIS child from another thread. The returned handle
    /// can outlive the `ExecChild` — after the child exits, calls
    /// on the signaler become no-ops on the closed socket (no panic,
    /// no deadlock).
    ///
    /// Typical use: register the signaler in a cancellation registry,
    /// move the child into a wait thread, fire the registry slot
    /// (e.g. on user-cancel or pool-shutdown) to interrupt.
    ///
    /// Cost: one atomic refcount increment.
    ///
    /// [`Send + Sync + Clone`]: ExecSignaler
    pub fn signaler(&self) -> ExecSignaler {
        ExecSignaler {
            sock_w: Arc::clone(&self.sock_w),
            interrupt_fd: Arc::clone(&self.interrupt_fd),
            we_signaled: Arc::clone(&self.we_signaled),
            aborted: Arc::clone(&self.aborted),
        }
    }

    /// Close the host-side socket (`shutdown(Both)`) WITHOUT waiting
    /// for a clean exit. The demux thread observes EOF and exits;
    /// a concurrent [`wait`] / [`output`] returns
    /// `Err(UnexpectedEof)`. Use this when you've already decided
    /// the child should stop right now — e.g. a pool shutdown
    /// killed the worker and the in-flight wait would otherwise
    /// hang.
    ///
    /// Equivalent to `self.signaler().abort()`. Provided here for
    /// the common case where you have the child handle directly.
    ///
    /// [`wait`]: Self::wait
    /// [`output`]: ExecBuilder::output
    pub fn abort(&self) -> io::Result<()> {
        self.aborted
            .store(true, std::sync::atomic::Ordering::SeqCst);
        match self.interrupt_fd.shutdown(Shutdown::Both) {
            Ok(()) => Ok(()),
            Err(e) if e.kind() == io::ErrorKind::NotConnected => Ok(()),
            Err(e) => Err(e),
        }
    }

    /// Non-consuming poll. Returns `Ok(Some(_))` if the child has
    /// exited (and reaps the exit status), `Ok(None)` if it's still
    /// running, `Err(_)` on demux failure. Mirrors
    /// [`std::process::Child::try_wait`] — useful when you want to
    /// register the signaler with a registry AND periodically check
    /// completion without committing to a blocking [`wait`].
    ///
    /// [`wait`]: Self::wait
    pub fn try_wait(&self) -> io::Result<Option<(ExitStatus, Option<u64>)>> {
        match self.exit_rx.try_recv() {
            Ok(DemuxExit::Status { code, peak_rss_kib }) => {
                Ok(Some((synthesize_exit(code), peak_rss_kib)))
            }
            Ok(DemuxExit::EofBeforeExit { frames, bytes }) => {
                // 0.7.44+ if we sent SIGNAL before EOF, the host
                // explicitly asked the child to die. The missing
                // EXIT is a vsock-congestion wire artifact, not a
                // semantic loss — synthesize SIGKILL exit so the
                // caller observes a clean "killed by signal 9".
                if self.we_signaled.load(std::sync::atomic::Ordering::SeqCst) {
                    Ok(Some((synthesize_killed_by_signal(libc::SIGKILL), None)))
                } else {
                    Err(eof_before_exit_error(
                        self.aborted.load(std::sync::atomic::Ordering::SeqCst),
                        frames,
                        bytes,
                    ))
                }
            }
            Ok(DemuxExit::Error(msg)) => Err(io::Error::new(io::ErrorKind::Other, msg)),
            Ok(DemuxExit::Io(e)) => Err(e),
            Err(TryRecvError::Empty) => Ok(None),
            Err(TryRecvError::Disconnected) => Err(io::Error::new(
                io::ErrorKind::Other,
                "exec: demux thread died",
            )),
        }
    }

    /// Resize the pty (tty mode only). No-op in pipe mode (the
    /// agent ignores RESIZE frames if no pty is attached).
    pub fn resize(&self, cols: u16, rows: u16) -> io::Result<()> {
        let mut payload = [0u8; 4];
        payload[0..2].copy_from_slice(&cols.to_be_bytes());
        payload[2..4].copy_from_slice(&rows.to_be_bytes());
        send_frame(&self.sock_w, FRAME_RESIZE, &payload)
    }

    /// Block until the guest process exits. Returns the exit
    /// status (or a synthesized one if the agent disconnected).
    pub fn wait(self) -> io::Result<ExitStatus> {
        Ok(self.wait_with_rss()?.0)
    }

    /// Like [`ExecChild::wait`] but also surfaces the peak RSS
    /// the agent reports (`Some(kib)` for v0.2.x+ agents,
    /// `None` for older ones). Used internally by
    /// [`ExecBuilder::output`] to populate
    /// [`ExecOutcome::peak_rss_kib`].
    pub fn wait_with_rss(mut self) -> io::Result<(ExitStatus, Option<u64>)> {
        // Drop stdin if the caller didn't already; the child needs
        // EOF to exit cleanly for any program that reads stdin.
        self.stdin.take();

        let exit = self
            .exit_rx
            .recv()
            .map_err(|_| io::Error::new(io::ErrorKind::Other, "exec: demux thread died"))?;
        if let Some(h) = self.demux.take() {
            let _ = h.join();
        }
        match exit {
            DemuxExit::Status { code, peak_rss_kib } => Ok((synthesize_exit(code), peak_rss_kib)),
            DemuxExit::EofBeforeExit { frames, bytes } => {
                // See `try_wait` for rationale: signal-then-EOF
                // is a successful kill from the caller's POV; the
                // missing EXIT frame is a wire-level congestion
                // artifact.
                if self.we_signaled.load(std::sync::atomic::Ordering::SeqCst) {
                    Ok((synthesize_killed_by_signal(libc::SIGKILL), None))
                } else {
                    Err(eof_before_exit_error(
                        self.aborted.load(std::sync::atomic::Ordering::SeqCst),
                        frames,
                        bytes,
                    ))
                }
            }
            DemuxExit::Error(msg) => Err(io::Error::new(io::ErrorKind::Other, msg)),
            DemuxExit::Io(e) => Err(e),
        }
    }
}

impl Drop for ExecChild {
    fn drop(&mut self) {
        // Two-step teardown:
        //   1. Send EOF on the child's stdin so well-behaved programs
        //      that read(0) (sh, cat, …) exit naturally.
        //   2. Shutdown the host-side socket. This wakes the demux
        //      thread from any in-flight blocking read AND signals
        //      to the agent that the controller is gone — the agent
        //      will SIGHUP the child if it hasn't exited yet.
        //
        // Step 2 is the fix for the 0.7.10 "drop doesn't propagate
        // to in-flight wait" bug an integrator reported: pre-0.7.10
        // Drop only did step 1, leaving the demux thread blocked in
        // a read that the agent never closed (e.g. because the
        // worker died via pool.shutdown() and the close-detection
        // didn't propagate). Step 2 makes Drop the canonical "I
        // give up, clean up everything" path.
        //
        // We do NOT join the demux thread here — Drop must not
        // block on a foreign thread. After shutdown the demux read
        // returns within microseconds; the thread exits on its own
        // and the OS reclaims its stack.
        self.stdin.take();
        // Mark the teardown local before the shutdown: if a concurrent
        // wait/try_wait is still racing, its resulting EofBeforeExit is
        // attributed to our Drop, not to an unexplained agent close.
        self.aborted
            .store(true, std::sync::atomic::Ordering::SeqCst);
        let _ = self.interrupt_fd.shutdown(Shutdown::Both);
    }
}

/// Stdin handle. Bytes you write here get framed into STDIN
/// frames and sent to the agent, which writes them to the child's
/// stdin. Drop or call [`close`] for EOF.
///
/// [`close`]: ExecStdin::close
pub struct ExecStdin {
    sock_w: Arc<Mutex<UnixStream>>,
    closed: bool,
}

impl ExecStdin {
    fn new(sock_w: Arc<Mutex<UnixStream>>) -> Self {
        Self {
            sock_w,
            closed: false,
        }
    }

    /// Send EOF on stdin. The agent half-closes the child's stdin
    /// pipe; programs that read until EOF (cat, sh, ...) will
    /// then exit.
    pub fn close(mut self) -> io::Result<()> {
        if self.closed {
            return Ok(());
        }
        self.closed = true;
        send_frame(&self.sock_w, FRAME_STDIN, &[])
    }
}

impl Write for ExecStdin {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        if buf.is_empty() {
            return Ok(0);
        }
        send_frame(&self.sock_w, FRAME_STDIN, buf)?;
        Ok(buf.len())
    }
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

impl Drop for ExecStdin {
    fn drop(&mut self) {
        if !self.closed {
            // Best-effort EOF; ignore errors — the socket may be
            // gone if the child already exited.
            let _ = send_frame(&self.sock_w, FRAME_STDIN, &[]);
        }
    }
}

/// Stdout handle. Wraps an in-memory chunk channel; the demux
/// thread sends STDOUT-frame payloads into it. `Read::read` drains
/// whole chunks at a time, holding any tail bytes that didn't fit
/// in the caller's `buf` in `leftover` for the next call.
///
/// 0.7.12: moved from a Unix pipe to an mpsc channel. The pipe
/// version had a 16 KiB ceiling on macOS that deadlocked the agent
/// writer mutex once a workload produced >~48 KiB before the
/// caller drained.
pub struct ExecStdout {
    rx: Receiver<Vec<u8>>,
    /// Tail of a previously-received chunk that didn't fit in the
    /// caller's last `buf`. Drained first on the next `read`.
    leftover: Vec<u8>,
}

impl ExecStdout {
    fn from_rx(rx: Receiver<Vec<u8>>) -> Self {
        Self {
            rx,
            leftover: Vec::new(),
        }
    }

    /// Nonblocking variant for poll-style bindings. Returns:
    /// - `Ok(Some(n))` when bytes were copied into `buf`;
    /// - `Ok(None)` when no chunk is currently queued;
    /// - `Ok(Some(0))` on EOF.
    pub fn try_read(&mut self, buf: &mut [u8]) -> io::Result<Option<usize>> {
        chunk_channel_try_read(&self.rx, &mut self.leftover, buf)
    }
}

impl Read for ExecStdout {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        chunk_channel_read(&self.rx, &mut self.leftover, buf)
    }
}

/// Stderr handle. Same shape as [`ExecStdout`]; empty in tty mode.
pub struct ExecStderr {
    rx: Receiver<Vec<u8>>,
    leftover: Vec<u8>,
}

impl ExecStderr {
    fn from_rx(rx: Receiver<Vec<u8>>) -> Self {
        Self {
            rx,
            leftover: Vec::new(),
        }
    }

    /// Nonblocking variant for poll-style bindings. See [`ExecStdout::try_read`].
    pub fn try_read(&mut self, buf: &mut [u8]) -> io::Result<Option<usize>> {
        chunk_channel_try_read(&self.rx, &mut self.leftover, buf)
    }
}

impl Read for ExecStderr {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        chunk_channel_read(&self.rx, &mut self.leftover, buf)
    }
}

/// Shared body for ExecStdout/ExecStderr's `Read::read`. Returns
/// `Ok(0)` when the sender has dropped AND no leftover remains —
/// that's the canonical EOF signal that lets caller loops like
/// `while (chunk = read(); chunk.len() > 0)` terminate.
fn chunk_channel_read(
    rx: &Receiver<Vec<u8>>,
    leftover: &mut Vec<u8>,
    buf: &mut [u8],
) -> io::Result<usize> {
    if buf.is_empty() {
        return Ok(0);
    }
    if leftover.is_empty() {
        match rx.recv() {
            Ok(chunk) => {
                if chunk.is_empty() {
                    // Treat an empty payload as a no-op — recurse to
                    // wait for the next real chunk.
                    return chunk_channel_read(rx, leftover, buf);
                }
                *leftover = chunk;
            }
            Err(_) => return Ok(0), // sender dropped → EOF
        }
    }
    let n = buf.len().min(leftover.len());
    buf[..n].copy_from_slice(&leftover[..n]);
    leftover.drain(..n);
    Ok(n)
}

fn chunk_channel_try_read(
    rx: &Receiver<Vec<u8>>,
    leftover: &mut Vec<u8>,
    buf: &mut [u8],
) -> io::Result<Option<usize>> {
    if buf.is_empty() {
        return Ok(Some(0));
    }
    while leftover.is_empty() {
        match rx.try_recv() {
            Ok(chunk) => {
                if chunk.is_empty() {
                    continue;
                }
                *leftover = chunk;
            }
            Err(TryRecvError::Empty) => return Ok(None),
            Err(TryRecvError::Disconnected) => return Ok(Some(0)),
        }
    }
    let n = buf.len().min(leftover.len());
    buf[..n].copy_from_slice(&leftover[..n]);
    leftover.drain(..n);
    Ok(Some(n))
}

/// Send a CONTROL frame to the in-guest exec agent and wait for
/// its single-frame ack. Used for `signal` (forward TERM/KILL to
/// the workload) and any future control actions.
///
/// Returns Ok if the agent ack'd success; Err with the agent's
/// reported reason otherwise. Network/socket errors surface as
/// `io::Error`; remote-side failures (e.g. "no workload pid yet"
/// during the bake-time window) surface as
/// `io::ErrorKind::Other` with the agent's message.
pub fn send_control(exec_path: &Path, action_json: &serde_json::Value) -> io::Result<()> {
    let _ = send_control_with_ack(exec_path, action_json, None)?;
    Ok(())
}

/// Like [`send_control`] but returns the parsed ack JSON for
/// callers that need ack fields (e.g. `read_file` reads the
/// `data_b64` field). `read_timeout` overrides the default 5s
/// receive timeout — pass a generous value when the agent might
/// take a while (large file reads).
///
/// Public so that `@supermachine/core` (the napi binding crate
/// in `npm/supermachine-core/`) can use the same wire-format
/// helper its Rust sibling does, without duplicating JSON-action
/// shapes. Embedders outside this workspace should prefer the
/// stable wrappers on [`crate::Vm`] (e.g. [`crate::Vm::read_file`])
/// — the action JSON shape may evolve.
pub fn send_control_with_ack(
    exec_path: &Path,
    action_json: &serde_json::Value,
    read_timeout: Option<std::time::Duration>,
) -> io::Result<serde_json::Value> {
    use std::io::Read;
    let mut sock = UnixStream::connect(exec_path)?;
    sock.set_read_timeout(read_timeout.or(Some(std::time::Duration::from_secs(5))))?;
    sock.set_write_timeout(Some(std::time::Duration::from_secs(5)))?;
    // Serialize the JSON body once.
    let body = serde_json::to_vec(action_json)
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("encode CONTROL: {e}")))?;
    let mut header = [0u8; 5];
    header[0] = FRAME_CONTROL;
    header[1..5].copy_from_slice(&(body.len() as u32).to_be_bytes());
    sock.write_all(&header)?;
    if !body.is_empty() {
        sock.write_all(&body)?;
    }

    // Read the ack frame. Bumped to 16 MiB to accommodate
    // read_file responses (base64-encoded ≈ 12 MiB raw bytes).
    let mut ack_hdr = [0u8; 5];
    sock.read_exact(&mut ack_hdr)?;
    let kind = ack_hdr[0];
    let len = u32::from_be_bytes([ack_hdr[1], ack_hdr[2], ack_hdr[3], ack_hdr[4]]) as usize;
    if len > 16 * 1024 * 1024 {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            format!("CONTROL ack frame too large: {len}"),
        ));
    }
    let mut ack_body = vec![0u8; len];
    if !ack_body.is_empty() {
        sock.read_exact(&mut ack_body)?;
    }

    if kind != FRAME_CONTROL {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            format!("expected CONTROL ack frame, got {kind:#x}"),
        ));
    }
    let ack: serde_json::Value = serde_json::from_slice(&ack_body).map_err(|e| {
        io::Error::new(io::ErrorKind::InvalidData, format!("CONTROL ack JSON: {e}"))
    })?;
    if ack.get("ok").and_then(|v| v.as_bool()) == Some(true) {
        return Ok(ack);
    }
    let msg = ack
        .get("error")
        .and_then(|v| v.as_str())
        .unwrap_or("CONTROL: agent reported failure")
        .to_owned();
    Err(io::Error::new(io::ErrorKind::Other, msg))
}

// ---------- spawn ----------

fn spawn(builder: ExecBuilder) -> io::Result<ExecChild> {
    let sock = UnixStream::connect(&builder.exec_path)?;
    let sock_r = sock.try_clone()?;
    // Third fd referencing the same socket, retained for the
    // host-side abort path (see ExecChild::interrupt_fd). Lives in
    // an Arc so it can be cheaply cloned into ExecSignaler handles
    // — Arc<UnixStream> is fine because UnixStream is Send + Sync.
    let interrupt_fd = Arc::new(sock.try_clone()?);
    let sock_w = Arc::new(Mutex::new(sock));

    // Send REQUEST.
    let stage_payload: Vec<StageFilePayload> = builder
        .stage_files
        .iter()
        .map(|s| StageFilePayload {
            path: &s.path,
            data_b64: crate::api::b64_encode(&s.data),
            mode: s.mode,
        })
        .collect();
    let payload = RequestPayload {
        argv: &builder.argv,
        env: &builder.env,
        cwd: builder.cwd.as_deref(),
        tty: builder.tty,
        cols: builder.cols,
        rows: builder.rows,
        stage_files: stage_payload,
        chain: &builder.chain,
    };
    let json = serde_json::to_vec(&payload).map_err(|e| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            format!("exec: encode REQUEST: {e}"),
        )
    })?;
    send_frame(&sock_w, FRAME_REQUEST, &json)?;

    // In-memory chunk channels for stdout/stderr. Replaces the
    // pre-0.7.12 Unix pipes — those had a 16 KiB macOS buffer
    // ceiling that capped streaming throughput. The channel is
    // unbounded: workloads that produce data faster than the
    // caller drains will accumulate bytes in memory until either
    // (a) the caller catches up or (b) the agent finishes and
    // demux exits, draining the queue on the receiver side. Same
    // memory profile as `output()`'s read_to_end accumulator.
    let (stdout_tx, stdout_r) = channel::<Vec<u8>>();
    let (stderr_tx, stderr_r) = channel::<Vec<u8>>();

    let (exit_tx, exit_rx) = channel::<DemuxExit>();
    let demux = thread::Builder::new()
        .name("supermachine-exec-demux".into())
        .spawn(move || {
            demux_loop(sock_r, stdout_tx, stderr_tx, exit_tx);
        })
        .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("exec: demux thread: {e}")))?;

    let stdin = ExecStdin::new(sock_w.clone());

    Ok(ExecChild {
        sock_w,
        we_signaled: Arc::new(AtomicBool::new(false)),
        aborted: Arc::new(AtomicBool::new(false)),
        interrupt_fd,
        stdout_r: Some(stdout_r),
        stderr_r: Some(stderr_r),
        stdin: Some(stdin),
        demux: Some(demux),
        exit_rx,
    })
}

fn demux_loop(
    mut sock: UnixStream,
    stdout_tx: Sender<Vec<u8>>,
    stderr_tx: Sender<Vec<u8>>,
    exit_tx: Sender<DemuxExit>,
) {
    // Push STDOUT/STDERR payloads into in-memory mpsc channels.
    // When this function returns the senders drop, signalling EOF
    // to ExecStdout/ExecStderr readers (their `recv()` returns
    // Err → `read` returns Ok(0)).
    //
    // Track how much the agent streamed before any premature EOF so
    // `EofBeforeExit` can self-diagnose (no output → never started /
    // VM gone; output-then-EOF → mid-stream drop).
    let mut frames: u64 = 0;
    let mut bytes: u64 = 0;
    loop {
        let (kind, payload) = match read_frame(&mut sock) {
            Ok(f) => f,
            Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
                let _ = exit_tx.send(DemuxExit::EofBeforeExit { frames, bytes });
                return;
            }
            Err(e) => {
                let _ = exit_tx.send(DemuxExit::Io(e));
                return;
            }
        };
        match kind {
            FRAME_STDOUT => {
                frames += 1;
                bytes += payload.len() as u64;
                let _ = stdout_tx.send(payload);
            }
            FRAME_STDERR => {
                frames += 1;
                bytes += payload.len() as u64;
                let _ = stderr_tx.send(payload);
            }
            FRAME_EXIT => {
                let _ = exit_tx.send(decode_exit_payload(&payload));
                return;
            }
            FRAME_ERROR => {
                let msg = String::from_utf8_lossy(&payload).into_owned();
                let _ = exit_tx.send(DemuxExit::Error(msg));
                return;
            }
            _ => {
                // STDIN / RESIZE / SIGNAL / REQUEST shouldn't come
                // from the agent. Ignore unknown frames so the
                // protocol can grow additive types without
                // breaking older hosts.
            }
        }
    }
}

// ---------- frame I/O ----------

fn send_frame(sock: &Arc<Mutex<UnixStream>>, kind: u8, payload: &[u8]) -> io::Result<()> {
    let mut header = [0u8; 5];
    header[0] = kind;
    header[1..5].copy_from_slice(&(payload.len() as u32).to_be_bytes());
    let mut g = sock
        .lock()
        .map_err(|_| io::Error::new(io::ErrorKind::Other, "exec: socket mutex poisoned"))?;
    g.write_all(&header)?;
    if !payload.is_empty() {
        g.write_all(payload)?;
    }
    Ok(())
}

fn read_frame<R: Read>(sock: &mut R) -> io::Result<(u8, Vec<u8>)> {
    let mut header = [0u8; 5];
    sock.read_exact(&mut header)?;
    let kind = header[0];
    let len = u32::from_be_bytes([header[1], header[2], header[3], header[4]]) as usize;
    // Ceiling generous enough that the agent can stream large
    // outputs without us tripping a guard, but tight enough that
    // a malformed length doesn't allocate gigabytes.
    if len > 16 * 1024 * 1024 {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            format!("exec: frame len {len} > 16 MiB"),
        ));
    }
    let mut payload = vec![0u8; len];
    if !payload.is_empty() {
        sock.read_exact(&mut payload)?;
    }
    Ok((kind, payload))
}

// ---------- exit status synthesis ----------

/// Structured detail for an EOF that arrived before the agent sent EXIT
/// (and the host did not signal the child). Wrapped inside the
/// `io::Error` returned by [`ExecChild::wait`] / [`ExecBuilder::output`]
/// (kind `UnexpectedEof`), so callers can `downcast_ref::<ExecEof>()` to
/// branch reliably instead of string-matching the message. The `Display`
/// text is the same self-diagnosing message as before, so existing
/// `to_string()` / log output is unchanged.
#[derive(Debug, Clone, Copy)]
pub struct ExecEof {
    /// The host tore the connection down locally (`abort()` / `Drop`) —
    /// not an agent-side failure.
    pub aborted: bool,
    /// STDOUT/STDERR frames the agent streamed before the close.
    pub frames: u64,
    /// Bytes the agent streamed before the close.
    pub bytes: u64,
}

impl ExecEof {
    /// True when this is a connection-SETUP failure: the agent closed
    /// before producing ANY output or EXIT, and we didn't abort locally.
    /// The workload had not yet streamed anything — re-establishing the
    /// connection and re-sending the request is the *only* class of EOF
    /// that may be safely retried (mid-stream drops, where output was
    /// already delivered, must not be — a retry could double-execute a
    /// side-effecting command). Even here, retry is at-least-once: see
    /// [`ExecBuilder::output_resilient`].
    pub fn is_setup_failure(&self) -> bool {
        !self.aborted && self.frames == 0 && self.bytes == 0
    }
}

impl std::fmt::Display for ExecEof {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        if self.aborted {
            f.write_str(
                "exec: connection closed before EXIT because the host aborted it locally \
                 (abort()/Drop) — not an agent-side failure",
            )
        } else if self.frames == 0 && self.bytes == 0 {
            f.write_str(
                "exec: agent closed the connection before producing any output or EXIT — the \
                 in-guest agent likely failed to start the process, or the VM went away \
                 (guest panic / reboot / OOM-kill)",
            )
        } else {
            write!(
                f,
                "exec: agent closed the connection after {} output frame(s) / {} byte(s) \
                 without sending EXIT — the channel dropped mid-stream (the host may be \
                 oversubscribed; set SUPERMACHINE_WORKER_LOG_FILE to capture worker logs \
                 for a post-mortem)",
                self.frames, self.bytes
            )
        }
    }
}

impl std::error::Error for ExecEof {}

/// Build the self-diagnosing error for a premature EOF (see [`ExecEof`]).
/// Returns an `io::Error` of kind `UnexpectedEof` wrapping the structured
/// detail, so callers can both read the message and `downcast_ref` it.
fn eof_before_exit_error(aborted: bool, frames: u64, bytes: u64) -> io::Error {
    io::Error::new(
        io::ErrorKind::UnexpectedEof,
        ExecEof {
            aborted,
            frames,
            bytes,
        },
    )
}

/// Decode an EXIT frame payload: a 4-byte BE exit code, optionally
/// followed by an 8-byte BE peak-RSS (KiB). A payload SHORTER than the
/// 4-byte code is a truncated/corrupt frame and is reported as an
/// `Error` — NOT silently as a clean exit-0, which would hide a failed
/// workload from a caller that relies on the exit code. The agent fully
/// controls this frame, so a buggy/hostile/congestion-truncated EXIT
/// must fail loud rather than masquerade as success.
fn decode_exit_payload(payload: &[u8]) -> DemuxExit {
    if payload.len() < 4 {
        return DemuxExit::Error(format!(
            "truncated EXIT frame: {} byte(s), need >= 4 for the exit code",
            payload.len()
        ));
    }
    let code = u32::from_be_bytes([payload[0], payload[1], payload[2], payload[3]]);
    let peak_rss_kib = if payload.len() >= 12 {
        Some(u64::from_be_bytes([
            payload[4],
            payload[5],
            payload[6],
            payload[7],
            payload[8],
            payload[9],
            payload[10],
            payload[11],
        ]))
    } else {
        None
    };
    DemuxExit::Status { code, peak_rss_kib }
}

#[cfg(unix)]
fn synthesize_exit(code: u32) -> ExitStatus {
    // ExitStatus on unix wraps a wait status int. The agent
    // already collapsed signal/exit into a u32 (signal -> 128+sig).
    // We re-encode as if WIFEXITED with that exit code: high byte
    // = exit code, low byte = 0. Calling `.code()` then returns
    // `Some(code)` which matches what callers expect.
    use std::os::unix::process::ExitStatusExt;
    ExitStatus::from_raw(((code as i32) & 0xff) << 8)
}

/// 0.7.44+ synthesize a "killed by signal N" ExitStatus when the
/// agent disconnects post-SIGNAL without sending an EXIT frame
/// (vsock data-channel congestion → EXIT can't queue before host
/// watchdog's ABORT_GRACE force-closes the socket).
///
/// Encoding: WIFSIGNALED uses the LOW 7 bits of the wait status
/// for the signal number, with bit 7 cleared (no core dump bit
/// set). `.code()` returns `None`, `.signal()` returns `Some(N)`,
/// and `process::Output::code()`'s legacy mapping returns
/// `Some(128 + N)` — matching the convention every test in the
/// suite uses (`expect(exitCode).toBe(137)` for SIGKILL).
#[cfg(unix)]
fn synthesize_killed_by_signal(sig: i32) -> ExitStatus {
    use std::os::unix::process::ExitStatusExt;
    // WIFSIGNALED wstatus: low 7 bits = signal, bit 7 = core dump.
    // Bit 8 (the WIFEXITED marker) must be CLEAR; bits 9-15 unused.
    ExitStatus::from_raw(sig & 0x7f)
}

// =====================================================================
// Tests — focused on the 0.7.10 API additions (ExecSignaler, abort,
// try_wait, output_with_cancel). Tests that need a real VM live in
// the integration suite under npm/supermachine-core/__test__/.
// =====================================================================
#[cfg(test)]
mod tests {
    use super::*;
    use std::os::unix::net::UnixListener;
    use std::os::unix::process::ExitStatusExt;
    use std::path::PathBuf;

    #[test]
    fn decode_exit_payload_reads_code_and_optional_rss() {
        // 4-byte code only.
        match decode_exit_payload(&[0, 0, 0, 42]) {
            DemuxExit::Status { code, peak_rss_kib } => {
                assert_eq!(code, 42);
                assert_eq!(peak_rss_kib, None);
            }
            _ => panic!("expected Status"),
        }
        // 12-byte: code + peak RSS.
        let mut p = vec![0, 0, 0, 7];
        p.extend_from_slice(&1234u64.to_be_bytes());
        match decode_exit_payload(&p) {
            DemuxExit::Status { code, peak_rss_kib } => {
                assert_eq!(code, 7);
                assert_eq!(peak_rss_kib, Some(1234));
            }
            _ => panic!("expected Status"),
        }
        // 4..11 bytes: code present, partial RSS tolerated as None.
        match decode_exit_payload(&[0, 0, 1, 0, 0xaa, 0xbb]) {
            DemuxExit::Status { code, peak_rss_kib } => {
                assert_eq!(code, 256);
                assert_eq!(peak_rss_kib, None);
            }
            _ => panic!("expected Status"),
        }
    }

    #[test]
    fn decode_exit_payload_truncated_is_error_not_exit_zero() {
        // The fix: a sub-4-byte EXIT frame must surface as an error, NOT
        // silently report a clean exit-0 that would mask a failed command.
        for short in [&[][..], &[0][..], &[0, 0][..], &[0, 0, 0][..]] {
            match decode_exit_payload(short) {
                DemuxExit::Error(msg) => assert!(msg.contains("truncated EXIT"), "{msg}"),
                _ => panic!("{}-byte EXIT frame must be an Error", short.len()),
            }
        }
    }

    #[test]
    fn synthesize_exit_round_trips_exit_code() {
        for c in [0u32, 1, 42, 255] {
            assert_eq!(synthesize_exit(c).code(), Some(c as i32), "code {c}");
        }
        assert!(synthesize_exit(0).signal().is_none());
    }

    #[test]
    fn synthesize_killed_by_signal_reports_signal_not_code() {
        // SIGKILL (9) / SIGTERM (15): `.signal()` set, `.code()` None.
        assert_eq!(synthesize_killed_by_signal(9).signal(), Some(9));
        assert_eq!(synthesize_killed_by_signal(9).code(), None);
        assert_eq!(synthesize_killed_by_signal(15).signal(), Some(15));
    }

    #[test]
    fn eof_before_exit_error_is_self_diagnosing() {
        // Every branch is an UnexpectedEof (callers match on .kind()),
        // but the message must point at the right cause.
        let aborted = eof_before_exit_error(true, 0, 0);
        assert_eq!(aborted.kind(), io::ErrorKind::UnexpectedEof);
        assert!(aborted.to_string().contains("aborted it locally"));

        let never_started = eof_before_exit_error(false, 0, 0);
        assert_eq!(never_started.kind(), io::ErrorKind::UnexpectedEof);
        assert!(never_started
            .to_string()
            .contains("before producing any output"));

        let mid_stream = eof_before_exit_error(false, 7, 4096);
        assert_eq!(mid_stream.kind(), io::ErrorKind::UnexpectedEof);
        let msg = mid_stream.to_string();
        assert!(
            msg.contains("7 output frame"),
            "should report frame count: {msg}"
        );
        assert!(msg.contains("4096 byte"), "should report byte count: {msg}");
        assert!(
            msg.contains("mid-stream"),
            "should flag a mid-stream drop: {msg}"
        );

        // `aborted` takes priority even if some output was seen — a local
        // teardown is a local teardown regardless of prior bytes.
        let aborted_after_output = eof_before_exit_error(true, 7, 4096);
        assert!(aborted_after_output
            .to_string()
            .contains("aborted it locally"));
    }

    #[test]
    fn exec_eof_is_structured_and_downcastable() {
        // The retry predicate must be reliable structured data, not a
        // brittle string match. Only the "never produced output, not
        // aborted" case is a retryable setup failure.
        let setup = eof_before_exit_error(false, 0, 0);
        let eof = setup
            .get_ref()
            .and_then(|e| e.downcast_ref::<ExecEof>())
            .expect("UnexpectedEof must wrap a downcastable ExecEof");
        assert!(
            eof.is_setup_failure(),
            "no output + not aborted ⇒ retryable setup failure"
        );

        // Mid-stream drop: output already delivered ⇒ NOT retryable.
        let mid = eof_before_exit_error(false, 3, 100);
        let mid_eof = mid.get_ref().unwrap().downcast_ref::<ExecEof>().unwrap();
        assert!(
            !mid_eof.is_setup_failure(),
            "mid-stream drop must never be retried"
        );

        // Local abort with zero output: NOT a setup failure (intentional).
        let aborted = eof_before_exit_error(true, 0, 0);
        let ab_eof = aborted
            .get_ref()
            .unwrap()
            .downcast_ref::<ExecEof>()
            .unwrap();
        assert!(
            !ab_eof.is_setup_failure(),
            "local abort is not an agent setup failure"
        );

        // `output_resilient(0)` is exactly `output()` — no retry behaviour
        // to reason about for the default.
        // (Behavioural retry needs a live agent; covered by the napi suite.)
    }

    /// Compile-time bound check: ExecSignaler must be Send + Sync +
    /// Clone. If a future refactor accidentally adds a !Sync field
    /// to ExecSignaler this test fails to build, surfacing the
    /// regression as a compile error rather than a runtime
    /// "your watchdog can't see the signaler" puzzle.
    #[test]
    fn exec_signaler_is_send_sync_clone() {
        fn assert_send<T: Send>() {}
        fn assert_sync<T: Sync>() {}
        fn assert_clone<T: Clone>() {}
        assert_send::<ExecSignaler>();
        assert_sync::<ExecSignaler>();
        assert_clone::<ExecSignaler>();
    }

    /// Compile-time bound check: ExecChild stays Send (the existing
    /// guarantee). Sync was never offered (exit_rx is mpsc::Receiver,
    /// which is !Sync); embedders who need cross-thread control go
    /// through ExecSignaler.
    #[test]
    fn exec_child_is_send() {
        fn assert_send<T: Send>() {}
        assert_send::<ExecChild>();
    }

    /// Spin up a UnixListener that accepts a connection, swallows the
    /// REQUEST frame, then HANGS — never sends EXIT. Without
    /// `abort()`, a `wait()` on the resulting child blocks forever
    /// (or until the test harness kills the process). With
    /// `abort()`, the host-side shutdown wakes the demux thread,
    /// `wait()` returns Err(UnexpectedEof) within milliseconds.
    ///
    /// This is the unit-test analogue of the integrator's "drop the
    /// Vm to interrupt an exec doesn't work" bug — pre-fix the
    /// blocking wait wouldn't unblock; post-fix it does.
    /// A short, unique Unix-socket directory under `/tmp`. macOS caps
    /// `sockaddr_un.sun_path` at 104 bytes; building the socket path
    /// from `std::env::temp_dir()` (a long `$TMPDIR` like
    /// `/var/folders/…`) plus a descriptive subdir name plus
    /// `exec.sock` overflows that → `bind()` fails with ENAMETOOLONG.
    /// Anchoring under `/tmp` with a compact name keeps the path well
    /// under the limit regardless of `$TMPDIR`. The atomic seq gives
    /// per-process uniqueness without a long nanosecond suffix.
    fn short_sock_dir() -> PathBuf {
        use std::sync::atomic::{AtomicU64, Ordering};
        static SEQ: AtomicU64 = AtomicU64::new(0);
        let dir = PathBuf::from("/tmp").join(format!(
            "sm-exec-{}-{}",
            std::process::id(),
            SEQ.fetch_add(1, Ordering::Relaxed)
        ));
        std::fs::create_dir_all(&dir).unwrap();
        dir
    }

    #[test]
    fn signaler_abort_unblocks_blocked_wait() {
        let dir = short_sock_dir();
        let sock_path: PathBuf = dir.join("exec.sock");
        let listener = UnixListener::bind(&sock_path).unwrap();
        // Accept-and-park thread: the moment a client connects, we
        // sleep "forever" (= 60s, the test's outer timeout) without
        // ever sending bytes back. The post-fix abort path is what
        // unblocks the client's wait.
        let listener_handle = thread::spawn(move || {
            if let Ok((mut conn, _addr)) = listener.accept() {
                // Consume the REQUEST frame so the client's send
                // doesn't block. Then hang.
                let mut hdr = [0u8; 5];
                if conn.read_exact(&mut hdr).is_ok() {
                    let len = u32::from_be_bytes([hdr[1], hdr[2], hdr[3], hdr[4]]) as usize;
                    let mut payload = vec![0u8; len];
                    let _ = conn.read_exact(&mut payload);
                }
                // Park forever (or until the connection drops).
                thread::sleep(Duration::from_secs(60));
            }
        });

        // Spawn a child against the hung listener.
        let builder = ExecBuilder::new(sock_path.clone()).argv(["true"]);
        let child = builder.spawn().expect("spawn against listener");
        let signaler = child.signaler();

        // From another thread, abort after a small delay.
        let abort_handle = thread::spawn(move || {
            thread::sleep(Duration::from_millis(50));
            signaler.abort().expect("abort must succeed");
        });

        // Wait. Pre-fix this hangs (until the listener's 60s sleep
        // ends or the test runner times out). Post-fix it returns
        // Err(UnexpectedEof) within ~50ms + RTT.
        let t0 = Instant::now();
        let res = child.wait();
        let elapsed = t0.elapsed();
        let _ = abort_handle.join();
        // Don't join the listener thread — it's parked.

        assert!(
            elapsed < Duration::from_secs(5),
            "wait must unblock quickly after abort; took {:?}",
            elapsed
        );
        assert!(
            res.is_err(),
            "wait after abort must return Err (the agent didn't send EXIT); got {:?}",
            res.as_ref().ok()
        );
        let err = res.unwrap_err();
        assert_eq!(
            err.kind(),
            io::ErrorKind::UnexpectedEof,
            "expected UnexpectedEof, got {:?}: {err}",
            err.kind()
        );

        // Cleanup.
        let _ = std::fs::remove_file(&sock_path);
        let _ = std::fs::remove_dir(&dir);
        // listener thread will be reaped when its sleep ends; in
        // test environment it gets killed on process exit.
        std::mem::forget(listener_handle);
    }

    /// `try_wait()` returns None while the child is running, Some
    /// after it exits (or the connection is aborted). Mirrors
    /// std::process::Child::try_wait's contract.
    #[test]
    fn try_wait_returns_none_then_some_after_abort() {
        let dir = short_sock_dir();
        let sock_path: PathBuf = dir.join("exec.sock");
        let listener = UnixListener::bind(&sock_path).unwrap();
        let listener_handle = thread::spawn(move || {
            if let Ok((mut conn, _)) = listener.accept() {
                let mut hdr = [0u8; 5];
                if conn.read_exact(&mut hdr).is_ok() {
                    let len = u32::from_be_bytes([hdr[1], hdr[2], hdr[3], hdr[4]]) as usize;
                    let mut payload = vec![0u8; len];
                    let _ = conn.read_exact(&mut payload);
                }
                thread::sleep(Duration::from_secs(60));
            }
        });

        let builder = ExecBuilder::new(sock_path.clone()).argv(["true"]);
        let child = builder.spawn().expect("spawn");

        // While the listener parks (no EXIT sent), try_wait sees
        // nothing.
        assert!(
            child.try_wait().expect("try_wait pre-exit").is_none(),
            "try_wait must return None while child is running"
        );

        // Abort the connection. The demux thread observes EOF and
        // posts DemuxExit::EofBeforeExit on exit_rx. try_wait then
        // surfaces that as an Err.
        child.abort().expect("abort");

        // Give the demux thread a moment to observe shutdown and
        // post to exit_rx. Tight bound — actual latency is <1ms in
        // practice.
        let t0 = Instant::now();
        let result = loop {
            match child.try_wait() {
                Ok(Some(_)) => break Ok(()), // unexpected but tolerable
                Ok(None) => {
                    if t0.elapsed() > Duration::from_secs(2) {
                        panic!("try_wait never observed post-abort completion");
                    }
                    thread::sleep(Duration::from_millis(5));
                }
                Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
                    break Err(e);
                }
                Err(e) => panic!("unexpected try_wait error: {e}"),
            }
        };
        // We expect the Err path (no EXIT was sent).
        assert!(result.is_err(), "try_wait must surface UnexpectedEof");

        let _ = std::fs::remove_file(&sock_path);
        let _ = std::fs::remove_dir(&dir);
        std::mem::forget(listener_handle);
    }

    /// Cloning a signaler then calling abort on the clone affects
    /// the original ExecChild. Verifies the Arc-sharing is
    /// connected end-to-end.
    #[test]
    fn signaler_clone_shares_underlying_socket() {
        let dir = short_sock_dir();
        let sock_path: PathBuf = dir.join("exec.sock");
        let listener = UnixListener::bind(&sock_path).unwrap();
        let listener_handle = thread::spawn(move || {
            if let Ok((mut conn, _)) = listener.accept() {
                let mut hdr = [0u8; 5];
                if conn.read_exact(&mut hdr).is_ok() {
                    let len = u32::from_be_bytes([hdr[1], hdr[2], hdr[3], hdr[4]]) as usize;
                    let mut payload = vec![0u8; len];
                    let _ = conn.read_exact(&mut payload);
                }
                thread::sleep(Duration::from_secs(60));
            }
        });

        let builder = ExecBuilder::new(sock_path.clone()).argv(["true"]);
        let child = builder.spawn().expect("spawn");
        let s1 = child.signaler();
        let s2 = s1.clone(); // Arc clone — same underlying socket.
        drop(s1); // Original goes away; abort from the clone still works.

        thread::spawn(move || {
            thread::sleep(Duration::from_millis(50));
            let _ = s2.abort();
        });

        let res = child.wait();
        assert!(res.is_err(), "cloned signaler's abort must unblock wait");

        let _ = std::fs::remove_file(&sock_path);
        let _ = std::fs::remove_dir(&dir);
        std::mem::forget(listener_handle);
    }
}

// Property-based hardening of the exec wire framing (`send_frame` /
// `read_frame`, the 1-byte-type + 4-byte-BE-len + payload protocol the
// host and the guest agent both speak). Two properties:
//
//   1. Roundtrip: anything send_frame writes, read_frame reads back
//      byte-for-byte (kind + payload), for any kind and any payload up
//      to the frame size cap.
//   2. Garbage tolerance: read_frame fed ARBITRARY bytes followed by a
//      peer close never panics and never tries to allocate past the
//      16 MiB ceiling — it returns Ok or a clean Err. This is the
//      hostile-input contract; a missing length cap here would be a
//      remote OOM from a compromised/buggy guest agent.
#[cfg(test)]
mod frame_proptests {
    use super::{read_frame, send_frame};
    use proptest::prelude::*;
    use std::io::Write;
    use std::os::unix::net::UnixStream;
    use std::sync::{Arc, Mutex};

    proptest! {
        #![proptest_config(ProptestConfig::with_cases(512))]

        /// send_frame → read_frame is a lossless roundtrip for any
        /// (kind, payload). Payload kept under the socket buffer so the
        /// single-threaded write can't deadlock against the read.
        #[test]
        fn frame_roundtrips(kind in any::<u8>(), payload in prop::collection::vec(any::<u8>(), 0..8192)) {
            let (tx, mut rx) = UnixStream::pair().unwrap();
            let tx = Arc::new(Mutex::new(tx));
            send_frame(&tx, kind, &payload).unwrap();
            let (got_kind, got_payload) = read_frame(&mut rx).unwrap();
            prop_assert_eq!(got_kind, kind);
            prop_assert_eq!(got_payload, payload);
        }

        /// read_frame must survive arbitrary bytes + peer close: no
        /// panic, no over-cap allocation, no hang. The writer closes
        /// after dumping the bytes, so a length field that over-promises
        /// resolves to a clean UnexpectedEof rather than blocking.
        #[test]
        fn read_frame_tolerates_garbage(raw in prop::collection::vec(any::<u8>(), 0..512)) {
            let (mut tx, mut rx) = UnixStream::pair().unwrap();
            tx.write_all(&raw).ok();
            drop(tx); // EOF — unblocks a read that out-runs the bytes.
            // The contract is "doesn't panic / doesn't hang / doesn't
            // honor a >16 MiB length". Result may be Ok or Err.
            if let Ok((_kind, payload)) = read_frame(&mut rx) {
                prop_assert!(payload.len() <= 16 * 1024 * 1024);
            }
        }
    }

    const CAP: usize = 16 * 1024 * 1024;

    /// A length field of exactly the cap is accepted; one byte over is
    /// rejected with InvalidData BEFORE any payload allocation. This
    /// pins the precise boundary of the remote-OOM guard.
    #[test]
    fn read_frame_length_cap_boundary() {
        use std::io::ErrorKind;

        // cap + 1 → rejected immediately on the header (no payload sent).
        let (mut tx, mut rx) = UnixStream::pair().unwrap();
        let mut hdr = [0u8; 5];
        hdr[0] = 1; // kind
        hdr[1..5].copy_from_slice(&((CAP + 1) as u32).to_be_bytes());
        tx.write_all(&hdr).unwrap();
        drop(tx);
        let err = read_frame(&mut rx).unwrap_err();
        assert_eq!(
            err.kind(),
            ErrorKind::InvalidData,
            "over-cap must be InvalidData"
        );
        assert!(err.to_string().contains("16 MiB"), "got: {err}");

        // Exactly cap → accepted. Write from a thread since a 16 MiB
        // payload exceeds the socket buffer and would deadlock a
        // single-threaded write against the read.
        let (tx, mut rx) = UnixStream::pair().unwrap();
        let tx = Arc::new(Mutex::new(tx));
        let writer = std::thread::spawn(move || {
            send_frame(&tx, 7, &vec![0xabu8; CAP]).unwrap();
        });
        let (kind, payload) = read_frame(&mut rx).unwrap();
        writer.join().unwrap();
        assert_eq!(kind, 7);
        assert_eq!(payload.len(), CAP);
        assert!(payload.iter().all(|&b| b == 0xab));
    }
}