microsandbox 0.5.8

//! Spawning the sandbox process.
//!
//! [`spawn_sandbox`] assembles CLI arguments from [`SandboxConfig`],
//! fork+execs `msb sandbox`, and reads the startup JSON to obtain the
//! sandbox process PID. The sandbox process runs the VMM and agent relay
//! internally.

#[cfg(unix)]
use std::os::fd::AsRawFd;
#[cfg(unix)]
use std::os::unix::ffi::OsStrExt;
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
use std::{
    collections::HashMap,
    ffi::OsString,
    fmt::Write,
    os::fd::{FromRawFd, OwnedFd},
    path::{Path, PathBuf},
    process::Stdio,
};

use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use rand::RngExt;
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, Set};
use serde::{Deserialize, Serialize};
use sha2::{Digest as Sha2Digest, Sha256};
use tempfile::TempDir;
use tokio::{
    io::{AsyncBufRead, AsyncBufReadExt},
    process::Command,
};

use microsandbox_image::{Digest, GlobalCache};
use microsandbox_metrics::{MetricsRegistry, ReserveSlot, SlotReservation};
use microsandbox_protocol::{
    ENV_BLOCK_ROOT, ENV_DIR_MOUNTS, ENV_DISK_MOUNTS, ENV_FILE_MOUNTS, ENV_HANDOFF_INIT,
    ENV_HANDOFF_INIT_ARGS, ENV_HANDOFF_INIT_CWD, ENV_HANDOFF_INIT_ENV, ENV_HOSTNAME,
    ENV_SECURITY_PROFILE, ENV_TMPFS, ENV_USER,
};
use microsandbox_types::SandboxLogLevel;
use microsandbox_utils::{DB_FILENAME, DB_SUBDIR};

use crate::{
    MicrosandboxError, MicrosandboxResult,
    backend::LocalBackend,
    config,
    db::entity::volume as volume_entity,
    runtime::handle::{MetricsReservationCleanup, ProcessHandle},
    sandbox::{
        DiskImageFormat, HostPermissions, MountOptions, NamedVolumeMode, Rlimit, RootfsSource,
        SandboxConfig, StatVirtualization, VolumeMount,
    },
    volume::{
        VolumeConfig, VolumeKind, provision_volume_path, validate_volume_config,
        validate_volume_name,
    },
};

//--------------------------------------------------------------------------------------------------
// Constants
//--------------------------------------------------------------------------------------------------

#[cfg(target_os = "linux")]
static SIGCHLD_ALT_STACK_INIT: tokio::sync::OnceCell<()> = tokio::sync::OnceCell::const_new();

const AGENT_SOCKET_HASH_HEX_LEN: usize = 32;

//--------------------------------------------------------------------------------------------------
// Types
//--------------------------------------------------------------------------------------------------

/// JSON structure read from the sandbox process stdout on startup.
#[derive(Debug, Deserialize)]
struct StartupInfo {
    pid: u32,
}

#[derive(Debug, Clone)]
struct MetricsReservation {
    shm_name: String,
    slot: u32,
    generation: u64,
}

struct Pipe {
    read_fd: OwnedFd,
    write_fd: OwnedFd,
}

/// How the sandbox process should behave relative to the creating process.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum SpawnMode {
    /// The creating process keeps the sandbox handle and agent bridge alive.
    Attached,

    /// The sandbox must survive after the creating process exits.
    Detached,
}

//--------------------------------------------------------------------------------------------------
// Functions
//--------------------------------------------------------------------------------------------------

/// Spawn the sandbox process for a sandbox.
///
/// Returns a [`ProcessHandle`] and the path to the agent relay socket.
///
/// The function:
/// 1. Resolves the `msb` binary path
/// 2. Creates sandbox directories (logs, runtime, scripts)
/// 3. Builds CLI arguments from the config
/// 4. Spawns the hidden `msb sandbox` process with `--agent-sock` for the relay
/// 5. Reads startup JSON from stdout to get child PIDs
pub async fn spawn_sandbox(
    local: &LocalBackend,
    config: &SandboxConfig,
    sandbox_id: i32,
    mode: SpawnMode,
) -> MicrosandboxResult<(ProcessHandle, PathBuf)> {
    // libkrunfw is process-level (one dylib per process address space). The
    // resolver consults MSB_LIBKRUNFW_PATH env, then SDK_LIBKRUNFW_PATH static,
    // then config.paths.libkrunfw, then filesystem fallbacks — see
    // `config::resolve_libkrunfw_path` for the full precedence ladder.
    let global = local.config();
    let msb_path = config::resolve_msb_path(global)?;
    let libkrunfw_path = config::resolve_libkrunfw_path(global)?;
    tracing::debug!(
        msb = %msb_path.display(),
        libkrunfw = %libkrunfw_path.display(),
        sandbox = %config.spec.name,
        cpus = config.spec.resources.cpus,
        memory_mib = config.spec.resources.memory_mib,
        mode = ?mode,
        "spawn_sandbox: resolved paths"
    );

    let sandbox_dir = global.sandboxes_dir().join(&config.spec.name);
    let log_dir = sandbox_dir.join("logs");
    let runtime_dir = sandbox_dir.join("runtime");
    let scripts_dir = runtime_dir.join("scripts");
    let db_dir = global.home().join(DB_SUBDIR);
    let db_path = db_dir.join(DB_FILENAME);

    // Create directories concurrently.
    tokio::try_join!(
        tokio::fs::create_dir_all(&log_dir),
        tokio::fs::create_dir_all(&scripts_dir),
    )?;

    // Write scripts to the runtime scripts directory.
    for (name, content) in &config.spec.runtime.scripts {
        // Prevent path traversal: only use the filename component.
        let safe_name = Path::new(name).file_name().ok_or_else(|| {
            crate::MicrosandboxError::InvalidConfig(format!("invalid script name: {name}"))
        })?;
        let script_path = scripts_dir.join(safe_name);
        tokio::fs::write(&script_path, content).await?;
        #[cfg(unix)]
        tokio::fs::set_permissions(&script_path, std::fs::Permissions::from_mode(0o755)).await?;
    }

    // Compute the agent relay socket path.
    let agent_sock_path = resolve_sandbox_agent_socket_path(&config.spec.name)?;

    // Stage file bind mounts: each file gets its own isolated directory so
    // that virtio-fs (which requires directories) can share it without
    // exposing adjacent files on the host.
    let (staged_file_mounts, file_mounts_staging) = stage_file_mounts(config).await?;
    ensure_named_volumes(local, config).await?;
    let metrics_reservation = if config.effective_metrics_interval().is_some() {
        reserve_metrics_slot(local, config, sandbox_id)
    } else {
        None
    };
    let parent_watchdog = match mode {
        SpawnMode::Attached => match create_parent_watchdog_pipe() {
            Ok(pipe) => Some(pipe),
            Err(err) => {
                release_metrics_reservation(config, metrics_reservation.as_ref());
                return Err(err);
            }
        },
        SpawnMode::Detached => None,
    };
    let startup_pipe = match mode {
        SpawnMode::Attached => None,
        SpawnMode::Detached => match create_startup_pipe() {
            Ok(pipe) => Some(pipe),
            Err(err) => {
                release_metrics_reservation(config, metrics_reservation.as_ref());
                return Err(err);
            }
        },
    };

    // Build the command.
    let mut cmd = Command::new(&msb_path);
    cmd.args(sandbox_cli_args(
        local,
        config,
        sandbox_id,
        &db_path,
        global.database.connect_timeout_secs,
        &log_dir,
        &runtime_dir,
        &agent_sock_path,
        &libkrunfw_path,
        &staged_file_mounts,
        metrics_reservation.as_ref(),
        parent_watchdog
            .as_ref()
            .map(|_| microsandbox_runtime::vm::PARENT_WATCH_FD),
        startup_pipe
            .as_ref()
            .map(|_| microsandbox_runtime::vm::STARTUP_FD),
    ));

    // Prevent the sandbox process from inheriting the parent's terminal on
    // stdin — the VMM's implicit console auto-detects terminals and sets raw
    // mode, which corrupts the parent's terminal output (\n without \r).
    cmd.stdin(Stdio::null());

    if parent_watchdog.is_some() || startup_pipe.is_some() {
        let parent_watch_fd = parent_watchdog
            .as_ref()
            .map(|pipe| pipe.read_fd.as_raw_fd());
        let startup_write_fd = startup_pipe.as_ref().map(|pipe| pipe.write_fd.as_raw_fd());
        unsafe {
            cmd.pre_exec(move || {
                if startup_write_fd.is_some() {
                    detach_from_launcher_session()?;
                }
                if let Some(fd) = parent_watch_fd {
                    dup_inherited_fd(fd, microsandbox_runtime::vm::PARENT_WATCH_FD)?;
                }
                if let Some(fd) = startup_write_fd {
                    dup_inherited_fd(fd, microsandbox_runtime::vm::STARTUP_FD)?;
                }
                Ok(())
            });
        }
    }

    // Capture stdout for attached startup JSON. Detached mode uses a
    // dedicated startup fd so stdio can be severed from the launcher.
    if startup_pipe.is_some() {
        cmd.stdout(Stdio::null());
        cmd.stderr(Stdio::null());
    } else {
        cmd.stdout(Stdio::piped());
        cmd.stderr(Stdio::inherit());
    }

    ensure_sigchld_handler_uses_alt_stack_before_spawn().await?;

    // Spawn the sandbox process.
    let mut child = match cmd.spawn() {
        Ok(child) => child,
        Err(err) => {
            release_metrics_reservation(config, metrics_reservation.as_ref());
            return Err(err.into());
        }
    };

    let _pid = match child.id() {
        Some(pid) => pid,
        None => {
            release_metrics_reservation(config, metrics_reservation.as_ref());
            return Err(crate::MicrosandboxError::Runtime(
                "sandbox process exited immediately".into(),
            ));
        }
    };
    tracing::debug!(pid = _pid, sandbox = %config.spec.name, "spawn_sandbox: process started");

    // Read the startup JSON from the dedicated startup pipe in detached
    // mode, otherwise stdout.
    let mut reader: Box<dyn AsyncBufRead + Send + Unpin> = match startup_pipe {
        Some(pipe) => {
            let Pipe { read_fd, write_fd } = pipe;
            drop(write_fd);
            Box::new(tokio::io::BufReader::new(tokio::fs::File::from_std(
                std::fs::File::from(read_fd),
            )))
        }
        None => {
            let stdout = child.stdout.take().ok_or_else(|| {
                release_metrics_reservation(config, metrics_reservation.as_ref());
                crate::MicrosandboxError::Runtime("failed to capture sandbox stdout".into())
            })?;
            Box::new(tokio::io::BufReader::new(stdout))
        }
    };
    let mut line = String::new();
    match tokio::time::timeout(
        std::time::Duration::from_secs(30),
        reader.read_line(&mut line),
    )
    .await
    {
        Ok(Ok(_)) => {}
        Ok(Err(err)) => {
            terminate_startup_process(&mut child).await;
            release_metrics_reservation(config, metrics_reservation.as_ref());
            return Err(err.into());
        }
        Err(_) => {
            terminate_startup_process(&mut child).await;
            release_metrics_reservation(config, metrics_reservation.as_ref());
            return Err(crate::MicrosandboxError::Runtime(
                "sandbox startup timeout: no JSON received within 30 seconds".into(),
            ));
        }
    }

    let startup: StartupInfo = match serde_json::from_str(line.trim()) {
        Ok(info) => info,
        Err(_) => {
            let status = terminate_startup_process(&mut child).await;
            release_metrics_reservation(config, metrics_reservation.as_ref());
            tracing::debug!(
                raw_line = ?line,
                exit_status = ?status,
                "spawn_sandbox: failed to parse startup JSON"
            );
            return Err(crate::MicrosandboxError::Runtime(format!(
                "sandbox process exited ({status:?}) before sending startup info \
                 (line: {line:?}, check stderr above for details)"
            )));
        }
    };

    tracing::debug!(
        vm_pid = startup.pid,
        agent_sock = %agent_sock_path.display(),
        "spawn_sandbox: startup JSON received"
    );

    let handle = ProcessHandle::new(
        startup.pid,
        config.spec.name.clone(),
        child,
        file_mounts_staging,
        Vec::new(),
        parent_watchdog.map(|pipe| pipe.write_fd),
        metrics_reservation.as_ref().map(|reservation| {
            MetricsReservationCleanup::new(
                reservation.shm_name.clone(),
                reservation.slot,
                reservation.generation,
            )
        }),
    );

    Ok((handle, agent_sock_path))
}

//--------------------------------------------------------------------------------------------------
// Functions: Helpers
//--------------------------------------------------------------------------------------------------

fn reserve_metrics_slot(
    local: &LocalBackend,
    config: &SandboxConfig,
    sandbox_id: i32,
) -> Option<MetricsReservation> {
    let shm_name = local.config().metrics_registry_shm_name();
    let capacity = local.config().metrics_registry_capacity();
    let registry = match MetricsRegistry::open_or_create(&shm_name, capacity) {
        Ok(registry) => registry,
        Err(err) => {
            tracing::warn!(error = %err, sandbox = %config.spec.name, "failed to open metrics registry");
            return None;
        }
    };
    let memory_limit_bytes = u64::from(config.spec.resources.memory_mib) * 1024 * 1024;
    match registry.reserve(ReserveSlot {
        sandbox_id,
        name: &config.spec.name,
        memory_limit_bytes,
    }) {
        Ok(SlotReservation { slot, generation }) => Some(MetricsReservation {
            shm_name,
            slot,
            generation,
        }),
        Err(err) => {
            tracing::warn!(error = %err, sandbox = %config.spec.name, "failed to reserve metrics slot");
            None
        }
    }
}

fn create_parent_watchdog_pipe() -> MicrosandboxResult<Pipe> {
    create_pipe()
}

fn create_startup_pipe() -> MicrosandboxResult<Pipe> {
    create_pipe()
}

fn create_pipe() -> MicrosandboxResult<Pipe> {
    let mut fds = [0; 2];
    let rc = create_cloexec_pipe(&mut fds);
    if rc != 0 {
        return Err(std::io::Error::last_os_error().into());
    }

    let read_fd = unsafe { OwnedFd::from_raw_fd(fds[0]) };
    let write_fd = unsafe { OwnedFd::from_raw_fd(fds[1]) };

    #[cfg(not(target_os = "linux"))]
    {
        set_cloexec(&read_fd, true)?;
        set_cloexec(&write_fd, true)?;
    }

    Ok(Pipe { read_fd, write_fd })
}

fn dup_inherited_fd(src: i32, dst: i32) -> std::io::Result<()> {
    if unsafe { libc::dup2(src, dst) } < 0 {
        return Err(std::io::Error::last_os_error());
    }
    if src != dst && unsafe { libc::close(src) } < 0 {
        return Err(std::io::Error::last_os_error());
    }
    let flags = unsafe { libc::fcntl(dst, libc::F_GETFD) };
    if flags < 0 {
        return Err(std::io::Error::last_os_error());
    }
    if unsafe { libc::fcntl(dst, libc::F_SETFD, flags & !libc::FD_CLOEXEC) } < 0 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(())
}

fn detach_from_launcher_session() -> std::io::Result<()> {
    if unsafe { libc::setsid() } < 0 {
        return Err(std::io::Error::last_os_error());
    }

    let mut action: libc::sigaction = unsafe { std::mem::zeroed() };
    action.sa_sigaction = libc::SIG_IGN;
    if unsafe { libc::sigemptyset(&mut action.sa_mask) } < 0 {
        return Err(std::io::Error::last_os_error());
    }
    if unsafe { libc::sigaction(libc::SIGHUP, &action, std::ptr::null_mut()) } < 0 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(())
}

#[cfg(target_os = "linux")]
fn create_cloexec_pipe(fds: &mut [i32; 2]) -> i32 {
    unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC) }
}

#[cfg(not(target_os = "linux"))]
fn create_cloexec_pipe(fds: &mut [i32; 2]) -> i32 {
    unsafe { libc::pipe(fds.as_mut_ptr()) }
}

#[cfg(not(target_os = "linux"))]
fn set_cloexec(fd: &OwnedFd, enabled: bool) -> MicrosandboxResult<()> {
    let current = unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_GETFD) };
    if current < 0 {
        return Err(std::io::Error::last_os_error().into());
    }

    let mut next = current;
    if enabled {
        next |= libc::FD_CLOEXEC;
    } else {
        next &= !libc::FD_CLOEXEC;
    }

    if unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_SETFD, next) } < 0 {
        return Err(std::io::Error::last_os_error().into());
    }

    Ok(())
}

fn release_metrics_reservation(config: &SandboxConfig, reservation: Option<&MetricsReservation>) {
    let Some(reservation) = reservation else {
        return;
    };
    let registry = match MetricsRegistry::open(&reservation.shm_name) {
        Ok(registry) => registry,
        Err(err) => {
            tracing::debug!(error = %err, sandbox = %config.spec.name, "release: failed to open metrics registry");
            return;
        }
    };
    if let Err(err) = registry.release_reserved(reservation.slot, reservation.generation) {
        tracing::debug!(error = %err, sandbox = %config.spec.name, "release: metrics slot release failed");
    }
}

#[cfg(target_os = "linux")]
async fn ensure_sigchld_handler_uses_alt_stack_before_spawn() -> MicrosandboxResult<()> {
    SIGCHLD_ALT_STACK_INIT
        .get_or_try_init(|| async {
            install_tokio_sigchld_handler()?;
            patch_sigchld_handler_uses_alt_stack();
            Ok::<(), MicrosandboxError>(())
        })
        .await?;
    Ok(())
}

#[cfg(not(target_os = "linux"))]
async fn ensure_sigchld_handler_uses_alt_stack_before_spawn() -> MicrosandboxResult<()> {
    Ok(())
}

#[cfg(target_os = "linux")]
fn install_tokio_sigchld_handler() -> MicrosandboxResult<()> {
    let signal = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::child())?;
    let _ = Box::leak(Box::new(signal));
    Ok(())
}

#[cfg(target_os = "linux")]
fn patch_sigchld_handler_uses_alt_stack() {
    unsafe {
        let mut action = std::mem::MaybeUninit::<libc::sigaction>::uninit();
        if libc::sigaction(libc::SIGCHLD, std::ptr::null(), action.as_mut_ptr()) != 0 {
            return;
        }

        let mut action = action.assume_init();
        if action.sa_flags & libc::SA_ONSTACK != 0 {
            return;
        }

        action.sa_flags |= libc::SA_ONSTACK;
        let _ = libc::sigaction(libc::SIGCHLD, &action, std::ptr::null_mut());
    }
}

async fn ensure_named_volumes(
    local: &LocalBackend,
    config: &SandboxConfig,
) -> MicrosandboxResult<()> {
    for mount in &config.spec.mounts {
        let Some(create) = mount.named_create() else {
            continue;
        };

        validate_volume_name(create.name())?;
        let pools = local.db().await?;
        let existing = volume_entity::Entity::find()
            .filter(volume_entity::Column::Name.eq(create.name()))
            .one(pools.read())
            .await?;

        if existing.is_some() {
            match create.mode() {
                NamedVolumeMode::Create => {
                    return Err(MicrosandboxError::VolumeAlreadyExists(
                        create.name().to_string(),
                    ));
                }
                NamedVolumeMode::EnsureExists | NamedVolumeMode::Existing => continue,
            }
        }

        if create.mode() == NamedVolumeMode::Existing {
            return Err(MicrosandboxError::VolumeNotFound(create.name().to_string()));
        }

        let volume_config = VolumeConfig {
            name: create.name().to_string(),
            kind: create.kind(),
            quota_mib: create.quota_mib(),
            capacity_mib: create.capacity_mib(),
            labels: create.labels().to_vec(),
        };
        validate_volume_config(&volume_config)?;

        let labels_json = if create.labels().is_empty() {
            None
        } else {
            Some(serde_json::to_string(create.labels())?)
        };
        let now = chrono::Utc::now().naive_utc();
        let capacity_bytes = volume_config
            .capacity_mib
            .map(|mib| i64::from(mib) * 1024 * 1024);
        let model = volume_entity::ActiveModel {
            name: Set(volume_config.name.clone()),
            kind: Set(volume_config.kind.as_str().to_string()),
            quota_mib: Set(volume_config.quota_mib.map(|value| value as i32)),
            size_bytes: Set(None),
            capacity_bytes: Set(capacity_bytes),
            disk_format: Set((volume_config.kind == VolumeKind::Disk).then(|| "raw".to_string())),
            disk_fstype: Set((volume_config.kind == VolumeKind::Disk).then(|| "ext4".to_string())),
            labels: Set(labels_json),
            created_at: Set(Some(now)),
            updated_at: Set(Some(now)),
            ..Default::default()
        };
        volume_entity::Entity::insert(model)
            .exec(pools.write())
            .await?;
        provision_volume_path(&volume_config, &local.volume_path(&volume_config.name)).await?;
    }

    Ok(())
}

/// Return agent relay socket paths in preferred connection order.
pub(crate) fn sandbox_agent_socket_path_candidates(name: &str) -> [PathBuf; 2] {
    let (run_dir, sandboxes_dir) = crate::backend::default_backend()
        .as_local()
        .map(|local| (local.config().run_dir(), local.config().sandboxes_dir()))
        .unwrap_or_else(|| {
            let home = microsandbox_utils::resolve_home();
            (
                home.join(microsandbox_utils::RUN_SUBDIR),
                home.join(microsandbox_utils::SANDBOXES_SUBDIR),
            )
        });
    sandbox_agent_socket_path_candidates_with_roots(&run_dir, &sandboxes_dir, name)
}

pub(crate) fn sandbox_agent_socket_path_candidates_for(
    local: &LocalBackend,
    name: &str,
) -> [PathBuf; 2] {
    sandbox_agent_socket_path_candidates_with_roots(
        &local.config().run_dir(),
        &local.config().sandboxes_dir(),
        name,
    )
}

fn sandbox_agent_socket_path_candidates_with_roots(
    run_dir: &Path,
    sandboxes_dir: &Path,
    name: &str,
) -> [PathBuf; 2] {
    [
        sandbox_agent_socket_path(run_dir, name),
        legacy_sandbox_agent_socket_path(sandboxes_dir, name),
    ]
}

/// Pick the first socket path usable on this platform.
pub(crate) fn resolve_sandbox_agent_socket_path(name: &str) -> MicrosandboxResult<PathBuf> {
    let candidates = sandbox_agent_socket_path_candidates(name);
    for path in &candidates {
        if sandbox_agent_socket_path_fits(path) {
            return Ok(path.clone());
        }
    }

    let shortest = candidates
        .iter()
        .map(|path| sandbox_agent_socket_path_len(path))
        .min()
        .unwrap_or(0);
    Err(crate::MicrosandboxError::InvalidConfig(format!(
        "agent relay socket path is too long: shortest derived path is {shortest} bytes, \
         but Unix socket paths on this platform must be shorter than {} bytes; set \
         MSB_HOME or paths.sandboxes to a shorter directory",
        unix_socket_path_capacity()
    )))
}

fn sandbox_agent_socket_path(run_dir: &Path, name: &str) -> PathBuf {
    let mut hasher = Sha256::new();
    hasher.update(name.as_bytes());
    let digest = hasher.finalize();

    let mut filename = String::with_capacity(AGENT_SOCKET_HASH_HEX_LEN + ".sock".len());
    for byte in digest.iter().take(AGENT_SOCKET_HASH_HEX_LEN / 2) {
        let _ = Write::write_fmt(&mut filename, format_args!("{byte:02x}"));
    }
    filename.push_str(".sock");

    run_dir.join("agent").join(filename)
}

fn legacy_sandbox_agent_socket_path(sandboxes_dir: &Path, name: &str) -> PathBuf {
    sandboxes_dir.join(name).join("runtime").join("agent.sock")
}

#[cfg(unix)]
fn sandbox_agent_socket_path_fits(path: &Path) -> bool {
    sandbox_agent_socket_path_len(path) < unix_socket_path_capacity()
}

#[cfg(not(unix))]
fn sandbox_agent_socket_path_fits(_path: &Path) -> bool {
    true
}

#[cfg(unix)]
fn sandbox_agent_socket_path_len(path: &Path) -> usize {
    path.as_os_str().as_bytes().len()
}

#[cfg(not(unix))]
fn sandbox_agent_socket_path_len(_path: &Path) -> usize {
    0
}

#[cfg(unix)]
fn unix_socket_path_capacity() -> usize {
    let storage = unsafe { std::mem::zeroed::<libc::sockaddr_un>() };
    storage.sun_path.len()
}

#[cfg(not(unix))]
fn unix_socket_path_capacity() -> usize {
    usize::MAX
}

async fn terminate_startup_process(
    child: &mut tokio::process::Child,
) -> Option<std::process::ExitStatus> {
    let _ = child.start_kill();
    child.wait().await.ok()
}

/// Scan `config.spec.mounts` for file bind mounts and stage each file in its own
/// isolated directory inside an ephemeral [`TempDir`].
///
/// Returns a map from guest path to `(file_mount_dir, filename, tag)` for
/// each staged file, plus the `TempDir` handle that must be kept alive for
/// the VM's lifetime.
async fn stage_file_mounts(
    config: &SandboxConfig,
) -> MicrosandboxResult<(HashMap<String, (PathBuf, String, String)>, Option<TempDir>)> {
    // Collect file bind mounts first so we can skip TempDir creation when
    // there are none.
    let file_mounts: Vec<_> = config
        .spec
        .mounts
        .iter()
        .filter_map(|m| match m {
            VolumeMount::Bind {
                host,
                guest,
                options,
                ..
            } if host.is_file() => Some((host, guest, options.readonly)),
            _ => None,
        })
        .collect();

    if file_mounts.is_empty() {
        return Ok((HashMap::new(), None));
    }

    let tempdir = tempfile::tempdir()?;
    let mut staged = HashMap::new();

    for (host, guest, readonly) in file_mounts {
        // Generate a random tag to avoid collisions.
        let id: u32 = rand::rng().random();
        let tag = format!("fm_{id:08x}");

        let file_mount_dir = tempdir.path().join(&tag);
        tokio::fs::create_dir_all(&file_mount_dir).await?;

        let filename_os = host.file_name().ok_or_else(|| {
            crate::MicrosandboxError::InvalidConfig(format!(
                "file mount has no filename: {}",
                host.display()
            ))
        })?;

        let filename = filename_os.to_str().ok_or_else(|| {
            crate::MicrosandboxError::InvalidConfig(format!(
                "file mount filename is not valid UTF-8: {}",
                host.display()
            ))
        })?;

        // The MSB_FILE_MOUNTS protocol uses `:` and `;` as delimiters.
        if filename.contains(':') || filename.contains(';') {
            return Err(crate::MicrosandboxError::InvalidConfig(format!(
                "file mount filename must not contain ':' or ';': {filename}"
            )));
        }

        let target = file_mount_dir.join(filename);

        // Hard-link preserves the same inode — writes in the guest propagate
        // to the host and vice-versa. Falls back to copy for cross-filesystem
        // mounts (different device IDs).
        match tokio::fs::hard_link(host, &target).await {
            Ok(()) => {
                tracing::debug!(
                    host = %host.display(),
                    file_mount_dir = %target.display(),
                    "file mount: hard-linked"
                );
            }
            Err(e) if e.raw_os_error() == Some(libc::EXDEV) => {
                if !readonly {
                    tracing::warn!(
                        host = %host.display(),
                        file_mount_dir = %target.display(),
                        "file mount: cross-filesystem, falling back to copy \
                         (guest writes will NOT propagate to host)"
                    );
                } else {
                    tracing::debug!(
                        host = %host.display(),
                        file_mount_dir = %target.display(),
                        "file mount: cross-filesystem, copying (read-only)"
                    );
                }
                tokio::fs::copy(host, &target).await?;
            }
            Err(e) => return Err(e.into()),
        }

        staged.insert(guest.clone(), (file_mount_dir, filename.to_string(), tag));
    }

    Ok((staged, Some(tempdir)))
}

/// Push a `--mount tag:host_path[:ro]` arg pair.
fn push_dir_mount_arg(
    args: &mut Vec<OsString>,
    guest: &str,
    host_display: &impl std::fmt::Display,
    options: MountOptions,
    stat_virtualization: StatVirtualization,
    host_permissions: HostPermissions,
) {
    let tag = guest_mount_tag(guest);
    let mut arg = format!("{tag}:{host_display}");
    let mut opts = mount_option_tokens(options);
    append_policy_options(&mut opts, stat_virtualization, host_permissions);
    append_option_block(&mut arg, opts);
    args.push(OsString::from("--mount"));
    args.push(OsString::from(arg));
}

/// Append a `tag:guest_path[:ro]` entry to the `MSB_DIR_MOUNTS` env var value.
fn push_dir_mounts_spec(dir_mounts_val: &mut String, guest: &str, options: MountOptions) {
    if !dir_mounts_val.is_empty() {
        dir_mounts_val.push(';');
    }
    let tag = guest_mount_tag(guest);
    dir_mounts_val.push_str(&tag);
    dir_mounts_val.push(':');
    dir_mounts_val.push_str(guest);
    append_option_block(dir_mounts_val, mount_option_tokens(options));
}

/// Push a `--mount fm_tag:file_mount_dir[:ro]` arg pair.
fn push_file_mount_arg(
    args: &mut Vec<OsString>,
    tag: &str,
    file_mount_dir: &Path,
    options: MountOptions,
    stat_virtualization: StatVirtualization,
    host_permissions: HostPermissions,
) {
    let mut arg = format!("{tag}:{}", file_mount_dir.display());
    let mut opts = mount_option_tokens(options);
    append_policy_options(&mut opts, stat_virtualization, host_permissions);
    append_option_block(&mut arg, opts);
    args.push(OsString::from("--mount"));
    args.push(OsString::from(arg));
}

/// Push a `--disk id:host_path:format[:ro]` arg pair.
fn push_disk_mount_arg(
    args: &mut Vec<OsString>,
    id: &str,
    host_display: &impl std::fmt::Display,
    format: &DiskImageFormat,
    options: MountOptions,
) {
    let mut arg = format!("{id}:{host_display}:{}", format.as_str());
    if options.readonly {
        arg.push_str(":ro");
    }
    args.push(OsString::from("--disk"));
    args.push(OsString::from(arg));
}

/// Append a `id:guest_path[:opts]` entry to the `MSB_DISK_MOUNTS` env var value.
fn push_disk_mounts_spec(
    disk_mounts_val: &mut String,
    id: &str,
    guest: &str,
    fstype: Option<&str>,
    options: MountOptions,
) {
    if !disk_mounts_val.is_empty() {
        disk_mounts_val.push(';');
    }
    disk_mounts_val.push_str(id);
    disk_mounts_val.push(':');
    disk_mounts_val.push_str(guest);
    let mut opts = mount_option_tokens(options);
    if let Some(fs) = fstype {
        opts.push(format!("fstype={fs}"));
    }
    append_option_block(disk_mounts_val, opts);
}

/// Append a `tag:filename:guest_path[:ro]` entry to the `MSB_FILE_MOUNTS` env var value.
fn push_file_mounts_spec(
    file_mounts_val: &mut String,
    tag: &str,
    filename: &str,
    guest: &str,
    options: MountOptions,
) {
    if !file_mounts_val.is_empty() {
        file_mounts_val.push(';');
    }
    file_mounts_val.push_str(tag);
    file_mounts_val.push(':');
    file_mounts_val.push_str(filename);
    file_mounts_val.push(':');
    file_mounts_val.push_str(guest);
    append_option_block(file_mounts_val, mount_option_tokens(options));
}

fn mount_option_tokens(options: MountOptions) -> Vec<String> {
    let mut tokens = Vec::new();
    if options.readonly {
        tokens.push("ro".to_string());
    }
    if options.noexec {
        tokens.push("noexec".to_string());
    }
    if options.nosuid {
        tokens.push("nosuid".to_string());
    }
    if options.nodev {
        tokens.push("nodev".to_string());
    }
    tokens
}

fn append_policy_options(
    opts: &mut Vec<String>,
    stat_virtualization: StatVirtualization,
    host_permissions: HostPermissions,
) {
    match stat_virtualization {
        StatVirtualization::Strict => {}
        StatVirtualization::Relaxed => opts.push("stat-virt=relaxed".to_string()),
        StatVirtualization::Off => opts.push("stat-virt=off".to_string()),
    }
    match host_permissions {
        HostPermissions::Private => {}
        HostPermissions::Mirror => opts.push("host-perms=mirror".to_string()),
    }
}

fn append_option_block(spec: &mut String, opts: Vec<String>) {
    if opts.is_empty() {
        return;
    }
    spec.push(':');
    spec.push_str(&opts.join(","));
}

/// Encodes sandbox-wide rlimits for the guest init environment.
fn encode_rlimits(rlimits: &[Rlimit]) -> String {
    use std::fmt::Write;

    let mut out = String::with_capacity(rlimits.len() * 32);
    for (i, rlimit) in rlimits.iter().enumerate() {
        if i > 0 {
            out.push(';');
        }
        write!(
            out,
            "{}={}:{}",
            rlimit.resource.as_str(),
            rlimit.soft,
            rlimit.hard
        )
        .expect("writing to String cannot fail");
    }
    out
}

/// Encodes a handoff-init argv/env payload into printable env-var text.
fn encode_handoff_json<T: Serialize>(value: &T) -> String {
    let json = serde_json::to_vec(value).expect("handoff init payload is JSON-serializable");
    URL_SAFE_NO_PAD.encode(json)
}

/// Derive a stable, collision-resistant identifier from a guest mount path.
///
/// Used for virtiofs tags and for virtio-blk `serial` fields (the block id
/// agentd resolves via `/dev/disk/by-id/virtio-<id>`). The naive `/` → `_`
/// mangling collides for adversarial inputs (`/var/log` and `/var_log` both
/// produce `var_log`), so we append a short sha256-derived suffix.
///
/// Output is at most 20 bytes — the kernel's virtio-blk serial length limit.
/// Layout: `<slug[..11]>_<8-hex>`. The slug-part is a debugging hint; the
/// 8-hex suffix is what actually disambiguates.
fn guest_mount_tag(guest_path: &str) -> String {
    use std::fmt::Write as _;

    const SLUG_MAX: usize = 11;
    const HASH_HEX_LEN: usize = 8;

    let slug: String = guest_path
        .replace('/', "_")
        .trim_start_matches('_')
        .chars()
        .take(SLUG_MAX)
        .collect();

    let mut hasher = Sha256::new();
    hasher.update(guest_path.as_bytes());
    let digest = hasher.finalize();

    // Total layout: optional `<slug>_` prefix + HASH_HEX_LEN hex chars.
    let mut out = String::with_capacity(slug.len() + 1 + HASH_HEX_LEN);
    if !slug.is_empty() {
        out.push_str(&slug);
        out.push('_');
    }
    for byte in digest.iter().take(HASH_HEX_LEN / 2) {
        // write! to a String can't fail.
        let _ = write!(out, "{byte:02x}");
    }
    out
}

/// Build the `msb sandbox` CLI args for a sandbox.
#[allow(clippy::too_many_arguments)]
fn sandbox_cli_args(
    local: &LocalBackend,
    config: &SandboxConfig,
    sandbox_id: i32,
    db_path: &Path,
    db_connect_timeout_secs: u64,
    log_dir: &Path,
    runtime_dir: &Path,
    agent_sock_path: &Path,
    libkrunfw_path: &Path,
    staged_file_mounts: &HashMap<String, (PathBuf, String, String)>,
    metrics_reservation: Option<&MetricsReservation>,
    parent_watch_fd: Option<i32>,
    startup_fd: Option<i32>,
) -> Vec<OsString> {
    let mut args = vec![OsString::from("sandbox")];

    if let Some(log_level) = config.spec.runtime.log_level {
        args.push(OsString::from(sandbox_log_level_cli_flag(log_level)));
    }

    args.push(OsString::from("--name"));
    args.push(OsString::from(&config.spec.name));
    args.push(OsString::from("--sandbox-id"));
    args.push(OsString::from(sandbox_id.to_string()));
    args.push(OsString::from("--db-path"));
    args.push(db_path.as_os_str().to_os_string());
    args.push(OsString::from("--db-connect-timeout-secs"));
    args.push(OsString::from(db_connect_timeout_secs.to_string()));
    args.push(OsString::from("--log-dir"));
    args.push(log_dir.as_os_str().to_os_string());
    args.push(OsString::from("--runtime-dir"));
    args.push(runtime_dir.as_os_str().to_os_string());
    args.push(OsString::from("--sandboxes-dir"));
    args.push(local.sandboxes_dir().as_os_str().to_os_string());
    args.push(OsString::from("--agent-sock"));
    args.push(agent_sock_path.as_os_str().to_os_string());
    if let Some(fd) = parent_watch_fd {
        args.push(OsString::from("--parent-watch-fd"));
        args.push(OsString::from(fd.to_string()));
    }
    if let Some(fd) = startup_fd {
        args.push(OsString::from("--startup-fd"));
        args.push(OsString::from(fd.to_string()));
    }
    push_startup_command_args(&mut args, config);

    let sp = &config.spec.lifecycle;
    if let Some(max_dur) = sp.max_duration_secs {
        args.push(OsString::from("--max-duration"));
        args.push(OsString::from(max_dur.to_string()));
    }
    if let Some(idle) = sp.idle_timeout_secs {
        args.push(OsString::from("--idle-timeout"));
        args.push(OsString::from(idle.to_string()));
    }

    args.push(OsString::from("--libkrunfw-path"));
    args.push(libkrunfw_path.as_os_str().to_os_string());
    args.push(OsString::from("--vcpus"));
    args.push(OsString::from(config.spec.resources.cpus.to_string()));
    args.push(OsString::from("--memory-mib"));
    args.push(OsString::from(config.spec.resources.memory_mib.to_string()));
    match config.effective_metrics_interval() {
        Some(ms) => {
            args.push(OsString::from("--metrics-sample-interval-ms"));
            args.push(OsString::from(ms.get().to_string()));
        }
        None => args.push(OsString::from("--disable-metrics-sample")),
    }
    if let Some(reservation) = metrics_reservation {
        args.push(OsString::from("--metrics-shm-name"));
        args.push(OsString::from(&reservation.shm_name));
        args.push(OsString::from("--metrics-slot"));
        args.push(OsString::from(reservation.slot.to_string()));
        args.push(OsString::from("--metrics-generation"));
        args.push(OsString::from(reservation.generation.to_string()));
    }

    match &config.spec.image {
        RootfsSource::Bind(path) => {
            args.push(OsString::from("--rootfs-path"));
            args.push(path.as_os_str().to_os_string());
        }
        RootfsSource::Oci(_) => {
            // Derive VMDK + upper paths from the stored manifest digest.
            if let Some(ref digest_str) = config.manifest_digest {
                let cache_dir = local.cache_dir();
                let cache = GlobalCache::new(&cache_dir).expect("cache init");
                let digest: Digest = digest_str.parse().expect("invalid manifest digest");
                let vmdk_path = cache.vmdk_path(&digest);

                let sandbox_dir = local.sandboxes_dir().join(&config.spec.name);
                let upper_path = sandbox_dir.join("upper.ext4");

                // VMDK (fsmeta + layers) as read-only block device.
                args.push(OsString::from("--rootfs-disk"));
                args.push(vmdk_path.as_os_str().to_os_string());
                args.push(OsString::from("--rootfs-disk-format"));
                args.push(OsString::from("vmdk"));

                // upper.ext4 as writable block device.
                args.push(OsString::from("--rootfs-blk"));
                args.push(upper_path.as_os_str().to_os_string());

                // MSB_BLOCK_ROOT: always 2 devices.
                let block_root = "kind=oci-erofs,lower=/dev/vda,upper=/dev/vdb,upper_fstype=ext4";
                args.push(OsString::from("--env"));
                args.push(OsString::from(format!("{}={block_root}", ENV_BLOCK_ROOT)));
            }
        }
        RootfsSource::DiskImage {
            path,
            format,
            fstype,
        } => {
            args.push(OsString::from("--rootfs-disk"));
            args.push(path.as_os_str().to_os_string());
            args.push(OsString::from("--rootfs-disk-format"));
            args.push(OsString::from(format.as_str()));

            // Build MSB_BLOCK_ROOT env var value.
            let mut block_root_val = String::from("kind=disk-image,device=/dev/vda");
            if let Some(ft) = fstype {
                block_root_val.push_str(&format!(",fstype={ft}"));
            }
            args.push(OsString::from("--env"));
            args.push(OsString::from(format!(
                "{}={block_root_val}",
                ENV_BLOCK_ROOT
            )));
        }
    }

    // Process mounts: emit --mount args for virtiofs mounts, --disk args
    // for disk-image mounts, and collect guest-side mount specs as env
    // vars for agentd.
    let mut tmpfs_val = String::new();
    let mut dir_mounts_val = String::new();
    let mut file_mounts_val = String::new();
    let mut disk_mounts_val = String::new();
    for mount in &config.spec.mounts {
        match mount {
            VolumeMount::Bind {
                host,
                guest,
                options,
                stat_virtualization,
                host_permissions,
            } => {
                if let Some((file_mount_dir, filename, tag)) = staged_file_mounts.get(guest) {
                    push_file_mount_arg(
                        &mut args,
                        tag,
                        file_mount_dir,
                        *options,
                        *stat_virtualization,
                        *host_permissions,
                    );
                    push_file_mounts_spec(&mut file_mounts_val, tag, filename, guest, *options);
                } else {
                    push_dir_mount_arg(
                        &mut args,
                        guest,
                        &host.display(),
                        *options,
                        *stat_virtualization,
                        *host_permissions,
                    );
                    push_dir_mounts_spec(&mut dir_mounts_val, guest, *options);
                }
            }
            VolumeMount::Named {
                name,
                guest,
                options,
                stat_virtualization,
                host_permissions,
                create: _,
            } => {
                let vol_path = local.volume_path(name);
                push_dir_mount_arg(
                    &mut args,
                    guest,
                    &vol_path.display(),
                    *options,
                    *stat_virtualization,
                    *host_permissions,
                );
                push_dir_mounts_spec(&mut dir_mounts_val, guest, *options);
            }
            VolumeMount::Tmpfs {
                guest,
                size_mib,
                options,
            } => {
                if !tmpfs_val.is_empty() {
                    tmpfs_val.push(';');
                }
                tmpfs_val.push_str(guest);
                let mut opts = Vec::new();
                if let Some(s) = size_mib {
                    opts.push(format!("size={s}"));
                }
                opts.extend(mount_option_tokens(*options));
                append_option_block(&mut tmpfs_val, opts);
            }
            VolumeMount::DiskImage {
                host,
                guest,
                format,
                fstype,
                options,
            } => {
                let id = guest_mount_tag(guest);
                push_disk_mount_arg(&mut args, &id, &host.display(), format, *options);
                push_disk_mounts_spec(
                    &mut disk_mounts_val,
                    &id,
                    guest,
                    fstype.as_deref(),
                    *options,
                );
            }
        }
    }

    if !tmpfs_val.is_empty() {
        args.push(OsString::from("--env"));
        args.push(OsString::from(format!("{}={tmpfs_val}", ENV_TMPFS)));
    }

    if !dir_mounts_val.is_empty() {
        args.push(OsString::from("--env"));
        args.push(OsString::from(format!(
            "{}={dir_mounts_val}",
            ENV_DIR_MOUNTS
        )));
    }

    if !file_mounts_val.is_empty() {
        args.push(OsString::from("--env"));
        args.push(OsString::from(format!(
            "{}={file_mounts_val}",
            ENV_FILE_MOUNTS
        )));
    }

    if !disk_mounts_val.is_empty() {
        args.push(OsString::from("--env"));
        args.push(OsString::from(format!(
            "{}={disk_mounts_val}",
            ENV_DISK_MOUNTS
        )));
    }

    if !config.spec.rlimits.is_empty() {
        args.push(OsString::from("--env"));
        args.push(OsString::from(format!(
            "{}={}",
            microsandbox_protocol::ENV_RLIMITS,
            encode_rlimits(&config.spec.rlimits)
        )));
    }

    // Network configuration.
    #[cfg(feature = "net")]
    {
        let network = config
            .local_network_config()
            .expect("sandbox network spec should decode to local network config");
        let net_json = serde_json::to_string(&network).expect("failed to serialize network config");
        args.push(OsString::from("--network-config"));
        args.push(OsString::from(net_json));
        args.push(OsString::from("--sandbox-slot"));
        args.push(OsString::from(sandbox_id.to_string()));
    }

    for var in &config.spec.env {
        args.push(OsString::from("--env"));
        args.push(OsString::from(format!("{}={}", var.key, var.value)));
    }

    if let Some(ref user) = config.spec.runtime.user {
        args.push(OsString::from("--env"));
        args.push(OsString::from(format!("{}={user}", ENV_USER)));
    }

    args.push(OsString::from("--env"));
    args.push(OsString::from(format!(
        "{}={}",
        ENV_SECURITY_PROFILE,
        match config.spec.security_profile {
            crate::sandbox::SecurityProfile::Default => "default",
            crate::sandbox::SecurityProfile::Restricted => "restricted",
        }
    )));

    // Hostname: explicit value or fall back to a sandbox-name-derived form
    // that fits within the Linux UTS limit.
    {
        let hostname = match config.spec.runtime.hostname.as_deref() {
            Some(h) => h.to_string(),
            None => crate::sandbox::hostname_from_sandbox_name(&config.spec.name),
        };
        args.push(OsString::from("--env"));
        args.push(OsString::from(format!("{}={hostname}", ENV_HOSTNAME)));
    }

    // Handoff-init: PID 1 hand-off to a user-supplied init binary.
    // The builder's `validate()` rejects non-UTF-8 cmd paths, args/env
    // containing NUL, and env keys containing `=`, so the JSON payloads
    // below can't produce a corrupted execve wire format.
    if let Some(ref init) = config.spec.init {
        let cmd = init
            .cmd
            .to_str()
            .expect("validate() rejects non-UTF-8 cmd paths");
        args.push(OsString::from("--env"));
        args.push(OsString::from(format!("{ENV_HANDOFF_INIT}={cmd}")));

        if !init.args.is_empty() {
            let argv_val = encode_handoff_json(&init.args);
            args.push(OsString::from("--env"));
            args.push(OsString::from(format!(
                "{ENV_HANDOFF_INIT_ARGS}={argv_val}"
            )));
        }

        if let Some(ref workdir) = config.spec.runtime.workdir {
            args.push(OsString::from("--env"));
            args.push(OsString::from(format!("{ENV_HANDOFF_INIT_CWD}={workdir}")));
        }

        if !init.env.is_empty() {
            let env_val = encode_handoff_json(&init.env);
            args.push(OsString::from("--env"));
            args.push(OsString::from(format!("{ENV_HANDOFF_INIT_ENV}={env_val}")));
        }
    }

    if let Some(ref workdir) = config.spec.runtime.workdir {
        args.push(OsString::from("--workdir"));
        args.push(OsString::from(workdir));
    }

    args
}

fn push_startup_command_args(args: &mut Vec<OsString>, config: &SandboxConfig) {
    let Some((cmd, cmd_args)) = resolve_startup_command(config) else {
        return;
    };

    args.push(OsString::from(format!("--startup-cmd={cmd}")));
    for arg in cmd_args {
        args.push(OsString::from(format!("--startup-arg={arg}")));
    }
    for var in &config.spec.env {
        args.push(OsString::from(format!(
            "--startup-env={}={}",
            var.key, var.value
        )));
    }
    if let Some(workdir) = &config.spec.runtime.workdir {
        args.push(OsString::from(format!("--startup-cwd={workdir}")));
    }
    if let Some(user) = &config.spec.runtime.user {
        args.push(OsString::from(format!("--startup-user={user}")));
    }
}

fn resolve_startup_command(config: &SandboxConfig) -> Option<(String, Vec<String>)> {
    if !config.startup_command_requested {
        return None;
    }

    match (&config.spec.runtime.entrypoint, &config.spec.runtime.cmd) {
        (Some(entrypoint), cmd) if !entrypoint.is_empty() => {
            let bin = entrypoint[0].clone();
            let args = entrypoint[1..]
                .iter()
                .chain(cmd.iter().flatten())
                .cloned()
                .collect();
            Some((bin, args))
        }
        (_, Some(cmd)) if !cmd.is_empty() => {
            let bin = cmd[0].clone();
            let args = cmd[1..].to_vec();
            Some((bin, args))
        }
        _ => None,
    }
}

fn sandbox_log_level_cli_flag(level: SandboxLogLevel) -> &'static str {
    match level {
        SandboxLogLevel::Error => "--error",
        SandboxLogLevel::Warn => "--warn",
        SandboxLogLevel::Info => "--info",
        SandboxLogLevel::Debug => "--debug",
        SandboxLogLevel::Trace => "--trace",
    }
}

//--------------------------------------------------------------------------------------------------
// Tests
//--------------------------------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::ffi::OsString;
    use std::path::{Path, PathBuf};

    use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
    use microsandbox_types::HandoffInit;
    use serde::de::DeserializeOwned;
    use tempfile::tempdir;

    use super::sandbox_cli_args;
    use crate::{
        LogLevel,
        backend::LocalBackend,
        sandbox::{
            DiskImageFormat, OciRootfsSource, Rlimit, RlimitResource, RootfsSource, SandboxBuilder,
            SandboxConfig,
        },
    };

    //----------------------------------------------------------------------------------------------
    // Functions: Helpers
    //----------------------------------------------------------------------------------------------

    /// Build a `LocalBackend` for tests. Uses `lazy()` since these tests only
    /// exercise the pure-rendering `sandbox_cli_args` path — no DB / FS
    /// touches.
    fn test_local_backend() -> LocalBackend {
        LocalBackend::lazy()
    }

    fn render_args(config: &SandboxConfig) -> Vec<String> {
        let local = test_local_backend();
        sandbox_cli_args(
            &local,
            config,
            42,
            Path::new("/tmp/msb.db"),
            30,
            Path::new("/tmp/logs"),
            Path::new("/tmp/runtime"),
            Path::new("/tmp/agent.sock"),
            Path::new("/tmp/libkrunfw.dylib"),
            &HashMap::new(),
            None,
            None,
            None,
        )
        .iter()
        .map(|arg| arg.to_string_lossy().into_owned())
        .collect()
    }

    fn decode_handoff_json<T: DeserializeOwned>(value: &str) -> T {
        let json = URL_SAFE_NO_PAD.decode(value).expect("base64url payload");
        serde_json::from_slice(&json).expect("handoff JSON payload")
    }

    fn render_args_with_file_mounts(
        config: &SandboxConfig,
        staged_file_mounts: &HashMap<String, (PathBuf, String, String)>,
    ) -> Vec<String> {
        let local = test_local_backend();
        sandbox_cli_args(
            &local,
            config,
            42,
            Path::new("/tmp/msb.db"),
            30,
            Path::new("/tmp/logs"),
            Path::new("/tmp/runtime"),
            Path::new("/tmp/agent.sock"),
            Path::new("/tmp/libkrunfw.dylib"),
            staged_file_mounts,
            None,
            None,
            None,
        )
        .iter()
        .map(|arg| arg.to_string_lossy().into_owned())
        .collect()
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_include_selected_log_level() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .log_level(LogLevel::Debug)
            .build()
            .await
            .unwrap();

        let args = render_args(&config);

        assert!(args.iter().any(|arg| arg == "--debug"));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_are_silent_by_default() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .build()
            .await
            .unwrap();

        let args = render_args(&config);

        assert!(!args.iter().any(|arg| {
            matches!(
                arg.as_str(),
                "--error" | "--warn" | "--info" | "--debug" | "--trace"
            )
        }));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_include_agent_sock_path() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .build()
            .await
            .unwrap();

        let rendered = render_args(&config);

        assert!(
            rendered
                .windows(2)
                .any(|pair| pair == ["--agent-sock", "/tmp/agent.sock"])
        );
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_include_startup_fd_when_supplied() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .build()
            .await
            .unwrap();

        let local = test_local_backend();
        let args = sandbox_cli_args(
            &local,
            &config,
            42,
            Path::new("/tmp/msb.db"),
            30,
            Path::new("/tmp/logs"),
            Path::new("/tmp/runtime"),
            Path::new("/tmp/agent.sock"),
            Path::new("/tmp/libkrunfw.dylib"),
            &HashMap::new(),
            None,
            None,
            Some(microsandbox_runtime::vm::STARTUP_FD),
        );

        assert!(args.windows(2).any(|pair| pair
            == [
                OsString::from("--startup-fd"),
                OsString::from(microsandbox_runtime::vm::STARTUP_FD.to_string()),
            ]));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_include_detached_startup_command() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .entrypoint(["/entrypoint"])
            .env("APP_ENV", "test")
            .workdir("/workspace")
            .user("nobody")
            .persistent_initial_command(["/bin/sh", "-lc", "echo detached"])
            .build()
            .await
            .unwrap();

        let rendered = render_args(&config);

        assert!(rendered.contains(&"--startup-cmd=/entrypoint".to_string()));
        assert!(rendered.contains(&"--startup-arg=/bin/sh".to_string()));
        assert!(rendered.contains(&"--startup-arg=-lc".to_string()));
        assert!(rendered.contains(&"--startup-arg=echo detached".to_string()));
        assert!(rendered.contains(&"--startup-env=APP_ENV=test".to_string()));
        assert!(rendered.contains(&"--startup-cwd=/workspace".to_string()));
        assert!(rendered.contains(&"--startup-user=nobody".to_string()));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_skip_startup_exec_when_init_owns_argv() {
        let mut config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .workdir("/opt/hermes")
            .persistent_initial_command(["gateway", "run"])
            .build()
            .await
            .unwrap();
        config.spec.init = Some(HandoffInit {
            cmd: PathBuf::from("/init"),
            args: vec![
                "/opt/hermes/docker/main-wrapper.sh".to_string(),
                "gateway".to_string(),
                "run".to_string(),
            ],
            env: Vec::new(),
        });
        config.startup_command_requested = false;

        let rendered = render_args(&config);

        assert_eq!(
            find_env(&rendered, "MSB_HANDOFF_INIT").as_deref(),
            Some("/init")
        );
        let argv = find_env(&rendered, "MSB_HANDOFF_INIT_ARGS").expect("argv env present");
        let decoded: Vec<String> = decode_handoff_json(&argv);
        assert_eq!(
            decoded,
            vec![
                "/opt/hermes/docker/main-wrapper.sh".to_string(),
                "gateway".to_string(),
                "run".to_string(),
            ]
        );
        assert_eq!(
            find_env(&rendered, "MSB_HANDOFF_INIT_CWD").as_deref(),
            Some("/opt/hermes")
        );
        assert!(!rendered.iter().any(|arg| arg.starts_with("--startup-cmd")));
        assert!(!rendered.iter().any(|arg| arg.starts_with("--startup-arg")));
    }

    #[tokio::test]
    async fn test_agent_socket_candidates_follow_explicit_local_backend_paths() {
        let temp = tempdir().unwrap();
        let home = temp.path().join("msb-home");
        let backend = LocalBackend::builder().home(&home).build().await.unwrap();

        let [hashed, legacy] =
            super::sandbox_agent_socket_path_candidates_for(&backend, "sdk-socket-test");

        assert!(hashed.starts_with(backend.config().run_dir().join("agent")));
        assert_eq!(
            legacy,
            backend
                .config()
                .sandboxes_dir()
                .join("sdk-socket-test")
                .join("runtime")
                .join("agent.sock")
        );
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_include_rlimits_env() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .rlimit(RlimitResource::Nofile, 65_535)
            .build()
            .await
            .unwrap();

        let local = test_local_backend();
        let args = sandbox_cli_args(
            &local,
            &config,
            42,
            Path::new("/tmp/msb.db"),
            30,
            Path::new("/tmp/logs"),
            Path::new("/tmp/runtime"),
            Path::new("/tmp/agent.sock"),
            Path::new("/tmp/libkrunfw.dylib"),
            &HashMap::new(),
            None,
            None,
            None,
        );

        let rendered = args
            .iter()
            .map(|arg| arg.to_string_lossy().into_owned())
            .collect::<Vec<_>>();

        assert!(rendered.windows(2).any(|pair| {
            pair[0] == "--env"
                && pair[1] == format!("{}=nofile=65535:65535", microsandbox_protocol::ENV_RLIMITS)
        }));
    }

    #[tokio::test]
    async fn test_encode_rlimits_round_trips_through_protocol_parser() {
        use microsandbox_protocol::exec::ExecRlimit;

        let rlimits = vec![
            Rlimit {
                resource: RlimitResource::Nofile,
                soft: 4096,
                hard: 65_535,
            },
            Rlimit {
                resource: RlimitResource::Nproc,
                soft: 1024,
                hard: 1024,
            },
        ];

        let encoded = super::encode_rlimits(&rlimits);
        let parsed: Vec<ExecRlimit> = encoded
            .split(';')
            .map(|entry| entry.parse::<ExecRlimit>().unwrap())
            .collect();

        assert_eq!(parsed.len(), 2);
        assert_eq!(parsed[0].resource, "nofile");
        assert_eq!(parsed[0].soft, 4096);
        assert_eq!(parsed[0].hard, 65_535);
        assert_eq!(parsed[1].resource, "nproc");
        assert_eq!(parsed[1].soft, 1024);
        assert_eq!(parsed[1].hard, 1024);
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_emit_metrics_interval_flag() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .metrics_sample_interval(std::time::Duration::from_millis(1000))
            .build()
            .await
            .unwrap();

        let rendered = render_args(&config);

        assert!(
            rendered
                .windows(2)
                .any(|pair| pair == ["--metrics-sample-interval-ms", "1000"]),
            "expected metrics interval flag in {rendered:?}"
        );
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_include_custom_metrics_sample_interval() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .metrics_sample_interval(std::time::Duration::from_millis(2500))
            .build()
            .await
            .unwrap();

        let rendered = render_args(&config);

        assert!(
            rendered
                .windows(2)
                .any(|pair| pair == ["--metrics-sample-interval-ms", "2500"]),
            "expected custom metrics interval flag in {rendered:?}"
        );
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_disabled_metrics_emit_disable_flag() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .metrics_sample_interval(std::time::Duration::ZERO)
            .build()
            .await
            .unwrap();

        let rendered = render_args(&config);

        assert!(
            rendered.iter().any(|arg| arg == "--disable-metrics-sample"),
            "expected `--disable-metrics-sample` flag; got {rendered:?}"
        );
        assert!(
            !rendered
                .iter()
                .any(|arg| arg == "--metrics-sample-interval-ms"),
            "should not also emit interval flag; got {rendered:?}"
        );
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_disable_overrides_positive_interval() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .metrics_sample_interval(std::time::Duration::from_millis(2500))
            .disable_metrics_sample()
            .build()
            .await
            .unwrap();

        let rendered = render_args(&config);

        assert!(
            rendered.iter().any(|arg| arg == "--disable-metrics-sample"),
            "expected disable flag to win over positive interval; got {rendered:?}"
        );
        assert!(
            !rendered
                .iter()
                .any(|arg| arg == "--metrics-sample-interval-ms"),
            "should not emit interval flag when disable is set; got {rendered:?}"
        );
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_include_db_connect_timeout() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .build()
            .await
            .unwrap();

        let rendered = render_args(&config);

        assert!(
            rendered
                .windows(2)
                .any(|pair| pair == ["--db-connect-timeout-secs", "30"])
        );
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_use_passthrough_for_bind_rootfs() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .build()
            .await
            .unwrap();

        let rendered = render_args(&config);
        assert!(rendered.contains(&"--rootfs-path".to_string()));
        assert!(rendered.contains(&"/tmp/rootfs".to_string()));
        assert!(!rendered.contains(&"--rootfs-lower".to_string()));
        assert!(!rendered.contains(&"--rootfs-upper".to_string()));
        assert!(!rendered.contains(&"--rootfs-staging".to_string()));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_oci_without_manifest_digest_emits_no_block_root() {
        let config = SandboxBuilder::new("test")
            .image("alpine")
            .build()
            .await
            .unwrap();
        assert!(matches!(config.spec.image, RootfsSource::Oci(_)));

        let rendered = render_args(&config);
        // Without a manifest_digest set, no block root args should be emitted.
        assert!(!rendered.contains(&"--rootfs-blk".to_string()));
        assert!(!rendered.contains(&"--rootfs-disk".to_string()));
        assert!(!rendered.iter().any(|a| a.starts_with("MSB_BLOCK_ROOT=")));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_inject_tmpfs_env_var() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .volume("/tmp", |m| m.tmpfs().size(256u32))
            .volume("/var/tmp", |m| m.tmpfs())
            .build()
            .await
            .unwrap();

        let rendered = render_args(&config);

        assert!(rendered.contains(&"MSB_TMPFS=/tmp:size=256;/var/tmp".to_string()));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_tmpfs_readonly_appends_ro() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .volume("/seed", |m| m.tmpfs().size(64u32).readonly())
            .build()
            .await
            .unwrap();

        let rendered = render_args(&config);

        assert!(rendered.contains(&"MSB_TMPFS=/seed:size=64,ro".to_string()));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_apply_default_oci_tmpfs() {
        let mut config = SandboxConfig {
            spec: microsandbox_types::SandboxSpec {
                name: "test".into(),
                image: RootfsSource::Oci(OciRootfsSource {
                    reference: "alpine".into(),
                    upper_size_mib: None,
                }),
                resources: microsandbox_types::SandboxResources {
                    memory_mib: 1024,
                    ..Default::default()
                },
                ..Default::default()
            },
            manifest_digest: Some(
                "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".into(),
            ),
            ..Default::default()
        };
        config.apply_runtime_defaults();

        let rendered = render_args(&config);

        assert!(rendered.contains(&"MSB_TMPFS=/tmp:size=256".to_string()));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_omit_tmpfs_env_var_when_no_tmpfs() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .build()
            .await
            .unwrap();

        let rendered = render_args(&config);

        assert!(!rendered.iter().any(|a| a.starts_with("MSB_TMPFS=")));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_disk_image_with_fstype() {
        let config = SandboxBuilder::new("test")
            .image_with(|i| i.disk("/tmp/ubuntu.qcow2").fstype("ext4"))
            .build()
            .await
            .unwrap();

        assert!(matches!(config.spec.image, RootfsSource::DiskImage { .. }));

        let rendered = render_args(&config);

        assert!(rendered.contains(&"--rootfs-disk".to_string()));
        assert!(rendered.contains(&"/tmp/ubuntu.qcow2".to_string()));
        assert!(rendered.contains(&"--rootfs-disk-format".to_string()));
        assert!(rendered.contains(&"qcow2".to_string()));
        assert!(
            rendered.contains(
                &"MSB_BLOCK_ROOT=kind=disk-image,device=/dev/vda,fstype=ext4".to_string()
            )
        );

        // Should not contain bind or overlay args.
        assert!(!rendered.contains(&"--rootfs-path".to_string()));
        assert!(!rendered.contains(&"--rootfs-lower".to_string()));
        assert!(!rendered.contains(&"--rootfs-upper".to_string()));
        assert!(!rendered.contains(&"--rootfs-staging".to_string()));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_disk_image_without_fstype() {
        let config = SandboxBuilder::new("test")
            .image_with(|i| i.disk("/tmp/alpine.raw"))
            .build()
            .await
            .unwrap();

        assert!(matches!(config.spec.image, RootfsSource::DiskImage { .. }));

        let rendered = render_args(&config);

        assert!(rendered.contains(&"--rootfs-disk".to_string()));
        assert!(rendered.contains(&"/tmp/alpine.raw".to_string()));
        assert!(rendered.contains(&"--rootfs-disk-format".to_string()));
        assert!(rendered.contains(&"raw".to_string()));
        assert!(rendered.contains(&"MSB_BLOCK_ROOT=kind=disk-image,device=/dev/vda".to_string()));

        // Should not contain bind or overlay args.
        assert!(!rendered.contains(&"--rootfs-path".to_string()));
        assert!(!rendered.contains(&"--rootfs-lower".to_string()));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_file_mount_generates_correct_args() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .volume("/guest/config.txt", |m| {
                m.bind("/host/config.txt").readonly().noexec()
            })
            .build()
            .await
            .unwrap();

        let mut staged_file_mounts = HashMap::new();
        staged_file_mounts.insert(
            "/guest/config.txt".to_string(),
            (
                PathBuf::from("/tmp/staging/fm_aabbccdd"),
                "config.txt".to_string(),
                "fm_aabbccdd".to_string(),
            ),
        );

        let rendered = render_args_with_file_mounts(&config, &staged_file_mounts);

        // File mount should use staging dir in --mount.
        assert!(rendered.windows(2).any(|pair| pair[0] == "--mount"
            && pair[1] == "fm_aabbccdd:/tmp/staging/fm_aabbccdd:ro,noexec"));
        // MSB_FILE_MOUNTS should contain the spec.
        assert!(rendered.contains(
            &"MSB_FILE_MOUNTS=fm_aabbccdd:config.txt:/guest/config.txt:ro,noexec".to_string()
        ));
        // MSB_DIR_MOUNTS should NOT contain the file mount.
        assert!(!rendered.iter().any(|a| a.starts_with("MSB_DIR_MOUNTS=")));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_mixed_file_and_dir_mounts() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .volume("/data", |m| m.bind("/host/data"))
            .volume("/guest/file.txt", |m| m.bind("/host/file.txt"))
            .build()
            .await
            .unwrap();

        let mut staged_file_mounts = HashMap::new();
        staged_file_mounts.insert(
            "/guest/file.txt".to_string(),
            (
                PathBuf::from("/tmp/staging/fm_11223344"),
                "file.txt".to_string(),
                "fm_11223344".to_string(),
            ),
        );

        let rendered = render_args_with_file_mounts(&config, &staged_file_mounts);

        // Directory mount in MSB_DIR_MOUNTS.
        let data_tag = super::guest_mount_tag("/data");
        assert!(rendered.contains(&format!("MSB_DIR_MOUNTS={data_tag}:/data")));
        // File mount in MSB_FILE_MOUNTS.
        assert!(
            rendered.contains(&"MSB_FILE_MOUNTS=fm_11223344:file.txt:/guest/file.txt".to_string())
        );
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_disk_image_volume() {
        // SandboxBuilder::validate canonicalizes disk hosts, so the file
        // must exist. Stage one in a tempdir.
        let dir = tempfile::tempdir().unwrap();
        let host = dir.path().join("data.qcow2");
        std::fs::write(&host, []).unwrap();

        let host_clone = host.clone();
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .volume("/data", |m| {
                m.disk(host_clone)
                    .format(DiskImageFormat::Qcow2)
                    .fstype("ext4")
            })
            .build()
            .await
            .unwrap();

        let rendered = render_args(&config);

        // --disk arg present with correct layout.
        let data_tag = super::guest_mount_tag("/data");
        let expected_disk_arg = format!("{data_tag}:{}:qcow2", host.display());
        assert!(
            rendered
                .windows(2)
                .any(|pair| pair[0] == "--disk" && pair[1] == expected_disk_arg),
            "missing --disk arg in {rendered:?}"
        );

        // MSB_DISK_MOUNTS env entry carries the guest path and fstype.
        let expected_env = format!("MSB_DISK_MOUNTS={data_tag}:/data:fstype=ext4");
        assert!(rendered.contains(&expected_env));
    }

    #[tokio::test]
    async fn test_sandbox_cli_args_disk_image_readonly() {
        let dir = tempfile::tempdir().unwrap();
        let host = dir.path().join("seed.raw");
        std::fs::write(&host, []).unwrap();

        let host_clone = host.clone();
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .volume("/seed", |m| m.disk(host_clone).readonly().noexec())
            .build()
            .await
            .unwrap();

        let rendered = render_args(&config);
        let tag = super::guest_mount_tag("/seed");

        assert!(rendered.windows(2).any(
            |pair| pair[0] == "--disk" && pair[1] == format!("{tag}:{}:raw:ro", host.display())
        ));
        assert!(rendered.contains(&format!("MSB_DISK_MOUNTS={tag}:/seed:ro,noexec")));
    }

    #[tokio::test]
    async fn test_guest_mount_tag_is_deterministic() {
        let a = super::guest_mount_tag("/data");
        let b = super::guest_mount_tag("/data");
        assert_eq!(a, b);
    }

    #[tokio::test]
    async fn test_guest_mount_tag_disambiguates_colliding_paths() {
        // The naive `/` → `_` mangling treats these as identical. The
        // slug+hash form must not.
        let a = super::guest_mount_tag("/var/log");
        let b = super::guest_mount_tag("/var_log");
        assert_ne!(a, b);
        assert!(a.starts_with("var_log_"));
        assert!(b.starts_with("var_log_"));
    }

    #[tokio::test]
    async fn test_guest_mount_tag_fits_virtio_blk_serial_limit() {
        // virtio-blk serial is capped at 20 bytes. Long guest paths must still fit.
        let long = "/a/very/deeply/nested/guest/mount/point/that/exceeds/the/slug/cap";
        let tag = super::guest_mount_tag(long);
        assert!(tag.len() <= 20, "tag {tag:?} exceeds 20 bytes");
    }

    #[tokio::test]
    async fn test_guest_mount_tag_slug_prefix_is_readable() {
        assert!(super::guest_mount_tag("/data").starts_with("data_"));
        assert!(super::guest_mount_tag("/var/log").starts_with("var_log_"));
    }

    //----------------------------------------------------------------------------------------------
    // Tests: Handoff init env-var construction
    //----------------------------------------------------------------------------------------------

    /// Helper to grep the rendered args for an `--env KEY=...` entry.
    fn find_env(args: &[String], key: &str) -> Option<String> {
        let prefix = format!("{key}=");
        args.windows(2).find_map(|pair| {
            if pair[0] == "--env" && pair[1].starts_with(&prefix) {
                Some(pair[1][prefix.len()..].to_string())
            } else {
                None
            }
        })
    }

    #[tokio::test]
    async fn test_handoff_init_emits_only_cmd_when_args_and_env_empty() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .init("/lib/systemd/systemd")
            .build()
            .await
            .unwrap();

        let args = render_args(&config);

        assert_eq!(
            find_env(&args, "MSB_HANDOFF_INIT").as_deref(),
            Some("/lib/systemd/systemd")
        );
        assert!(find_env(&args, "MSB_HANDOFF_INIT_ARGS").is_none());
        assert!(find_env(&args, "MSB_HANDOFF_INIT_CWD").is_none());
        assert!(find_env(&args, "MSB_HANDOFF_INIT_ENV").is_none());
    }

    #[tokio::test]
    async fn test_handoff_init_emits_cwd_when_workdir_set() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .init("/init")
            .workdir("/opt/hermes")
            .build()
            .await
            .unwrap();

        let args = render_args(&config);

        assert_eq!(
            find_env(&args, "MSB_HANDOFF_INIT_CWD").as_deref(),
            Some("/opt/hermes")
        );
    }

    #[tokio::test]
    async fn test_handoff_init_encodes_argv_as_base64url_json() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .init_with("/lib/systemd/systemd", |i| {
                i.args([
                    "--unit=multi-user.target",
                    "--log-level=warning",
                    "literal\x1funit-separator",
                ])
            })
            .build()
            .await
            .unwrap();

        let args = render_args(&config);
        let argv = find_env(&args, "MSB_HANDOFF_INIT_ARGS").expect("argv env present");
        let decoded: Vec<String> = decode_handoff_json(&argv);

        assert_eq!(
            decoded,
            vec![
                "--unit=multi-user.target",
                "--log-level=warning",
                "literal\x1funit-separator"
            ]
        );
    }

    #[tokio::test]
    async fn test_handoff_init_encodes_env_pairs_as_base64url_json() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .init_with("/sbin/init", |i| {
                i.env("container", "microsandbox")
                    .env("LANG", "C.UTF-8")
                    .env("TOKEN", "a=b;c\x1fd")
            })
            .build()
            .await
            .unwrap();

        let args = render_args(&config);
        let env_val = find_env(&args, "MSB_HANDOFF_INIT_ENV").expect("env present");
        let decoded: Vec<(String, String)> = decode_handoff_json(&env_val);

        assert_eq!(
            decoded,
            vec![
                ("container".to_string(), "microsandbox".to_string()),
                ("LANG".to_string(), "C.UTF-8".to_string()),
                ("TOKEN".to_string(), "a=b;c\x1fd".to_string())
            ]
        );
    }

    #[tokio::test]
    async fn test_handoff_init_omitted_when_unset() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .build()
            .await
            .unwrap();

        let args = render_args(&config);

        assert!(find_env(&args, "MSB_HANDOFF_INIT").is_none());
        assert!(find_env(&args, "MSB_HANDOFF_INIT_ARGS").is_none());
        assert!(find_env(&args, "MSB_HANDOFF_INIT_ENV").is_none());
    }

    #[tokio::test]
    async fn test_handoff_init_unit_separator_in_arg_allowed() {
        let config = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .init_with("/sbin/init", |i| i.args(["foo\x1fbar"]))
            .build()
            .await
            .unwrap();
        let args = render_args(&config);
        let argv = find_env(&args, "MSB_HANDOFF_INIT_ARGS").expect("argv env present");
        let decoded: Vec<String> = decode_handoff_json(&argv);

        assert_eq!(decoded, vec!["foo\x1fbar"]);
    }

    #[tokio::test]
    async fn test_handoff_init_equals_in_env_key_rejected_at_build_time() {
        let err = SandboxBuilder::new("test")
            .image("/tmp/rootfs")
            .init_with("/sbin/init", |i| i.env("BAD=KEY", "v"))
            .build()
            .await
            .unwrap_err();
        assert!(format!("{err}").contains("must not contain '='"));
    }
}