zlayer-agent 0.14.1

//! OCI Bundle Creation
//!
//! Creates OCI-compliant bundles for container runtimes using libcontainer (youki).
//! A bundle consists of a directory with:
//! - config.json: OCI runtime specification
//! - rootfs/: Container filesystem (symlink or bind mount target)

use crate::cdi::{self, CdiContainerEdits, CdiRegistry};
use crate::error::{AgentError, Result};
use crate::runtime::ContainerId;
use oci_spec::runtime::{
    Capability, Hook, HookBuilder, Hooks, HooksBuilder, LinuxBuilder, LinuxCapabilitiesBuilder,
    LinuxCpuBuilder, LinuxDeviceBuilder, LinuxDeviceCgroupBuilder, LinuxDeviceType,
    LinuxMemoryBuilder, LinuxNamespaceBuilder, LinuxNamespaceType, LinuxResourcesBuilder, Mount,
    MountBuilder, PosixRlimit, PosixRlimitBuilder, PosixRlimitType, ProcessBuilder, RootBuilder,
    Spec, SpecBuilder, UserBuilder,
};
// `LinuxIdMappingBuilder` is only used by the unix-gated rootless user-namespace
// helpers below; importing it unconditionally trips dead-code lints on Windows.
#[cfg(unix)]
use oci_spec::runtime::LinuxIdMappingBuilder;
use std::collections::{HashMap, HashSet};
// `MetadataExt` is only meaningful on Unix-like hosts where `/dev/*` nodes exist
// and have major/minor numbers. On Windows this module is still built so that
// `BundleBuilder::build_spec_only` (cross-platform OCI Spec generation) can be
// called from the WSL2 delegate runtime, which then pipes the generated
// `config.json` into a Linux WSL2 distro that owns the actual device
// fingerprint. See G-1 / G-2 in the Windows plan. The import is performed
// inside `get_device_major_minor` itself to avoid an unused-import warning on
// non-Unix platforms.
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::Arc;
use tokio::fs;
use zlayer_secrets::{SecretScope, SecretsProvider};
use zlayer_spec::{GpuSharingMode, ServiceSpec, ShardingSpec, StorageSpec, StorageTier, SwarmRole};

/// Default host directory for the NVIDIA MPS control pipe when the spec
/// doesn't override [`zlayer_spec::GpuSpec::mps_pipe_dir`].
const DEFAULT_MPS_PIPE_DIR: &str = "/tmp/nvidia-mps";

/// Default host directory for NVIDIA MPS log output when the spec doesn't
/// override [`zlayer_spec::GpuSpec::mps_log_dir`].
const DEFAULT_MPS_LOG_DIR: &str = "/tmp/nvidia-log";

/// Container path where a host-supplied NVIDIA time-slicing config YAML is
/// surfaced (read-only). The file is informational — `ZLayer` doesn't interpret
/// it; tools running inside the container can read it to discover slice
/// topology.
const TIMESLICE_CONFIG_CONTAINER_PATH: &str = "/etc/nvidia/gpu-time-slicing.yaml";

/// Resolved MPS host directories (pipe + log), validated to exist on disk.
///
/// Returned by [`resolve_mps_dirs`] only when `GpuSpec.sharing == Mps`. Both
/// paths are absolute and guaranteed to be directories at the time the
/// helper ran — callers can bind-mount them directly.
struct MpsDirs {
    pipe_dir: PathBuf,
    log_dir: PathBuf,
}

/// Resolve and validate the MPS pipe / log directories for a GPU spec.
///
/// Returns `Ok(None)` when sharing is not MPS (or absent), `Ok(Some(...))`
/// when both directories exist on the host, or
/// [`AgentError::GpuSharingUnavailable`] when either directory is missing.
///
/// Defaults to [`DEFAULT_MPS_PIPE_DIR`] / [`DEFAULT_MPS_LOG_DIR`] when the
/// spec omits explicit paths, matching the convention used by
/// `nvidia-cuda-mps-control` out of the box.
fn resolve_mps_dirs(gpu: &zlayer_spec::GpuSpec) -> Result<Option<MpsDirs>> {
    if gpu.sharing != Some(GpuSharingMode::Mps) {
        return Ok(None);
    }

    let pipe_dir = PathBuf::from(gpu.mps_pipe_dir.as_deref().unwrap_or(DEFAULT_MPS_PIPE_DIR));
    let log_dir = PathBuf::from(gpu.mps_log_dir.as_deref().unwrap_or(DEFAULT_MPS_LOG_DIR));

    if !pipe_dir.is_dir() {
        return Err(AgentError::GpuSharingUnavailable {
            mode: "mps".to_string(),
            reason: format!(
                "MPS pipe directory {} does not exist; ensure nvidia-cuda-mps-control is running",
                pipe_dir.display()
            ),
        });
    }
    if !log_dir.is_dir() {
        return Err(AgentError::GpuSharingUnavailable {
            mode: "mps".to_string(),
            reason: format!(
                "MPS log directory {} does not exist; ensure nvidia-cuda-mps-control is running",
                log_dir.display()
            ),
        });
    }

    Ok(Some(MpsDirs { pipe_dir, log_dir }))
}

/// Build the swarm ring-neighbor discovery env vars for a pipeline-parallel
/// inference stage/coordinator container.
///
/// The overlay is a flat DIRECT full-mesh, so this injects no routing — it only
/// tells a node WHO its ring neighbors are, by their bare `ZLayer` service
/// names. Those names resolve over the existing per-service overlay DNS exactly
/// like the distributed-training `MASTER_ADDR={service}` hint does.
///
/// Returned pairs (any whose value is absent are simply omitted, never emitted
/// as empty strings):
/// - `ZLAYER_SWARM_ID`
/// - `ZLAYER_SWARM_ROLE` (`"stage"` | `"coordinator"`)
/// - `ZLAYER_SWARM_LAYER_START` / `ZLAYER_SWARM_LAYER_END` /
///   `ZLAYER_SWARM_TOTAL_LAYERS` (stages only)
/// - `ZLAYER_SWARM_COORDINATOR` (when `sharding.coordinator` is set)
/// - `ZLAYER_SWARM_NEXT_PEER` / `ZLAYER_SWARM_PREV_PEER`
///
/// ## Ring derivation
///
/// The full ring is every stage ordered by `layer_start`. For a STAGE we build
/// that order from `sharding.peers` (each carries its own `service` name and
/// band) plus a self-marker for this node's own band (`sharding.layer_start ..
/// layer_end`, no name). This node's index `i` in the sorted list gives
/// `prev = list[i-1]` and `next = list[i+1]`, with ring-wrap to the coordinator:
/// the first stage's prev and the last stage's next are the coordinator (when
/// one is set). For the COORDINATOR role: `next` is the first stage (lowest
/// `layer_start`) and `prev` is the last stage (highest `layer_end`).
///
/// A single-stage swarm (no peers) emits neither NEXT nor PREV unless a
/// coordinator is present, in which case both point at it.
#[must_use]
pub(crate) fn swarm_ring_env(sharding: &ShardingSpec) -> Vec<(String, String)> {
    /// One slot in the sorted ring. `service == None` marks THIS node's own
    /// band (it never appears as its own neighbor); named slots are `peers`.
    struct Slot {
        service: Option<String>,
        layer_start: u32,
        layer_end: u32,
    }

    let mut out: Vec<(String, String)> = Vec::new();
    out.push(("ZLAYER_SWARM_ID".to_string(), sharding.swarm_id.clone()));

    let role_str = match sharding.role {
        SwarmRole::Stage => "stage",
        SwarmRole::Coordinator => "coordinator",
    };
    out.push(("ZLAYER_SWARM_ROLE".to_string(), role_str.to_string()));

    if matches!(sharding.role, SwarmRole::Stage) {
        out.push((
            "ZLAYER_SWARM_LAYER_START".to_string(),
            sharding.layer_start.to_string(),
        ));
        out.push((
            "ZLAYER_SWARM_LAYER_END".to_string(),
            sharding.layer_end.to_string(),
        ));
        out.push((
            "ZLAYER_SWARM_TOTAL_LAYERS".to_string(),
            sharding.layer_count.to_string(),
        ));
    }

    let coordinator = sharding.coordinator.clone();
    if let Some(ref coord) = coordinator {
        out.push(("ZLAYER_SWARM_COORDINATOR".to_string(), coord.clone()));
    }

    // Build the named stage list (the peers), sorted by layer_start.
    let mut peers: Vec<Slot> = sharding
        .peers
        .iter()
        .map(|p| Slot {
            service: Some(p.service.clone()),
            layer_start: p.layer_start,
            layer_end: p.layer_end,
        })
        .collect();
    peers.sort_by_key(|s| s.layer_start);

    let (next, prev): (Option<String>, Option<String>) = match sharding.role {
        SwarmRole::Coordinator => {
            // next = first stage (lowest start); prev = last stage (highest end).
            let first = peers.iter().min_by_key(|s| s.layer_start);
            let last = peers.iter().max_by_key(|s| s.layer_end);
            (
                first.and_then(|s| s.service.clone()),
                last.and_then(|s| s.service.clone()),
            )
        }
        SwarmRole::Stage => {
            // Full ring = peers + this node's own band (self-marker, unnamed).
            let mut ring: Vec<Slot> = peers;
            ring.push(Slot {
                service: None,
                layer_start: sharding.layer_start,
                layer_end: sharding.layer_end,
            });
            ring.sort_by_key(|s| s.layer_start);

            // Locate self: the only unnamed slot.
            let self_idx = ring.iter().position(|s| s.service.is_none());
            match self_idx {
                Some(i) => {
                    // prev: previous slot, else wrap to coordinator (first stage).
                    let prev = if i == 0 {
                        coordinator.clone()
                    } else {
                        ring[i - 1].service.clone()
                    };
                    // next: following slot, else wrap to coordinator (last stage).
                    let next = if i + 1 >= ring.len() {
                        coordinator.clone()
                    } else {
                        ring[i + 1].service.clone()
                    };
                    (next, prev)
                }
                // Self-marker is always inserted, so this is unreachable in
                // practice; degrade gracefully to no neighbors.
                None => (None, None),
            }
        }
    };

    if let Some(next) = next {
        out.push(("ZLAYER_SWARM_NEXT_PEER".to_string(), next));
    }
    if let Some(prev) = prev {
        out.push(("ZLAYER_SWARM_PREV_PEER".to_string(), prev));
    }

    out
}

/// Convert a CDI device node descriptor into the OCI [`LinuxDevice`] used by
/// the runtime.
///
/// CDI device nodes may omit `type`, `major`, and `minor` — in that case we
/// probe the host (via `get_device_type` / `get_device_major_minor`) using
/// the resolved host path, falling back to character device with zero
/// major/minor when the file is unavailable (typical for test fixtures
/// that reference paths that don't exist on the build host).
fn cdi_node_to_oci_device(
    node: &crate::cdi::CdiDeviceNode,
) -> Result<oci_spec::runtime::LinuxDevice> {
    let host_path = node.host_path.as_deref().unwrap_or(&node.path);

    let dev_type = match node.device_type.as_deref() {
        Some("c" | "u") => LinuxDeviceType::C,
        Some("b") => LinuxDeviceType::B,
        Some("p") => LinuxDeviceType::P,
        _ => get_device_type(host_path).unwrap_or(LinuxDeviceType::C),
    };

    let (major, minor) = if let (Some(maj), Some(min)) = (node.major, node.minor) {
        (maj, min)
    } else {
        get_device_major_minor(host_path).unwrap_or((0, 0))
    };

    let mut builder = LinuxDeviceBuilder::default()
        .path(node.path.clone())
        .typ(dev_type)
        .major(major)
        .minor(minor);
    if let Some(mode) = node.file_mode {
        builder = builder.file_mode(mode);
    } else {
        builder = builder.file_mode(0o666u32);
    }
    builder = builder.uid(node.uid.unwrap_or(0));
    builder = builder.gid(node.gid.unwrap_or(0));

    builder.build().map_err(|e| {
        AgentError::InvalidSpec(format!(
            "failed to build CDI device {path}: {e}",
            path = node.path
        ))
    })
}

/// Convert a CDI hook descriptor into the OCI [`Hook`] used by the runtime.
fn convert_cdi_hook(cdi_hook: &crate::cdi::CdiHook) -> Result<Hook> {
    let mut builder = HookBuilder::default().path(PathBuf::from(&cdi_hook.path));
    if !cdi_hook.args.is_empty() {
        builder = builder.args(cdi_hook.args.clone());
    }
    if !cdi_hook.env.is_empty() {
        builder = builder.env(cdi_hook.env.clone());
    }
    builder
        .build()
        .map_err(|e| AgentError::InvalidSpec(format!("failed to build CDI hook: {e}")))
}

/// All Linux capabilities for privileged mode
const ALL_CAPABILITIES: &[Capability] = &[
    Capability::AuditControl,
    Capability::AuditRead,
    Capability::AuditWrite,
    Capability::BlockSuspend,
    Capability::Bpf,
    Capability::CheckpointRestore,
    Capability::Chown,
    Capability::DacOverride,
    Capability::DacReadSearch,
    Capability::Fowner,
    Capability::Fsetid,
    Capability::IpcLock,
    Capability::IpcOwner,
    Capability::Kill,
    Capability::Lease,
    Capability::LinuxImmutable,
    Capability::MacAdmin,
    Capability::MacOverride,
    Capability::Mknod,
    Capability::NetAdmin,
    Capability::NetBindService,
    Capability::NetBroadcast,
    Capability::NetRaw,
    Capability::Perfmon,
    Capability::Setfcap,
    Capability::Setgid,
    Capability::Setpcap,
    Capability::Setuid,
    Capability::SysAdmin,
    Capability::SysBoot,
    Capability::SysChroot,
    Capability::SysModule,
    Capability::SysNice,
    Capability::SysPacct,
    Capability::SysPtrace,
    Capability::SysRawio,
    Capability::SysResource,
    Capability::SysTime,
    Capability::SysTtyConfig,
    Capability::Syslog,
    Capability::WakeAlarm,
];

/// Parse memory string like "512Mi", "1Gi" to bytes
///
/// Supports both IEC (binary) and SI (decimal) units:
/// - IEC: Ki, Mi, Gi, Ti (powers of 1024)
/// - SI: K/k, M/m, G/g, T/t (powers of 1000)
/// - No suffix: bytes
///
/// # Examples
/// ```ignore
/// assert_eq!(parse_memory_string("512Mi").unwrap(), 512 * 1024 * 1024);
/// assert_eq!(parse_memory_string("1Gi").unwrap(), 1024 * 1024 * 1024);
/// assert_eq!(parse_memory_string("2G").unwrap(), 2 * 1000 * 1000 * 1000);
/// ```
///
/// Render the contents of an `/etc/resolv.conf` for the given resolver
/// addresses.
///
/// One `nameserver <ip>` line per entry, then a single `search <domains>` line
/// when `search_domains` is non-empty (space-joined), then a single
/// `options edns0` line (enables EDNS(0) so larger UDP responses — e.g. the
/// overlay resolver forwarding A/AAAA records — are not truncated). We emit
/// ONLY the explicit overlay search domains passed here, never the ones that
/// would otherwise be inherited from the (hijacked) host resolv.conf we are
/// replacing — the per-deployment `<deployment>.<zone> <zone>` search domain is
/// what lets a container resolve a bare `<svc>` / `<svc>.service`.
///
/// This exists because youki/libcontainer performs NO resolv.conf handling of
/// its own — without an explicit bind mount the container sees only whatever
/// `/etc/resolv.conf` shipped in the image (often empty or absent). The caller
/// writes this string into the bundle directory and bind-mounts it read-only at
/// `/etc/resolv.conf`.
#[must_use]
pub fn generate_resolv_conf(nameservers: &[String], search_domains: &[String]) -> String {
    let mut out = String::new();
    for ns in nameservers {
        out.push_str("nameserver ");
        out.push_str(ns);
        out.push('\n');
    }
    if !search_domains.is_empty() {
        out.push_str("search ");
        out.push_str(&search_domains.join(" "));
        out.push('\n');
    }
    out.push_str("options edns0\n");
    out
}

/// # Errors
/// Returns an error if the string cannot be parsed as a memory size.
pub fn parse_memory_string(s: &str) -> std::result::Result<u64, String> {
    let s = s.trim();
    if s.is_empty() {
        return Err("empty memory string".to_string());
    }

    let (num_str, multiplier) = if let Some(n) = s.strip_suffix("Ki") {
        (n, 1024u64)
    } else if let Some(n) = s.strip_suffix("Mi") {
        (n, 1024u64 * 1024)
    } else if let Some(n) = s.strip_suffix("Gi") {
        (n, 1024u64 * 1024 * 1024)
    } else if let Some(n) = s.strip_suffix("Ti") {
        (n, 1024u64 * 1024 * 1024 * 1024)
    } else if let Some(n) = s.strip_suffix('K').or_else(|| s.strip_suffix('k')) {
        (n, 1000u64)
    } else if let Some(n) = s.strip_suffix('M').or_else(|| s.strip_suffix('m')) {
        (n, 1000u64 * 1000)
    } else if let Some(n) = s.strip_suffix('G').or_else(|| s.strip_suffix('g')) {
        (n, 1000u64 * 1000 * 1000)
    } else if let Some(n) = s.strip_suffix('T').or_else(|| s.strip_suffix('t')) {
        (n, 1000u64 * 1000 * 1000 * 1000)
    } else {
        (s, 1u64)
    };

    let num: u64 = num_str
        .parse()
        .map_err(|e| format!("invalid number: {e}"))?;

    Ok(num * multiplier)
}

/// Get major and minor device numbers from a device path
///
/// Unix-only: relies on `MetadataExt::rdev()` which isn't available on Windows.
/// When `bundle.rs` is compiled for a Windows host (for the WSL2 delegate's
/// cross-platform `build_spec_only` path), device probing is skipped entirely —
/// the Linux side of the delegate is responsible for its own device fingerprint.
/// The non-Unix stub below returns `Unsupported` so the `if let Ok(..)` /
/// `.unwrap_or(..)` call sites at the CDI / GPU passthrough paths skip cleanly.
#[cfg(unix)]
#[allow(clippy::cast_possible_wrap)]
fn get_device_major_minor(path: &str) -> std::io::Result<(i64, i64)> {
    use std::os::unix::fs::MetadataExt;
    let metadata = std::fs::metadata(path)?;
    let rdev = metadata.rdev();
    // Major is upper 8 bits (after shifting), minor is lower 8 bits
    let major = ((rdev >> 8) & 0xff) as i64;
    let minor = (rdev & 0xff) as i64;
    Ok((major, minor))
}

/// Non-Unix stub: device-cgroup probes require Unix; callers use `if let Ok(..)` to skip.
#[cfg(not(unix))]
fn get_device_major_minor(_path: &str) -> std::io::Result<(i64, i64)> {
    Err(std::io::Error::new(
        std::io::ErrorKind::Unsupported,
        "device-cgroup probes require Unix",
    ))
}

/// Translate the Docker `--ulimit <name>` style key into the OCI
/// `PosixRlimitType` enum. Returns `None` for unknown names so the caller
/// can surface a clean error.
fn ulimit_name_to_posix(name: &str) -> Option<PosixRlimitType> {
    Some(match name.to_ascii_lowercase().as_str() {
        "cpu" => PosixRlimitType::RlimitCpu,
        "fsize" => PosixRlimitType::RlimitFsize,
        "data" => PosixRlimitType::RlimitData,
        "stack" => PosixRlimitType::RlimitStack,
        "core" => PosixRlimitType::RlimitCore,
        "rss" => PosixRlimitType::RlimitRss,
        "nproc" => PosixRlimitType::RlimitNproc,
        "nofile" => PosixRlimitType::RlimitNofile,
        "memlock" => PosixRlimitType::RlimitMemlock,
        "as" => PosixRlimitType::RlimitAs,
        "locks" => PosixRlimitType::RlimitLocks,
        "sigpending" => PosixRlimitType::RlimitSigpending,
        "msgqueue" => PosixRlimitType::RlimitMsgqueue,
        "nice" => PosixRlimitType::RlimitNice,
        "rtprio" => PosixRlimitType::RlimitRtprio,
        "rttime" => PosixRlimitType::RlimitRttime,
        _ => return None,
    })
}

#[cfg(test)]
mod ulimit_translation_tests {
    use super::{ulimit_name_to_posix, PosixRlimitType};

    #[test]
    fn known_names_map() {
        assert_eq!(
            ulimit_name_to_posix("nofile"),
            Some(PosixRlimitType::RlimitNofile)
        );
        assert_eq!(
            ulimit_name_to_posix("NOFILE"),
            Some(PosixRlimitType::RlimitNofile)
        );
        assert_eq!(
            ulimit_name_to_posix("nproc"),
            Some(PosixRlimitType::RlimitNproc)
        );
        assert_eq!(ulimit_name_to_posix("as"), Some(PosixRlimitType::RlimitAs));
    }

    #[test]
    fn unknown_names_return_none() {
        assert!(ulimit_name_to_posix("not_a_real_ulimit").is_none());
        assert!(ulimit_name_to_posix("").is_none());
    }
}

/// Detect device type from path
///
/// Unix-only: uses `FileTypeExt::is_char_device` / `is_block_device` which are
/// not available on Windows. See `get_device_major_minor` for the rationale.
#[cfg(unix)]
fn get_device_type(path: &str) -> std::io::Result<LinuxDeviceType> {
    use std::os::unix::fs::FileTypeExt;
    let metadata = std::fs::metadata(path)?;
    let file_type = metadata.file_type();
    if file_type.is_char_device() {
        Ok(LinuxDeviceType::C)
    } else if file_type.is_block_device() {
        Ok(LinuxDeviceType::B)
    } else {
        Ok(LinuxDeviceType::U) // Unknown/other
    }
}

/// Non-Unix stub: device-cgroup probes require Unix; callers use `.unwrap_or(..)` to skip.
#[cfg(not(unix))]
fn get_device_type(_path: &str) -> std::io::Result<LinuxDeviceType> {
    Err(std::io::Error::new(
        std::io::ErrorKind::Unsupported,
        "device-cgroup probes require Unix",
    ))
}

/// Builder for OCI container bundles
///
/// Creates the directory structure and config.json required for OCI-compliant
/// container runtimes like runc or youki.
///
/// # Example
/// ```ignore
/// let dirs = zlayer_paths::ZLayerDirs::system_default();
/// let builder = BundleBuilder::new(dirs.bundles().join("mycontainer"))
///     .with_rootfs(dirs.rootfs().join("myimage"));
///
/// let bundle_path = builder.build(&container_id, &service_spec).await?;
/// ```
#[derive(Clone)]
pub struct BundleBuilder {
    /// Base directory for the bundle
    bundle_dir: PathBuf,
    /// Path to the unpacked rootfs (from image layers)
    rootfs_path: Option<PathBuf>,
    /// Custom hostname (defaults to container ID)
    hostname: Option<String>,
    /// Additional environment variables
    extra_env: Vec<(String, String)>,
    /// Custom working directory
    cwd: Option<String>,
    /// Custom command/args to run (overrides image default)
    args: Option<Vec<String>>,
    /// Pre-resolved volume paths from `StorageManager`
    volume_paths: HashMap<String, PathBuf>,
    /// Image configuration from the OCI registry (entrypoint, cmd, env, workdir, user)
    image_config: Option<zlayer_registry::ImageConfig>,
    /// Use host networking (skip Network namespace, container shares host network)
    host_network: bool,
    /// Join an existing network namespace by path (Docker `--network container:<id>`).
    ///
    /// When `Some`, the container's Network namespace is built WITH this path so
    /// libcontainer `setns()`es into the target's netns (JOIN) instead of
    /// unsharing a fresh one. Mutually exclusive with `host_network`: when
    /// `host_network` is true the container shares the host stack and no Network
    /// namespace is emitted at all, so this field is ignored.
    netns_path: Option<PathBuf>,
    /// Secrets provider for resolving $S: prefixed env vars
    secrets_provider: Option<Arc<dyn SecretsProvider>>,
    /// Deployment scope for secret lookups (e.g., deployment name)
    deployment_scope: Option<SecretScope>,
    /// Host-side Unix socket path to bind-mount into the container
    socket_path: Option<String>,
    /// Host-side per-container Docker Engine API socket to bind-mount at
    /// `/var/run/docker.sock`.
    docker_socket_path: Option<String>,
    /// Optional shared per-platform toolchain cache dir (host path). When set,
    /// it is RW-bind-mounted into the container at `/opt/zlayer/toolchains` and
    /// the runner tool-cache env (`RUNNER_TOOL_CACHE`, `AGENT_TOOLSDIRECTORY`)
    /// points there, so CI toolchains are shared across containers instead of
    /// re-downloaded. `None` => no toolchain mount (default).
    toolchain_cache: Option<PathBuf>,
    /// Optional CDI registry override (defaults to discovery from system paths).
    ///
    /// Wrapped in `Arc` so [`BundleBuilder`] can stay [`Clone`]. Primarily set
    /// in tests via [`BundleBuilder::with_cdi_registry`]; production paths
    /// leave this `None` and lazy-discover via [`CdiRegistry::discover`] when
    /// a `GpuSpec` is present.
    cdi_registry: Option<Arc<CdiRegistry>>,
}

impl std::fmt::Debug for BundleBuilder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BundleBuilder")
            .field("bundle_dir", &self.bundle_dir)
            .field("rootfs_path", &self.rootfs_path)
            .field("hostname", &self.hostname)
            .field("extra_env", &self.extra_env)
            .field("cwd", &self.cwd)
            .field("args", &self.args)
            .field("volume_paths", &self.volume_paths)
            .field("image_config", &self.image_config)
            .field("host_network", &self.host_network)
            .field("netns_path", &self.netns_path)
            .field("secrets_provider", &self.secrets_provider.is_some())
            .field("deployment_scope", &self.deployment_scope)
            .field("socket_path", &self.socket_path)
            .field("docker_socket_path", &self.docker_socket_path)
            .field("toolchain_cache", &self.toolchain_cache)
            .field("cdi_registry", &self.cdi_registry.is_some())
            .finish()
    }
}

/// Build OCI `uid_mappings` (or `gid_mappings` — same structure) for a rootless
/// container. Always emits a single-id mapping (container 0 → `host_id`, size 1).
/// If `username` has an entry in `subid_path` (e.g. /etc/subuid), appends a
/// range mapping (container 1 → range start, size = range count).
///
/// Rootless user-namespace mapping is a Linux/libcontainer concept; on Windows
/// containers run via HCS so this helper is unix-only.
#[cfg(unix)]
fn build_rootless_id_mappings(
    host_id: u32,
    subid_path: &str,
    username: &str,
) -> Vec<oci_spec::runtime::LinuxIdMapping> {
    let mut mappings = vec![LinuxIdMappingBuilder::default()
        .container_id(0_u32)
        .host_id(host_id)
        .size(1_u32)
        .build()
        .unwrap()];
    if !username.is_empty() {
        if let Some((start, count)) = read_subid_range(subid_path, username) {
            mappings.push(
                LinuxIdMappingBuilder::default()
                    .container_id(1_u32)
                    .host_id(start)
                    .size(count)
                    .build()
                    .unwrap(),
            );
        }
    }
    mappings
}

/// Build a single-id OCI mapping (container 0 → `host_id`, size 1) with NO
/// subordinate range. Used when the daemon runs in its own single-uid userns
/// and cannot sub-delegate a /etc/subuid range to nested containers (a single
/// own-uid map is written directly to `/proc/<pid>/uid_map`, no newuidmap needed).
#[cfg(unix)]
fn build_single_id_mapping(host_id: u32) -> Vec<oci_spec::runtime::LinuxIdMapping> {
    vec![LinuxIdMappingBuilder::default()
        .container_id(0_u32)
        .host_id(host_id)
        .size(1_u32)
        .build()
        .unwrap()]
}

/// Read /etc/subuid (or /etc/subgid) and return the (start, count) range
/// allocated to the given username, if any. Returns None on any I/O error
/// or when the user has no entry — callers must fall back to a single-id
/// mapping in that case.
///
/// Subuid files are a Linux concept and the only caller is the unix-gated
/// `build_rootless_id_mappings`, so this helper is unix-only as well.
#[cfg(unix)]
fn read_subid_range(path: &str, username: &str) -> Option<(u32, u32)> {
    let contents = std::fs::read_to_string(path).ok()?;
    for line in contents.lines() {
        let mut parts = line.splitn(3, ':');
        let user = parts.next()?;
        if user != username {
            continue;
        }
        let start: u32 = parts.next()?.parse().ok()?;
        let count: u32 = parts.next()?.parse().ok()?;
        return Some((start, count));
    }
    None
}

impl BundleBuilder {
    /// Create a new `BundleBuilder` with the specified bundle directory
    ///
    /// The bundle directory will be created if it doesn't exist.
    /// The structure will be:
    /// ```text
    /// {bundle_dir}/
    /// ├── config.json
    /// └── rootfs/  (symlink to actual rootfs or mount point)
    /// ```
    #[must_use]
    pub fn new(bundle_dir: PathBuf) -> Self {
        Self {
            bundle_dir,
            rootfs_path: None,
            hostname: None,
            extra_env: Vec::new(),
            cwd: None,
            args: None,
            volume_paths: HashMap::new(),
            image_config: None,
            host_network: false,
            netns_path: None,
            secrets_provider: None,
            deployment_scope: None,
            socket_path: None,
            docker_socket_path: None,
            toolchain_cache: None,
            cdi_registry: None,
        }
    }

    /// Override the CDI registry used for GPU device resolution.
    ///
    /// When unset, [`build_oci_spec`](Self::build_oci_spec) discovers CDI
    /// specs lazily from the standard system search paths (`/etc/cdi`,
    /// `/var/run/cdi`, plus `$CDI_SPEC_DIRS`). Tests use this setter to
    /// inject fixture-backed registries pointed at a temp directory.
    #[must_use]
    pub fn with_cdi_registry(mut self, registry: Arc<CdiRegistry>) -> Self {
        self.cdi_registry = Some(registry);
        self
    }

    /// Create a `BundleBuilder` for a container in the default bundle location
    #[must_use]
    pub fn for_container(container_id: &ContainerId) -> Self {
        let bundle_dir = zlayer_paths::ZLayerDirs::system_default()
            .bundles()
            .join(container_id.to_string());
        Self::new(bundle_dir)
    }

    /// Set the rootfs path (from unpacked image layers)
    ///
    /// This path will be symlinked into the bundle as `rootfs/`
    #[must_use]
    pub fn with_rootfs(mut self, rootfs_path: PathBuf) -> Self {
        self.rootfs_path = Some(rootfs_path);
        self
    }

    /// Set a custom hostname for the container
    #[must_use]
    pub fn with_hostname(mut self, hostname: String) -> Self {
        self.hostname = Some(hostname);
        self
    }

    /// Add extra environment variables
    #[must_use]
    pub fn with_env(mut self, key: String, value: String) -> Self {
        self.extra_env.push((key, value));
        self
    }

    /// Set the working directory
    #[must_use]
    pub fn with_cwd(mut self, cwd: String) -> Self {
        self.cwd = Some(cwd);
        self
    }

    /// Set the command/args to run
    #[must_use]
    pub fn with_args(mut self, args: Vec<String>) -> Self {
        self.args = Some(args);
        self
    }

    /// Set pre-resolved volume paths from `StorageManager`
    ///
    /// These are used to map named/anonymous/S3 volumes to their host paths
    /// when building storage mounts in the OCI spec.
    #[must_use]
    pub fn with_volume_paths(mut self, volume_paths: HashMap<String, PathBuf>) -> Self {
        self.volume_paths = volume_paths;
        self
    }

    /// Set the OCI image configuration (entrypoint, cmd, env, workdir, user)
    ///
    /// When set, the image config provides defaults for the container process
    /// that are used when the deployment spec doesn't override them.
    #[must_use]
    pub fn with_image_config(mut self, config: zlayer_registry::ImageConfig) -> Self {
        self.image_config = Some(config);
        self
    }

    /// Enable host networking mode
    ///
    /// When true, the container will NOT get its own network namespace and will
    /// share the host's network stack. This is equivalent to Docker's `--network host`.
    /// Use this when overlay networking is unavailable or not desired.
    #[must_use]
    pub fn with_host_network(mut self, host_network: bool) -> Self {
        self.host_network = host_network;
        self
    }

    /// Join an existing network namespace by path (Docker `--network container:<id>`).
    ///
    /// When `Some(path)`, the container's Network namespace is emitted WITH that
    /// `path`, which causes libcontainer to `setns()` into the target's netns
    /// (sharing its interfaces, overlay attach, and DNS) instead of unsharing a
    /// fresh one. The container therefore does NOT set up its own veth/overlay;
    /// it inherits the target's. Pass `None` (the default) for the normal "new
    /// netns" behaviour. Ignored when [`Self::with_host_network`] is `true`,
    /// since host networking emits no Network namespace at all.
    #[must_use]
    pub fn with_netns_path(mut self, netns_path: Option<PathBuf>) -> Self {
        self.netns_path = netns_path;
        self
    }

    /// Set the secrets provider for resolving `$S:` prefixed environment variables
    ///
    /// When set, environment variables with `$S:secret-name` syntax will be resolved
    /// from this provider at bundle creation time.
    #[must_use]
    pub fn with_secrets_provider(mut self, provider: Arc<dyn SecretsProvider>) -> Self {
        self.secrets_provider = Some(provider);
        self
    }

    /// Set the deployment scope for secret lookups
    ///
    /// This is typically the deployment name and is used as the scope when
    /// resolving `$S:` prefixed environment variables.
    #[must_use]
    pub fn with_deployment_scope(mut self, scope: SecretScope) -> Self {
        self.deployment_scope = Some(scope);
        self
    }

    /// Set a host-side Unix socket path to bind-mount into the container at
    /// the default `ZLayer` socket path (read-only).
    #[must_use]
    pub fn with_socket_mount(mut self, path: impl Into<String>) -> Self {
        self.socket_path = Some(path.into());
        self
    }

    /// Set a host-side per-container Docker Engine API socket to bind-mount into
    /// the container at `/var/run/docker.sock` (read-write, so the container can
    /// `connect()`).
    #[must_use]
    pub fn with_docker_socket_mount(mut self, path: impl Into<String>) -> Self {
        self.docker_socket_path = Some(path.into());
        self
    }

    /// Set the shared toolchain-cache host dir (see [`Self::toolchain_cache`]).
    #[must_use]
    pub fn with_toolchain_cache(mut self, dir: impl Into<PathBuf>) -> Self {
        self.toolchain_cache = Some(dir.into());
        self
    }

    /// Get the bundle directory path
    #[must_use]
    pub fn bundle_dir(&self) -> &Path {
        &self.bundle_dir
    }

    /// Build the OCI bundle from a `ServiceSpec`
    ///
    /// Creates the bundle directory structure and generates config.json
    /// based on the provided service specification.
    ///
    /// # Returns
    /// The path to the bundle directory on success
    ///
    /// # Errors
    /// - `AgentError::CreateFailed` if directory creation fails
    /// - `AgentError::InvalidSpec` if the OCI spec generation fails
    ///
    /// # Platform
    /// Unix-only. Uses `tokio::fs::symlink` which is defined in terms of
    /// `std::os::unix::fs::symlink` and does not exist on Windows. The Windows
    /// WSL2 delegate path should call [`BundleBuilder::build_spec_only`] to
    /// obtain the OCI [`Spec`] and pipe it into the WSL2 distro, where the
    /// Linux side of the delegate handles bundle directory creation.
    #[cfg(unix)]
    pub async fn build(&self, container_id: &ContainerId, spec: &ServiceSpec) -> Result<PathBuf> {
        // Create bundle directory
        fs::create_dir_all(&self.bundle_dir)
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: container_id.to_string(),
                reason: format!("failed to create bundle directory: {e}"),
            })?;

        // Set up rootfs (symlink or create empty directory)
        let rootfs_in_bundle = self.bundle_dir.join("rootfs");
        if let Some(ref rootfs_path) = self.rootfs_path {
            // Remove existing rootfs symlink/dir if present
            let _ = fs::remove_file(&rootfs_in_bundle).await;
            let _ = fs::remove_dir(&rootfs_in_bundle).await;

            // Create symlink to actual rootfs.
            // On Unix: `tokio::fs::symlink` (unified file/dir symlink).
            // On Windows: `tokio::fs::symlink_dir` (wraps CreateSymbolicLinkW with
            // SYMBOLIC_LINK_FLAG_DIRECTORY) — rootfs is always an OCI layer directory.
            #[cfg(unix)]
            tokio::fs::symlink(rootfs_path, &rootfs_in_bundle)
                .await
                .map_err(|e| AgentError::CreateFailed {
                    id: container_id.to_string(),
                    reason: format!(
                        "failed to symlink rootfs from {} to {}: {}",
                        rootfs_path.display(),
                        rootfs_in_bundle.display(),
                        e
                    ),
                })?;

            #[cfg(windows)]
            tokio::fs::symlink_dir(rootfs_path, &rootfs_in_bundle)
                .await
                .map_err(|e| AgentError::CreateFailed {
                    id: container_id.to_string(),
                    reason: format!(
                        "failed to symlink rootfs from {} to {}: {}",
                        rootfs_path.display(),
                        rootfs_in_bundle.display(),
                        e
                    ),
                })?;
        } else {
            // Create empty rootfs directory (for bind mounts)
            fs::create_dir_all(&rootfs_in_bundle)
                .await
                .map_err(|e| AgentError::CreateFailed {
                    id: container_id.to_string(),
                    reason: format!("failed to create rootfs directory: {e}"),
                })?;
        }

        // Generate OCI runtime spec
        let oci_spec = self
            .build_spec_only(container_id, spec, &self.volume_paths)
            .await?;

        // Write config.json
        let config_path = self.bundle_dir.join("config.json");
        let config_json =
            serde_json::to_string_pretty(&oci_spec).map_err(|e| AgentError::CreateFailed {
                id: container_id.to_string(),
                reason: format!("failed to serialize OCI spec: {e}"),
            })?;

        fs::write(&config_path, config_json)
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: container_id.to_string(),
                reason: format!("failed to write config.json: {e}"),
            })?;

        tracing::debug!(
            "Created OCI bundle at {} for container {}",
            self.bundle_dir.display(),
            container_id
        );

        Ok(self.bundle_dir.clone())
    }

    /// Render the OCI runtime spec without creating a bundle directory
    /// or writing `config.json`.
    ///
    /// This is the cross-platform entry point for OCI spec generation and is
    /// the only bundle-builder method that is callable on Windows. Used by the
    /// WSL2 delegate runtime (`runtimes/wsl2_delegate.rs`): the Windows host
    /// renders the spec, then streams the JSON into the WSL distro filesystem
    /// where `youki` will consume it. The bundle path passed to
    /// `BundleBuilder::new` is purely informational in that flow; this method
    /// never touches the filesystem.
    ///
    /// Unix hosts that want both the spec *and* the on-disk bundle layout
    /// (rootfs symlink, `config.json`, parent directories) should continue to
    /// use [`BundleBuilder::build`] or [`BundleBuilder::write_config`].
    ///
    /// # Errors
    /// Returns [`AgentError::InvalidSpec`] if any of the OCI `*Builder` types
    /// reject the configuration, or if environment-variable secret resolution
    /// fails.
    pub async fn build_spec_only(
        &self,
        container_id: &ContainerId,
        spec: &ServiceSpec,
        volume_paths: &std::collections::HashMap<String, PathBuf>,
    ) -> Result<oci_spec::runtime::Spec> {
        self.build_oci_spec(container_id, spec, volume_paths).await
    }

    /// Resolve CDI edits for a service spec's GPU request, if any.
    ///
    /// Returns:
    /// - `Ok(None)` when the spec has no `GpuSpec`, when the vendor isn't a
    ///   known CDI-published kind (e.g. `"apple"`), or when no explicit
    ///   registry was set and lazy discovery turned up no installed specs
    ///   (production fallback — baked-in defaults take over).
    /// - `Ok(Some(vec))` with one entry per requested device when CDI specs
    ///   are available and resolution succeeds.
    /// - `Err(AgentError::InvalidSpec(...))` when the caller explicitly opted
    ///   into CDI (via `with_cdi_registry`) but the resolution fails —
    ///   surfaces [`cdi::CdiError::SpecMissing`] /
    ///   [`cdi::CdiError::DeviceMissing`] / [`cdi::CdiError::NoDevices`] as
    ///   actionable strings.
    fn resolve_cdi_edits(&self, spec: &ServiceSpec) -> Result<Option<Vec<CdiContainerEdits>>> {
        let Some(ref gpu) = spec.resources.gpu else {
            return Ok(None);
        };

        // Map short vendor to CDI kind. Unknown vendors (e.g. "apple") fall
        // back to baked-in behavior.
        let Some(kind) = cdi::vendor_to_cdi_kind(&gpu.vendor) else {
            return Ok(None);
        };

        // Decide registry source:
        // - Explicit override: strict mode. Missing kind/device == hard error.
        // - Lazy discover: opportunistic. Missing kind == silent fallback to
        //   baked-in defaults so prod hosts without CDI installed keep
        //   working.
        let (registry, strict) = if let Some(reg) = &self.cdi_registry {
            (reg.clone(), true)
        } else {
            let reg = Arc::new(CdiRegistry::discover());
            if reg.is_empty() {
                return Ok(None);
            }
            (reg, false)
        };

        let device_names: Vec<String> = (0..gpu.count).map(|i| i.to_string()).collect();

        match registry.resolve_for_kind(kind, &device_names) {
            Ok(edits) => Ok(Some(edits)),
            Err(err) => {
                if strict {
                    Err(AgentError::InvalidSpec(format!(
                        "CDI resolution failed for vendor '{}': {err}",
                        gpu.vendor
                    )))
                } else {
                    tracing::warn!(
                        vendor = %gpu.vendor,
                        kind = %kind,
                        error = %err,
                        "CDI resolution failed; falling back to baked-in GPU device passthrough"
                    );
                    Ok(None)
                }
            }
        }
    }

    /// Build the OCI runtime spec from `ServiceSpec`.
    ///
    /// The full, CDI-aware implementation that backs both
    /// [`BundleBuilder::build_spec_only`] (cross-platform, public) and the
    /// Unix-only [`BundleBuilder::build`] / [`BundleBuilder::write_config`]
    /// paths that additionally manage the bundle directory on disk.
    ///
    /// # Errors
    /// Returns [`AgentError::InvalidSpec`] if any of the OCI `*Builder` types
    /// reject the configuration, or if environment-variable secret resolution
    /// fails.
    ///
    /// # Panics
    /// Panics if the builder-internal `MountBuilder::build()` call fails for
    /// the optional `ZLayer` API socket bind-mount. This is only reachable when
    /// [`BundleBuilder::with_socket_mount`] has been used with a malformed
    /// path, and is treated as a programmer error because all fields are
    /// statically constructed from known-good inputs.
    #[allow(clippy::too_many_lines)]
    async fn build_oci_spec(
        &self,
        container_id: &ContainerId,
        spec: &ServiceSpec,
        volume_paths: &std::collections::HashMap<String, PathBuf>,
    ) -> Result<Spec> {
        // Resolve CDI edits up front. When present, these replace the
        // baked-in vendor device-node / env injection below; when absent
        // (no CDI installed, unknown vendor), the legacy code paths run.
        let cdi_edits = self.resolve_cdi_edits(spec)?;

        // Build user: image config user > root (spec doesn't currently have user override)
        let user = {
            let (uid, gid) = if let Some(user_str) = self
                .image_config
                .as_ref()
                .and_then(|c| c.user.as_ref())
                .filter(|u| !u.is_empty())
            {
                // Parse "uid:gid" or "uid" format from image config
                let parts: Vec<&str> = user_str.splitn(2, ':').collect();
                let uid = parts[0].parse::<u32>().unwrap_or(0);
                let gid = if parts.len() > 1 {
                    parts[1].parse::<u32>().unwrap_or(0)
                } else {
                    uid
                };
                (uid, gid)
            } else {
                (0u32, 0u32)
            };

            UserBuilder::default()
                .uid(uid)
                .gid(gid)
                .build()
                .map_err(|e| AgentError::InvalidSpec(format!("failed to build user: {e}")))?
        };

        // Build environment variables
        // Layer: image config env (base) -> defaults -> spec env -> builder extra env
        let mut env: Vec<String> = Vec::new();
        let mut env_keys: HashSet<String> = HashSet::new();

        // Seed with image config env first (lowest priority)
        if let Some(img_env) = self.image_config.as_ref().and_then(|c| c.env.as_ref()) {
            for entry in img_env {
                if let Some(key) = entry.split('=').next() {
                    env_keys.insert(key.to_string());
                }
                env.push(entry.clone());
            }
        }

        // If image config didn't provide PATH, add the default
        if !env_keys.contains("PATH") {
            env.push(
                "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin".to_string(),
            );
            env_keys.insert("PATH".to_string());
        }

        // Add TERM for interactive compatibility (if not already set)
        if !env_keys.contains("TERM") {
            env.push("TERM=xterm".to_string());
            env_keys.insert("TERM".to_string());
        }

        // When a shared toolchain cache is mounted, point the CI runner
        // tool-cache env at it. Seeded as a low-priority default (before spec
        // env and builder extras) so an image/spec/user value can override.
        if self.toolchain_cache.is_some() {
            if !env_keys.contains("RUNNER_TOOL_CACHE") {
                env.push("RUNNER_TOOL_CACHE=/opt/zlayer/toolchains".to_string());
                env_keys.insert("RUNNER_TOOL_CACHE".to_string());
            }
            if !env_keys.contains("AGENT_TOOLSDIRECTORY") {
                env.push("AGENT_TOOLSDIRECTORY=/opt/zlayer/toolchains".to_string());
                env_keys.insert("AGENT_TOOLSDIRECTORY".to_string());
            }
        }

        // Add service-specific env vars, resolving $S: and $E: prefixed references
        // These override image config env for same keys
        //
        // When a secrets provider is available, use the full secrets-aware resolver
        // that handles both $S: (secret) and $E: (env) prefixed values.
        // Otherwise fall back to the env-only resolver.
        if let (Some(secrets_provider), Some(scope)) =
            (&self.secrets_provider, &self.deployment_scope)
        {
            let resolved_map = crate::env::resolve_env_with_secrets(
                &spec.env,
                secrets_provider.as_ref(),
                &scope.to_storage_scope(),
            )
            .await
            .map_err(|e| {
                AgentError::InvalidSpec(format!("environment variable resolution failed: {e}"))
            })?;

            for (key, value) in &resolved_map {
                if env_keys.contains(key.as_str()) {
                    env.retain(|e| e.split('=').next() != Some(key.as_str()));
                }
                env_keys.insert(key.clone());
                env.push(format!("{key}={value}"));
            }
        } else {
            let resolved = crate::env::resolve_env_vars_with_warnings(&spec.env).map_err(|e| {
                AgentError::InvalidSpec(format!("environment variable resolution failed: {e}"))
            })?;

            // Log any warnings about resolved env vars
            for warning in &resolved.warnings {
                tracing::warn!(container = %container_id, "{}", warning);
            }

            // Merge spec env: spec values take precedence over image config for same keys
            for var in &resolved.vars {
                if let Some(key) = var.split('=').next() {
                    if env_keys.contains(key) {
                        // Remove the old entry from image config
                        env.retain(|e| e.split('=').next() != Some(key));
                    }
                    env_keys.insert(key.to_string());
                }
                env.push(var.clone());
            }
        }

        // Add extra env vars from builder (highest priority)
        for (key, value) in &self.extra_env {
            if env_keys.contains(key.as_str()) {
                env.retain(|e| e.split('=').next() != Some(key.as_str()));
            }
            env_keys.insert(key.clone());
            env.push(format!("{key}={value}"));
        }

        // GPU device visibility environment variables.
        //
        // When CDI edits are available, the vendor-supplied spec is the
        // source of truth (e.g. NVIDIA's `nvidia-ctk cdi generate` emits
        // `NVIDIA_VISIBLE_DEVICES` plus driver-capability env on every
        // device entry). Otherwise fall back to the historical baked-in
        // strings so non-CDI hosts continue to advertise the right devices
        // to CUDA/ROCm/oneAPI runtimes.
        if let Some(ref edits_per_device) = cdi_edits {
            for edits in edits_per_device {
                for entry in &edits.env {
                    if let Some(key) = entry.split('=').next() {
                        if env_keys.contains(key) {
                            env.retain(|e| e.split('=').next() != Some(key));
                        }
                        env_keys.insert(key.to_string());
                    }
                    env.push(entry.clone());
                }
            }
        } else if let Some(ref gpu) = spec.resources.gpu {
            // Default to 0..count when no explicit indices are provided
            let indices: Vec<String> = (0..gpu.count).map(|i| i.to_string()).collect();
            let device_list = indices.join(",");
            match gpu.vendor.as_str() {
                "nvidia" => {
                    env.push(format!("NVIDIA_VISIBLE_DEVICES={device_list}"));
                    env.push(format!("CUDA_VISIBLE_DEVICES={device_list}"));
                }
                "amd" => {
                    env.push(format!("ROCR_VISIBLE_DEVICES={device_list}"));
                    env.push(format!("HIP_VISIBLE_DEVICES={device_list}"));
                }
                "intel" => {
                    env.push(format!("ZE_AFFINITY_MASK={device_list}"));
                }
                _ => {}
            }
        }

        // GPU sharing (MPS / time-slicing) env injection.
        //
        // Layered on top of the CDI / baked-in `*_VISIBLE_DEVICES` block above:
        // * MPS: validate host pipe/log dirs exist (error otherwise) and
        //   export `CUDA_MPS_PIPE_DIRECTORY` / `CUDA_MPS_LOG_DIRECTORY`.
        // * Time-slicing: override `CUDA_VISIBLE_DEVICES` to the configured
        //   slice index so the workload sees a single virtualised GPU rather
        //   than the full 0..count list emitted above.
        //
        // The mount side (bind-mounting the MPS dirs / time-slicing config
        // file) is handled further down where the rest of the mounts get
        // assembled.
        let mps_dirs = if let Some(ref gpu) = spec.resources.gpu {
            resolve_mps_dirs(gpu)?
        } else {
            None
        };
        if let Some(ref dirs) = mps_dirs {
            let pipe = format!("CUDA_MPS_PIPE_DIRECTORY={}", dirs.pipe_dir.display());
            let log = format!("CUDA_MPS_LOG_DIRECTORY={}", dirs.log_dir.display());
            if env_keys.contains("CUDA_MPS_PIPE_DIRECTORY") {
                env.retain(|e| e.split('=').next() != Some("CUDA_MPS_PIPE_DIRECTORY"));
            }
            if env_keys.contains("CUDA_MPS_LOG_DIRECTORY") {
                env.retain(|e| e.split('=').next() != Some("CUDA_MPS_LOG_DIRECTORY"));
            }
            env_keys.insert("CUDA_MPS_PIPE_DIRECTORY".to_string());
            env_keys.insert("CUDA_MPS_LOG_DIRECTORY".to_string());
            env.push(pipe);
            env.push(log);
        }
        if let Some(ref gpu) = spec.resources.gpu {
            if gpu.sharing == Some(GpuSharingMode::TimeSlice) {
                if let Some(idx) = gpu.time_slice_index {
                    // Time-slicing virtualises a single physical GPU as N
                    // slices; the workload sees one device, addressed by
                    // its slice index. Override whatever the CDI / baked-in
                    // path emitted earlier.
                    env.retain(|e| e.split('=').next() != Some("CUDA_VISIBLE_DEVICES"));
                    env_keys.insert("CUDA_VISIBLE_DEVICES".to_string());
                    env.push(format!("CUDA_VISIBLE_DEVICES={idx}"));
                }
            }
        }

        // Inject distributed training coordination env vars when configured.
        // MASTER_ADDR uses the service DNS name (resolved by the overlay DNS).
        // RANK defaults to 0 (overridden by the agent when placing specific replicas).
        if let Some(ref gpu) = spec.resources.gpu {
            if let Some(ref dist) = gpu.distributed {
                env.push(format!("MASTER_PORT={}", dist.master_port));
                env.push(format!("MASTER_ADDR={}", container_id.service));
                env.push("WORLD_SIZE=1".to_string());
                env.push("RANK=0".to_string());
                env.push("LOCAL_RANK=0".to_string());
                match dist.backend.as_str() {
                    "nccl" => env.push("NCCL_SOCKET_IFNAME=eth0".to_string()),
                    "gloo" => env.push("GLOO_SOCKET_IFNAME=eth0".to_string()),
                    _ => {}
                }
            }
        }

        // Inject swarm ring-neighbor discovery env vars for pipeline-parallel
        // inference stages/coordinators. Bare peer service names resolve over
        // the existing per-service overlay DNS (same as MASTER_ADDR above).
        if let Some(ref gpu) = spec.resources.gpu {
            if let Some(ref sharding) = gpu.sharding {
                for (k, v) in crate::bundle::swarm_ring_env(sharding) {
                    env.push(format!("{k}={v}"));
                }
            }
        }

        // Build capabilities
        let capabilities = self.build_capabilities(spec)?;

        // Determine working directory: builder override > spec.command.workdir > image config > "/"
        let cwd = self
            .cwd
            .clone()
            .or_else(|| spec.command.workdir.clone())
            .or_else(|| {
                self.image_config
                    .as_ref()
                    .and_then(|c| c.working_dir.as_ref())
                    .filter(|w| !w.is_empty())
                    .cloned()
            })
            .unwrap_or_else(|| "/".to_string());

        // Resolve process args: builder override > spec command > image config > /bin/sh
        let process_args = if let Some(ref args) = self.args {
            args.clone()
        } else {
            Self::resolve_command_from_spec(spec, self.image_config.as_ref())
        };

        // Build process
        let mut process_builder = ProcessBuilder::default()
            .terminal(false)
            .user(user)
            .env(env)
            .args(process_args)
            .cwd(cwd)
            .no_new_privileges(!spec.privileged && spec.capabilities.is_empty());

        // Set capabilities if we have them
        if let Some(caps) = capabilities {
            process_builder = process_builder.capabilities(caps);
        }

        // Translate `spec.ulimits` (Docker --ulimit style, lowercase keys) into
        // OCI `process.rlimits`. Without this libcontainer never calls
        // setrlimit and the container inherits the launching daemon's
        // defaults — typically nofile=1024, which saturates sharded-storage
        // workloads (PlatformStore, etc.) within seconds of boot.
        let mut rlimits: Vec<PosixRlimit> = Vec::with_capacity(spec.ulimits.len());
        for (name, limit) in &spec.ulimits {
            let typ = ulimit_name_to_posix(name).ok_or_else(|| {
                AgentError::InvalidSpec(format!(
                    "unknown ulimit name `{name}` (expected one of: cpu, fsize, data, stack, \
                     core, rss, nproc, nofile, memlock, as, locks, sigpending, msgqueue, nice, \
                     rtprio, rttime)"
                ))
            })?;
            let entry = PosixRlimitBuilder::default()
                .typ(typ)
                .soft(u64::try_from(limit.soft.max(0)).unwrap_or(0))
                .hard(u64::try_from(limit.hard.max(0)).unwrap_or(0))
                .build()
                .map_err(|e| {
                    AgentError::InvalidSpec(format!("failed to build rlimit `{name}`: {e}"))
                })?;
            rlimits.push(entry);
        }
        if !rlimits.is_empty() {
            process_builder = process_builder.rlimits(rlimits);
        }

        let process = process_builder
            .build()
            .map_err(|e| AgentError::InvalidSpec(format!("failed to build process: {e}")))?;

        // Build root filesystem config
        // Note: "rootfs" is relative to the bundle directory per OCI spec
        let root = RootBuilder::default()
            .path("rootfs".to_string())
            .readonly(false)
            .build()
            .map_err(|e| AgentError::InvalidSpec(format!("failed to build root: {e}")))?;

        // Build default mounts
        let mut mounts = self.build_default_mounts(spec)?;

        // Add storage mounts from spec
        let storage_mounts = self.build_storage_mounts(spec, volume_paths)?;
        mounts.extend(storage_mounts);

        // Add ZLayer API socket bind-mount if configured.
        // Use typ("bind") so libcontainer's mount code handles the source path
        // correctly for sockets (canonicalize + file-based mount point creation).
        if let Some(ref socket_path) = self.socket_path {
            mounts.push(
                MountBuilder::default()
                    .destination(zlayer_paths::ZLayerDirs::default_socket_path())
                    .typ("bind")
                    .source(socket_path.clone())
                    .options(vec!["rbind".into(), "ro".into()])
                    .build()
                    .expect("valid socket mount"),
            );
        }

        // Per-container Docker Engine API socket bind-mount. Read-write (no
        // "ro"): connecting to a Unix socket needs write access to the socket
        // file. `typ("bind")` so libcontainer creates the file mount point.
        if let Some(ref docker_socket_path) = self.docker_socket_path {
            mounts.push(
                MountBuilder::default()
                    .destination("/var/run/docker.sock")
                    .typ("bind")
                    .source(docker_socket_path.clone())
                    .options(vec!["rbind".into()])
                    .build()
                    .expect("valid docker socket mount"),
            );
        }

        // Shared toolchain cache RW bind-mount. When set, CI runner toolchains
        // (installed under `/opt/zlayer/toolchains` via `RUNNER_TOOL_CACHE`) are
        // shared across containers instead of re-downloaded per run.
        if let Some(ref toolchain_cache) = self.toolchain_cache {
            mounts.push(
                MountBuilder::default()
                    .destination("/opt/zlayer/toolchains")
                    .typ("bind")
                    .source(toolchain_cache.clone())
                    .options(vec!["rbind".into(), "rw".into()])
                    .build()
                    .expect("valid toolchain cache mount"),
            );
        }

        // Container DNS resolver injection.
        //
        // youki/libcontainer does no resolv.conf handling on its own: the
        // container sees whatever `/etc/resolv.conf` the image shipped (often
        // empty/absent). When the spec carries explicit resolver addresses
        // (`spec.dns`, populated upstream in `ServiceManager` with the overlay
        // resolver's node-IP — the host's own resolv.conf is unusable because
        // the netbird `~.` systemd-resolved hijack swallows container queries),
        // we materialize a minimal resolv.conf alongside the bundle and
        // bind-mount it read-only at `/etc/resolv.conf`.
        //
        // The `resolv.conf` `nameserver` directive has no port syntax (always
        // port 53), which is exactly why the overlay DNS server must already be
        // bound on `<node_ip>:53` for this address to be useful.
        //
        // Host-network containers share the host's `/etc/resolv.conf` directly,
        // so we skip injection for them (matching the Docker runtime). On the
        // WSL2-on-Windows render path `build_spec_only` is called without an
        // on-disk bundle directory; the `bundle_dir.exists()` guard skips the
        // file write + mount there, preserving today's behavior.
        // Track the resolv.conf source so the host-bind validation pass below
        // can exempt it: a missing/failed resolv.conf source must warn+skip,
        // never abort the container start (DNS injection is best-effort).
        let mut resolv_conf_source: Option<PathBuf> = None;
        if !spec.host_network && !spec.dns.is_empty() && self.bundle_dir.exists() {
            let resolv_path = self.bundle_dir.join("resolv.conf");
            let contents = generate_resolv_conf(&spec.dns, &spec.dns_search);

            // Defensively ensure the bundle directory exists rather than relying
            // solely on an external caller having created it; `bundle_dir.exists()`
            // above only proves it existed at check time.
            let mut wrote = true;
            if let Some(parent) = resolv_path.parent() {
                if let Err(e) = fs::create_dir_all(parent).await {
                    tracing::warn!(
                        bundle_dir = %parent.display(),
                        error = %e,
                        "failed to ensure bundle dir for resolv.conf; skipping DNS injection"
                    );
                    wrote = false;
                }
            }

            if wrote {
                if let Err(e) = fs::write(&resolv_path, contents).await {
                    tracing::warn!(
                        path = %resolv_path.display(),
                        error = %e,
                        "failed to write resolv.conf to bundle; skipping DNS injection"
                    );
                    wrote = false;
                }
            }

            // Verify the write actually landed on disk before pushing the mount;
            // a phantom source would otherwise blow up libcontainer's rootfs
            // canonicalize with the opaque "failed to prepare rootfs".
            if wrote && !resolv_path.exists() {
                tracing::warn!(
                    path = %resolv_path.display(),
                    "resolv.conf write reported success but file is absent; skipping DNS injection"
                );
                wrote = false;
            }

            if wrote {
                resolv_conf_source = Some(resolv_path.clone());
                mounts.push(
                    MountBuilder::default()
                        .destination("/etc/resolv.conf".to_string())
                        .typ("bind")
                        .source(resolv_path.to_string_lossy().to_string())
                        .options(vec!["rbind".to_string(), "ro".to_string()])
                        .build()
                        .map_err(|e| {
                            AgentError::InvalidSpec(format!(
                                "failed to build resolv.conf mount: {e}"
                            ))
                        })?,
                );
            }
        }

        // Append CDI-provided mounts (e.g. vendor driver libraries that the
        // GPU runtime needs to expose to the container).
        if let Some(ref edits_per_device) = cdi_edits {
            for edits in edits_per_device {
                for cdi_mount in &edits.mounts {
                    let mut opts = cdi_mount.options.clone();
                    if !opts.iter().any(|o| o == "bind" || o == "rbind") {
                        opts.push("rbind".to_string());
                    }
                    mounts.push(
                        MountBuilder::default()
                            .destination(cdi_mount.container_path.clone())
                            .typ("bind")
                            .source(cdi_mount.host_path.clone())
                            .options(opts)
                            .build()
                            .map_err(|e| {
                                AgentError::InvalidSpec(format!("failed to build CDI mount: {e}"))
                            })?,
                    );
                }
            }
        }

        // GPU sharing mounts.
        //
        // MPS: bind-mount the host pipe / log directories into the container
        // at the same path so the in-container CUDA runtime can talk to the
        // MPS daemon over its UNIX socket and append to the shared log.
        // The env vars (`CUDA_MPS_PIPE_DIRECTORY` / `CUDA_MPS_LOG_DIRECTORY`)
        // are exported earlier in the env-assembly block.
        //
        // Time-slicing: optionally surface the host's slicing config YAML at
        // a well-known read-only path so introspection tools inside the
        // container can read it.
        if let Some(ref dirs) = mps_dirs {
            mounts.push(
                MountBuilder::default()
                    .destination(dirs.pipe_dir.clone())
                    .typ("bind")
                    .source(dirs.pipe_dir.clone())
                    .options(vec!["rbind".into(), "rw".into()])
                    .build()
                    .map_err(|e| {
                        AgentError::InvalidSpec(format!("failed to build MPS pipe mount: {e}"))
                    })?,
            );
            mounts.push(
                MountBuilder::default()
                    .destination(dirs.log_dir.clone())
                    .typ("bind")
                    .source(dirs.log_dir.clone())
                    .options(vec!["rbind".into(), "rw".into()])
                    .build()
                    .map_err(|e| {
                        AgentError::InvalidSpec(format!("failed to build MPS log mount: {e}"))
                    })?,
            );
        }
        if let Some(ref gpu) = spec.resources.gpu {
            if gpu.sharing == Some(GpuSharingMode::TimeSlice) {
                if let Some(ref cfg_path) = gpu.time_slicing_config_path {
                    let host = PathBuf::from(cfg_path);
                    if !host.is_file() {
                        return Err(AgentError::GpuSharingUnavailable {
                            mode: "time-slice".to_string(),
                            reason: format!(
                                "time-slicing config {} is not a regular file on the host",
                                host.display()
                            ),
                        });
                    }
                    mounts.push(
                        MountBuilder::default()
                            .destination(PathBuf::from(TIMESLICE_CONFIG_CONTAINER_PATH))
                            .typ("bind")
                            .source(host)
                            .options(vec!["rbind".into(), "ro".into()])
                            .build()
                            .map_err(|e| {
                                AgentError::InvalidSpec(format!(
                                    "failed to build time-slicing config mount: {e}"
                                ))
                            })?,
                    );
                }
            }
        }

        // Validate host-source bind mounts before handing the spec to
        // libcontainer. libcontainer canonicalizes the source of every bind
        // mount during rootfs prep; a missing source aborts the whole start
        // with the opaque `failed to prepare rootfs` and no indication of which
        // mount is at fault. Catch it here with an actionable error naming the
        // source AND the destination.
        //
        // Virtual filesystems (proc/tmpfs/sysfs/devpts/mqueue/cgroup*/...) have
        // no real host source path — their `source` is a label like "proc" or
        // "tmpfs" — so they are skipped. The daemon-generated resolv.conf
        // convenience mount is exempt: if its source is missing it would have
        // been warn-skipped above (never reaching the mount list), but we guard
        // explicitly so a regression there degrades gracefully rather than
        // hard-failing a start.
        Self::validate_host_bind_sources(&mounts, resolv_conf_source.as_deref())?;

        // Build Linux-specific config
        let linux = self.build_linux_config(container_id, spec, cdi_edits.as_deref())?;

        // Determine hostname
        let hostname = self
            .hostname
            .clone()
            .unwrap_or_else(|| container_id.to_string());

        // Build the complete spec, attaching any CDI-provided hooks.
        let mut spec_builder = SpecBuilder::default()
            .version("1.0.2".to_string())
            .root(root)
            .process(process)
            .hostname(hostname)
            .mounts(mounts)
            .linux(linux);

        if let Some(ref edits_per_device) = cdi_edits {
            if let Some(hooks) = Self::build_hooks_from_cdi(edits_per_device)? {
                spec_builder = spec_builder.hooks(hooks);
            }
        }

        let oci_spec = spec_builder
            .build()
            .map_err(|e| AgentError::InvalidSpec(format!("failed to build OCI spec: {e}")))?;

        Ok(oci_spec)
    }

    /// Validate that every host-source bind mount in `mounts` points at a path
    /// that actually exists on disk.
    ///
    /// libcontainer (youki) canonicalizes the `source` of every `bind`/`rbind`
    /// mount while preparing the rootfs. When the source is missing the call
    /// fails deep inside libcontainer with the opaque, source-less message
    /// `failed to prepare rootfs` — leaving no clue which of the socket, CDI,
    /// GPU, or storage-volume mounts is at fault. This pre-flight check turns
    /// that into an actionable [`AgentError::MountSourceMissing`] naming both
    /// the host `source` and the in-container `dest`.
    ///
    /// Distinguishing *virtual* mounts from *host bind* mounts robustly:
    /// - A host bind has an **absolute** filesystem `source` path. The kernel
    ///   pseudo-filesystems (`proc`, `tmpfs`, `sysfs`, `devpts`, `mqueue`,
    ///   `cgroup`/`cgroup2`, ...) carry a *relative label* source like `"proc"`
    ///   or `"tmpfs"`, never an absolute path, so `Path::is_absolute` cleanly
    ///   separates the two — independent of how the `typ`/`options` express the
    ///   bind. This builder emits both `typ="bind"` for the socket/CDI/GPU
    ///   mounts and `typ="none"` + an `rbind` option for storage volume binds;
    ///   keying off the absolute source catches both shapes.
    /// - A secondary guard skips any explicit virtual-FS `typ` even if it were
    ///   (incorrectly) handed an absolute source, so a real pseudo-filesystem
    ///   is never wrongly rejected.
    ///
    /// `resolv_conf_source`, when set, is the host path of the daemon-generated
    /// `/etc/resolv.conf` convenience mount. It is exempt from the hard check:
    /// DNS injection is best-effort and a missing resolv.conf source must never
    /// block a container start (it is warn-skipped at write time).
    fn validate_host_bind_sources(
        mounts: &[Mount],
        resolv_conf_source: Option<&Path>,
    ) -> Result<()> {
        // Virtual filesystem types whose `source` is a label, not a host path.
        // libcontainer does not canonicalize these, so a missing "source" is
        // irrelevant; anything else with an absolute source is a real host bind
        // that libcontainer *will* canonicalize.
        const VIRTUAL_FS_TYPES: &[&str] = &[
            "proc",
            "tmpfs",
            "sysfs",
            "devpts",
            "mqueue",
            "cgroup",
            "cgroup2",
            "devtmpfs",
            "ramfs",
            "securityfs",
            "debugfs",
            "tracefs",
            "fusectl",
            "configfs",
            "pstore",
            "bpf",
            "binfmt_misc",
            "hugetlbfs",
        ];

        for mount in mounts {
            let Some(source) = mount.source() else {
                continue;
            };

            // Primary discriminator: a real host bind source is always an
            // absolute path; virtual-FS sources are relative labels.
            if !source.is_absolute() {
                continue;
            }

            // Secondary guard: skip explicit virtual-FS types regardless.
            if let Some(typ) = mount.typ().as_deref() {
                if VIRTUAL_FS_TYPES.contains(&typ) {
                    continue;
                }
            }

            // Exempt the daemon-generated resolv.conf convenience mount.
            if let Some(resolv) = resolv_conf_source {
                if source == resolv {
                    continue;
                }
            }

            if !source.exists() {
                return Err(AgentError::MountSourceMissing {
                    src_path: source.to_string_lossy().into_owned(),
                    dest: mount.destination().to_string_lossy().into_owned(),
                });
            }
        }
        Ok(())
    }

    /// Convert the union of CDI hooks across all resolved devices into an
    /// OCI [`Hooks`] block.
    ///
    /// Returns `Ok(None)` when no device contributed hooks (so the spec
    /// builder skips the empty block — `oci-spec` treats `null` as "no
    /// hooks" while serializers may emit empty arrays otherwise).
    fn build_hooks_from_cdi(edits_per_device: &[CdiContainerEdits]) -> Result<Option<Hooks>> {
        let mut prestart: Vec<Hook> = Vec::new();
        let mut create_runtime: Vec<Hook> = Vec::new();
        let mut create_container: Vec<Hook> = Vec::new();
        let mut start_container: Vec<Hook> = Vec::new();
        let mut poststart: Vec<Hook> = Vec::new();
        let mut poststop: Vec<Hook> = Vec::new();

        for edits in edits_per_device {
            let Some(ref h) = edits.hooks else { continue };
            for hook in &h.prestart {
                prestart.push(convert_cdi_hook(hook)?);
            }
            for hook in &h.create_runtime {
                create_runtime.push(convert_cdi_hook(hook)?);
            }
            for hook in &h.create_container {
                create_container.push(convert_cdi_hook(hook)?);
            }
            for hook in &h.start_container {
                start_container.push(convert_cdi_hook(hook)?);
            }
            for hook in &h.poststart {
                poststart.push(convert_cdi_hook(hook)?);
            }
            for hook in &h.poststop {
                poststop.push(convert_cdi_hook(hook)?);
            }
        }

        if prestart.is_empty()
            && create_runtime.is_empty()
            && create_container.is_empty()
            && start_container.is_empty()
            && poststart.is_empty()
            && poststop.is_empty()
        {
            return Ok(None);
        }

        let mut builder = HooksBuilder::default();
        if !prestart.is_empty() {
            #[allow(deprecated)]
            {
                builder = builder.prestart(prestart);
            }
        }
        if !create_runtime.is_empty() {
            builder = builder.create_runtime(create_runtime);
        }
        if !create_container.is_empty() {
            builder = builder.create_container(create_container);
        }
        if !start_container.is_empty() {
            builder = builder.start_container(start_container);
        }
        if !poststart.is_empty() {
            builder = builder.poststart(poststart);
        }
        if !poststop.is_empty() {
            builder = builder.poststop(poststop);
        }

        let hooks = builder
            .build()
            .map_err(|e| AgentError::InvalidSpec(format!("failed to build CDI hooks: {e}")))?;
        Ok(Some(hooks))
    }

    /// Build Linux capabilities configuration
    #[allow(clippy::unused_self)]
    fn build_capabilities(
        &self,
        spec: &ServiceSpec,
    ) -> Result<Option<oci_spec::runtime::LinuxCapabilities>> {
        if spec.privileged {
            // Privileged mode: all capabilities
            let all_caps: HashSet<Capability> = ALL_CAPABILITIES.iter().copied().collect();
            let empty_caps: HashSet<Capability> = HashSet::new();

            let caps = LinuxCapabilitiesBuilder::default()
                .bounding(all_caps.clone())
                .effective(all_caps.clone())
                .permitted(all_caps)
                .inheritable(empty_caps.clone())
                .ambient(empty_caps)
                .build()
                .map_err(|e| {
                    AgentError::InvalidSpec(format!("failed to build capabilities: {e}"))
                })?;

            Ok(Some(caps))
        } else if !spec.capabilities.is_empty() {
            // Specific capabilities requested
            let caps: HashSet<Capability> = spec
                .capabilities
                .iter()
                .filter_map(|c| {
                    // Normalize capability name (add CAP_ prefix if missing, uppercase)
                    let cap_name = if c.starts_with("CAP_") {
                        c.to_uppercase()
                    } else {
                        format!("CAP_{}", c.to_uppercase())
                    };
                    Capability::from_str(&cap_name).ok()
                })
                .collect();

            let empty_caps: HashSet<Capability> = HashSet::new();

            let built_caps = LinuxCapabilitiesBuilder::default()
                .bounding(caps.clone())
                .effective(caps.clone())
                .permitted(caps)
                .inheritable(empty_caps.clone())
                .ambient(empty_caps)
                .build()
                .map_err(|e| {
                    AgentError::InvalidSpec(format!("failed to build capabilities: {e}"))
                })?;

            Ok(Some(built_caps))
        } else {
            // Default: minimal capabilities for basic container operation
            let default_caps: HashSet<Capability> = [
                Capability::Chown,
                Capability::DacOverride,
                Capability::Fsetid,
                Capability::Fowner,
                Capability::Mknod,
                Capability::NetRaw,
                Capability::Setgid,
                Capability::Setuid,
                Capability::Setfcap,
                Capability::Setpcap,
                Capability::NetBindService,
                Capability::SysChroot,
                Capability::Kill,
                Capability::AuditWrite,
            ]
            .into_iter()
            .collect();

            let empty_caps: HashSet<Capability> = HashSet::new();

            let built_caps = LinuxCapabilitiesBuilder::default()
                .bounding(default_caps.clone())
                .effective(default_caps.clone())
                .permitted(default_caps)
                .inheritable(empty_caps.clone())
                .ambient(empty_caps)
                .build()
                .map_err(|e| {
                    AgentError::InvalidSpec(format!("failed to build capabilities: {e}"))
                })?;

            Ok(Some(built_caps))
        }
    }

    /// Build default filesystem mounts for the container
    #[allow(clippy::unused_self, clippy::too_many_lines)]
    fn build_default_mounts(&self, spec: &ServiceSpec) -> Result<Vec<Mount>> {
        let mut mounts = Vec::new();

        // /proc
        mounts.push(
            MountBuilder::default()
                .destination("/proc".to_string())
                .typ("proc".to_string())
                .source("proc".to_string())
                .options(vec![
                    "nosuid".to_string(),
                    "noexec".to_string(),
                    "nodev".to_string(),
                ])
                .build()
                .map_err(|e| {
                    AgentError::InvalidSpec(format!("failed to build /proc mount: {e}"))
                })?,
        );

        // /dev
        mounts.push(
            MountBuilder::default()
                .destination("/dev".to_string())
                .typ("tmpfs".to_string())
                .source("tmpfs".to_string())
                .options(vec![
                    "nosuid".to_string(),
                    "strictatime".to_string(),
                    "mode=755".to_string(),
                    "size=65536k".to_string(),
                ])
                .build()
                .map_err(|e| AgentError::InvalidSpec(format!("failed to build /dev mount: {e}")))?,
        );

        // /dev/pts
        mounts.push(
            MountBuilder::default()
                .destination("/dev/pts".to_string())
                .typ("devpts".to_string())
                .source("devpts".to_string())
                .options(vec![
                    "nosuid".to_string(),
                    "noexec".to_string(),
                    "newinstance".to_string(),
                    "ptmxmode=0666".to_string(),
                    "mode=0620".to_string(),
                    "gid=5".to_string(),
                ])
                .build()
                .map_err(|e| {
                    AgentError::InvalidSpec(format!("failed to build /dev/pts mount: {e}"))
                })?,
        );

        // /dev/shm
        mounts.push(
            MountBuilder::default()
                .destination("/dev/shm".to_string())
                .typ("tmpfs".to_string())
                .source("shm".to_string())
                .options(vec![
                    "nosuid".to_string(),
                    "noexec".to_string(),
                    "nodev".to_string(),
                    "mode=1777".to_string(),
                    "size=65536k".to_string(),
                ])
                .build()
                .map_err(|e| {
                    AgentError::InvalidSpec(format!("failed to build /dev/shm mount: {e}"))
                })?,
        );

        // /dev/mqueue
        mounts.push(
            MountBuilder::default()
                .destination("/dev/mqueue".to_string())
                .typ("mqueue".to_string())
                .source("mqueue".to_string())
                .options(vec![
                    "nosuid".to_string(),
                    "noexec".to_string(),
                    "nodev".to_string(),
                ])
                .build()
                .map_err(|e| {
                    AgentError::InvalidSpec(format!("failed to build /dev/mqueue mount: {e}"))
                })?,
        );

        // /sys - read-only unless privileged
        let sys_options = if spec.privileged {
            vec![
                "nosuid".to_string(),
                "noexec".to_string(),
                "nodev".to_string(),
            ]
        } else {
            vec![
                "nosuid".to_string(),
                "noexec".to_string(),
                "nodev".to_string(),
                "ro".to_string(),
            ]
        };

        mounts.push(
            MountBuilder::default()
                .destination("/sys".to_string())
                .typ("sysfs".to_string())
                .source("sysfs".to_string())
                .options(sys_options)
                .build()
                .map_err(|e| AgentError::InvalidSpec(format!("failed to build /sys mount: {e}")))?,
        );

        // /sys/fs/cgroup - for cgroup access
        mounts.push(
            MountBuilder::default()
                .destination("/sys/fs/cgroup".to_string())
                .typ("cgroup2".to_string())
                .source("cgroup".to_string())
                .options(vec![
                    "nosuid".to_string(),
                    "noexec".to_string(),
                    "nodev".to_string(),
                    "relatime".to_string(),
                ])
                .build()
                .map_err(|e| {
                    AgentError::InvalidSpec(format!("failed to build cgroup mount: {e}"))
                })?,
        );

        Ok(mounts)
    }

    /// Build storage mounts from `ServiceSpec` storage entries
    ///
    /// Converts `StorageSpec` entries to OCI Mount entries.
    /// Note: Named and Anonymous volumes require `StorageManager` to prepare paths.
    /// S3 volumes require s3fs FUSE mount (handled separately).
    #[allow(clippy::unused_self, clippy::too_many_lines)]
    fn build_storage_mounts(
        &self,
        spec: &ServiceSpec,
        volume_paths: &std::collections::HashMap<String, PathBuf>,
    ) -> Result<Vec<Mount>> {
        let mut mounts = Vec::new();

        for storage in &spec.storage {
            let mount = match storage {
                StorageSpec::Bind {
                    source,
                    target,
                    readonly,
                } => {
                    let mut options = vec!["rbind".to_string()];
                    if *readonly {
                        options.push("ro".to_string());
                    } else {
                        options.push("rw".to_string());
                    }

                    MountBuilder::default()
                        .destination(target.clone())
                        .typ("none".to_string())
                        .source(source.clone())
                        .options(options)
                        .build()
                        .map_err(|e| {
                            AgentError::InvalidSpec(format!(
                                "failed to build bind mount for {target}: {e}"
                            ))
                        })?
                }

                StorageSpec::Named {
                    name,
                    target,
                    readonly,
                    tier,
                    ..
                } => {
                    // Get the prepared volume path from StorageManager
                    let source = volume_paths.get(name).ok_or_else(|| {
                        AgentError::InvalidSpec(format!(
                            "volume '{name}' not prepared - ensure StorageManager.ensure_volume() was called"
                        ))
                    })?;

                    // Warn about SQLite safety for non-local tiers
                    if matches!(tier, StorageTier::Network) {
                        tracing::warn!(
                            volume = %name,
                            tier = ?tier,
                            "Network storage tier is NOT SQLite-safe. Avoid using SQLite databases on this volume."
                        );
                    }

                    let mut options = vec!["rbind".to_string()];
                    if *readonly {
                        options.push("ro".to_string());
                    } else {
                        options.push("rw".to_string());
                    }

                    MountBuilder::default()
                        .destination(target.clone())
                        .typ("none".to_string())
                        .source(source.to_string_lossy().to_string())
                        .options(options)
                        .build()
                        .map_err(|e| {
                            AgentError::InvalidSpec(format!(
                                "failed to build named volume mount for {target}: {e}"
                            ))
                        })?
                }

                StorageSpec::Anonymous { target, tier } => {
                    // Anonymous volumes should have been created by StorageManager
                    // and the path passed in volume_paths with key "_anon_{target}"
                    let key = format!("_anon_{}", target.trim_start_matches('/').replace('/', "_"));
                    let source = volume_paths.get(&key).ok_or_else(|| {
                        AgentError::InvalidSpec(format!(
                            "anonymous volume for '{target}' not prepared"
                        ))
                    })?;

                    if matches!(tier, StorageTier::Network) {
                        tracing::warn!(
                            target = %target,
                            tier = ?tier,
                            "Network storage tier is NOT SQLite-safe."
                        );
                    }

                    let options = vec!["rbind".to_string(), "rw".to_string()];

                    MountBuilder::default()
                        .destination(target.clone())
                        .typ("none".to_string())
                        .source(source.to_string_lossy().to_string())
                        .options(options)
                        .build()
                        .map_err(|e| {
                            AgentError::InvalidSpec(format!(
                                "failed to build anonymous volume mount for {target}: {e}"
                            ))
                        })?
                }

                StorageSpec::Tmpfs { target, size, mode } => {
                    let mut options = vec!["nosuid".to_string(), "nodev".to_string()];

                    if let Some(size_str) = size {
                        options.push(format!("size={size_str}"));
                    }

                    if let Some(mode_val) = mode {
                        options.push(format!("mode={mode_val:o}"));
                    }

                    MountBuilder::default()
                        .destination(target.clone())
                        .typ("tmpfs".to_string())
                        .source("tmpfs".to_string())
                        .options(options)
                        .build()
                        .map_err(|e| {
                            AgentError::InvalidSpec(format!(
                                "failed to build tmpfs mount for {target}: {e}"
                            ))
                        })?
                }

                StorageSpec::S3 {
                    bucket,
                    prefix,
                    target,
                    readonly,
                    endpoint: _,
                    credentials: _,
                } => {
                    // S3 mounts are handled via s3fs FUSE
                    // The StorageManager should have mounted the bucket and passed the path
                    let key = format!("_s3_{}_{}", bucket, prefix.as_deref().unwrap_or(""));
                    let source = volume_paths.get(&key).ok_or_else(|| {
                        AgentError::InvalidSpec(format!(
                            "S3 volume for bucket '{bucket}' not mounted - ensure StorageManager.mount_s3() was called"
                        ))
                    })?;

                    tracing::warn!(
                        bucket = %bucket,
                        target = %target,
                        "S3 storage is NOT SQLite-safe. Use for read-heavy workloads only."
                    );

                    let mut options = vec!["rbind".to_string()];
                    if *readonly {
                        options.push("ro".to_string());
                    } else {
                        options.push("rw".to_string());
                    }

                    MountBuilder::default()
                        .destination(target.clone())
                        .typ("none".to_string())
                        .source(source.to_string_lossy().to_string())
                        .options(options)
                        .build()
                        .map_err(|e| {
                            AgentError::InvalidSpec(format!(
                                "failed to build S3 mount for {target}: {e}"
                            ))
                        })?
                }
            };

            mounts.push(mount);
        }

        Ok(mounts)
    }

    /// Build Linux-specific configuration
    #[allow(clippy::similar_names)] // euid/egid are POSIX-standard paired names
    #[allow(clippy::too_many_lines)]
    fn build_linux_config(
        &self,
        container_id: &ContainerId,
        spec: &ServiceSpec,
        cdi_edits: Option<&[CdiContainerEdits]>,
    ) -> Result<oci_spec::runtime::Linux> {
        // Build namespaces
        let mut namespaces = vec![
            LinuxNamespaceBuilder::default()
                .typ(LinuxNamespaceType::Pid)
                .build()
                .unwrap(),
            LinuxNamespaceBuilder::default()
                .typ(LinuxNamespaceType::Ipc)
                .build()
                .unwrap(),
            LinuxNamespaceBuilder::default()
                .typ(LinuxNamespaceType::Uts)
                .build()
                .unwrap(),
            LinuxNamespaceBuilder::default()
                .typ(LinuxNamespaceType::Mount)
                .build()
                .unwrap(),
        ];

        // Network namespace handling — three mutually-exclusive cases:
        //
        //   * `host_network`            → emit NO Network namespace; the container
        //                                 shares the host stack (Docker `--network
        //                                 host`). `netns_path` is ignored here.
        //   * `netns_path = Some(path)` → emit a Network namespace WITH `path`,
        //                                 which makes libcontainer `setns()` into
        //                                 the target container's netns (JOIN —
        //                                 Docker `--network container:<id>`). The
        //                                 joined container inherits the target's
        //                                 interfaces/overlay/DNS and does NOT set
        //                                 up its own veth/overlay.
        //   * otherwise                 → emit a Network namespace WITHOUT a path
        //                                 (unshare a fresh netns — default).
        if self.host_network {
            // share host: no Network namespace.
        } else if let Some(netns_path) = self.netns_path.as_ref() {
            namespaces.push(
                LinuxNamespaceBuilder::default()
                    .typ(LinuxNamespaceType::Network)
                    .path(netns_path.clone())
                    .build()
                    .unwrap(),
            );
        } else {
            namespaces.push(
                LinuxNamespaceBuilder::default()
                    .typ(LinuxNamespaceType::Network)
                    .build()
                    .unwrap(),
            );
        }

        // `nix::unistd` is unix-only. On non-unix targets (Windows), libcontainer
        // is not the runtime path (HCS is) and this function is effectively dead
        // code — so we statically force `rootless = false` there and skip the
        // user-namespace mapping block entirely.
        // When the daemon itself runs inside a rootless userns (uid 0 mapped to host
        // 1000), geteuid() returns 0 *inside that userns*, so the geteuid check would
        // wrongly conclude rootless=false and build the container with NO user
        // namespace. ZLAYER_ROOTLESS=1 (set by the daemon at startup, inherited
        // in-process) is authoritative: force a nested single-uid userns so each
        // container netns is a DESCENDANT of the daemon userns and overlayd can setns
        // in to attach veth.
        #[cfg(unix)]
        let daemon_rootless = std::env::var_os("ZLAYER_ROOTLESS").is_some();
        #[cfg(unix)]
        let rootless = daemon_rootless || !nix::unistd::geteuid().is_root();
        #[cfg(not(unix))]
        let rootless = false;

        if rootless {
            namespaces.push(
                LinuxNamespaceBuilder::default()
                    .typ(LinuxNamespaceType::User)
                    .build()
                    .unwrap(),
            );
            namespaces.push(
                LinuxNamespaceBuilder::default()
                    .typ(LinuxNamespaceType::Cgroup)
                    .build()
                    .unwrap(),
            );
        }

        let mut linux_builder = LinuxBuilder::default().namespaces(namespaces);

        #[cfg(unix)]
        if rootless {
            if daemon_rootless {
                // The daemon runs in its own single-uid userns and cannot
                // sub-delegate a /etc/subuid range to a nested container, so map
                // only container 0 → host 0 (size 1) for both uid and gid. This
                // single own-uid map is written directly to /proc/<pid>/uid_map
                // (no newuidmap / subuid range needed).
                linux_builder = linux_builder
                    .uid_mappings(build_single_id_mapping(0))
                    .gid_mappings(build_single_id_mapping(0));
            } else {
                let euid = nix::unistd::geteuid();
                let egid = nix::unistd::getegid();
                let username = nix::unistd::User::from_uid(euid)
                    .ok()
                    .flatten()
                    .map(|u| u.name)
                    .unwrap_or_default();
                linux_builder = linux_builder
                    .uid_mappings(build_rootless_id_mappings(
                        euid.as_raw(),
                        "/etc/subuid",
                        &username,
                    ))
                    .gid_mappings(build_rootless_id_mappings(
                        egid.as_raw(),
                        "/etc/subgid",
                        &username,
                    ));
            }
        }

        // Build resources (CPU, memory, devices)
        let resources = self.build_resources(spec)?;
        if let Some(resources) = resources {
            linux_builder = linux_builder.resources(resources);
        }

        // Build device entries for passthrough.
        //
        // When CDI edits are present, the vendor-supplied device-node list
        // replaces our baked-in vendor-specific defaults — CDI knows the
        // host's exact device geometry (which majors/minors map to which
        // GPUs) so we trust it over our static `/dev/nvidiaN` enumeration.
        let mut devices = self.build_devices(spec, None, cdi_edits.is_some())?;
        if let Some(edits_per_device) = cdi_edits {
            for edits in edits_per_device {
                for node in &edits.device_nodes {
                    devices.push(cdi_node_to_oci_device(node)?);
                }
            }
        }
        if !devices.is_empty() {
            linux_builder = linux_builder.devices(devices);
        }

        // Set rootfs propagation (matches Docker default)
        linux_builder = linux_builder.rootfs_propagation("private".to_string());

        // Set masked/readonly paths based on privileged mode
        if spec.privileged {
            // Privileged containers get no masked paths (full access)
            linux_builder = linux_builder.masked_paths(vec![]).readonly_paths(vec![]);
        } else {
            // Set masked paths for security (hide sensitive host info)
            let masked_paths = vec![
                "/proc/acpi".to_string(),
                "/proc/asound".to_string(),
                "/proc/kcore".to_string(),
                "/proc/keys".to_string(),
                "/proc/latency_stats".to_string(),
                "/proc/timer_list".to_string(),
                "/proc/timer_stats".to_string(),
                "/proc/sched_debug".to_string(),
                "/proc/scsi".to_string(),
                "/sys/firmware".to_string(),
            ];

            // Set readonly paths for security
            let readonly_paths = vec![
                "/proc/bus".to_string(),
                "/proc/fs".to_string(),
                "/proc/irq".to_string(),
                "/proc/sys".to_string(),
                "/proc/sysrq-trigger".to_string(),
            ];

            linux_builder = linux_builder
                .masked_paths(masked_paths)
                .readonly_paths(readonly_paths);
        }

        // Determine cgroups_path so libcontainer creates the container cgroup
        // under the current process's cgroup rather than at the v2 root. This
        // is required when running inside another container (e.g. Forgejo CI
        // `container:` block) where `/sys/fs/cgroup/cgroup.subtree_control` is
        // read-only. Precedence:
        //   1. spec.cgroup_parent (per-service override)         — all platforms
        //   2. ZLAYER_CGROUP_PARENT env var (host-wide override) — all platforms
        //   3. /proc/self/cgroup (auto-detect when nested)       — Linux only
        //   4. unset (default — bare-metal happy path; also the WSL2-delegate
        //      case on non-Linux hosts, where libcontainer inside the WSL
        //      distro resolves the parent at `zlayer runtime create` time)
        let cid = container_id.to_string();

        // Explicit overrides are honored on every platform: a user might pin a
        // cgroup_parent for a WSL-delegate-bound spec even when this process
        // is running on Windows.
        let explicit_parent: Option<(String, &'static str)> =
            if let Some(p) = spec.cgroup_parent.as_deref().filter(|s| !s.is_empty()) {
                Some((p.to_string(), "spec"))
            } else if let Some(p) = std::env::var("ZLAYER_CGROUP_PARENT")
                .ok()
                .filter(|s| !s.is_empty())
            {
                Some((p, "env"))
            } else {
                None
            };

        // Auto-detect (and the "no writable parent" hard error below) are
        // Linux-only: they inspect /proc/self/cgroup and /sys/fs/cgroup, which
        // don't exist on Windows hosts. When the bundle is destined for the
        // WSL2 delegate, cgroup-parent resolution happens inside the distro
        // at `zlayer runtime create` time, not here on the host.
        #[cfg(target_os = "linux")]
        let auto_parent: Option<(String, &'static str)> = {
            // A writable cgroup-v2 root means this is a root host daemon (NOT a
            // nested CI `container:` block, where the root is read-only). In
            // that case root containers OUTSIDE the daemon's own systemd unit
            // cgroup, under a top-level `/zlayer/containers` node. Keeping them
            // out of `/system.slice/zlayer.service/...` is what stops a
            // KillMode=process survivor from turning the unit cgroup into a
            // populated inner node and wedging `systemctl restart zlayer` with
            // EBUSY (`Result: resources`). Note: `is_nested` is NOT the right
            // discriminator — a normal systemd service is always "nested"
            // under `/system.slice/...`; only `can_write_cgroup_root`
            // distinguishes the host daemon from the read-only nested case.
            let host_mode = crate::capability::DaemonCapabilities::get().can_write_cgroup_root;
            if host_mode {
                if let Some(p) = crate::capability::ensure_host_container_parent() {
                    Some((p, "auto-host"))
                } else if let Some(p) = crate::capability::ensure_daemon_leaf_and_container_parent()
                {
                    // Top-level setup failed (unexpected for a writable root);
                    // fall back to in-scope placement rather than the v2 root.
                    Some((p, "auto-init"))
                } else if let Some(p) = crate::capability::current_cgroup_v2_path() {
                    Some((p, "auto"))
                } else {
                    None
                }
            } else if let Some(p) = crate::capability::ensure_daemon_leaf_and_container_parent() {
                // Nested / read-only-root case (CI `container:` block): the
                // delegated `<scope>/containers` subtree is the only writable
                // place, and the daemon here is not a KillMode=process systemd
                // unit, so the EBUSY-on-restart scenario does not apply.
                Some((p, "auto-init"))
            } else if let Some(p) = crate::capability::current_cgroup_v2_path() {
                // Fallback: migration failed (likely cgroup root is read-only); use the
                // raw scope path. Pre-fix behaviour — surfaces the original error.
                Some((p, "auto"))
            } else {
                None
            }
        };
        #[cfg(not(target_os = "linux"))]
        let auto_parent: Option<(String, &'static str)> = None;

        let (cgroup_parent_value, cgroup_parent_source): (Option<String>, &'static str) =
            explicit_parent
                .or(auto_parent)
                .map_or((None, "none"), |(p, s)| (Some(p), s));

        // Diagnostic guard rail: capability survey says we're nested, but we
        // couldn't resolve a cgroup parent here. This combination should not
        // normally happen because both code paths consult the same
        // `current_cgroup_v2_path()` helper. Surface it so an operator can
        // investigate; do not fail container creation. Linux-only — the
        // capability survey is itself a no-op on non-Linux.
        #[cfg(target_os = "linux")]
        if cgroup_parent_value.is_none() && crate::capability::DaemonCapabilities::get().is_nested {
            tracing::warn!(
                container_id = %cid,
                "capability survey reports nested daemon but cgroup_parent could not be resolved — proceeding with v2 root"
            );
        }

        if let Some(parent) = cgroup_parent_value {
            let parent = parent.trim_end_matches('/');
            let full = format!("{parent}/{cid}");
            match cgroup_parent_source {
                "spec" => tracing::info!(
                    container_id = %cid,
                    source = "spec",
                    path = %full,
                    "cgroup_parent selected"
                ),
                "env" => tracing::info!(
                    container_id = %cid,
                    source = "env",
                    path = %full,
                    "cgroup_parent selected"
                ),
                "auto" => tracing::info!(
                    container_id = %cid,
                    source = "auto",
                    path = %full,
                    "cgroup_parent selected (from /proc/self/cgroup)"
                ),
                "auto-init" => tracing::info!(
                    container_id = %cid,
                    source = "auto-init",
                    path = %full,
                    "cgroup_parent selected (migrated daemon to <scope>/init; containers go under <scope>/containers)"
                ),
                "auto-host" => tracing::info!(
                    container_id = %cid,
                    source = "auto-host",
                    path = %full,
                    "cgroup_parent selected (host daemon; containers rooted at top-level /zlayer/containers, outside the unit cgroup)"
                ),
                _ => unreachable!(),
            }
            linux_builder = linux_builder.cgroups_path(std::path::PathBuf::from(full));
        } else {
            // Auto-detect found nothing AND no explicit override. Behaviour
            // differs by platform:
            //   - Linux: this is a real error in nested-container envs where
            //     the cgroup root is read-only. Emit the hard error so an
            //     operator fixes the env.
            //   - Non-Linux (Windows host building a bundle for the WSL2
            //     delegate): expected path; cgroup setup happens inside the
            //     distro at runtime-create time.
            #[cfg(target_os = "linux")]
            {
                let caps = crate::capability::DaemonCapabilities::get();
                if !caps.can_write_cgroup_root {
                    return Err(AgentError::InvalidSpec(format!(
                        "cannot create container {cid}: no writable cgroup parent. \
                         /proc/self/cgroup reports the cgroup-v2 root, and \
                         /sys/fs/cgroup is read-only to this process. Fix one of: \
                         (a) run the daemon's outer container with --cgroupns=host \
                         so /proc/self/cgroup reports a real parent; \
                         (b) set ZLAYER_CGROUP_PARENT=/path/to/writable/cgroup; \
                         (c) grant the daemon write access to /sys/fs/cgroup."
                    )));
                }
                tracing::info!(
                    container_id = %cid,
                    "cgroup_parent unset — libcontainer will use v2 root (cgroup root is writable here)"
                );
            }
            #[cfg(not(target_os = "linux"))]
            tracing::debug!(
                container_id = %cid,
                "non-Linux host — cgroup_parent unset; libcontainer inside the WSL distro will resolve a parent from its cgroup-v2 root"
            );
        }

        linux_builder
            .build()
            .map_err(|e| AgentError::InvalidSpec(format!("failed to build linux config: {e}")))
    }

    /// Build resource limits (CPU, memory, device cgroups)
    #[allow(clippy::cast_possible_truncation, clippy::cast_possible_wrap)]
    fn build_resources(
        &self,
        spec: &ServiceSpec,
    ) -> Result<Option<oci_spec::runtime::LinuxResources>> {
        let mut resources_builder = LinuxResourcesBuilder::default();
        let mut has_resources = false;

        // CPU limits
        if let Some(cpu_limit) = spec.resources.cpu {
            // Convert CPU cores to microseconds quota
            // 100000 microseconds = 1 core's worth of time per period
            let quota = (cpu_limit * 100_000.0) as i64;
            let cpu = LinuxCpuBuilder::default()
                .quota(quota)
                .period(100_000u64)
                .build()
                .map_err(|e| AgentError::InvalidSpec(format!("failed to build CPU limits: {e}")))?;

            resources_builder = resources_builder.cpu(cpu);
            has_resources = true;
        }

        // Memory limits
        if let Some(ref memory_str) = spec.resources.memory {
            let bytes = parse_memory_string(memory_str)
                .map_err(|e| AgentError::InvalidSpec(format!("invalid memory limit: {e}")))?;

            let memory = LinuxMemoryBuilder::default()
                .limit(bytes as i64)
                .build()
                .map_err(|e| {
                    AgentError::InvalidSpec(format!("failed to build memory limits: {e}"))
                })?;

            resources_builder = resources_builder.memory(memory);
            has_resources = true;
        }

        // Device cgroup rules
        let device_rules = self.build_device_cgroup_rules(spec, None)?;
        if !device_rules.is_empty() {
            resources_builder = resources_builder.devices(device_rules);
            has_resources = true;
        }

        if has_resources {
            let resources = resources_builder
                .build()
                .map_err(|e| AgentError::InvalidSpec(format!("failed to build resources: {e}")))?;
            Ok(Some(resources))
        } else {
            Ok(None)
        }
    }

    /// Build device cgroup rules
    #[allow(clippy::unused_self, clippy::too_many_lines)]
    fn build_device_cgroup_rules(
        &self,
        spec: &ServiceSpec,
        _gpu_indices: Option<&[u32]>,
    ) -> Result<Vec<oci_spec::runtime::LinuxDeviceCgroup>> {
        let mut rules = Vec::new();

        if spec.privileged {
            // Privileged mode: allow all devices
            let rule = LinuxDeviceCgroupBuilder::default()
                .allow(true)
                .access("rwm".to_string())
                .build()
                .map_err(|e| {
                    AgentError::InvalidSpec(format!("failed to build device cgroup rule: {e}"))
                })?;
            rules.push(rule);
        } else {
            // Default: deny all, then allow specific devices
            let deny_all = LinuxDeviceCgroupBuilder::default()
                .allow(false)
                .access("rwm".to_string())
                .build()
                .map_err(|e| AgentError::InvalidSpec(format!("failed to build deny rule: {e}")))?;
            rules.push(deny_all);

            // Allow standard container devices
            // /dev/null, /dev/zero, /dev/full, /dev/random, /dev/urandom, /dev/tty
            let standard_char_devices = [
                (1, 3, "rwm"),    // /dev/null
                (1, 5, "rwm"),    // /dev/zero
                (1, 7, "rwm"),    // /dev/full
                (1, 8, "rwm"),    // /dev/random
                (1, 9, "rwm"),    // /dev/urandom
                (5, 0, "rwm"),    // /dev/tty
                (5, 1, "rwm"),    // /dev/console
                (5, 2, "rwm"),    // /dev/ptmx
                (136, -1, "rwm"), // /dev/pts/* (wildcard minor)
            ];

            for (major, minor, access) in standard_char_devices {
                let mut builder = LinuxDeviceCgroupBuilder::default()
                    .allow(true)
                    .typ(LinuxDeviceType::C)
                    .major(i64::from(major))
                    .access(access.to_string());

                if minor >= 0 {
                    builder = builder.minor(i64::from(minor));
                }

                let rule = builder.build().map_err(|e| {
                    AgentError::InvalidSpec(format!("failed to build char device rule: {e}"))
                })?;
                rules.push(rule);
            }

            // Allow specific devices from spec (Unix-only: requires /dev/* fs
            // probing via `MetadataExt::rdev`). On Windows the WSL2 delegate
            // path regenerates these inside the Linux distro, so we skip here.
            #[cfg(unix)]
            for device in &spec.devices {
                if let Ok((major, minor)) = get_device_major_minor(&device.path) {
                    let dev_type = get_device_type(&device.path).unwrap_or(LinuxDeviceType::C);

                    // Build access string
                    let mut access = String::new();
                    if device.read {
                        access.push('r');
                    }
                    if device.write {
                        access.push('w');
                    }
                    if device.mknod {
                        access.push('m');
                    }
                    if access.is_empty() {
                        access = "rw".to_string();
                    }

                    let rule = LinuxDeviceCgroupBuilder::default()
                        .allow(true)
                        .typ(dev_type)
                        .major(major)
                        .minor(minor)
                        .access(access)
                        .build()
                        .map_err(|e| {
                            AgentError::InvalidSpec(format!(
                                "failed to build device rule for {}: {}",
                                device.path, e
                            ))
                        })?;
                    rules.push(rule);
                } else {
                    tracing::warn!("Failed to get device info for {}, skipping", device.path);
                }
            }

            // Auto-allow GPU devices in cgroup when gpu spec is set
            if let Some(ref gpu) = spec.resources.gpu {
                match gpu.vendor.as_str() {
                    "nvidia" => {
                        // Allow all nvidia devices (major 195 for nvidia GPUs)
                        let rule = LinuxDeviceCgroupBuilder::default()
                            .allow(true)
                            .typ(LinuxDeviceType::C)
                            .major(195i64)
                            .access("rwm".to_string())
                            .build()
                            .map_err(|e| {
                                AgentError::InvalidSpec(format!(
                                    "failed to build GPU cgroup rule: {e}"
                                ))
                            })?;
                        rules.push(rule);

                        // nvidia-uvm (major 510 or check dynamically)
                        let uvm_rule = LinuxDeviceCgroupBuilder::default()
                            .allow(true)
                            .typ(LinuxDeviceType::C)
                            .major(510i64)
                            .access("rwm".to_string())
                            .build()
                            .map_err(|e| {
                                AgentError::InvalidSpec(format!(
                                    "failed to build GPU UVM cgroup rule: {e}"
                                ))
                            })?;
                        rules.push(uvm_rule);
                    }
                    "amd" => {
                        // AMD ROCm: /dev/dri/renderD* and /dev/dri/card* (major 226)
                        let dri_rule = LinuxDeviceCgroupBuilder::default()
                            .allow(true)
                            .typ(LinuxDeviceType::C)
                            .major(226i64)
                            .access("rwm".to_string())
                            .build()
                            .map_err(|e| {
                                AgentError::InvalidSpec(format!(
                                    "failed to build AMD DRI cgroup rule: {e}"
                                ))
                            })?;
                        rules.push(dri_rule);

                        // /dev/kfd - AMD Kernel Fusion Driver for compute (major 234)
                        let kfd_rule = LinuxDeviceCgroupBuilder::default()
                            .allow(true)
                            .typ(LinuxDeviceType::C)
                            .major(234i64)
                            .access("rwm".to_string())
                            .build()
                            .map_err(|e| {
                                AgentError::InvalidSpec(format!(
                                    "failed to build AMD KFD cgroup rule: {e}"
                                ))
                            })?;
                        rules.push(kfd_rule);
                    }
                    "intel" => {
                        // Intel GPU: /dev/dri/renderD* and /dev/dri/card* (major 226)
                        let dri_rule = LinuxDeviceCgroupBuilder::default()
                            .allow(true)
                            .typ(LinuxDeviceType::C)
                            .major(226i64)
                            .access("rwm".to_string())
                            .build()
                            .map_err(|e| {
                                AgentError::InvalidSpec(format!(
                                    "failed to build Intel DRI cgroup rule: {e}"
                                ))
                            })?;
                        rules.push(dri_rule);
                    }
                    other => {
                        // Unknown vendor - allow DRI devices as a reasonable default
                        tracing::warn!(
                            vendor = %other,
                            "Unknown GPU vendor, allowing DRI devices (major 226)"
                        );
                        let dri_rule = LinuxDeviceCgroupBuilder::default()
                            .allow(true)
                            .typ(LinuxDeviceType::C)
                            .major(226i64)
                            .access("rwm".to_string())
                            .build()
                            .map_err(|e| {
                                AgentError::InvalidSpec(format!(
                                    "failed to build GPU DRI cgroup rule: {e}"
                                ))
                            })?;
                        rules.push(dri_rule);
                    }
                }
            }
        }

        Ok(rules)
    }

    /// Build Linux device entries for passthrough
    ///
    /// # Platform
    /// Every branch below walks `/dev/*` on the host to resolve major/minor
    /// numbers via `MetadataExt::rdev`. On Windows (where this module is
    /// compiled only to feed the WSL2 delegate's cross-platform spec path) we
    /// skip device discovery and return an empty list — the Linux side of the
    /// delegate re-runs this step inside the WSL2 distro.
    #[allow(clippy::unused_self, clippy::too_many_lines)]
    #[cfg_attr(not(unix), allow(clippy::unnecessary_wraps, clippy::needless_return))]
    fn build_devices(
        &self,
        spec: &ServiceSpec,
        gpu_indices: Option<&[u32]>,
        skip_gpu_defaults: bool,
    ) -> Result<Vec<oci_spec::runtime::LinuxDevice>> {
        #[cfg(not(unix))]
        {
            let _ = (spec, gpu_indices, skip_gpu_defaults);
            return Ok(Vec::new());
        }

        #[cfg(unix)]
        {
            let mut devices = Vec::new();

            for device in &spec.devices {
                if let Ok((major, minor)) = get_device_major_minor(&device.path) {
                    let dev_type = get_device_type(&device.path).unwrap_or(LinuxDeviceType::C);

                    let linux_device = LinuxDeviceBuilder::default()
                        .path(device.path.clone())
                        .typ(dev_type)
                        .major(major)
                        .minor(minor)
                        .file_mode(0o666u32)
                        .uid(0u32)
                        .gid(0u32)
                        .build()
                        .map_err(|e| {
                            AgentError::InvalidSpec(format!(
                                "failed to build device {}: {}",
                                device.path, e
                            ))
                        })?;

                    devices.push(linux_device);
                }
            }

            // When CDI is providing GPU device descriptors the caller will
            // append the vendor-supplied entries; skip our hard-coded
            // `/dev/nvidiaN` enumeration so we don't end up with both sources
            // of truth.
            if skip_gpu_defaults {
                return Ok(devices);
            }

            // Auto-inject GPU devices when gpu spec is set
            if let Some(ref gpu) = spec.resources.gpu {
                let indices: Vec<u32> =
                    gpu_indices.map_or_else(|| (0..gpu.count).collect(), <[u32]>::to_vec);

                match gpu.vendor.as_str() {
                    "nvidia" => {
                        // Always needed: nvidiactl, nvidia-uvm, nvidia-uvm-tools
                        let always_devices =
                            ["/dev/nvidiactl", "/dev/nvidia-uvm", "/dev/nvidia-uvm-tools"];
                        for dev_path in &always_devices {
                            if let Ok((major, minor)) = get_device_major_minor(dev_path) {
                                let dev_type =
                                    get_device_type(dev_path).unwrap_or(LinuxDeviceType::C);
                                let linux_device = LinuxDeviceBuilder::default()
                                    .path((*dev_path).to_string())
                                    .typ(dev_type)
                                    .major(major)
                                    .minor(minor)
                                    .file_mode(0o666u32)
                                    .uid(0u32)
                                    .gid(0u32)
                                    .build()
                                    .map_err(|e| {
                                        AgentError::InvalidSpec(format!(
                                            "failed to build GPU device {dev_path}: {e}"
                                        ))
                                    })?;
                                devices.push(linux_device);
                            } else {
                                tracing::warn!(
                                    "GPU device {} not found on host, skipping",
                                    dev_path
                                );
                            }
                        }

                        // Per-GPU devices: /dev/nvidia0, /dev/nvidia1, etc.
                        for i in &indices {
                            let dev_path = format!("/dev/nvidia{i}");
                            if let Ok((major, minor)) = get_device_major_minor(&dev_path) {
                                let dev_type =
                                    get_device_type(&dev_path).unwrap_or(LinuxDeviceType::C);
                                let linux_device = LinuxDeviceBuilder::default()
                                    .path(dev_path.clone())
                                    .typ(dev_type)
                                    .major(major)
                                    .minor(minor)
                                    .file_mode(0o666u32)
                                    .uid(0u32)
                                    .gid(0u32)
                                    .build()
                                    .map_err(|e| {
                                        AgentError::InvalidSpec(format!(
                                            "failed to build GPU device {dev_path}: {e}"
                                        ))
                                    })?;
                                devices.push(linux_device);
                            } else {
                                tracing::warn!(
                                    "GPU device {} not found on host, skipping",
                                    dev_path
                                );
                            }
                        }
                    }
                    "amd" => {
                        // AMD ROCm: /dev/kfd is always required for compute
                        let amd_always_devices = ["/dev/kfd"];
                        for dev_path in &amd_always_devices {
                            if let Ok((major, minor)) = get_device_major_minor(dev_path) {
                                let dev_type =
                                    get_device_type(dev_path).unwrap_or(LinuxDeviceType::C);
                                let linux_device = LinuxDeviceBuilder::default()
                                    .path((*dev_path).to_string())
                                    .typ(dev_type)
                                    .major(major)
                                    .minor(minor)
                                    .file_mode(0o666u32)
                                    .uid(0u32)
                                    .gid(0u32)
                                    .build()
                                    .map_err(|e| {
                                        AgentError::InvalidSpec(format!(
                                            "failed to build GPU device {dev_path}: {e}"
                                        ))
                                    })?;
                                devices.push(linux_device);
                            } else {
                                tracing::warn!(
                                    "GPU device {} not found on host, skipping",
                                    dev_path
                                );
                            }
                        }

                        // DRI render nodes: /dev/dri/renderD128, renderD129, etc.
                        for i in &indices {
                            let dev_path = format!("/dev/dri/renderD{}", 128 + i);
                            if let Ok((major, minor)) = get_device_major_minor(&dev_path) {
                                let dev_type =
                                    get_device_type(&dev_path).unwrap_or(LinuxDeviceType::C);
                                let linux_device = LinuxDeviceBuilder::default()
                                    .path(dev_path.clone())
                                    .typ(dev_type)
                                    .major(major)
                                    .minor(minor)
                                    .file_mode(0o666u32)
                                    .uid(0u32)
                                    .gid(0u32)
                                    .build()
                                    .map_err(|e| {
                                        AgentError::InvalidSpec(format!(
                                            "failed to build GPU device {dev_path}: {e}"
                                        ))
                                    })?;
                                devices.push(linux_device);
                            } else {
                                tracing::warn!(
                                    "GPU device {} not found on host, skipping",
                                    dev_path
                                );
                            }
                        }

                        // DRI card nodes: /dev/dri/card0, card1, etc.
                        for i in &indices {
                            let dev_path = format!("/dev/dri/card{i}");
                            if let Ok((major, minor)) = get_device_major_minor(&dev_path) {
                                let dev_type =
                                    get_device_type(&dev_path).unwrap_or(LinuxDeviceType::C);
                                let linux_device = LinuxDeviceBuilder::default()
                                    .path(dev_path.clone())
                                    .typ(dev_type)
                                    .major(major)
                                    .minor(minor)
                                    .file_mode(0o666u32)
                                    .uid(0u32)
                                    .gid(0u32)
                                    .build()
                                    .map_err(|e| {
                                        AgentError::InvalidSpec(format!(
                                            "failed to build GPU device {dev_path}: {e}"
                                        ))
                                    })?;
                                devices.push(linux_device);
                            } else {
                                tracing::warn!(
                                    "GPU device {} not found on host, skipping",
                                    dev_path
                                );
                            }
                        }
                    }
                    "intel" => {
                        // Intel GPU: DRI render nodes /dev/dri/renderD128, etc.
                        for i in &indices {
                            let dev_path = format!("/dev/dri/renderD{}", 128 + i);
                            if let Ok((major, minor)) = get_device_major_minor(&dev_path) {
                                let dev_type =
                                    get_device_type(&dev_path).unwrap_or(LinuxDeviceType::C);
                                let linux_device = LinuxDeviceBuilder::default()
                                    .path(dev_path.clone())
                                    .typ(dev_type)
                                    .major(major)
                                    .minor(minor)
                                    .file_mode(0o666u32)
                                    .uid(0u32)
                                    .gid(0u32)
                                    .build()
                                    .map_err(|e| {
                                        AgentError::InvalidSpec(format!(
                                            "failed to build GPU device {dev_path}: {e}"
                                        ))
                                    })?;
                                devices.push(linux_device);
                            } else {
                                tracing::warn!(
                                    "GPU device {} not found on host, skipping",
                                    dev_path
                                );
                            }
                        }

                        // Intel DRI card nodes: /dev/dri/card0, card1, etc.
                        for i in &indices {
                            let dev_path = format!("/dev/dri/card{i}");
                            if let Ok((major, minor)) = get_device_major_minor(&dev_path) {
                                let dev_type =
                                    get_device_type(&dev_path).unwrap_or(LinuxDeviceType::C);
                                let linux_device = LinuxDeviceBuilder::default()
                                    .path(dev_path.clone())
                                    .typ(dev_type)
                                    .major(major)
                                    .minor(minor)
                                    .file_mode(0o666u32)
                                    .uid(0u32)
                                    .gid(0u32)
                                    .build()
                                    .map_err(|e| {
                                        AgentError::InvalidSpec(format!(
                                            "failed to build GPU device {dev_path}: {e}"
                                        ))
                                    })?;
                                devices.push(linux_device);
                            } else {
                                tracing::warn!(
                                    "GPU device {} not found on host, skipping",
                                    dev_path
                                );
                            }
                        }
                    }
                    other => {
                        // Unknown vendor - try DRI render nodes as default
                        tracing::warn!(
                            vendor = %other,
                            "Unknown GPU vendor, attempting DRI device passthrough"
                        );
                        for i in &indices {
                            let dev_path = format!("/dev/dri/renderD{}", 128 + i);
                            if let Ok((major, minor)) = get_device_major_minor(&dev_path) {
                                let dev_type =
                                    get_device_type(&dev_path).unwrap_or(LinuxDeviceType::C);
                                let linux_device = LinuxDeviceBuilder::default()
                                    .path(dev_path.clone())
                                    .typ(dev_type)
                                    .major(major)
                                    .minor(minor)
                                    .file_mode(0o666u32)
                                    .uid(0u32)
                                    .gid(0u32)
                                    .build()
                                    .map_err(|e| {
                                        AgentError::InvalidSpec(format!(
                                            "failed to build GPU device {dev_path}: {e}"
                                        ))
                                    })?;
                                devices.push(linux_device);
                            } else {
                                tracing::warn!(
                                    "GPU device {} not found on host, skipping",
                                    dev_path
                                );
                            }
                        }
                    }
                }
            }

            Ok(devices)
        } // end #[cfg(unix)]
    }

    /// Generate the OCI spec and write config.json to the bundle directory
    ///
    /// Unlike `build()`, this does NOT create the bundle directory or set up rootfs.
    /// Use this when the bundle directory and rootfs already exist (e.g., rootfs was
    /// extracted directly by `LayerUnpacker`).
    ///
    /// # Errors
    /// Returns an error if the OCI spec cannot be built or config.json cannot be written.
    ///
    /// # Returns
    /// The path to the bundle directory on success
    pub async fn write_config(
        &self,
        container_id: &ContainerId,
        spec: &ServiceSpec,
    ) -> Result<PathBuf> {
        // Generate OCI runtime spec
        let oci_spec = self
            .build_spec_only(container_id, spec, &self.volume_paths)
            .await?;

        // Write config.json
        let config_path = self.bundle_dir.join("config.json");
        let config_json =
            serde_json::to_string_pretty(&oci_spec).map_err(|e| AgentError::CreateFailed {
                id: container_id.to_string(),
                reason: format!("failed to serialize OCI spec: {e}"),
            })?;

        fs::write(&config_path, config_json)
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: container_id.to_string(),
                reason: format!("failed to write config.json: {e}"),
            })?;

        tracing::debug!(
            "Wrote OCI config.json at {} for container {}",
            config_path.display(),
            container_id
        );

        Ok(self.bundle_dir.clone())
    }

    /// Resolve command from `ServiceSpec` and optional image config following Docker/OCI semantics
    ///
    /// Resolution order:
    /// 1. spec entrypoint + args -> use those
    /// 2. spec entrypoint only -> use entrypoint
    /// 3. spec args only -> use args
    /// 4. `image_config` entrypoint/cmd -> use `image_config.full_command()`
    /// 5. fallback to /bin/sh
    fn resolve_command_from_spec(
        spec: &ServiceSpec,
        image_config: Option<&zlayer_registry::ImageConfig>,
    ) -> Vec<String> {
        let mut args = Vec::new();

        match (&spec.command.entrypoint, &spec.command.args) {
            (Some(entrypoint), Some(cmd_args)) => {
                args.extend_from_slice(entrypoint);
                args.extend_from_slice(cmd_args);
            }
            (Some(entrypoint), None) => {
                args.extend_from_slice(entrypoint);
            }
            (None, Some(cmd_args)) if !cmd_args.is_empty() => {
                args.extend_from_slice(cmd_args);
            }
            _ => {
                // No spec command - try image config
                if let Some(img_cmd) =
                    image_config.and_then(zlayer_registry::ImageConfig::full_command)
                {
                    if img_cmd.is_empty() {
                        args.push("/bin/sh".to_string());
                    } else {
                        args.extend(img_cmd);
                    }
                } else {
                    args.push("/bin/sh".to_string());
                }
            }
        }

        args
    }

    /// Clean up a bundle directory
    ///
    /// Removes the bundle directory and all its contents.
    ///
    /// # Errors
    /// Returns an error if the bundle directory cannot be removed.
    pub async fn cleanup(&self) -> Result<()> {
        if self.bundle_dir.exists() {
            fs::remove_dir_all(&self.bundle_dir)
                .await
                .map_err(|e| AgentError::CreateFailed {
                    id: "cleanup".to_string(),
                    reason: format!(
                        "failed to remove bundle directory {}: {}",
                        self.bundle_dir.display(),
                        e
                    ),
                })?;
        }
        Ok(())
    }
}

/// Create a bundle for a container
///
/// Convenience function that creates a bundle in the default location.
///
/// # Errors
/// Returns an error if bundle creation fails.
///
/// # Platform
/// Unix-only — wraps [`BundleBuilder::build`], which uses
/// `tokio::fs::symlink` (not available on Windows). Windows callers should
/// use [`BundleBuilder::build_spec_only`] directly and pipe the result into
/// a WSL2 delegate.
#[cfg(unix)]
pub async fn create_bundle(
    container_id: &ContainerId,
    spec: &ServiceSpec,
    rootfs_path: Option<PathBuf>,
) -> Result<PathBuf> {
    let mut builder =
        BundleBuilder::for_container(container_id).with_host_network(spec.host_network);

    if let Some(rootfs) = rootfs_path {
        builder = builder.with_rootfs(rootfs);
    }

    builder.build(container_id, spec).await
}

/// Clean up a container's bundle
///
/// Convenience function to remove a bundle from the default location.
///
/// # Errors
/// Returns an error if cleanup fails.
pub async fn cleanup_bundle(container_id: &ContainerId) -> Result<()> {
    let builder = BundleBuilder::for_container(container_id);
    builder.cleanup().await
}

#[cfg(test)]
mod tests {
    use super::*;
    use zlayer_spec::*;

    fn mock_spec() -> ServiceSpec {
        serde_yaml::from_str::<DeploymentSpec>(
            r"
version: v1
deployment: test
services:
  test:
    rtype: service
    image:
      name: test:latest
    endpoints:
      - name: http
        protocol: http
        port: 8080
",
        )
        .unwrap()
        .services
        .remove("test")
        .unwrap()
    }

    #[cfg(target_os = "linux")]
    fn mock_spec_with_resources() -> ServiceSpec {
        serde_yaml::from_str::<DeploymentSpec>(
            r"
version: v1
deployment: test
services:
  test:
    rtype: service
    image:
      name: test:latest
    resources:
      cpu: 0.5
      memory: 512Mi
    env:
      MY_VAR: my_value
      ANOTHER: value2
    endpoints:
      - name: http
        protocol: http
        port: 8080
",
        )
        .unwrap()
        .services
        .remove("test")
        .unwrap()
    }

    #[cfg(target_os = "linux")]
    fn mock_privileged_spec() -> ServiceSpec {
        serde_yaml::from_str::<DeploymentSpec>(
            r"
version: v1
deployment: test
services:
  test:
    rtype: service
    image:
      name: test:latest
    privileged: true
    endpoints:
      - name: http
        protocol: http
        port: 8080
",
        )
        .unwrap()
        .services
        .remove("test")
        .unwrap()
    }

    #[test]
    fn test_parse_memory_string() {
        assert_eq!(parse_memory_string("512Mi").unwrap(), 512 * 1024 * 1024);
        assert_eq!(parse_memory_string("1Gi").unwrap(), 1024 * 1024 * 1024);
        assert_eq!(parse_memory_string("2G").unwrap(), 2 * 1000 * 1000 * 1000);
        assert_eq!(parse_memory_string("1024").unwrap(), 1024);
        assert_eq!(parse_memory_string("512Ki").unwrap(), 512 * 1024);
    }

    #[test]
    fn test_parse_memory_string_errors() {
        assert!(parse_memory_string("").is_err());
        assert!(parse_memory_string("abc").is_err());
        assert!(parse_memory_string("12.5Mi").is_err());
    }

    #[test]
    fn test_generate_resolv_conf_single_nameserver() {
        let out = generate_resolv_conf(&["10.42.0.1".to_string()], &[]);
        assert_eq!(out, "nameserver 10.42.0.1\noptions edns0\n");
    }

    #[test]
    fn test_generate_resolv_conf_two_nameservers() {
        let out = generate_resolv_conf(&["10.42.0.1".to_string(), "fd00::1".to_string()], &[]);
        assert_eq!(
            out,
            "nameserver 10.42.0.1\nnameserver fd00::1\noptions edns0\n"
        );
    }

    #[test]
    fn test_generate_resolv_conf_emits_search_domains() {
        let out = generate_resolv_conf(
            &["10.200.0.5".to_string()],
            &[
                "forgejo-stack.zlayer.local".to_string(),
                "zlayer.local".to_string(),
            ],
        );
        assert_eq!(
            out,
            "nameserver 10.200.0.5\nsearch forgejo-stack.zlayer.local zlayer.local\noptions edns0\n"
        );
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_oci_spec_injects_resolv_conf_mount() {
        let dir = tempfile::tempdir().unwrap();
        let id = ContainerId::new("test".to_string(), 1);
        let mut spec = mock_spec();
        spec.dns = vec!["10.42.0.1".to_string()];
        let builder = BundleBuilder::new(dir.path().to_path_buf());

        let oci_spec = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .unwrap();

        let mounts = oci_spec.mounts().as_ref().expect("mounts present");
        let resolv_mount = mounts
            .iter()
            .find(|m| m.destination() == Path::new("/etc/resolv.conf"))
            .expect("resolv.conf mount injected");
        let source = resolv_mount.source().as_ref().unwrap();
        let written = std::fs::read_to_string(source).unwrap();
        assert_eq!(written, "nameserver 10.42.0.1\noptions edns0\n");
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_oci_spec_no_resolv_conf_when_dns_empty() {
        let dir = tempfile::tempdir().unwrap();
        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_spec(); // spec.dns defaults to empty
        let builder = BundleBuilder::new(dir.path().to_path_buf());

        let oci_spec = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .unwrap();

        let mounts = oci_spec.mounts().as_ref().expect("mounts present");
        assert!(
            !mounts
                .iter()
                .any(|m| m.destination() == Path::new("/etc/resolv.conf")),
            "no resolv.conf mount should be injected for empty spec.dns"
        );
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_oci_spec_no_resolv_conf_when_host_network() {
        let dir = tempfile::tempdir().unwrap();
        let id = ContainerId::new("test".to_string(), 1);
        let mut spec = mock_spec();
        spec.dns = vec!["10.42.0.1".to_string()];
        spec.host_network = true;
        let builder = BundleBuilder::new(dir.path().to_path_buf());

        let oci_spec = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .unwrap();

        let mounts = oci_spec.mounts().as_ref().expect("mounts present");
        assert!(
            !mounts
                .iter()
                .any(|m| m.destination() == Path::new("/etc/resolv.conf")),
            "host_network containers must inherit the host resolv.conf"
        );
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_oci_spec_errors_on_missing_bind_source() {
        // A storage bind mount whose host source does not exist must abort the
        // bundle build with an actionable error naming the source, rather than
        // letting libcontainer fail later with "failed to prepare rootfs".
        let dir = tempfile::tempdir().unwrap();
        let id = ContainerId::new("test".to_string(), 1);
        let mut spec = mock_spec();
        let missing = dir.path().join("definitely-does-not-exist");
        spec.storage.push(StorageSpec::Bind {
            source: missing.to_string_lossy().into_owned(),
            target: "/data".to_string(),
            readonly: false,
        });
        let builder = BundleBuilder::new(dir.path().to_path_buf());

        let err = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .expect_err("missing bind source should error");

        match err {
            AgentError::MountSourceMissing { src_path, dest } => {
                assert_eq!(src_path, missing.to_string_lossy());
                assert_eq!(dest, "/data");
            }
            other => panic!("expected MountSourceMissing, got {other:?}"),
        }
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_oci_spec_accepts_existing_bind_source() {
        // An existing host source must pass validation and produce the mount.
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("present");
        std::fs::create_dir_all(&src).unwrap();
        let id = ContainerId::new("test".to_string(), 1);
        let mut spec = mock_spec();
        spec.storage.push(StorageSpec::Bind {
            source: src.to_string_lossy().into_owned(),
            target: "/data".to_string(),
            readonly: false,
        });
        let builder = BundleBuilder::new(dir.path().to_path_buf());

        let oci_spec = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .expect("existing bind source should build");

        let mounts = oci_spec.mounts().as_ref().expect("mounts present");
        assert!(
            mounts.iter().any(|m| m.destination() == Path::new("/data")),
            "bind mount with existing source should be present"
        );
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_resolv_conf_present_when_source_writable() {
        // The bundle dir is writable, so the daemon-generated resolv.conf is
        // written and its bind mount is injected (and passes validation).
        let dir = tempfile::tempdir().unwrap();
        let id = ContainerId::new("test".to_string(), 1);
        let mut spec = mock_spec();
        spec.dns = vec!["10.42.0.1".to_string()];
        let builder = BundleBuilder::new(dir.path().to_path_buf());

        let oci_spec = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .expect("writable bundle dir should build with resolv.conf");

        let mounts = oci_spec.mounts().as_ref().expect("mounts present");
        let resolv = mounts
            .iter()
            .find(|m| m.destination() == Path::new("/etc/resolv.conf"))
            .expect("resolv.conf mount injected when source is writable");
        let source = resolv.source().as_ref().unwrap();
        assert!(source.exists(), "resolv.conf source must exist on disk");
    }

    // Unix-only: this test asserts Unix path semantics — a `/…` source is
    // "absolute" (a real host bind that must exist). `validate_host_bind_sources`
    // keys off `Path::is_absolute()`, which is `false` for `/…` on Windows (no
    // drive prefix), so a Unix-style missing source is treated as a relative
    // virtual-FS label and skipped there. The function validates LINUX container
    // bind sources (the OCI spec's mount sources are Linux paths even when the
    // Windows host delegates to WSL), so the missing-source assertion only holds
    // under Unix path semantics.
    #[cfg(unix)]
    #[test]
    fn test_validate_host_bind_sources_skips_virtual_and_resolv() {
        // Virtual filesystems (relative-label sources) and the exempt
        // resolv.conf source must never trigger MountSourceMissing, even when
        // a real host bind alongside them is missing-or-present.
        let resolv = std::path::PathBuf::from("/nonexistent/bundle/resolv.conf");

        // proc (relative label) + missing resolv (exempt) -> Ok.
        let proc = MountBuilder::default()
            .destination("/proc".to_string())
            .typ("proc".to_string())
            .source("proc".to_string())
            .build()
            .unwrap();
        let resolv_mount = MountBuilder::default()
            .destination("/etc/resolv.conf".to_string())
            .typ("bind".to_string())
            .source(resolv.to_string_lossy().to_string())
            .options(vec!["rbind".to_string(), "ro".to_string()])
            .build()
            .unwrap();
        BundleBuilder::validate_host_bind_sources(&[proc, resolv_mount], Some(resolv.as_path()))
            .expect("virtual + exempt-resolv mounts must validate");

        // A non-resolv missing host bind (typ=none storage shape) -> error.
        let storage = MountBuilder::default()
            .destination("/data".to_string())
            .typ("none".to_string())
            .source("/definitely/not/here".to_string())
            .options(vec!["rbind".to_string(), "rw".to_string()])
            .build()
            .unwrap();
        let err = BundleBuilder::validate_host_bind_sources(&[storage], None)
            .expect_err("missing storage bind source must error");
        assert!(matches!(err, AgentError::MountSourceMissing { .. }));
    }

    #[test]
    fn test_bundle_builder_new() {
        let builder = BundleBuilder::new("/tmp/test-bundle".into());
        assert_eq!(builder.bundle_dir(), Path::new("/tmp/test-bundle"));
        assert!(builder.rootfs_path.is_none());
    }

    #[test]
    fn test_bundle_builder_for_container() {
        let dirs = zlayer_paths::ZLayerDirs::system_default();
        let id = ContainerId::new("myservice".to_string(), 1);
        let builder = BundleBuilder::for_container(&id);
        assert_eq!(builder.bundle_dir(), dirs.bundles().join("myservice-rep-1"));
    }

    #[test]
    fn test_bundle_builder_with_rootfs() {
        let dirs = zlayer_paths::ZLayerDirs::system_default();
        let builder = BundleBuilder::new("/tmp/test-bundle".into())
            .with_rootfs(dirs.rootfs().join("myimage"));
        assert_eq!(builder.rootfs_path, Some(dirs.rootfs().join("myimage")));
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_oci_spec_basic() {
        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_spec();
        let builder = BundleBuilder::new("/tmp/test-bundle".into());

        let oci_spec = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .unwrap();

        assert_eq!(oci_spec.version(), "1.0.2");
        assert!(oci_spec.root().is_some());
        assert_eq!(
            oci_spec.root().as_ref().unwrap().path(),
            std::path::Path::new("rootfs")
        );
        assert!(oci_spec.process().is_some());
        assert!(oci_spec.linux().is_some());
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_oci_spec_with_resources() {
        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_spec_with_resources();
        let builder = BundleBuilder::new("/tmp/test-bundle".into());

        let oci_spec = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .unwrap();

        // Check that resources are set
        let linux = oci_spec.linux().as_ref().unwrap();
        let resources = linux.resources().as_ref().unwrap();

        // Check CPU
        let cpu = resources.cpu().as_ref().unwrap();
        assert_eq!(cpu.quota(), Some(50_000)); // 0.5 cores * 100000
        assert_eq!(cpu.period(), Some(100_000));

        // Check memory
        let memory = resources.memory().as_ref().unwrap();
        assert_eq!(memory.limit(), Some(512 * 1024 * 1024)); // 512Mi
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_oci_spec_translates_ulimits() {
        let id = ContainerId::new("test".to_string(), 1);
        let mut spec = mock_spec();
        spec.ulimits.insert(
            "nofile".to_string(),
            UlimitSpec {
                soft: 100_000,
                hard: 200_000,
            },
        );
        // Negative limits must clamp to 0 (matches the `.max(0)` conversion).
        spec.ulimits
            .insert("nproc".to_string(), UlimitSpec { soft: -1, hard: -5 });
        let builder = BundleBuilder::new("/tmp/test-bundle".into());

        let oci_spec = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .unwrap();

        let process = oci_spec.process().as_ref().expect("process present");
        let rlimits = process.rlimits().as_ref().expect("rlimits present");

        // Exactly one nofile entry: our override fully replaces the oci
        // default (1024), it does not append a duplicate the kernel would
        // resolve ambiguously.
        let nofile: Vec<_> = rlimits
            .iter()
            .filter(|r| r.typ() == PosixRlimitType::RlimitNofile)
            .collect();
        assert_eq!(nofile.len(), 1, "nofile must not be duplicated");
        assert_eq!(nofile[0].soft(), 100_000);
        assert_eq!(nofile[0].hard(), 200_000);

        let nproc = rlimits
            .iter()
            .find(|r| r.typ() == PosixRlimitType::RlimitNproc)
            .expect("nproc rlimit present");
        assert_eq!(nproc.soft(), 0, "negative soft clamps to 0");
        assert_eq!(nproc.hard(), 0, "negative hard clamps to 0");
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_oci_spec_rejects_unknown_ulimit() {
        let id = ContainerId::new("test".to_string(), 1);
        let mut spec = mock_spec();
        spec.ulimits.insert(
            "not_a_real_ulimit".to_string(),
            UlimitSpec { soft: 1, hard: 1 },
        );
        let builder = BundleBuilder::new("/tmp/test-bundle".into());

        let err = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .expect_err("unknown ulimit name must be rejected");
        assert!(
            err.to_string().contains("not_a_real_ulimit"),
            "error should name the unknown ulimit: {err}"
        );
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_oci_spec_keeps_oci_default_rlimits_when_ulimits_empty() {
        // When `spec.ulimits` is empty we must NOT touch the process builder's
        // rlimits — the OCI default (`ProcessBuilder::default()` ships a single
        // `RLIMIT_NOFILE` of 1024, the kernel default). This documents the
        // exact baseline the ulimits override replaces, so a regression that
        // wipes the default (or, worse, our override) is caught here.
        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_spec();
        let builder = BundleBuilder::new("/tmp/test-bundle".into());

        let oci_spec = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .unwrap();

        let process = oci_spec.process().as_ref().expect("process present");
        let rlimits = process
            .rlimits()
            .as_ref()
            .expect("oci default rlimits present");
        let nofile = rlimits
            .iter()
            .find(|r| r.typ() == PosixRlimitType::RlimitNofile)
            .expect("default nofile rlimit present");
        // The oci-spec default the daemon would otherwise leak into the
        // container: 1024 — the exact value that EMFILE'd PlatformStore.
        assert_eq!(nofile.soft(), 1024);
        assert_eq!(nofile.hard(), 1024);
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_oci_spec_privileged() {
        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_privileged_spec();
        let builder = BundleBuilder::new("/tmp/test-bundle".into());

        let oci_spec = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .unwrap();

        // Check that all capabilities are set
        let process = oci_spec.process().as_ref().unwrap();
        let caps = process.capabilities().as_ref().unwrap();
        let bounding = caps.bounding().as_ref().unwrap();

        // Should have all capabilities
        assert!(bounding.contains(&Capability::SysAdmin));
        assert!(bounding.contains(&Capability::NetAdmin));

        // Check that masked paths are NOT set for privileged
        let linux = oci_spec.linux().as_ref().unwrap();
        assert!(
            linux.masked_paths().is_none() || linux.masked_paths().as_ref().unwrap().is_empty()
        );
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_oci_spec_environment() {
        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_spec_with_resources();
        let builder = BundleBuilder::new("/tmp/test-bundle".into())
            .with_env("EXTRA_VAR".to_string(), "extra_value".to_string());

        let oci_spec = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .unwrap();

        let process = oci_spec.process().as_ref().unwrap();
        let env = process.env().as_ref().unwrap();

        // Check service env vars are present
        assert!(env.iter().any(|e| e == "MY_VAR=my_value"));
        assert!(env.iter().any(|e| e == "ANOTHER=value2"));
        // Check extra env var is present
        assert!(env.iter().any(|e| e == "EXTRA_VAR=extra_value"));
        // Check PATH is present
        assert!(env.iter().any(|e| e.starts_with("PATH=")));
    }

    /// Regression for the production bug where a container env var
    /// `KEY=$S:NAME` reached the container as the LITERAL string `$S:NAME`
    /// instead of the resolved secret value, because the bundle build never
    /// resolved `$S:` against the per-service / per-environment secret scope.
    ///
    /// The fix wired `secrets_provider` + `deployment_scope` into the bundle
    /// build (bundle.rs ~1152), so that `$S:` values are resolved via
    /// `crate::env::resolve_env_with_secrets(&spec.env, provider,
    /// &scope.to_storage_scope())`. This test exercises that exact gate
    /// through `build_spec_only` (the same testable seam the other
    /// `build_oci_spec` tests use — no real rootfs required) and asserts the
    /// resolved value lands in the OCI process env, NOT the literal `$S:...`.
    ///
    /// The fake provider only returns the secret for the CORRECT storage scope
    /// (`project:zatabase:env:env-uuid`). Before the fix the env was passed
    /// through unresolved (no provider/scope wired), so `$S:ZATA_SECRETS_KEY_HEX`
    /// would survive verbatim; with the wrong scope the provider returns
    /// `NotFound` and the build errors — either way this test would FAIL.
    mod secret_scope_regression {
        use super::*;
        use async_trait::async_trait;
        use zlayer_secrets::{Secret, SecretMetadata, SecretsError, SecretsProvider};

        /// Fake provider that yields a known value for EXACTLY one (scope, name)
        /// pair and [`NotFound`](SecretsError::NotFound) for anything else. This makes the test prove the
        /// CORRECT storage-scope string is threaded through the bundle gate: a
        /// missing or wrong scope can never accidentally succeed.
        struct ScopePinnedProvider {
            expect_scope: String,
            expect_name: String,
            value: String,
        }

        #[async_trait]
        impl SecretsProvider for ScopePinnedProvider {
            async fn get_secret(&self, scope: &str, name: &str) -> zlayer_secrets::Result<Secret> {
                if scope == self.expect_scope && name == self.expect_name {
                    Ok(Secret::new(&self.value))
                } else {
                    Err(SecretsError::NotFound {
                        name: name.to_string(),
                    })
                }
            }

            async fn get_secrets(
                &self,
                scope: &str,
                names: &[&str],
            ) -> zlayer_secrets::Result<HashMap<String, Secret>> {
                let mut out = HashMap::new();
                for name in names {
                    if let Ok(secret) = self.get_secret(scope, name).await {
                        out.insert((*name).to_string(), secret);
                    }
                }
                Ok(out)
            }

            async fn list_secrets(
                &self,
                _scope: &str,
            ) -> zlayer_secrets::Result<Vec<SecretMetadata>> {
                Ok(Vec::new())
            }

            async fn exists(&self, scope: &str, name: &str) -> zlayer_secrets::Result<bool> {
                Ok(self.get_secret(scope, name).await.is_ok())
            }
        }

        /// A spec whose env contains the exact production shape: a `$S:` ref.
        fn spec_with_secret_env() -> ServiceSpec {
            serde_yaml::from_str::<DeploymentSpec>(
                r"
version: v1
deployment: zatabase
services:
  test:
    rtype: service
    image:
      name: test:latest
    env:
      ZATA_SECRETS_KEY_HEX: $S:ZATA_SECRETS_KEY_HEX
    endpoints:
      - name: http
        protocol: http
        port: 8080
",
            )
            .unwrap()
            .services
            .remove("test")
            .unwrap()
        }

        /// Pins the scope grammar the bundle wiring depends on. If
        /// `to_storage_scope()` ever drifts, the whole resolution path silently
        /// breaks — so assert the exact string the provider is keyed on.
        #[test]
        fn for_env_storage_scope_grammar_is_stable() {
            assert_eq!(
                SecretScope::for_env(Some("zatabase"), "env-uuid").to_storage_scope(),
                "project:zatabase:env:env-uuid"
            );
            // The wrong/missing-project form yields a DIFFERENT scope, which is
            // why passing it (the old behavior) fails to find the secret.
            assert_eq!(
                SecretScope::for_env(None, "env-uuid").to_storage_scope(),
                "env:env-uuid"
            );
        }

        /// POSITIVE: with the provider + correct scope wired into the builder,
        /// `$S:ZATA_SECRETS_KEY_HEX` resolves to the real value in the OCI env.
        #[tokio::test]
        async fn bundle_gate_resolves_secret_with_correct_scope() {
            let provider = Arc::new(ScopePinnedProvider {
                expect_scope: "project:zatabase:env:env-uuid".to_string(),
                expect_name: "ZATA_SECRETS_KEY_HEX".to_string(),
                value: "deadbeef".to_string(),
            });

            let id = ContainerId::new("test".to_string(), 1);
            let spec = spec_with_secret_env();
            let builder = BundleBuilder::new("/tmp/test-bundle-secret-ok".into())
                .with_secrets_provider(provider)
                .with_deployment_scope(SecretScope::for_env(Some("zatabase"), "env-uuid"));

            let oci_spec = builder
                .build_spec_only(&id, &spec, &std::collections::HashMap::new())
                .await
                .expect("build should succeed when the secret resolves");

            let env = oci_spec
                .process()
                .as_ref()
                .unwrap()
                .env()
                .as_ref()
                .unwrap()
                .clone();

            // The resolved secret value must be present...
            assert!(
                env.iter().any(|e| e == "ZATA_SECRETS_KEY_HEX=deadbeef"),
                "expected resolved secret in OCI env, got: {env:?}"
            );
            // ...and the LITERAL unresolved ref must NOT survive (the bug).
            assert!(
                !env.iter()
                    .any(|e| e == "ZATA_SECRETS_KEY_HEX=$S:ZATA_SECRETS_KEY_HEX"),
                "literal $S: ref leaked into container env (the original bug): {env:?}"
            );
        }

        /// NEGATIVE: the same provider, but the builder is given the WRONG scope
        /// (`env:env-uuid`, i.e. project dropped — the old broken behavior).
        /// The provider returns `NotFound`, so the bundle build must error rather
        /// than silently shipping the literal `$S:` ref. This proves the test is
        /// genuinely exercising scope-wiring: a wrong scope cannot succeed.
        #[tokio::test]
        async fn bundle_gate_wrong_scope_fails_to_resolve() {
            let provider = Arc::new(ScopePinnedProvider {
                expect_scope: "project:zatabase:env:env-uuid".to_string(),
                expect_name: "ZATA_SECRETS_KEY_HEX".to_string(),
                value: "deadbeef".to_string(),
            });

            let id = ContainerId::new("test".to_string(), 1);
            let spec = spec_with_secret_env();
            // Wrong scope: env:env-uuid (project dropped) != project:zatabase:env:env-uuid
            let builder = BundleBuilder::new("/tmp/test-bundle-secret-wrong".into())
                .with_secrets_provider(provider)
                .with_deployment_scope(SecretScope::for_env(None, "env-uuid"));

            let result = builder
                .build_spec_only(&id, &spec, &std::collections::HashMap::new())
                .await;

            assert!(
                result.is_err(),
                "wrong scope must fail secret resolution, not silently pass the literal ref"
            );
        }
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_namespaces() {
        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_spec();
        let builder = BundleBuilder::new("/tmp/test-bundle".into());

        let oci_spec = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .unwrap();
        let linux = oci_spec.linux().as_ref().unwrap();
        let namespaces = linux.namespaces().as_ref().unwrap();

        // Check we have the expected namespaces
        let namespace_types: Vec<_> = namespaces
            .iter()
            .map(oci_spec::runtime::LinuxNamespace::typ)
            .collect();
        assert!(namespace_types.contains(&LinuxNamespaceType::Pid));
        assert!(namespace_types.contains(&LinuxNamespaceType::Ipc));
        assert!(namespace_types.contains(&LinuxNamespaceType::Uts));
        assert!(namespace_types.contains(&LinuxNamespaceType::Mount));
        assert!(namespace_types.contains(&LinuxNamespaceType::Network));
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_namespaces_host_network() {
        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_spec();
        let builder = BundleBuilder::new("/tmp/test-bundle".into()).with_host_network(true);

        let oci_spec = builder
            .build_spec_only(&id, &spec, &std::collections::HashMap::new())
            .await
            .unwrap();
        let linux = oci_spec.linux().as_ref().unwrap();
        let namespaces = linux.namespaces().as_ref().unwrap();

        // Check we have the expected namespaces (NO Network namespace)
        let namespace_types: Vec<_> = namespaces
            .iter()
            .map(oci_spec::runtime::LinuxNamespace::typ)
            .collect();
        assert!(namespace_types.contains(&LinuxNamespaceType::Pid));
        assert!(namespace_types.contains(&LinuxNamespaceType::Ipc));
        assert!(namespace_types.contains(&LinuxNamespaceType::Uts));
        assert!(namespace_types.contains(&LinuxNamespaceType::Mount));
        assert!(
            !namespace_types.contains(&LinuxNamespaceType::Network),
            "Network namespace should NOT be present in host_network mode"
        );
    }

    /// The three network-namespace cases must be honoured exactly:
    ///   * `host_network`          → no Network namespace at all.
    ///   * `netns_path` = `Some(p)` → Network namespace WITH that path (JOIN).
    ///   * neither                  → Network namespace WITHOUT a path (fresh).
    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_build_namespaces_netns_path_cases() {
        use std::path::PathBuf;

        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_spec();
        let empty = std::collections::HashMap::new();

        let net_ns = |oci: &oci_spec::runtime::Spec| -> Option<Option<PathBuf>> {
            oci.linux()
                .as_ref()
                .and_then(|l| l.namespaces().as_ref())
                .and_then(|ns| {
                    ns.iter()
                        .find(|n| n.typ() == LinuxNamespaceType::Network)
                        .map(|n| n.path().clone())
                })
        };

        // host_network → NO Network namespace.
        let host = BundleBuilder::new("/tmp/test-bundle".into())
            .with_host_network(true)
            .with_netns_path(Some(PathBuf::from("/proc/1234/ns/net")))
            .build_spec_only(&id, &spec, &empty)
            .await
            .unwrap();
        assert!(
            net_ns(&host).is_none(),
            "host_network must emit no Network namespace even when netns_path is set"
        );

        // netns_path set → Network namespace WITH that exact path (JOIN).
        let join_path = PathBuf::from("/proc/4242/ns/net");
        let joined = BundleBuilder::new("/tmp/test-bundle".into())
            .with_netns_path(Some(join_path.clone()))
            .build_spec_only(&id, &spec, &empty)
            .await
            .unwrap();
        assert_eq!(
            net_ns(&joined),
            Some(Some(join_path)),
            "netns_path must produce a Network namespace carrying that path"
        );

        // neither → Network namespace WITHOUT a path (fresh unshare).
        let fresh = BundleBuilder::new("/tmp/test-bundle".into())
            .build_spec_only(&id, &spec, &empty)
            .await
            .unwrap();
        assert_eq!(
            net_ns(&fresh),
            Some(None),
            "default mode must produce a Network namespace with no path"
        );
    }

    #[test]
    fn test_build_default_mounts() {
        let spec = mock_spec();
        let builder = BundleBuilder::new("/tmp/test-bundle".into());

        let mounts = builder.build_default_mounts(&spec).unwrap();

        // Check we have the expected mounts
        let mount_destinations: Vec<_> = mounts
            .iter()
            .map(|m| m.destination().to_string_lossy().to_string())
            .collect();
        assert!(mount_destinations.contains(&"/proc".to_string()));
        assert!(mount_destinations.contains(&"/dev".to_string()));
        assert!(mount_destinations.contains(&"/dev/pts".to_string()));
        assert!(mount_destinations.contains(&"/dev/shm".to_string()));
        assert!(mount_destinations.contains(&"/sys".to_string()));
    }

    #[test]
    fn test_build_storage_mounts_bind() {
        let spec = serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(
            r"
version: v1
deployment: test
services:
  test:
    image:
      name: test:latest
    storage:
      - type: bind
        source: /host/data
        target: /app/data
        readonly: true
",
        )
        .unwrap()
        .services
        .remove("test")
        .unwrap();

        let builder = BundleBuilder::new("/tmp/test-bundle".into());
        let volume_paths = std::collections::HashMap::new();

        let mounts = builder.build_storage_mounts(&spec, &volume_paths).unwrap();

        assert_eq!(mounts.len(), 1);
        assert_eq!(mounts[0].destination().to_string_lossy(), "/app/data");
        assert_eq!(
            mounts[0]
                .source()
                .as_ref()
                .map(|s| s.to_string_lossy().to_string()),
            Some("/host/data".to_string())
        );
        let options = mounts[0].options().as_ref().unwrap();
        assert!(options.contains(&"rbind".to_string()));
        assert!(options.contains(&"ro".to_string()));
    }

    #[test]
    fn test_build_storage_mounts_named() {
        let spec = serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(
            r"
version: v1
deployment: test
services:
  test:
    image:
      name: test:latest
    storage:
      - type: named
        name: my-volume
        target: /app/data
",
        )
        .unwrap()
        .services
        .remove("test")
        .unwrap();

        let dirs = zlayer_paths::ZLayerDirs::system_default();
        let builder = BundleBuilder::new("/tmp/test-bundle".into());
        let mut volume_paths = std::collections::HashMap::new();
        volume_paths.insert("my-volume".to_string(), dirs.volumes().join("my-volume"));

        let mounts = builder.build_storage_mounts(&spec, &volume_paths).unwrap();

        assert_eq!(mounts.len(), 1);
        assert_eq!(mounts[0].destination().to_string_lossy(), "/app/data");
        assert_eq!(
            mounts[0]
                .source()
                .as_ref()
                .map(|s| s.to_string_lossy().to_string()),
            Some(
                dirs.volumes()
                    .join("my-volume")
                    .to_string_lossy()
                    .into_owned()
            )
        );
    }

    #[test]
    fn test_build_storage_mounts_tmpfs() {
        let spec = serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(
            r"
version: v1
deployment: test
services:
  test:
    image:
      name: test:latest
    storage:
      - type: tmpfs
        target: /app/tmp
        size: 256Mi
        mode: 1777
",
        )
        .unwrap()
        .services
        .remove("test")
        .unwrap();

        let builder = BundleBuilder::new("/tmp/test-bundle".into());
        let volume_paths = std::collections::HashMap::new();

        let mounts = builder.build_storage_mounts(&spec, &volume_paths).unwrap();

        assert_eq!(mounts.len(), 1);
        assert_eq!(mounts[0].destination().to_string_lossy(), "/app/tmp");
        assert_eq!(mounts[0].typ().as_ref().map(String::as_str), Some("tmpfs"));
        let options = mounts[0].options().as_ref().unwrap();
        assert!(options.iter().any(|o| o.starts_with("size=")));
        assert!(options.iter().any(|o| o.starts_with("mode=")));
    }

    #[test]
    fn test_build_storage_mounts_multiple() {
        let spec = serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(
            r"
version: v1
deployment: test
services:
  test:
    image:
      name: test:latest
    storage:
      - type: bind
        source: /etc/config
        target: /app/config
        readonly: true
      - type: named
        name: app-data
        target: /app/data
      - type: tmpfs
        target: /app/tmp
",
        )
        .unwrap()
        .services
        .remove("test")
        .unwrap();

        let dirs = zlayer_paths::ZLayerDirs::system_default();
        let builder = BundleBuilder::new("/tmp/test-bundle".into());
        let mut volume_paths = std::collections::HashMap::new();
        volume_paths.insert("app-data".to_string(), dirs.volumes().join("app-data"));

        let mounts = builder.build_storage_mounts(&spec, &volume_paths).unwrap();

        assert_eq!(mounts.len(), 3);

        // Verify each mount is correct type
        let destinations: Vec<String> = mounts
            .iter()
            .map(|m| m.destination().to_string_lossy().to_string())
            .collect();
        assert!(destinations.contains(&"/app/config".to_string()));
        assert!(destinations.contains(&"/app/data".to_string()));
        assert!(destinations.contains(&"/app/tmp".to_string()));
    }

    #[test]
    fn test_build_storage_mounts_anonymous_missing_path() {
        let spec = serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(
            r"
version: v1
deployment: test
services:
  test:
    image:
      name: test:latest
    storage:
      - type: anonymous
        target: /app/cache
",
        )
        .unwrap()
        .services
        .remove("test")
        .unwrap();

        let builder = BundleBuilder::new("/tmp/test-bundle".into());
        let volume_paths = std::collections::HashMap::new(); // No path provided

        let result = builder.build_storage_mounts(&spec, &volume_paths);

        // Should fail because anonymous volume path not prepared
        assert!(result.is_err());
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn test_oci_spec_includes_storage_mounts() {
        let id = ContainerId::new("test".to_string(), 1);
        // The full spec-build path validates bind-mount sources up front
        // (AgentError::MountSourceMissing), so the host source must exist on
        // disk. Use a real temp dir as the source to exercise the happy path.
        let src_dir = tempfile::tempdir().unwrap();
        let src_path = src_dir.path().to_string_lossy().into_owned();
        let spec = serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(&format!(
            r"
version: v1
deployment: test
services:
  test:
    image:
      name: test:latest
    storage:
      - type: bind
        source: {src_path}
        target: /app/data
      - type: tmpfs
        target: /app/tmp
"
        ))
        .unwrap()
        .services
        .remove("test")
        .unwrap();

        let builder = BundleBuilder::new("/tmp/test-bundle".into());
        let volume_paths = std::collections::HashMap::new();

        let oci_spec = builder
            .build_spec_only(&id, &spec, &volume_paths)
            .await
            .unwrap();

        // Verify the OCI spec includes storage mounts
        let mounts = oci_spec.mounts().as_ref().unwrap();
        let destinations: Vec<String> = mounts
            .iter()
            .map(|m| m.destination().to_string_lossy().to_string())
            .collect();

        // Should include both default mounts and storage mounts
        assert!(destinations.contains(&"/proc".to_string())); // default
        assert!(destinations.contains(&"/dev".to_string())); // default
        assert!(destinations.contains(&"/app/data".to_string())); // storage bind
        assert!(destinations.contains(&"/app/tmp".to_string())); // storage tmpfs
    }

    fn mock_gpu_spec(vendor: &str, count: u32) -> ServiceSpec {
        let yaml = format!(
            "
version: v1
deployment: test
services:
  test:
    rtype: service
    image:
      name: test:latest
    resources:
      gpu:
        count: {count}
        vendor: {vendor}
    endpoints:
      - name: http
        protocol: http
        port: 8080
"
        );
        serde_yaml::from_str::<DeploymentSpec>(&yaml)
            .unwrap()
            .services
            .remove("test")
            .unwrap()
    }

    fn write_nvidia_cdi_fixture(dir: &std::path::Path, json: &str) {
        std::fs::write(dir.join("nvidia.json"), json).unwrap();
    }

    fn nvidia_cdi_fixture() -> &'static str {
        r#"{
            "cdiVersion": "0.6.0",
            "kind": "nvidia.com/gpu",
            "devices": [{
                "name": "0",
                "containerEdits": {
                    "deviceNodes": [
                        {"path": "/dev/nvidia0", "type": "c", "major": 195, "minor": 0}
                    ],
                    "env": ["NVIDIA_VISIBLE_DEVICES=0"],
                    "hooks": {
                        "createContainer": [{
                            "path": "/usr/bin/nvidia-container-runtime-hook",
                            "args": ["nvidia-container-runtime-hook", "prestart"]
                        }]
                    }
                }
            }]
        }"#
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn gpu_spec_translates_to_cdi_device_nodes() {
        let dir = tempfile::tempdir().unwrap();
        write_nvidia_cdi_fixture(dir.path(), nvidia_cdi_fixture());
        let registry = std::sync::Arc::new(crate::cdi::CdiRegistry::discover_from(&[dir.path()]));

        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_gpu_spec("nvidia", 1);
        let builder = BundleBuilder::new("/tmp/test-bundle-cdi".into()).with_cdi_registry(registry);

        let oci_spec = builder
            .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
            .await
            .expect("build with CDI fixture");

        // CDI device node merged into linux.devices
        let linux = oci_spec.linux().as_ref().expect("linux config present");
        let devices = linux.devices().as_ref().expect("devices present");
        assert!(
            devices
                .iter()
                .any(|d| d.path() == std::path::Path::new("/dev/nvidia0")),
            "expected /dev/nvidia0 from CDI fixture; got {:?}",
            devices
                .iter()
                .map(oci_spec::runtime::LinuxDevice::path)
                .collect::<Vec<_>>()
        );

        // CDI env var merged into process.env
        let process = oci_spec.process().as_ref().expect("process present");
        let env = process.env().as_ref().expect("env present");
        assert!(
            env.iter().any(|e| e == "NVIDIA_VISIBLE_DEVICES=0"),
            "expected NVIDIA_VISIBLE_DEVICES=0 in env; got {env:?}"
        );

        // CDI hook merged into hooks.createContainer
        let hooks = oci_spec.hooks().as_ref().expect("hooks present");
        let create_container = hooks
            .create_container()
            .as_ref()
            .expect("createContainer hooks present");
        assert_eq!(create_container.len(), 1);
        assert_eq!(
            create_container[0].path(),
            &std::path::PathBuf::from("/usr/bin/nvidia-container-runtime-hook")
        );
    }

    #[tokio::test]
    async fn gpu_spec_with_missing_cdi_returns_error() {
        // Empty tempdir — no CDI specs installed at all.
        let dir = tempfile::tempdir().unwrap();
        let registry = std::sync::Arc::new(crate::cdi::CdiRegistry::discover_from(&[dir.path()]));

        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_gpu_spec("nvidia", 1);
        let builder =
            BundleBuilder::new("/tmp/test-bundle-cdi-missing".into()).with_cdi_registry(registry);

        let err = builder
            .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
            .await
            .expect_err("should fail when CDI registry is empty");

        match err {
            AgentError::InvalidSpec(msg) => {
                assert!(
                    msg.contains("nvidia") || msg.contains("CDI"),
                    "error should mention CDI / vendor; got: {msg}"
                );
            }
            other => panic!("expected InvalidSpec, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn gpu_spec_with_unknown_device_returns_error() {
        // Spec has device "0" but the request will ask for two GPUs (so the
        // resolver will look for "1" and fail).
        let dir = tempfile::tempdir().unwrap();
        write_nvidia_cdi_fixture(dir.path(), nvidia_cdi_fixture());
        let registry = std::sync::Arc::new(crate::cdi::CdiRegistry::discover_from(&[dir.path()]));

        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_gpu_spec("nvidia", 2);
        let builder =
            BundleBuilder::new("/tmp/test-bundle-cdi-unknown".into()).with_cdi_registry(registry);

        let err = builder
            .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
            .await
            .expect_err("should fail when device '1' is not declared");
        match err {
            AgentError::InvalidSpec(msg) => {
                assert!(
                    msg.contains("'1'") || msg.contains("device"),
                    "error should mention the missing device; got: {msg}"
                );
            }
            other => panic!("expected InvalidSpec, got {other:?}"),
        }
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn gpu_spec_with_all_devices_expands_to_all_in_spec() {
        // Fixture with two declared devices ("0" and "1").
        let dir = tempfile::tempdir().unwrap();
        let fixture = r#"{
            "cdiVersion": "0.6.0",
            "kind": "nvidia.com/gpu",
            "devices": [
                {
                    "name": "0",
                    "containerEdits": {
                        "env": ["NVIDIA_VISIBLE_DEVICES=0"],
                        "deviceNodes": [
                            {"path": "/dev/nvidia0", "type": "c", "major": 195, "minor": 0}
                        ]
                    }
                },
                {
                    "name": "1",
                    "containerEdits": {
                        "env": ["NVIDIA_VISIBLE_DEVICES=1"],
                        "deviceNodes": [
                            {"path": "/dev/nvidia1", "type": "c", "major": 195, "minor": 1}
                        ]
                    }
                }
            ]
        }"#;
        write_nvidia_cdi_fixture(dir.path(), fixture);
        let registry = std::sync::Arc::new(crate::cdi::CdiRegistry::discover_from(&[dir.path()]));

        // Resolve "all" via the registry directly to validate expansion
        // semantics independently of how we map count -> names.
        let edits = registry
            .resolve_for_kind("nvidia.com/gpu", &["all".to_string()])
            .expect("resolve all");
        assert_eq!(edits.len(), 2);

        // Now build the bundle for a 2-GPU service and confirm both nodes
        // land in linux.devices.
        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_gpu_spec("nvidia", 2);
        let builder =
            BundleBuilder::new("/tmp/test-bundle-cdi-all".into()).with_cdi_registry(registry);

        let oci_spec = builder
            .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
            .await
            .expect("build with 2-device fixture");

        let devices = oci_spec
            .linux()
            .as_ref()
            .unwrap()
            .devices()
            .as_ref()
            .expect("devices present");
        let paths: Vec<_> = devices.iter().map(|d| d.path().clone()).collect();
        assert!(paths.contains(&std::path::PathBuf::from("/dev/nvidia0")));
        assert!(paths.contains(&std::path::PathBuf::from("/dev/nvidia1")));
    }

    /// Build the standard fixture-backed CDI registry used by the MPS /
    /// time-slicing tests. Identical to the helper used by the 5.A CDI
    /// tests above but expressed as a closure-style helper to keep each test
    /// self-contained.
    fn build_nvidia_cdi_registry(dir: &std::path::Path) -> std::sync::Arc<crate::cdi::CdiRegistry> {
        write_nvidia_cdi_fixture(dir, nvidia_cdi_fixture());
        std::sync::Arc::new(crate::cdi::CdiRegistry::discover_from(&[dir]))
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn gpu_spec_with_mps_sharing_injects_env_and_mounts() {
        // Stage host-side MPS directories in a tempdir so the resolver's
        // `is_dir()` check passes without touching /tmp/nvidia-mps on the
        // real host.
        let cdi_dir = tempfile::tempdir().unwrap();
        let mps_root = tempfile::tempdir().unwrap();
        let pipe_dir = mps_root.path().join("nvidia-mps");
        let log_dir = mps_root.path().join("nvidia-log");
        std::fs::create_dir(&pipe_dir).unwrap();
        std::fs::create_dir(&log_dir).unwrap();
        let registry = build_nvidia_cdi_registry(cdi_dir.path());

        let id = ContainerId::new("test".to_string(), 1);
        let mut spec = mock_gpu_spec("nvidia", 1);
        let gpu = spec.resources.gpu.as_mut().expect("gpu spec set");
        gpu.sharing = Some(zlayer_spec::GpuSharingMode::Mps);
        gpu.mps_pipe_dir = Some(pipe_dir.to_string_lossy().into_owned());
        gpu.mps_log_dir = Some(log_dir.to_string_lossy().into_owned());

        let builder =
            BundleBuilder::new("/tmp/test-bundle-mps-env".into()).with_cdi_registry(registry);
        let oci_spec = builder
            .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
            .await
            .expect("build with MPS sharing");

        let env = oci_spec
            .process()
            .as_ref()
            .and_then(|p| p.env().as_ref())
            .expect("env present");
        let pipe_expect = format!("CUDA_MPS_PIPE_DIRECTORY={}", pipe_dir.display());
        let log_expect = format!("CUDA_MPS_LOG_DIRECTORY={}", log_dir.display());
        assert!(
            env.iter().any(|e| e == &pipe_expect),
            "expected {pipe_expect} in env; got {env:?}"
        );
        assert!(
            env.iter().any(|e| e == &log_expect),
            "expected {log_expect} in env; got {env:?}"
        );

        let mounts = oci_spec.mounts().as_ref().expect("mounts present");
        assert!(
            mounts
                .iter()
                .any(|m| m.destination() == &pipe_dir && m.source().as_ref() == Some(&pipe_dir)),
            "expected bind mount of MPS pipe dir {}; got destinations {:?}",
            pipe_dir.display(),
            mounts.iter().map(Mount::destination).collect::<Vec<_>>()
        );
        assert!(
            mounts
                .iter()
                .any(|m| m.destination() == &log_dir && m.source().as_ref() == Some(&log_dir)),
            "expected bind mount of MPS log dir {}",
            log_dir.display()
        );
    }

    #[tokio::test]
    async fn gpu_spec_with_mps_sharing_fails_when_pipe_dir_missing() {
        let cdi_dir = tempfile::tempdir().unwrap();
        let registry = build_nvidia_cdi_registry(cdi_dir.path());

        let id = ContainerId::new("test".to_string(), 1);
        let mut spec = mock_gpu_spec("nvidia", 1);
        let gpu = spec.resources.gpu.as_mut().expect("gpu spec set");
        gpu.sharing = Some(zlayer_spec::GpuSharingMode::Mps);
        // Path that demonstrably does not exist — tempdir() returns a unique
        // path so appending "definitely-not-here" gives a guaranteed miss.
        let missing = tempfile::tempdir().unwrap();
        let missing_path = missing.path().join("definitely-not-here");
        gpu.mps_pipe_dir = Some(missing_path.to_string_lossy().into_owned());

        let builder =
            BundleBuilder::new("/tmp/test-bundle-mps-missing".into()).with_cdi_registry(registry);
        let err = builder
            .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
            .await
            .expect_err("should fail when MPS pipe dir is missing");
        match err {
            AgentError::GpuSharingUnavailable { mode, reason } => {
                assert_eq!(mode, "mps");
                assert!(
                    reason.contains("pipe") || reason.contains(&missing_path.display().to_string()),
                    "reason should mention the missing path; got: {reason}"
                );
            }
            other => panic!("expected GpuSharingUnavailable, got {other:?}"),
        }
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn gpu_spec_with_timeslicing_injects_visible_devices() {
        let cdi_dir = tempfile::tempdir().unwrap();
        let registry = build_nvidia_cdi_registry(cdi_dir.path());

        let id = ContainerId::new("test".to_string(), 1);
        let mut spec = mock_gpu_spec("nvidia", 1);
        let gpu = spec.resources.gpu.as_mut().expect("gpu spec set");
        gpu.sharing = Some(zlayer_spec::GpuSharingMode::TimeSlice);
        gpu.time_slice_index = Some(2);

        let builder =
            BundleBuilder::new("/tmp/test-bundle-timeslice".into()).with_cdi_registry(registry);
        let oci_spec = builder
            .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
            .await
            .expect("build with time-slicing");

        let env = oci_spec
            .process()
            .as_ref()
            .and_then(|p| p.env().as_ref())
            .expect("env present");
        // Time-slicing must clobber any earlier `CUDA_VISIBLE_DEVICES` (e.g.
        // the CDI-emitted full-device list) to advertise exactly the slice.
        let cuda_entries: Vec<&String> = env
            .iter()
            .filter(|e| e.starts_with("CUDA_VISIBLE_DEVICES="))
            .collect();
        assert_eq!(
            cuda_entries.len(),
            1,
            "exactly one CUDA_VISIBLE_DEVICES expected; got {cuda_entries:?}"
        );
        assert_eq!(cuda_entries[0], "CUDA_VISIBLE_DEVICES=2");
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn gpu_spec_without_sharing_omits_mps_env() {
        let cdi_dir = tempfile::tempdir().unwrap();
        let registry = build_nvidia_cdi_registry(cdi_dir.path());

        let id = ContainerId::new("test".to_string(), 1);
        let spec = mock_gpu_spec("nvidia", 1);
        assert!(spec.resources.gpu.as_ref().unwrap().sharing.is_none());

        let builder =
            BundleBuilder::new("/tmp/test-bundle-no-sharing".into()).with_cdi_registry(registry);
        let oci_spec = builder
            .build_oci_spec(&id, &spec, &std::collections::HashMap::new())
            .await
            .expect("build without sharing");

        let env = oci_spec
            .process()
            .as_ref()
            .and_then(|p| p.env().as_ref())
            .expect("env present");
        assert!(
            !env.iter().any(|e| e.starts_with("CUDA_MPS_")),
            "no CUDA_MPS_* env should be present without sharing; got {env:?}"
        );

        // No MPS mount should be added either. The 5.A CDI fixture mounts a
        // /dev/nvidia0 device but never bind-mounts /tmp/nvidia-mps; verify
        // we don't sneak that in.
        let mounts = oci_spec.mounts().as_ref().expect("mounts present");
        assert!(
            !mounts
                .iter()
                .any(|m| { m.destination().to_string_lossy().contains("nvidia-mps") }),
            "no MPS pipe mount should be present without sharing"
        );
    }

    #[cfg(unix)]
    mod subid_tests {
        use super::super::{build_single_id_mapping, read_subid_range};
        use std::io::Write;

        #[test]
        fn build_single_id_mapping_yields_one_identity_mapping() {
            let mappings = build_single_id_mapping(0);
            assert_eq!(mappings.len(), 1);
            assert_eq!(mappings[0].container_id(), 0);
            assert_eq!(mappings[0].host_id(), 0);
            assert_eq!(mappings[0].size(), 1);
        }

        #[test]
        fn read_subid_range_returns_range_for_user() {
            let mut tmp = tempfile::NamedTempFile::new().unwrap();
            writeln!(tmp, "alice:100000:65536").unwrap();
            writeln!(tmp, "bob:165536:65536").unwrap();
            tmp.flush().unwrap();
            let path = tmp.path().to_str().unwrap();
            assert_eq!(read_subid_range(path, "bob"), Some((165_536, 65_536)));
            assert_eq!(read_subid_range(path, "alice"), Some((100_000, 65_536)));
        }

        #[test]
        fn read_subid_range_returns_none_for_unknown_user() {
            let mut tmp = tempfile::NamedTempFile::new().unwrap();
            writeln!(tmp, "alice:100000:65536").unwrap();
            tmp.flush().unwrap();
            assert_eq!(
                read_subid_range(tmp.path().to_str().unwrap(), "carol"),
                None
            );
        }

        #[test]
        fn read_subid_range_returns_none_on_missing_file() {
            assert_eq!(
                read_subid_range("/this/path/does/not/exist/subuid", "anyone"),
                None
            );
        }
    }

    /// Collect [`swarm_ring_env`] output into a lookup map for assertions.
    fn swarm_env_map(sharding: &ShardingSpec) -> std::collections::HashMap<String, String> {
        swarm_ring_env(sharding).into_iter().collect()
    }

    #[test]
    fn swarm_ring_env_middle_stage() {
        // Own band 12..24, peers svc-a (0..12, first) and svc-c (24..36, last),
        // coordinator svc-coord, 36 total layers.
        let sharding = ShardingSpec {
            swarm_id: "swarm-x".to_string(),
            layer_start: 12,
            layer_end: 24,
            layer_count: 36,
            role: SwarmRole::Stage,
            manifest_ref: None,
            peers: vec![
                SwarmPeer {
                    service: "svc-a".to_string(),
                    layer_start: 0,
                    layer_end: 12,
                },
                SwarmPeer {
                    service: "svc-c".to_string(),
                    layer_start: 24,
                    layer_end: 36,
                },
            ],
            coordinator: Some("svc-coord".to_string()),
        };
        let env = swarm_env_map(&sharding);
        assert_eq!(
            env.get("ZLAYER_SWARM_ID").map(String::as_str),
            Some("swarm-x")
        );
        assert_eq!(
            env.get("ZLAYER_SWARM_ROLE").map(String::as_str),
            Some("stage")
        );
        assert_eq!(
            env.get("ZLAYER_SWARM_LAYER_START").map(String::as_str),
            Some("12")
        );
        assert_eq!(
            env.get("ZLAYER_SWARM_LAYER_END").map(String::as_str),
            Some("24")
        );
        assert_eq!(
            env.get("ZLAYER_SWARM_TOTAL_LAYERS").map(String::as_str),
            Some("36")
        );
        assert_eq!(
            env.get("ZLAYER_SWARM_COORDINATOR").map(String::as_str),
            Some("svc-coord")
        );
        assert_eq!(
            env.get("ZLAYER_SWARM_NEXT_PEER").map(String::as_str),
            Some("svc-c")
        );
        assert_eq!(
            env.get("ZLAYER_SWARM_PREV_PEER").map(String::as_str),
            Some("svc-a")
        );
    }

    #[test]
    fn swarm_ring_env_first_stage_prev_is_coordinator() {
        // Own band 0..12 is the FIRST stage; prev wraps to the coordinator,
        // next is the following stage svc-b.
        let sharding = ShardingSpec {
            swarm_id: "swarm-y".to_string(),
            layer_start: 0,
            layer_end: 12,
            layer_count: 36,
            role: SwarmRole::Stage,
            manifest_ref: None,
            peers: vec![
                SwarmPeer {
                    service: "svc-b".to_string(),
                    layer_start: 12,
                    layer_end: 24,
                },
                SwarmPeer {
                    service: "svc-c".to_string(),
                    layer_start: 24,
                    layer_end: 36,
                },
            ],
            coordinator: Some("svc-coord".to_string()),
        };
        let env = swarm_env_map(&sharding);
        assert_eq!(
            env.get("ZLAYER_SWARM_ROLE").map(String::as_str),
            Some("stage")
        );
        assert_eq!(
            env.get("ZLAYER_SWARM_PREV_PEER").map(String::as_str),
            Some("svc-coord")
        );
        assert_eq!(
            env.get("ZLAYER_SWARM_NEXT_PEER").map(String::as_str),
            Some("svc-b")
        );
    }

    #[test]
    fn swarm_ring_env_last_stage_next_is_coordinator() {
        // Own band 24..36 is the LAST stage; next wraps to the coordinator,
        // prev is the preceding stage svc-b.
        let sharding = ShardingSpec {
            swarm_id: "swarm-z".to_string(),
            layer_start: 24,
            layer_end: 36,
            layer_count: 36,
            role: SwarmRole::Stage,
            manifest_ref: None,
            peers: vec![
                SwarmPeer {
                    service: "svc-a".to_string(),
                    layer_start: 0,
                    layer_end: 12,
                },
                SwarmPeer {
                    service: "svc-b".to_string(),
                    layer_start: 12,
                    layer_end: 24,
                },
            ],
            coordinator: Some("svc-coord".to_string()),
        };
        let env = swarm_env_map(&sharding);
        assert_eq!(
            env.get("ZLAYER_SWARM_NEXT_PEER").map(String::as_str),
            Some("svc-coord")
        );
        assert_eq!(
            env.get("ZLAYER_SWARM_PREV_PEER").map(String::as_str),
            Some("svc-b")
        );
    }

    #[test]
    fn swarm_ring_env_coordinator_role() {
        // Coordinator: next = first stage (lowest start), prev = last stage
        // (highest end). No stage layer-band vars are emitted.
        let sharding = ShardingSpec {
            swarm_id: "swarm-c".to_string(),
            layer_start: 0,
            layer_end: 0,
            layer_count: 36,
            role: SwarmRole::Coordinator,
            manifest_ref: None,
            peers: vec![
                SwarmPeer {
                    service: "svc-a".to_string(),
                    layer_start: 0,
                    layer_end: 12,
                },
                SwarmPeer {
                    service: "svc-b".to_string(),
                    layer_start: 12,
                    layer_end: 24,
                },
                SwarmPeer {
                    service: "svc-c".to_string(),
                    layer_start: 24,
                    layer_end: 36,
                },
            ],
            coordinator: None,
        };
        let env = swarm_env_map(&sharding);
        assert_eq!(
            env.get("ZLAYER_SWARM_ROLE").map(String::as_str),
            Some("coordinator")
        );
        assert_eq!(
            env.get("ZLAYER_SWARM_NEXT_PEER").map(String::as_str),
            Some("svc-a")
        );
        assert_eq!(
            env.get("ZLAYER_SWARM_PREV_PEER").map(String::as_str),
            Some("svc-c")
        );
        // Stages-only vars must be absent for the coordinator.
        assert!(!env.contains_key("ZLAYER_SWARM_LAYER_START"));
        assert!(!env.contains_key("ZLAYER_SWARM_TOTAL_LAYERS"));
    }

    #[test]
    fn swarm_ring_env_single_stage_no_peers_uses_coordinator() {
        // Single-stage swarm: no peers. Both neighbors point at the coordinator.
        let sharding = ShardingSpec {
            swarm_id: "solo".to_string(),
            layer_start: 0,
            layer_end: 36,
            layer_count: 36,
            role: SwarmRole::Stage,
            manifest_ref: None,
            peers: vec![],
            coordinator: Some("svc-coord".to_string()),
        };
        let env = swarm_env_map(&sharding);
        assert_eq!(
            env.get("ZLAYER_SWARM_NEXT_PEER").map(String::as_str),
            Some("svc-coord")
        );
        assert_eq!(
            env.get("ZLAYER_SWARM_PREV_PEER").map(String::as_str),
            Some("svc-coord")
        );
    }

    #[test]
    fn swarm_ring_env_single_stage_no_peers_no_coordinator_emits_no_neighbors() {
        // Single-stage swarm with no coordinator: emit neither NEXT nor PREV.
        let sharding = ShardingSpec {
            swarm_id: "solo2".to_string(),
            layer_start: 0,
            layer_end: 36,
            layer_count: 36,
            role: SwarmRole::Stage,
            manifest_ref: None,
            peers: vec![],
            coordinator: None,
        };
        let env = swarm_env_map(&sharding);
        assert!(!env.contains_key("ZLAYER_SWARM_NEXT_PEER"));
        assert!(!env.contains_key("ZLAYER_SWARM_PREV_PEER"));
        assert!(!env.contains_key("ZLAYER_SWARM_COORDINATOR"));
    }
}