zlayer-agent 0.14.0

//! WSL2 delegate runtime that executes Linux containers inside a configurable
//! WSL2 distro via shell-out to `zlayer runtime <verb>`.
//!
//! Used by [`super::composite::CompositeRuntime`] on Windows hosts to handle
//! Linux-image services alongside HCS-managed Windows-image services. There is
//! **no in-distro daemon**, **no HTTP server**, and **no new crate**: every
//! [`Runtime`] trait method maps to one or more `wsl.exe -d <distro> -- ...`
//! invocations, where `<distro>` and the in-distro `zlayer` binary path are
//! both resolved from [`Wsl2DelegateConfig`] at construction time rather than
//! hardcoded.
//!
//! # Scope
//!
//! Phase F-2 wired the dispatch path. Phase G-2 wired a real Windows-side
//! OCI bundle-write path. Phase G-3/G-4/G-5 adds:
//!
//! * **Real exec streaming.** [`Runtime::exec_stream`] now spawns `wsl.exe`
//!   with piped stdout/stderr and emits [`ExecEvent::Stdout`] /
//!   [`ExecEvent::Stderr`] line-by-line via `BufReader::lines()`, followed
//!   by a terminal [`ExecEvent::Exit`] once the child reaps.
//! * **Real youki log path.** `youki create` is invoked with
//!   `--log <log_root>/<slug>.youki.log`, and [`Runtime::container_logs`]
//!   tails the same path — so logs actually make it out of the distro
//!   instead of landing in a fabricated file youki never writes to.
//! * **Config-driven distro + runtime binary.** [`Wsl2DelegateConfig`] carries
//!   `distro`, `runtime_binary` (optional — defaults to
//!   `/usr/local/bin/zlayer`), `bundle_root`, `log_root`, and
//!   `oci_state_root`. [`Wsl2DelegateRuntime::try_new`] preserves the old
//!   `Ok(None)` "no WSL" contract; the explicit
//!   [`Wsl2DelegateRuntime::try_new_with_config`] surface returns hard
//!   errors for misconfigurations so operators catch typos early.
//!
//! WSL2 is a FIRST-CLASS overlay member via the guest-managed attach kind
//! (the same path the macOS VZ-Linux runtime uses):
//! [`Runtime::overlay_attach_kind`] reports
//! [`OverlayAttachKind::GuestManaged`], so the service layer allocates a guest
//! overlay identity from overlayd post-start and hands it to
//! [`Runtime::push_overlay_config`]. That method brings up a real kernel
//! `WireGuard` device INSIDE the distro and moves it into the container's
//! network namespace, so the container's overlay IP routes cross-node. The
//! container runs in a NAMED netns (`/run/netns/<slug>`) created at
//! `create_container` time and joined by youki (via `with_netns_path`), so
//! `push_overlay_config` can address it by name with `ip netns exec <slug>`.
//! [`Wsl2DelegateRuntime::teardown_container_netns`] deletes the `WireGuard`
//! device + named netns on `remove_container`. There is no silent
//! host-networking fallback: an overlay-required container that cannot be wired
//! up hard-errors so the service layer rolls the allocation back.
//!
//! # Error mapping
//!
//! Every shell-out failure maps to [`AgentError::Network`] with a message of
//! the form `zlayer runtime <subcommand> failed (status <code>): <stderr>` so the
//! user sees both the command that was run and the distro's stderr.
//! Configuration errors (missing zlayer binary, bad path override) surface as
//! [`AgentError::Configuration`] from `try_new_with_config` instead.

#![cfg(all(target_os = "windows", feature = "wsl"))]

use std::collections::HashMap;
use std::net::IpAddr;
use std::path::PathBuf;
use std::process::{Output, Stdio};
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use oci_spec::runtime::{
    LinuxDevice, LinuxDeviceBuilder, LinuxDeviceType, Mount, MountBuilder, Spec,
};
use serde::Deserialize;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::sync::mpsc;
use tokio::sync::RwLock;
use tokio_stream::wrappers::ReceiverStream;
use zlayer_observability::logs::{LogEntry, LogSource, LogStream};
use zlayer_registry::{CompressionType, ImageConfig};
use zlayer_spec::{GpuSpec, PullPolicy, RegistryAuth, ServiceSpec};

use crate::bundle::BundleBuilder;
use crate::cgroups_stats::ContainerStats;
use crate::error::{AgentError, Result};
use crate::overlay_manager::make_interface_name;
use crate::runtime::{
    validate_signal, ContainerId, ContainerInspectDetails, ContainerState, ExecEvent,
    ExecEventStream, ImageInfo, OverlayAttachKind, PruneResult, Runtime,
};

/// Default in-distro path under which per-container OCI bundles are rooted.
const DEFAULT_BUNDLE_ROOT: &str = "/var/lib/zlayer/bundles";

/// Default in-distro directory where per-container youki log files are written.
/// `create_container` passes `--log <dir>/<slug>.youki.log` to `youki create`,
/// and `container_logs` reads back from the same file.
const DEFAULT_LOG_ROOT: &str = "/var/lib/zlayer/logs";

/// Maximum time we'll poll [`Runtime::wait_container`] before giving up. Youki
/// exposes no blocking `wait` subcommand today, so we poll `state` and bail
/// after a day's worth of polling — plenty for batch jobs, cheap to rerun.
const WAIT_POLL_CAP: Duration = Duration::from_secs(24 * 60 * 60);

/// Polling interval used by [`Runtime::wait_container`] and
/// [`Runtime::stop_container`] while waiting for a container to stop.
const WAIT_POLL_INTERVAL: Duration = Duration::from_millis(500);

/// Default in-distro path to the `zlayer` binary that exposes the
/// runc-compatible `runtime <verb>` surface. Installed by
/// `zlayer_wsl::setup::install_binary` at distro provisioning time.
const DEFAULT_RUNTIME_BINARY: &str = "/usr/local/bin/zlayer";

/// Default in-distro state root for `zlayer runtime`. Matches the
/// `RuntimeGlobal::state_root` default in `bin/zlayer/src/cli.rs` so an
/// operator can poke at containers from a shell inside the distro without
/// having to pass `--state-root` explicitly.
const DEFAULT_OCI_STATE_ROOT: &str = "/var/lib/zlayer/oci/state";

/// Configuration for [`Wsl2DelegateRuntime`].
///
/// Lets callers override the distro name and the in-distro `zlayer` runtime
/// binary location so this delegate isn't locked to the hardcoded `zlayer`
/// distro or `/usr/local/bin/zlayer`. `try_new` validates each field by
/// shelling out to the distro before returning a runtime, so a misconfigured
/// `runtime_binary` or a missing distro fails fast with an actionable error.
#[derive(Clone, Debug)]
pub struct Wsl2DelegateConfig {
    /// Name of the WSL2 distro to dispatch container operations into.
    /// Defaults to [`zlayer_wsl::distro::DISTRO_NAME`] (`"zlayer"`).
    pub distro: String,
    /// Absolute in-distro path to the `zlayer` binary that exposes the
    /// `runtime <verb>` surface. When `None`, defaults to
    /// [`DEFAULT_RUNTIME_BINARY`] (`/usr/local/bin/zlayer`) — the location
    /// `zlayer_wsl::setup::install_binary` writes to. Explicit overrides are
    /// honoured verbatim and verified at `try_new` time by running
    /// `<binary> runtime --help` inside the distro.
    pub runtime_binary: Option<String>,
    /// In-distro directory where per-container OCI bundles are materialized
    /// under `<bundle_root>/<container-slug>/`. Defaults to
    /// [`DEFAULT_BUNDLE_ROOT`] (`/var/lib/zlayer/bundles`).
    pub bundle_root: String,
    /// In-distro directory where the per-container log files written by
    /// `zlayer runtime create --log <path>` live; each container gets
    /// `<log_root>/<slug>.youki.log`. Defaults to [`DEFAULT_LOG_ROOT`]
    /// (`/var/lib/zlayer/logs`).
    pub log_root: String,
    /// In-distro state root threaded into every `zlayer runtime --state-root <p>
    /// <verb>` invocation. Defaults to [`DEFAULT_OCI_STATE_ROOT`]
    /// (`/var/lib/zlayer/oci/state`) so it matches the
    /// `RuntimeGlobal::state_root` default in `bin/zlayer/src/cli.rs` and
    /// shells inside the distro can
    /// drop the flag.
    pub oci_state_root: PathBuf,
}

impl Default for Wsl2DelegateConfig {
    fn default() -> Self {
        Self {
            distro: zlayer_wsl::distro::configured_distro(),
            runtime_binary: Some(DEFAULT_RUNTIME_BINARY.to_string()),
            bundle_root: DEFAULT_BUNDLE_ROOT.to_string(),
            log_root: DEFAULT_LOG_ROOT.to_string(),
            oci_state_root: PathBuf::from(DEFAULT_OCI_STATE_ROOT),
        }
    }
}

/// Cached per-image data gathered on the Windows host and reused by
/// [`Wsl2DelegateRuntime::create_container`] to populate the rootfs in the
/// WSL2 distro.
#[derive(Clone, Debug)]
struct CachedImage {
    /// On-disk paths to the (still-compressed) layer files plus their OCI media
    /// types, in application order (base first). The layers are streamed to host
    /// disk at pull time (never fully buffered in RAM) and decompressed
    /// just-in-time, streaming, into the distro at `create_container`.
    layers: Vec<(PathBuf, String)>,
    /// Image configuration (entrypoint/cmd/env/workdir/user) so we can feed
    /// it to [`BundleBuilder::with_image_config`] when rendering `config.json`.
    config: ImageConfig,
}

/// Host directory holding a pulled image's streamed layer files, keyed by a
/// sanitized image reference. Lives under the OS temp dir; cleaned on re-pull
/// and on `remove_image`.
fn wsl2_layer_stage_dir(image: &str) -> PathBuf {
    let safe = image.replace([':', '/', '@'], "_");
    std::env::temp_dir().join("zlayer-wsl2-layers").join(safe)
}

/// Per-container netns lifecycle state, tracked so `remove_container` knows
/// whether to tear the named netns + `WireGuard` device down.
///
/// Populated with [`NetnsState::Created`] by [`Runtime::create_container`] when
/// it provisions the named netns the container joins, upgraded to
/// [`NetnsState::Configured`] by [`Runtime::push_overlay_config`] once the
/// guest `WireGuard` device is live, and consulted/cleared by
/// [`Wsl2DelegateRuntime::teardown_container_netns`] at remove time. Entries
/// only exist for containers that got a named netns (the overlay-capable
/// `NetworkMode::Default` / `Bridge` cases); host-network, network-none, and
/// container-joined containers never appear here.
#[derive(Debug, Clone)]
enum NetnsState {
    /// The named netns `/run/netns/<slug>` was created at create time and the
    /// container joins it; no overlay `WireGuard` device configured yet.
    Created,
    /// [`Runtime::push_overlay_config`] brought up the guest `WireGuard` device
    /// inside the netns and stamped the overlay IP. `ip` is retained for
    /// tracing / debugging introspection.
    Configured {
        #[allow(dead_code)]
        ip: IpAddr,
    },
}

/// Hook for injecting a custom `wsl.exe` runner in unit tests. Production
/// uses [`zlayer_wsl::distro::wsl_exec`] via the default trait method.
///
/// Kept as a trait object behind `Arc` so [`Wsl2DelegateRuntime`] stays
/// `Send + Sync + 'static`. All method bodies in tests push the `(cmd, args)`
/// pair into a log and return a canned `Output`.
#[async_trait]
pub trait WslRunner: Send + Sync + 'static {
    /// Run `cmd args...` inside the WSL2 helper distro and return the
    /// buffered `Output`. Implementations that fail to invoke `wsl.exe`
    /// return an error — the caller translates to [`AgentError::Network`].
    async fn run(&self, cmd: &str, args: &[&str]) -> anyhow::Result<Output>;
}

/// Default runner: hands off directly to [`zlayer_wsl::distro::wsl_exec`].
struct DefaultWslRunner;

#[async_trait]
impl WslRunner for DefaultWslRunner {
    async fn run(&self, cmd: &str, args: &[&str]) -> anyhow::Result<Output> {
        zlayer_wsl::distro::wsl_exec(cmd, args).await
    }
}

/// `Runtime` implementation that shells out to `zlayer runtime <verb>` inside
/// the `zlayer` WSL2 distro.
///
/// Construct via [`Wsl2DelegateRuntime::try_new`], which returns `Ok(None)`
/// when WSL2 (or the helper distro) is not available — callers should treat
/// that as "no Linux-container support on this node" rather than an error.
pub struct Wsl2DelegateRuntime {
    /// Resolved runtime configuration: distro name, in-distro `zlayer` binary
    /// path, bundle root, log root, OCI state root. Fully populated by
    /// [`Wsl2DelegateRuntime::try_new`] — in particular `runtime_binary` is
    /// the final absolute path (defaulted to [`DEFAULT_RUNTIME_BINARY`] when
    /// the caller left it `None`).
    config: ResolvedConfig,
    /// Per-container cache of the Linux PID youki reports after `start`.
    /// Populated lazily; used by [`Runtime::get_container_pid`].
    pids: Arc<RwLock<HashMap<ContainerId, u32>>>,
    /// Per-container cache of the assigned overlay IP, set by an external
    /// caller (the `OverlayManager` / agent's create flow) via
    /// [`Wsl2DelegateRuntime::record_container_ip`]. Returned by
    /// [`Runtime::get_container_ip`] when the netns setup succeeds.
    ips: Arc<RwLock<HashMap<ContainerId, IpAddr>>>,
    /// Per-container record of the in-distro bundle directory that was
    /// materialized by `create_container`. Used by `remove_container` to
    /// clean up after a container goes away, and kept alive so
    /// `start_container` doesn't have to re-derive the path.
    bundle_roots: Arc<RwLock<HashMap<ContainerId, String>>>,
    /// Per-image cache of pulled layer blobs + config, populated by
    /// [`Wsl2DelegateRuntime::pull_image_with_policy`] and consumed by
    /// [`Wsl2DelegateRuntime::create_container`] when it streams the rootfs
    /// into the WSL2 distro. `ServiceSpec::image.name` is the cache key.
    image_cache: Arc<RwLock<HashMap<String, CachedImage>>>,
    /// Per-container netns lifecycle state. Populated by
    /// [`Runtime::create_container`] (named netns provisioned) and
    /// [`Runtime::push_overlay_config`] (overlay device configured), and
    /// cleared by [`Wsl2DelegateRuntime::teardown_container_netns`].
    netns: Arc<RwLock<HashMap<ContainerId, NetnsState>>>,
    /// Runner used to execute in-distro overlay commands (`ip netns`, `ip
    /// link`, `wg`, …). Swapped in tests for a recording fake; defaults to
    /// [`DefaultWslRunner`]. Note: the non-overlay code paths (G-2/G-3/G-4/G-5)
    /// use the [`wsl_exec_in`] free helper directly via `self.wsl()` /
    /// `self.zlayer_runtime()` so they can target the configured distro name;
    /// the runner covers the overlay/netns commands.
    runner: Arc<dyn WslRunner>,
    /// Optional auth context used to mint + persist a per-container scoped
    /// daemon token, inject it into the OCI spec env, and (best-effort)
    /// revoke it on teardown. `None` disables the whole token path — the
    /// container then gets no `ZLAYER_TOKEN`/`ZLAYER_API_URL`. Mirrors the
    /// Linux youki runtime's `auth_context`.
    auth_context: Option<crate::runtime::ContainerAuthContext>,
    /// Secrets provider for `$S:` env resolution, wired post-construction via
    /// `set_secrets_provider` and fed into the `BundleBuilder`. `None` until the
    /// daemon wires one in. Mirrors `HcsRuntime`. Uses `parking_lot::RwLock` (NOT
    /// the file's aliased tokio `RwLock`) because `set_secrets_provider` is sync.
    secrets_provider:
        parking_lot::RwLock<Option<std::sync::Arc<dyn zlayer_secrets::SecretsProvider>>>,
}

/// Internal helper that replaces every `Option` in [`Wsl2DelegateConfig`] with
/// a concrete value. Populated by [`Wsl2DelegateRuntime::try_new`] once every
/// field has been validated against the live distro.
#[derive(Clone, Debug)]
struct ResolvedConfig {
    distro: String,
    runtime_binary: String,
    bundle_root: String,
    log_root: String,
    oci_state_root: PathBuf,
}

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

impl Wsl2DelegateRuntime {
    /// Create a [`Wsl2DelegateRuntime`] if WSL2 is available on this host.
    ///
    /// Returns `Ok(None)` when WSL2 is not installed or the helper distro
    /// cannot be set up — callers should treat this as "no Linux-container
    /// support on this node" rather than an error. The probe is best-effort;
    /// transient WSL failures log a warning and return `Ok(None)` so daemon
    /// boot proceeds with HCS-only.
    ///
    /// # Errors
    ///
    /// Never bubbles an error up to the caller — the whole point of
    /// `try_new` returning `Option` is that absence-of-WSL2 is a normal
    /// state on Windows hosts. The `Result` return type is kept to leave
    /// room for future fatal-config errors (e.g. malformed environment).
    pub async fn try_new(
        auth_context: Option<crate::runtime::ContainerAuthContext>,
    ) -> Result<Option<Self>> {
        Self::try_new_with_config(Wsl2DelegateConfig::default(), auth_context).await
    }

    /// Create a [`Wsl2DelegateRuntime`] using a caller-supplied config.
    ///
    /// Same best-effort contract as [`Self::try_new`] — returns `Ok(None)` if
    /// WSL2 or the requested distro is unavailable — but lets the caller pin
    /// a non-default distro name, runtime binary, bundle root, log root, or
    /// OCI state root.
    ///
    /// `runtime_binary` resolution rules:
    /// 1. If `config.runtime_binary` is `Some(path)`, use that absolute path.
    /// 2. If `config.runtime_binary` is `None`, use
    ///    [`DEFAULT_RUNTIME_BINARY`] (`/usr/local/bin/zlayer`) — the location
    ///    `zlayer_wsl::setup::install_binary` writes to.
    ///
    /// The resolved binary is then sanity-checked by running
    /// `<binary> runtime --help` inside the distro; failure surfaces as a
    /// hard [`AgentError::Configuration`] so a stale Windows-arch binary or
    /// a build without the `youki-runtime` feature is caught at boot rather
    /// than at first dispatch.
    ///
    /// # Errors
    ///
    /// Returns `Err(AgentError::Configuration(_))` when the resolved
    /// `runtime_binary` does not expose the `zlayer runtime` subcommand
    /// surface inside the distro — a misconfiguration that warrants a hard
    /// fail rather than silently disabling Linux support.
    #[allow(clippy::too_many_lines)]
    pub async fn try_new_with_config(
        config: Wsl2DelegateConfig,
        auth_context: Option<crate::runtime::ContainerAuthContext>,
    ) -> Result<Option<Self>> {
        // 1. Detect WSL2. Any failure of the detect call is treated as
        //    "no WSL" rather than a hard error.
        let status = match zlayer_wsl::detect::detect_wsl().await {
            Ok(s) => s,
            Err(e) => {
                tracing::warn!(
                    error = %e,
                    "WSL2 detection failed; Linux container support disabled"
                );
                return Ok(None);
            }
        };
        if !status.wsl2_available {
            tracing::info!(
                wsl_installed = status.wsl_installed,
                "WSL2 not available; Linux container support disabled on this node"
            );
            return Ok(None);
        }

        // 2. Bootstrap the helper distro if missing. Any failure here is
        //    also best-effort — we log and return Ok(None). This step only
        //    runs when the caller is using the default distro (the setup
        //    module only knows how to bootstrap the `zlayer` distro); for
        //    custom distros we assume the operator has already provisioned
        //    it out-of-band.
        if config.distro == zlayer_wsl::distro::DISTRO_NAME {
            if let Err(e) = zlayer_wsl::setup::ensure_wsl_backend_ready().await {
                tracing::warn!(
                    error = %e,
                    "failed to bootstrap WSL2 zlayer distro; Linux container support disabled"
                );
                return Ok(None);
            }
        }

        // 3. Resolve the in-distro `zlayer` runtime binary. Either honour the
        //    caller's override or fall back to `/usr/local/bin/zlayer`, then
        //    sanity-check by running `<binary> runtime --help` so a stale
        //    Windows-arch binary or one built without the `youki-runtime`
        //    feature is caught up front rather than at first dispatch.
        let runtime_binary = config
            .runtime_binary
            .clone()
            .unwrap_or_else(|| DEFAULT_RUNTIME_BINARY.to_string());
        match wsl_exec_in(&config.distro, &runtime_binary, &["runtime", "--help"]).await {
            Ok(out) if out.status.success() => {}
            Ok(out) => {
                let stderr = String::from_utf8_lossy(&out.stderr);
                return Err(AgentError::Configuration(format!(
                    "zlayer binary at '{runtime_binary}' in distro '{}' does not expose \
                     the `runtime` subcommand (status {:?}): {}",
                    config.distro,
                    out.status.code(),
                    stderr.trim(),
                )));
            }
            Err(e) => {
                return Err(AgentError::Configuration(format!(
                    "zlayer binary at '{runtime_binary}' in distro '{}' does not expose \
                     the `runtime` subcommand: {e}",
                    config.distro,
                )));
            }
        }

        // 4. Ensure the log directory exists inside the distro so that
        //    `youki create --log <log_root>/<slug>.youki.log` does not fail
        //    on first use. `mkdir -p` is idempotent and cheap.
        if let Err(e) = wsl_exec_in(&config.distro, "mkdir", &["-p", &config.log_root]).await {
            tracing::warn!(
                distro = %config.distro,
                log_root = %config.log_root,
                error = %e,
                "failed to pre-create youki log root; container_logs may be empty until youki creates it"
            );
        }

        Ok(Some(Self {
            config: ResolvedConfig {
                distro: config.distro,
                runtime_binary,
                bundle_root: config.bundle_root,
                log_root: config.log_root,
                oci_state_root: config.oci_state_root,
            },
            pids: Arc::new(RwLock::new(HashMap::new())),
            ips: Arc::new(RwLock::new(HashMap::new())),
            bundle_roots: Arc::new(RwLock::new(HashMap::new())),
            image_cache: Arc::new(RwLock::new(HashMap::new())),
            netns: Arc::new(RwLock::new(HashMap::new())),
            runner: Arc::new(DefaultWslRunner),
            auth_context,
            secrets_provider: parking_lot::RwLock::new(None),
        }))
    }

    /// Record an overlay IP that has been allocated for the given container.
    ///
    /// [`Runtime::push_overlay_config`] is the authoritative writer — it records
    /// the IP it actually stamped on the guest `WireGuard` device — but this
    /// stays public so an external overlay-attach flow can pre-seed the IP if it
    /// allocated one out-of-band; [`Runtime::get_container_ip`] returns whatever
    /// was last recorded.
    pub async fn record_container_ip(&self, id: &ContainerId, ip: IpAddr) {
        self.ips.write().await.insert(id.clone(), ip);
    }

    /// In-distro directory for the given container's OCI bundle.
    fn bundle_dir(&self, id: &ContainerId) -> String {
        format!("{}/{}", self.config.bundle_root, id_slug(id))
    }

    /// In-distro path where youki writes the given container's log file. Kept
    /// in sync with the `--log` argument passed to `youki create`.
    fn log_path(&self, id: &ContainerId) -> String {
        format!("{}/{}.youki.log", self.config.log_root, id_slug(id))
    }

    /// Per-container in-distro (ext4) directories backing a WRITABLE overlay
    /// view of the shared toolchain store: `(upper, work, merged)`. They live
    /// under the container's bundle dir (distro-native ext4) — **not** under
    /// `/mnt` — because `fuse-overlayfs` needs its upper/work dirs on a real
    /// POSIX filesystem; the read-only LOWER is the Windows toolchain cache
    /// projected into the distro as a `/mnt/<drive>/…` `DrvFs` path. Deterministic
    /// from `id` so `create_container` and the teardown path agree.
    fn toolchain_overlay_dirs(&self, id: &ContainerId) -> (String, String, String) {
        let base = self.bundle_dir(id);
        (
            format!("{base}/toolchain-upper"),
            format!("{base}/toolchain-work"),
            format!("{base}/toolchain-merged"),
        )
    }

    /// Best-effort unmount of the per-container `fuse-overlayfs` toolchain view.
    /// Safe to call unconditionally: when no overlay was mounted (fuse-overlayfs
    /// absent → we bound the raw `/mnt` path instead, or the container never had
    /// a toolchain cache) `fusermount -u` on a non-mountpoint just returns
    /// nonzero, which we ignore. Must run before the bundle dir is `rm -rf`'d so
    /// we don't recurse into the still-mounted lower.
    async fn unmount_toolchain_overlay(&self, id: &ContainerId) {
        let (_, _, tc_merged) = self.toolchain_overlay_dirs(id);
        let _ = self.wsl("fusermount", &["-u", &tc_merged]).await;
    }

    /// Run a `zlayer runtime <verb>` subcommand inside the configured distro.
    /// Thin wrapper around [`wsl_exec_in`] that prepends the resolved
    /// runtime binary path plus the canonical `runtime --state-root <root>`
    /// prefix so callers don't have to repeat any of that boilerplate.
    async fn zlayer_runtime(&self, args: &[&str]) -> Result<Output> {
        // Hold the borrow on an owned String so the &str produced by
        // `to_string_lossy` outlives the &[&str] we hand to `wsl_exec_in`.
        let state_root_owned = self.config.oci_state_root.to_string_lossy().into_owned();
        let mut full_args: Vec<&str> = Vec::with_capacity(args.len() + 3);
        full_args.push("runtime");
        full_args.push("--state-root");
        full_args.push(state_root_owned.as_str());
        full_args.extend(args.iter().copied());
        wsl_exec_in(&self.config.distro, &self.config.runtime_binary, &full_args).await
    }

    /// Run an arbitrary binary inside the configured distro. Used for the
    /// handful of call sites that need `mkdir`, `rm`, `tail`, etc. rather
    /// than the `zlayer runtime` surface itself.
    async fn wsl(&self, cmd: &str, args: &[&str]) -> Result<Output> {
        wsl_exec_in(&self.config.distro, cmd, args).await
    }

    /// Execute `cmd args...` inside the distro via the configured [`WslRunner`].
    ///
    /// Separate from [`Self::wsl`] so unit tests for the J-3 netns plumbing
    /// can swap in a recording runner without having to stub the whole
    /// `wsl.exe` surface. Production code paths (G-2/G-3/G-4/G-5) continue
    /// to go through [`Self::wsl`] / [`Self::zlayer_runtime`] so they honour
    /// the configured distro + runtime binary.
    async fn wsl_run(&self, cmd: &str, args: &[&str]) -> Result<Output> {
        self.runner.run(cmd, args).await.map_err(|e| {
            AgentError::Network(format!("wsl.exe -d {} -- {cmd}: {e}", self.config.distro))
        })
    }

    /// IFNAMSIZ-safe name for the guest `WireGuard` device backing `slug`.
    /// `make_interface_name` guarantees `<= 15` chars by hashing long inputs,
    /// and is deterministic so `push_overlay_config` and
    /// `teardown_container_netns` agree on the name.
    fn wg_iface_name(slug: &str) -> String {
        make_interface_name(&[slug], "wg")
    }

    /// Provision the named network namespace `/run/netns/<slug>` inside the
    /// distro so youki can JOIN it (`with_netns_path`) and the post-start
    /// [`Runtime::push_overlay_config`] can later address it by name with
    /// `ip netns exec <slug>`. Idempotent: a pre-existing namespace ("File
    /// exists") is reused. Records [`NetnsState::Created`] so
    /// [`Self::teardown_container_netns`] knows to delete it later.
    async fn provision_named_netns(&self, id: &ContainerId, slug: &str) -> Result<()> {
        let mk_ns = self.wsl_run("ip", &["netns", "add", slug]).await?;
        if !mk_ns.status.success() {
            let stderr = String::from_utf8_lossy(&mk_ns.stderr);
            if !stderr.to_ascii_lowercase().contains("file exists") {
                return Err(AgentError::Network(format!(
                    "ip netns add {slug} failed (status {:?}): {}",
                    mk_ns.status.code(),
                    stderr.trim()
                )));
            }
        }
        self.netns
            .write()
            .await
            .insert(id.clone(), NetnsState::Created);
        Ok(())
    }

    /// Resolve the distro's default gateway — the `vEthernet (WSL)` host
    /// adapter IP reachable from inside the NAT'd distro. The overlayd-advertised
    /// peer endpoints are NODE OVERLAY IPs (only reachable THROUGH the overlay,
    /// circular), so every peer endpoint host is rewritten to this gateway (the
    /// host underlay) before being handed to `wg set`.
    async fn distro_default_gateway(&self) -> Result<String> {
        let out = self.wsl_run("ip", &["route", "show", "default"]).await?;
        if !out.status.success() {
            return Err(AgentError::Network(format!(
                "ip route show default failed (status {:?}): {}",
                out.status.code(),
                String::from_utf8_lossy(&out.stderr).trim()
            )));
        }
        let stdout = String::from_utf8_lossy(&out.stdout);
        parse_default_gateway(&stdout).ok_or_else(|| {
            AgentError::Network(format!(
                "could not parse WSL distro default gateway from `ip route show default`: {:?}",
                stdout.trim()
            ))
        })
    }

    /// The fallible body of [`Runtime::push_overlay_config`]: bring up the guest
    /// `WireGuard` device in the distro's main netns (so its UDP socket stays
    /// bound where the WSL uplink lives), program its peers, then move the
    /// device into the container's named netns and assign the overlay address,
    /// route, and DNS.
    ///
    /// Every step is a distinct in-distro command with its own error message;
    /// factoring each into its own helper would only deepen the per-step
    /// branching without improving readability.
    #[allow(clippy::too_many_lines)]
    async fn push_overlay_config_inner(
        &self,
        id: &ContainerId,
        slug: &str,
        wg_iface: &str,
        config: &zlayer_types::overlayd::GuestOverlayConfig,
    ) -> Result<()> {
        // Resolve the NAT gateway BEFORE touching the netns so a parse failure
        // fails cheaply with nothing to clean up.
        let gateway = self.distro_default_gateway().await?;

        // 1. Write the private key to a distro temp file with umask 077 so it
        //    is never world-readable and never passed as a (ps-visible) arg.
        let keyfile = format!("/run/zlayer-wg-{slug}.key");
        let write_key = format!(
            "umask 077 && printf '%s' {} > {}",
            sh_single_quote(&config.private_key),
            keyfile
        );
        self.run_checked("sh", &["-c", &write_key], "write wg private key")
            .await?;

        // 2. Create the WireGuard device in the distro's main netns.
        self.run_checked(
            "ip",
            &["link", "add", wg_iface, "type", "wireguard"],
            "ip link add wireguard",
        )
        .await?;

        // 3. Apply the private key + listen port.
        let listen_port = config.listen_port.to_string();
        self.run_checked(
            "wg",
            &[
                "set",
                wg_iface,
                "private-key",
                &keyfile,
                "listen-port",
                &listen_port,
            ],
            "wg set private-key/listen-port",
        )
        .await?;
        // Key consumed by `wg set`; remove it so it doesn't linger.
        let _ = self.wsl_run("rm", &["-f", &keyfile]).await;

        // 4. Program each peer, rewriting the (overlay-IP) endpoint host to the
        //    NAT gateway the distro can actually reach. Roaming peers (empty
        //    endpoint) are added without an endpoint arg.
        for peer in &config.peers {
            let endpoint = rewrite_endpoint_host(&peer.endpoint, &gateway);
            let keepalive = peer.persistent_keepalive_secs.to_string();
            let mut args: Vec<&str> = vec!["set", wg_iface, "peer", &peer.public_key];
            if !endpoint.is_empty() {
                args.push("endpoint");
                args.push(&endpoint);
            }
            args.push("allowed-ips");
            args.push(&peer.allowed_ips);
            args.push("persistent-keepalive");
            args.push(&keepalive);
            self.run_checked("wg", &args, "wg set peer").await?;
        }

        // 5. Move the device into the container's named netns. The UDP socket
        //    stays bound in the main netns (where the WSL uplink lives).
        self.run_checked(
            "ip",
            &["link", "set", wg_iface, "netns", slug],
            "ip link set wg netns",
        )
        .await?;

        // 6. Assign the overlay address inside the container netns.
        let addr = format!("{}/{}", config.overlay_ip, config.prefix_len);
        self.run_checked(
            "ip",
            &[
                "netns", "exec", slug, "ip", "addr", "add", &addr, "dev", wg_iface,
            ],
            "ip addr add overlay",
        )
        .await?;

        // 7. Bring the device + loopback up.
        self.run_checked(
            "ip",
            &["netns", "exec", slug, "ip", "link", "set", wg_iface, "up"],
            "ip link set wg up",
        )
        .await?;
        self.run_checked(
            "ip",
            &["netns", "exec", slug, "ip", "link", "set", "lo", "up"],
            "ip link set lo up",
        )
        .await?;

        // 8. Default route out the overlay device.
        self.run_checked(
            "ip",
            &[
                "netns", "exec", slug, "ip", "route", "add", "default", "dev", wg_iface,
            ],
            "ip route add default",
        )
        .await?;

        // 9. Write the container's /etc/resolv.conf (in its rootfs, which the
        //    running container shares) with the overlay DNS resolver + search
        //    domain so `<svc>`/`<svc>.<domain>` resolve. Skipped when overlayd
        //    advertised no resolver.
        if let Some(dns) = config.dns_server {
            use std::fmt::Write as _;
            let mut content = format!("nameserver {dns}\n");
            if let Some(domain) = &config.dns_domain {
                let _ = writeln!(content, "search {domain}");
            }
            let etc_dir = format!("{}/rootfs/etc", self.bundle_dir(id));
            let write_resolv = format!(
                "mkdir -p {} && printf '%s' {} > {}/resolv.conf",
                etc_dir,
                sh_single_quote(&content),
                etc_dir
            );
            self.run_checked("sh", &["-c", &write_resolv], "write resolv.conf")
                .await?;
        }

        Ok(())
    }

    /// Run an in-distro overlay command via the runner and turn a non-zero exit
    /// into an [`AgentError::Network`] tagged with `what`.
    async fn run_checked(&self, cmd: &str, args: &[&str], what: &str) -> Result<()> {
        let out = self.wsl_run(cmd, args).await?;
        if out.status.success() {
            Ok(())
        } else {
            Err(AgentError::Network(format!(
                "{what} ({cmd} {}) failed (status {:?}): {}",
                args.join(" "),
                out.status.code(),
                String::from_utf8_lossy(&out.stderr).trim()
            )))
        }
    }

    /// Tear down the per-container named netns + guest `WireGuard` device.
    ///
    /// Called from [`Runtime::remove_container`]. Infallible: every command is
    /// best-effort because the user has already asked for the container to be
    /// gone, and a partial teardown is more useful than a hard error that
    /// leaves the cleanup half-done. Deleting the named netns also tears down
    /// the `WireGuard` device (and its UDP socket) if it was already moved in;
    /// the explicit deletes cover the case where push failed before the move.
    async fn teardown_container_netns(&self, id: &ContainerId) {
        let state = self.netns.write().await.remove(id);
        if state.is_none() {
            // Host-network / network-none / container-joined: never got a
            // named netns, nothing to tear down.
            return;
        }
        let slug = id_slug(id);
        let wg_iface = Self::wg_iface_name(&slug);
        // Delete the wg device from the main netns (push failed before the
        // move) and from the container netns (push succeeded) — both best-effort.
        let _ = self.wsl_run("ip", &["link", "delete", &wg_iface]).await;
        let _ = self
            .wsl_run(
                "ip",
                &["netns", "exec", &slug, "ip", "link", "delete", &wg_iface],
            )
            .await;
        let _ = self.wsl_run("ip", &["netns", "delete", &slug]).await;
    }

    /// Query `zlayer runtime state` for the given container and parse its JSON.
    async fn query_state(&self, id: &ContainerId) -> Result<YoukiState> {
        let slug = id_slug(id);
        let output = self.zlayer_runtime(&["state", &slug]).await?;
        if !output.status.success() {
            let stderr = String::from_utf8_lossy(&output.stderr);
            // The runtime emits a consistent "not found" message when the
            // container id is unknown; translate to NotFound so callers get
            // the conventional 404 surface from the API layer.
            if stderr.to_ascii_lowercase().contains("does not exist")
                || stderr.to_ascii_lowercase().contains("not found")
            {
                return Err(AgentError::NotFound {
                    container: id.to_string(),
                    reason: format!("zlayer runtime state reports container unknown: {stderr}"),
                });
            }
            return Err(zlayer_runtime_error("state", &output));
        }
        parse_youki_state(&output)
    }
}

// ---------------------------------------------------------------------------
// Runtime impl
// ---------------------------------------------------------------------------

#[async_trait]
impl Runtime for Wsl2DelegateRuntime {
    fn set_secrets_provider(&self, provider: std::sync::Arc<dyn zlayer_secrets::SecretsProvider>) {
        *self.secrets_provider.write() = Some(provider);
    }

    async fn pull_image(&self, image: &str) -> Result<()> {
        self.pull_image_with_policy(
            image,
            PullPolicy::IfNotPresent,
            None,
            zlayer_spec::SourcePolicy::default(),
        )
        .await
    }

    async fn pull_image_with_policy(
        &self,
        image: &str,
        policy: PullPolicy,
        auth: Option<&RegistryAuth>,
        _source: zlayer_spec::SourcePolicy,
    ) -> Result<()> {
        // Fast path: already cached and the policy allows reuse.
        if matches!(policy, PullPolicy::IfNotPresent | PullPolicy::Never)
            && self.image_cache.read().await.contains_key(image)
        {
            if matches!(policy, PullPolicy::Never) {
                return Ok(());
            }
            tracing::debug!(image, "image cache hit; skipping re-pull");
            return Ok(());
        }
        if matches!(policy, PullPolicy::Never) {
            return Err(AgentError::PullFailed {
                image: image.to_string(),
                reason: "PullPolicy::Never but image is not in the WSL2 delegate cache".to_string(),
            });
        }

        // Pulls on the Windows host go through `zlayer_registry::ImagePuller`
        // because the WSL2 distro is *not* a registry: it has no credential
        // store, no reusable blob cache across daemon restarts, and — more
        // importantly — upstream `youki` has no real `pull` subcommand. Doing
        // the HTTP work here keeps a single code path for all runtimes and
        // gives us a `Vec<(blob, media_type)>` we can later stream directly
        // into the distro at `create_container` time.
        let registry_auth = match auth {
            Some(a) => zlayer_registry::RegistryAuth::Basic(a.username.clone(), a.password.clone()),
            // Honor ~/.docker/config.json (AuthConfig default = DockerConfig) so
            // `zlayer login` creds / Docker Hub auth apply instead of anonymous.
            None => {
                zlayer_core::AuthResolver::new(zlayer_core::AuthConfig::default()).resolve(image)
            }
        };
        let cache = zlayer_registry::BlobCache::new().map_err(|e| AgentError::PullFailed {
            image: image.to_string(),
            reason: format!("failed to create blob cache: {e}"),
        })?;
        // The WSL2 delegate *always* runs linux/amd64 containers inside the
        // helper distro, even though the host is Windows. Pin the puller's
        // platform selector so multi-platform image indexes resolve to the
        // Linux manifest — otherwise `oci-client` would pick the Windows
        // variant (matching the host) and every pull would silently fail.
        let cache_arc: Arc<Box<dyn zlayer_registry::BlobCacheBackend>> = Arc::new(Box::new(cache));
        let puller = zlayer_registry::ImagePuller::with_platform(
            cache_arc,
            zlayer_spec::TargetPlatform::new(
                zlayer_spec::OsKind::Linux,
                zlayer_spec::ArchKind::Amd64,
            ),
        );

        // Stream every layer to host disk (no full layer buffered in RAM). The
        // files persist in a per-image host stage dir consumed by
        // `create_container` and cleaned on re-pull / `remove_image`.
        let stage = wsl2_layer_stage_dir(image);
        let _ = tokio::fs::remove_dir_all(&stage).await;
        let layers = puller
            .pull_image_to_files_with_policy(image, &registry_auth, &stage, policy)
            .await
            .map_err(|e| AgentError::PullFailed {
                image: image.to_string(),
                reason: format!("registry pull failed: {e}"),
            })?;

        let config = puller
            .pull_image_config(image, &registry_auth)
            .await
            .map_err(|e| AgentError::PullFailed {
                image: image.to_string(),
                reason: format!("image config fetch failed: {e}"),
            })?;

        self.image_cache
            .write()
            .await
            .insert(image.to_string(), CachedImage { layers, config });

        Ok(())
    }

    #[allow(clippy::too_many_lines)]
    async fn create_container(&self, id: &ContainerId, spec: &ServiceSpec) -> Result<()> {
        let bundle_dir = self.bundle_dir(id);
        let rootfs_dir = format!("{bundle_dir}/rootfs");
        let slug = id_slug(id);

        // 1. Make sure the bundle + rootfs directories exist. Idempotent.
        let mkdir = self.wsl("mkdir", &["-p", &rootfs_dir]).await?;
        if !mkdir.status.success() {
            return Err(AgentError::CreateFailed {
                id: id.to_string(),
                reason: format!(
                    "mkdir -p {rootfs_dir} failed (status {:?}): {}",
                    mkdir.status.code(),
                    String::from_utf8_lossy(&mkdir.stderr).trim()
                ),
            });
        }

        // 2. Fetch or reuse the pulled image data. When the composite layer
        //    has already pulled this image the cache hits immediately; when
        //    the caller skipped the pull step (e.g. direct `create_container`
        //    in a test) we fall through to the same pull path used by
        //    `pull_image_with_policy`.
        let image_name = spec.image.name.to_string();
        let cached = if let Some(c) = self.image_cache.read().await.get(&image_name).cloned() {
            c
        } else {
            self.pull_image_with_policy(
                &image_name,
                spec.image.pull_policy,
                None,
                spec.image.source_policy.unwrap_or_default(),
            )
            .await?;
            self.image_cache
                .read()
                .await
                .get(&image_name)
                .cloned()
                .ok_or_else(|| AgentError::CreateFailed {
                    id: id.to_string(),
                    reason: format!(
                        "image {} missing from WSL2 delegate cache after pull",
                        spec.image.name
                    ),
                })?
        };

        // 3. Extract every layer into <bundle_dir>/rootfs inside the distro.
        //    Decompression happens on the Windows host (synchronous, fast,
        //    already in RAM) so the tar stream we hand to `wsl.exe -- tar`
        //    is always a plain tarball — no gzip/zstd tooling required
        //    inside the distro. `--no-same-owner` keeps things working when
        //    WSL exposes a mismatched uid map.
        let tar_sh_cmd = format!("cd {rootfs_dir} && tar -xf - --no-same-owner");
        for (i, (layer_path, media_type)) in cached.layers.iter().enumerate() {
            wsl_extract_layer_file(
                &["-d", &self.config.distro, "--", "sh", "-c", &tar_sh_cmd],
                layer_path,
                media_type,
            )
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: id.to_string(),
                reason: format!(
                    "streaming layer {i} ({media_type}) of {} into WSL2 rootfs failed: {e}",
                    spec.image.name
                ),
            })?;
        }

        // 3b. Resolve container networking the same way the Linux youki runtime
        //     does (youki.rs `network_mode` handling), but use a NAMED netns
        //     (`/run/netns/<slug>`) for the overlay-capable cases so the
        //     post-start guest overlay attach (`push_overlay_config`) can bring
        //     up WireGuard inside it by name:
        //       * `Host`            → share the host stack; no netns_path.
        //       * `None`            → fresh/empty netns; no named netns, no wg.
        //       * `Container{id}`   → JOIN the target's netns at its in-distro
        //                             `/proc/<pid>/ns/net`.
        //       * `Default`/`Bridge`→ provision `/run/netns/<slug>` and JOIN it;
        //                             overlay wg pushed post-start.
        let netns_path: Option<PathBuf> = match &spec.network_mode {
            zlayer_spec::NetworkMode::Container { id: target } => {
                let target_cid =
                    ContainerId::parse_display(target).ok_or_else(|| AgentError::CreateFailed {
                        id: id.to_string(),
                        reason: format!(
                            "network target container {target:?} is not a resolvable container id"
                        ),
                    })?;
                let pid = self
                    .get_container_pid(&target_cid)
                    .await
                    .map_err(|e| AgentError::CreateFailed {
                        id: id.to_string(),
                        reason: format!(
                            "network target container {target} not found or not running: {e}"
                        ),
                    })?
                    .ok_or_else(|| AgentError::CreateFailed {
                        id: id.to_string(),
                        reason: format!(
                            "network target container {target} is not running (no pid)"
                        ),
                    })?;
                Some(PathBuf::from(format!("/proc/{pid}/ns/net")))
            }
            zlayer_spec::NetworkMode::Host | zlayer_spec::NetworkMode::None => None,
            // Default | Bridge: overlay-capable — provision a named netns.
            _ => {
                self.provision_named_netns(id, &slug).await.map_err(|e| {
                    AgentError::CreateFailed {
                        id: id.to_string(),
                        reason: format!("failed to provision overlay netns: {e}"),
                    }
                })?;
                Some(PathBuf::from(format!("/run/netns/{slug}")))
            }
        };

        // 3c. Shared toolchain cache: give the container a per-container WRITABLE
        //     overlay view of the shared toolchain store. The host (Windows)
        //     cache dir is projected into the distro via `/mnt/<drive>/…` and
        //     used as the read-only overlay LOWER. youki runs INSIDE the distro,
        //     so we set the overlay up here, before launch, and bind the merged
        //     mountpoint into the bundle. Because the lower lives on DrvFs,
        //     kernel overlayfs over it is unreliable — use `fuse-overlayfs` with
        //     the upper/work dirs on the distro's native ext4 (under the
        //     per-container bundle dir). Each container thus gets isolated writes
        //     while still reusing downloaded toolchains across runs. If
        //     `fuse-overlayfs` isn't available in the distro we fall back to
        //     binding the raw `/mnt` path directly (shared-RW, the prior
        //     behaviour). Best-effort throughout: a toolchain overlay failure
        //     must never fail container create.
        let toolchain_wsl: Option<String> = {
            let host = zlayer_paths::ZLayerDirs::system_default().toolchain_cache();
            if let Some(lower) = zlayer_wsl::paths::windows_to_wsl(&host) {
                // `mkdir -p` the projected lower so the DrvFs source exists on
                // the Windows host, plus the ext4 upper/work/merged dirs.
                let _ = self.wsl("mkdir", &["-p", &lower]).await;
                let (tc_upper, tc_work, tc_merged) = self.toolchain_overlay_dirs(id);
                let _ = self
                    .wsl("mkdir", &["-p", &tc_upper, &tc_work, &tc_merged])
                    .await;
                let overlay_opt = format!("lowerdir={lower},upperdir={tc_upper},workdir={tc_work}");
                match self
                    .wsl("fuse-overlayfs", &["-o", &overlay_opt, &tc_merged])
                    .await
                {
                    Ok(out) if out.status.success() => {
                        tracing::debug!(
                            container = %id,
                            lower = %lower,
                            merged = %tc_merged,
                            "mounted per-container fuse-overlayfs toolchain view"
                        );
                        Some(tc_merged)
                    }
                    Ok(out) => {
                        tracing::warn!(
                            container = %id,
                            lower = %lower,
                            status = ?out.status.code(),
                            stderr = %String::from_utf8_lossy(&out.stderr).trim(),
                            "fuse-overlayfs toolchain mount failed; falling back to raw /mnt bind"
                        );
                        Some(lower)
                    }
                    Err(e) => {
                        tracing::warn!(
                            container = %id,
                            lower = %lower,
                            error = %e,
                            "fuse-overlayfs unavailable in distro; falling back to raw /mnt bind"
                        );
                        Some(lower)
                    }
                }
            } else {
                None
            }
        };

        // 4. Render the OCI runtime spec on the host using the cross-platform
        //    `BundleBuilder::build_spec_only` entry point (G-1). The bundle
        //    path passed to `BundleBuilder::new` is purely informational here
        //    — `build_spec_only` never touches the filesystem, so the
        //    Windows-style path is fine even though it'll never exist.
        let mut builder = BundleBuilder::new(PathBuf::from(&bundle_dir))
            .with_image_config(cached.config.clone())
            .with_hostname(slug.clone())
            .with_host_network(spec.host_network)
            .with_netns_path(netns_path);
        if let Some(tc) = toolchain_wsl {
            builder = builder.with_toolchain_cache(PathBuf::from(tc));
        }
        // Route spec env through `$S:` secret resolution when a provider AND a
        // secret scope are present (BundleBuilder's assembly handles it). The
        // ZLAYER_TOKEN env pushed later (after build_spec_only) stays highest.
        if let (Some(provider), Some(scope)) = (
            self.secrets_provider.read().clone(),
            spec.secret_scope.clone(),
        ) {
            builder = builder
                .with_secrets_provider(provider)
                .with_deployment_scope(scope);
        }
        let mut oci_spec = builder
            .build_spec_only(id, spec, &HashMap::new())
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: id.to_string(),
                reason: format!("failed to build OCI spec on Windows host: {e}"),
            })?;

        // Phase 5.D: wire `/dev/dxg` + WSLg lib mounts into the bundle when
        // the service spec requests a GPU. No-op for CPU-only workloads.
        // A missing `/dev/dxg` is a hard error here; we don't want a silent
        // CPU fallback for users who explicitly asked for GPU.
        let gpu_probe = DefaultWslGpuHostProbe { runtime: self };
        apply_wsl_gpu_to_spec(&mut oci_spec, spec, &gpu_probe).await?;

        // Inject a LEAST-PRIVILEGE scoped daemon token into the container's
        // OCI spec env so the Linux workload running inside the WSL2 distro can
        // talk back to the host API without external credentials — exactly the
        // treatment the native Linux youki runtime gives its containers
        // (read-only on its own deployment by default; broaden via
        // `zlayer.io/api-scopes`). The host admin Unix socket is NOT
        // bind-mounted unless the service opts in via `zlayer.io/daemon-socket`
        // (that grants full daemon admin).
        if let Some(ref auth_ctx) = self.auth_context {
            let deployment = spec.deployment.as_deref().unwrap_or(&id.service);
            let access = crate::auth::resolve_container_api_access(deployment, &spec.labels);
            let container_id = format!("{}-{}", id.service, id.replica);
            let jti = format!("container:{}:{}", id.service, container_id);
            // Persist the token record BEFORE embedding its jti — the auth
            // layer is fail-closed and rejects a jti with no record. On
            // persistence failure, mint without a jti (token still accepted,
            // bounded by TTL, but not revocable).
            let token_jti = if let Some(sink) = auth_ctx.token_sink.as_ref() {
                let now = chrono::Utc::now();
                let rec = zlayer_types::storage::StoredAccessToken {
                    id: jti.clone(),
                    name: id.service.clone(),
                    subject: jti.clone(),
                    roles: Vec::new(),
                    scopes: access.scopes.clone(),
                    expires_at: now
                        + chrono::Duration::seconds(
                            i64::try_from(access.ttl.as_secs()).unwrap_or(i64::MAX),
                        ),
                    created_at: now,
                    created_by: deployment.to_string(),
                    revoked_at: None,
                };
                if sink.persist(rec).await {
                    Some(jti)
                } else {
                    None
                }
            } else {
                None
            };
            let token = crate::auth::mint_container_token(
                &auth_ctx.jwt_secret,
                &id.service,
                &container_id,
                access.scopes,
                access.ttl,
                token_jti,
            )
            .map_err(|e| AgentError::CreateFailed {
                id: id.to_string(),
                reason: format!("Failed to mint container token: {e}"),
            })?;

            // Push the auth env onto the spec's process env (mirrors
            // `apply_wsl_gpu_to_spec`'s env mutation pattern). `ZLAYER_SOCKET`
            // is only set — and the socket only bind-mounted — when the service
            // explicitly opted into the daemon socket.
            let mut env = oci_spec
                .process()
                .as_ref()
                .and_then(|p| p.env().clone())
                .unwrap_or_default();
            env.push(format!("ZLAYER_API_URL={}", auth_ctx.api_url));
            env.push(format!("ZLAYER_TOKEN={token}"));
            if access.mount_socket {
                let socket_dest = zlayer_paths::ZLayerDirs::default_socket_path();
                env.push(format!("ZLAYER_SOCKET={socket_dest}"));
            }
            if let Some(process) = oci_spec.process_mut().as_mut() {
                process.set_env(Some(env));
            }

            // Bind-mount the host admin socket into the bundle when opted in —
            // mirrors youki's `with_socket_mount` intent. The socket path is a
            // host (Windows) path here; the distro sees it via WSL's
            // `/mnt/<drive>` projection, so we mount it at its conventional
            // in-container destination.
            if access.mount_socket {
                let socket_dest = zlayer_paths::ZLayerDirs::default_socket_path();
                let mut mounts = oci_spec.mounts().clone().unwrap_or_default();
                let socket_mount = MountBuilder::default()
                    .destination(PathBuf::from(&socket_dest))
                    .typ("bind".to_string())
                    .source(PathBuf::from(&auth_ctx.socket_path))
                    .options(vec!["rbind".to_string(), "rw".to_string()])
                    .build()
                    .map_err(|e| AgentError::CreateFailed {
                        id: id.to_string(),
                        reason: format!("failed to build daemon socket bind mount: {e}"),
                    })?;
                mounts.push(socket_mount);
                oci_spec.set_mounts(Some(mounts));
            }
        }

        let config_json =
            serde_json::to_string_pretty(&oci_spec).map_err(|e| AgentError::CreateFailed {
                id: id.to_string(),
                reason: format!("failed to serialize OCI spec to JSON: {e}"),
            })?;

        // 5. Stream the rendered `config.json` into the distro via `tee`.
        //    `tee` buffers the write inside the distro so WSL's stdin
        //    handling does not truncate partial UTF-8 at the filesystem
        //    boundary; redirecting stdout to /dev/null avoids echoing the
        //    payload back over the pipe.
        let config_path = format!("{bundle_dir}/config.json");
        let tee_sh_cmd = format!("tee {config_path} > /dev/null");
        wsl_stdin_pipe(
            &["-d", &self.config.distro, "--", "sh", "-c", &tee_sh_cmd],
            config_json.as_bytes(),
        )
        .await
        .map_err(|e| AgentError::CreateFailed {
            id: id.to_string(),
            reason: format!("failed to write config.json into WSL2 bundle: {e}"),
        })?;

        // 6. Hand the bundle to `zlayer runtime create`. `--log <path>` points
        //    at the per-container log file inside the distro so `container_logs`
        //    has something to tail — without this flag the runtime just
        //    writes to stderr which we can't easily read back later. A
        //    non-zero exit here rolls back the bundle directory so a retry
        //    sees a clean slate.
        let log_path = self.log_path(id);
        let create = self
            .zlayer_runtime(&["create", &slug, "--bundle", &bundle_dir, "--log", &log_path])
            .await?;
        if !create.status.success() {
            let stderr = String::from_utf8_lossy(&create.stderr).trim().to_string();
            // Best-effort cleanup: we don't care if it succeeds. Unmount the
            // per-container toolchain overlay BEFORE `rm -rf` so we don't recurse
            // into the still-mounted lower. Also tear down the named netns we may
            // have provisioned so a retry starts clean.
            self.unmount_toolchain_overlay(id).await;
            let _ = self.wsl("rm", &["-rf", &bundle_dir]).await;
            self.teardown_container_netns(id).await;
            return Err(AgentError::CreateFailed {
                id: id.to_string(),
                reason: format!(
                    "zlayer runtime create failed (status {:?}): {stderr}",
                    create.status.code(),
                ),
            });
        }

        // 7. Record the bundle path so cleanup knows where to look later.
        self.bundle_roots
            .write()
            .await
            .insert(id.clone(), bundle_dir);
        Ok(())
    }

    async fn start_container(&self, id: &ContainerId) -> Result<()> {
        // The container's network namespace was provisioned at create time
        // (named netns the container JOINS); overlay membership is wired up
        // POST-start by the service layer via `push_overlay_config` (the
        // guest-managed attach kind). Nothing to do here before `start`.
        let slug = id_slug(id);
        let output = self.zlayer_runtime(&["start", &slug]).await?;
        if !output.status.success() {
            // zlayer runtime start failed; tear down the named netns we
            // provisioned at create time so we don't leak it.
            self.teardown_container_netns(id).await;
            return Err(AgentError::StartFailed {
                id: id.to_string(),
                reason: format!(
                    "zlayer runtime start failed (status {:?}): {}",
                    output.status.code(),
                    String::from_utf8_lossy(&output.stderr).trim()
                ),
            });
        }
        // Best-effort: cache the PID for later get_container_pid calls.
        if let Ok(state) = self.query_state(id).await {
            if let Some(pid) = state.pid {
                self.pids.write().await.insert(id.clone(), pid);
            }
        }
        Ok(())
    }

    async fn stop_container(&self, id: &ContainerId, timeout: Duration) -> Result<()> {
        let slug = id_slug(id);
        // SIGTERM first.
        let term = self
            .zlayer_runtime(&["kill", "--all", &slug, "SIGTERM"])
            .await?;
        if !term.status.success() {
            // If the container is already stopped the runtime returns
            // nonzero; treat that as success for stop semantics rather than
            // erroring.
            let stderr = String::from_utf8_lossy(&term.stderr).to_ascii_lowercase();
            if !(stderr.contains("stopped") || stderr.contains("not running")) {
                return Err(zlayer_runtime_error("kill", &term));
            }
        }

        // Poll `state` until status == "stopped" or timeout expires.
        let deadline = tokio::time::Instant::now() + timeout;
        loop {
            match self.query_state(id).await {
                Ok(state) if state.is_stopped() => return Ok(()),
                Ok(_) => {}
                Err(AgentError::NotFound { .. }) => return Ok(()),
                Err(e) => return Err(e),
            }
            if tokio::time::Instant::now() >= deadline {
                break;
            }
            tokio::time::sleep(WAIT_POLL_INTERVAL).await;
        }

        // Escalate to SIGKILL.
        let kill = self
            .zlayer_runtime(&["kill", "--all", &slug, "SIGKILL"])
            .await?;
        if !kill.status.success() {
            let stderr = String::from_utf8_lossy(&kill.stderr).to_ascii_lowercase();
            if !(stderr.contains("stopped") || stderr.contains("not running")) {
                return Err(zlayer_runtime_error("kill", &kill));
            }
        }
        Ok(())
    }

    async fn remove_container(&self, id: &ContainerId) -> Result<()> {
        let slug = id_slug(id);
        let output = self.zlayer_runtime(&["delete", &slug]).await?;
        // Revoke the per-container scoped daemon token (best-effort) so a
        // removed container's credential can't outlive it up to its TTL. The
        // jti is the same deterministic string `create_container` persisted.
        if let Some(auth_ctx) = self.auth_context.as_ref() {
            if let Some(sink) = auth_ctx.token_sink.as_ref() {
                let container_id = format!("{}-{}", id.service, id.replica);
                let jti = format!("container:{}:{}", id.service, container_id);
                sink.revoke(&jti).await;
            }
        }
        // Tear down the per-container named netns + WireGuard device before
        // clearing caches. Best-effort: we've already told youki to delete the
        // container, and leaving a netns leak is less bad than failing remove.
        self.teardown_container_netns(id).await;
        // Clear caches regardless of delete outcome — the container entry
        // is gone from our perspective once we've called `delete`.
        self.pids.write().await.remove(id);
        self.ips.write().await.remove(id);
        // Best-effort: unmount the per-container fuse-overlayfs toolchain view
        // before wiping the bundle dir, so `rm -rf` doesn't recurse into the
        // still-mounted (DrvFs) lower. No-op when the container used the raw
        // /mnt-bind fallback or had no toolchain cache.
        self.unmount_toolchain_overlay(id).await;
        // Best-effort: wipe the bundle directory in the distro so a future
        // create with the same id starts from a clean rootfs.
        if let Some(bundle_dir) = self.bundle_roots.write().await.remove(id) {
            if let Err(e) = self.wsl("rm", &["-rf", &bundle_dir]).await {
                tracing::debug!(
                    container = %id,
                    bundle_dir = %bundle_dir,
                    error = %e,
                    "failed to remove WSL2 bundle dir; leaving for GC"
                );
            }
        }
        // Same best-effort cleanup for the per-container youki log file —
        // otherwise re-creating a container with the same id would tail
        // lines from the previous incarnation.
        let log_path = self.log_path(id);
        if let Err(e) = self.wsl("rm", &["-f", &log_path]).await {
            tracing::debug!(
                container = %id,
                log_path = %log_path,
                error = %e,
                "failed to remove youki log file; leaving for GC"
            );
        }
        if output.status.success() {
            Ok(())
        } else {
            let stderr = String::from_utf8_lossy(&output.stderr).to_ascii_lowercase();
            if stderr.contains("does not exist") || stderr.contains("not found") {
                return Ok(());
            }
            Err(zlayer_runtime_error("delete", &output))
        }
    }

    async fn container_state(&self, id: &ContainerId) -> Result<ContainerState> {
        let state = self.query_state(id).await?;
        Ok(state.as_container_state())
    }

    async fn container_logs(&self, id: &ContainerId, tail: usize) -> Result<Vec<LogEntry>> {
        // Read from the per-container log file that `youki create --log
        // <path>` writes to inside the distro. Before G-4 this pointed at a
        // fabricated `/var/log/youki/<slug>.stdout.log` that youki never
        // touched — now it's the same path `create_container` just passed
        // as `--log`, so a tail actually returns something.
        let log_path = self.log_path(id);
        let tail_str = tail.to_string();
        let output = self.wsl("tail", &["-n", &tail_str, &log_path]).await?;
        if !output.status.success() {
            let stderr = String::from_utf8_lossy(&output.stderr).to_ascii_lowercase();
            // First-call-before-any-log-writes: treat missing file as an
            // empty log rather than an error so startup polling is quiet.
            if stderr.contains("no such file") || stderr.contains("cannot open") {
                return Ok(Vec::new());
            }
            return Err(zlayer_runtime_error("tail", &output));
        }

        let stdout = String::from_utf8_lossy(&output.stdout);
        let now = chrono::Utc::now();
        let source = LogSource::Container(id.to_string());
        let service = Some(id.service.clone());
        let entries = stdout
            .lines()
            .filter(|l| !l.is_empty())
            .map(|line| LogEntry {
                timestamp: now,
                stream: LogStream::Stdout,
                message: line.to_string(),
                source: source.clone(),
                service: service.clone(),
                deployment: None,
            })
            .collect();
        Ok(entries)
    }

    async fn exec(&self, id: &ContainerId, cmd: &[String]) -> Result<(i32, String, String)> {
        if cmd.is_empty() {
            return Err(AgentError::InvalidSpec(
                "exec command must not be empty".to_string(),
            ));
        }
        let slug = id_slug(id);
        let mut args: Vec<&str> = vec!["exec", &slug, "--"];
        args.extend(cmd.iter().map(String::as_str));
        let output = self.zlayer_runtime(&args).await?;
        let stdout = String::from_utf8_lossy(&output.stdout).into_owned();
        let stderr = String::from_utf8_lossy(&output.stderr).into_owned();
        let exit = output.status.code().unwrap_or(-1);
        Ok((exit, stdout, stderr))
    }

    async fn get_container_stats(&self, id: &ContainerId) -> Result<ContainerStats> {
        let slug = id_slug(id);
        let output = self.zlayer_runtime(&["events", &slug, "--stats"]).await;
        let now = std::time::Instant::now();
        match output {
            Ok(out) if out.status.success() => Ok(parse_youki_stats(&out, now)),
            Ok(out) => Err(zlayer_runtime_error("events", &out)),
            Err(e) => Err(e),
        }
    }

    async fn wait_container(&self, id: &ContainerId) -> Result<i32> {
        let start = tokio::time::Instant::now();
        loop {
            match self.query_state(id).await {
                Ok(state) if state.is_stopped() => {
                    return Ok(state.exit_code.unwrap_or(0));
                }
                Ok(_) => {}
                Err(e) => return Err(e),
            }
            if start.elapsed() >= WAIT_POLL_CAP {
                return Err(AgentError::Timeout {
                    timeout: WAIT_POLL_CAP,
                });
            }
            tokio::time::sleep(WAIT_POLL_INTERVAL).await;
        }
    }

    async fn get_logs(&self, id: &ContainerId) -> Result<Vec<LogEntry>> {
        self.container_logs(id, usize::MAX).await
    }

    async fn get_container_pid(&self, id: &ContainerId) -> Result<Option<u32>> {
        if let Some(pid) = self.pids.read().await.get(id).copied() {
            return Ok(Some(pid));
        }
        // Fallback: query youki state and update the cache.
        match self.query_state(id).await {
            Ok(state) => {
                if let Some(pid) = state.pid {
                    self.pids.write().await.insert(id.clone(), pid);
                    Ok(Some(pid))
                } else {
                    Ok(None)
                }
            }
            Err(AgentError::NotFound { .. }) => Ok(None),
            Err(e) => Err(e),
        }
    }

    async fn get_container_ip(&self, id: &ContainerId) -> Result<Option<IpAddr>> {
        // The IP is recorded only by `push_overlay_config` once the guest
        // WireGuard device is actually live in the container netns (or by an
        // out-of-band `record_container_ip`), so whatever is present is real —
        // there is no host-network fallback that could leave a stale IP.
        Ok(self.ips.read().await.get(id).copied())
    }

    fn overlay_attach_kind(&self) -> OverlayAttachKind {
        // A WSL2 distro is a Linux guest with no host-visible netns/PID the
        // Windows host can plumb a veth into. The service layer instead asks
        // overlayd for a guest-managed config and pushes it into the distro via
        // `push_overlay_config`, where a kernel WireGuard device is brought up
        // inside the container's named netns — exactly like the macOS VZ-Linux
        // runtime.
        OverlayAttachKind::GuestManaged
    }

    async fn push_overlay_config(
        &self,
        id: &ContainerId,
        config: &zlayer_types::overlayd::GuestOverlayConfig,
    ) -> Result<()> {
        let slug = id_slug(id);
        let wg_iface = Self::wg_iface_name(&slug);
        match self
            .push_overlay_config_inner(id, &slug, &wg_iface, config)
            .await
        {
            Ok(()) => {
                // Record the overlay IP so `get_container_ip` reports it (mesh
                // routing / DNS prefer the overlay address) and mark the netns
                // configured so teardown cleans the WireGuard device.
                self.ips.write().await.insert(id.clone(), config.overlay_ip);
                self.netns.write().await.insert(
                    id.clone(),
                    NetnsState::Configured {
                        ip: config.overlay_ip,
                    },
                );
                tracing::info!(
                    container = %id,
                    overlay_ip = %config.overlay_ip,
                    wg_iface = %wg_iface,
                    "configured WSL2 guest overlay WireGuard device"
                );
                Ok(())
            }
            Err(e) => {
                // Best-effort cleanup of any partial WireGuard device so the
                // service layer's allocation rollback isn't shadowed by a leak.
                let _ = self.wsl_run("ip", &["link", "delete", &wg_iface]).await;
                let _ = self
                    .wsl_run(
                        "ip",
                        &["netns", "exec", &slug, "ip", "link", "delete", &wg_iface],
                    )
                    .await;
                Err(e)
            }
        }
    }

    async fn list_images(&self) -> Result<Vec<ImageInfo>> {
        // Youki doesn't (yet) ship an image-list subcommand that we can rely
        // on. Be explicit about the unsupported surface rather than pretending
        // to return a successful empty list.
        Err(AgentError::Unsupported(
            "list_images is not supported by the WSL2 delegate runtime \
             (youki has no image registry; images are managed on the host)"
                .to_string(),
        ))
    }

    async fn remove_image(&self, _image: &str, _force: bool) -> Result<()> {
        Err(AgentError::Unsupported(
            "remove_image is not supported by the WSL2 delegate runtime".to_string(),
        ))
    }

    async fn prune_images(&self) -> Result<PruneResult> {
        Err(AgentError::Unsupported(
            "prune_images is not supported by the WSL2 delegate runtime".to_string(),
        ))
    }

    async fn kill_container(&self, id: &ContainerId, signal: Option<&str>) -> Result<()> {
        let canonical = validate_signal(signal.unwrap_or("SIGKILL"))?;
        let slug = id_slug(id);
        let output = self.zlayer_runtime(&["kill", &slug, &canonical]).await?;
        if output.status.success() {
            Ok(())
        } else {
            Err(zlayer_runtime_error("kill", &output))
        }
    }

    async fn tag_image(&self, _source: &str, _target: &str) -> Result<()> {
        Err(AgentError::Unsupported(
            "tag_image is not supported by the WSL2 delegate runtime".to_string(),
        ))
    }

    async fn inspect_detailed(&self, id: &ContainerId) -> Result<ContainerInspectDetails> {
        // wsl2 keeps no ServiceSpec/port-map/network record post-create, so
        // ports/networks/health are genuinely unavailable (same as HcsRuntime,
        // hcs.rs:5520). We DO surface the two fields wsl2 can source live: the
        // youki exit code and the overlay IP.
        let exit_code = match self.query_state(id).await {
            Ok(state) => state.exit_code,
            Err(e @ AgentError::NotFound { .. }) => return Err(e),
            Err(_) => None,
        };
        let ipv4 = self.get_container_ip(id).await?.map(|ip| ip.to_string());
        Ok(ContainerInspectDetails {
            ports: Vec::new(),
            networks: Vec::new(),
            ipv4,
            health: None,
            exit_code,
        })
    }

    /// Real line-by-line streaming over the `wsl.exe` boundary.
    ///
    /// Spawns `wsl.exe -d <distro> -- <runtime_binary> oci --state-root
    /// <root> exec <slug> -- <cmd...>` with piped stdout and stderr, then
    /// drives two `BufReader::lines()` loops on background tasks. Each line
    /// becomes one [`ExecEvent::Stdout`] / [`ExecEvent::Stderr`]; once both
    /// readers reach EOF and the child process exits, a final
    /// [`ExecEvent::Exit`] is emitted and the stream closes. Errors surfaced
    /// before the child is spawned (empty cmd, `wsl.exe` not launchable)
    /// flow through the outer `Result`; post-spawn errors are logged and
    /// the stream closes with `ExecEvent::Exit(-1)`.
    async fn exec_stream(&self, id: &ContainerId, cmd: &[String]) -> Result<ExecEventStream> {
        if cmd.is_empty() {
            return Err(AgentError::InvalidSpec(
                "exec command must not be empty".to_string(),
            ));
        }

        let slug = id_slug(id);
        // Build the `wsl.exe` argv:
        //   `-d <distro> -- <runtime_binary> runtime --state-root <root>
        //    exec <slug> -- <user_cmd...>`
        // Everything is owned strings because we hand them off to a background
        // task below and can't rely on the &[&str] borrow outliving the spawn.
        let state_root = self.config.oci_state_root.to_string_lossy().into_owned();
        let mut argv: Vec<String> = Vec::with_capacity(10 + cmd.len());
        argv.push("-d".to_string());
        argv.push(self.config.distro.clone());
        argv.push("--".to_string());
        argv.push(self.config.runtime_binary.clone());
        argv.push("runtime".to_string());
        argv.push("--state-root".to_string());
        argv.push(state_root);
        argv.push("exec".to_string());
        argv.push(slug);
        argv.push("--".to_string());
        argv.extend(cmd.iter().cloned());

        let mut child = tokio::process::Command::new("wsl.exe")
            .args(&argv)
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .map_err(|e| AgentError::Network(format!("wsl.exe spawn for exec_stream: {e}")))?;

        let stdout = child.stdout.take().ok_or_else(|| {
            AgentError::Internal("wsl.exe child did not expose a stdout handle".to_string())
        })?;
        let stderr = child.stderr.take().ok_or_else(|| {
            AgentError::Internal("wsl.exe child did not expose a stderr handle".to_string())
        })?;

        Ok(spawn_exec_event_stream(child, stdout, stderr))
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Execute `cmd` inside a specific WSL2 distro, converting any transport
/// error into [`AgentError::Network`]. Used by [`Wsl2DelegateRuntime`]'s
/// per-instance helpers so the distro name is threaded through configuration
/// rather than hardcoded against `zlayer_wsl::distro::DISTRO_NAME`.
///
/// This is the config-aware replacement for the old `wsl_exec_or` helper
/// (which wrapped [`zlayer_wsl::distro::wsl_exec`], a function hardcoded to
/// the `zlayer` distro).
async fn wsl_exec_in(distro: &str, cmd: &str, args: &[&str]) -> Result<Output> {
    let mut wsl_args: Vec<&str> = vec!["-d", distro, "--", cmd];
    wsl_args.extend_from_slice(args);
    tokio::process::Command::new("wsl.exe")
        .args(&wsl_args)
        .output()
        .await
        .map_err(|e| AgentError::Network(format!("wsl.exe -d {distro} -- {cmd}: {e}")))
}

/// Wire up the background tasks that turn a spawned `wsl.exe` child's
/// stdout + stderr pipes into an ordered [`ExecEventStream`].
///
/// Factored out of [`Wsl2DelegateRuntime::exec_stream`] so the task graph —
/// two line-reader tasks plus one waiter that sends the terminal
/// [`ExecEvent::Exit`] — is named and unit-testable. The returned stream
/// always terminates with exactly one `ExecEvent::Exit`.
fn spawn_exec_event_stream(
    mut child: tokio::process::Child,
    stdout: tokio::process::ChildStdout,
    stderr: tokio::process::ChildStderr,
) -> ExecEventStream {
    let (tx, rx) = mpsc::channel::<ExecEvent>(128);

    let tx_stdout = tx.clone();
    let stdout_task = tokio::spawn(async move {
        let mut reader = BufReader::new(stdout).lines();
        loop {
            match reader.next_line().await {
                Ok(Some(line)) => {
                    if tx_stdout.send(ExecEvent::Stdout(line)).await.is_err() {
                        break;
                    }
                }
                Ok(None) => break,
                Err(e) => {
                    tracing::warn!(error = %e, "exec_stream: stdout read error");
                    break;
                }
            }
        }
    });

    let tx_stderr = tx.clone();
    let stderr_task = tokio::spawn(async move {
        let mut reader = BufReader::new(stderr).lines();
        loop {
            match reader.next_line().await {
                Ok(Some(line)) => {
                    if tx_stderr.send(ExecEvent::Stderr(line)).await.is_err() {
                        break;
                    }
                }
                Ok(None) => break,
                Err(e) => {
                    tracing::warn!(error = %e, "exec_stream: stderr read error");
                    break;
                }
            }
        }
    });

    // Supervisor: wait for both readers to drain, reap the child, and emit
    // the terminal `Exit` event. Keeping `tx` alive in *this* task (and only
    // this task) is load-bearing — it's how the receiver side observes
    // stream termination after the readers have dropped their clones.
    tokio::spawn(async move {
        let _ = stdout_task.await;
        let _ = stderr_task.await;
        let exit_code = match child.wait().await {
            Ok(status) => status.code().unwrap_or(-1),
            Err(e) => {
                tracing::warn!(
                    error = %e,
                    "wsl.exe exec_stream child wait failed; reporting exit -1"
                );
                -1
            }
        };
        // `send` fails only when the receiver has been dropped — benign.
        let _ = tx.send(ExecEvent::Exit(exit_code)).await;
    });

    Box::pin(ReceiverStream::new(rx))
}

/// Spawn `wsl.exe` with the supplied argv and pipe `stdin_bytes` into its
/// standard input, returning an error if the process exits non-zero or
/// stdin cannot be written.
///
/// The existing [`zlayer_wsl::distro::wsl_exec`] helper uses `Command::output`
/// which does not expose stdin; we need real stdin streaming for
/// `tee config.json` + `tar -xf -`. Lives inside this module (rather than
/// as a cross-crate helper in `zlayer-wsl`) because it is only load-bearing
/// for the WSL2 delegate today.
async fn wsl_stdin_pipe(args: &[&str], stdin_bytes: &[u8]) -> std::io::Result<()> {
    let mut child = tokio::process::Command::new("wsl.exe")
        .args(args)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;

    if let Some(mut stdin) = child.stdin.take() {
        stdin.write_all(stdin_bytes).await?;
        stdin.shutdown().await?;
    } else {
        return Err(std::io::Error::other(
            "wsl.exe child did not expose a stdin handle",
        ));
    }

    let output = child.wait_with_output().await?;
    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        return Err(std::io::Error::other(format!(
            "wsl.exe exited with {:?}: {}",
            output.status.code(),
            stderr.trim()
        )));
    }
    Ok(())
}

/// Stream-decompress a layer FILE and pipe the raw tarball through
/// `wsl.exe -- tar -xf -`, never holding the whole (de)compressed layer in RAM.
///
/// Compression is taken from `media_type`, peeking magic bytes when unknown —
/// the same detection [`zlayer_registry::LayerUnpacker`] uses.
async fn wsl_extract_layer_file(
    args: &[&str],
    layer_path: &std::path::Path,
    media_type: &str,
) -> std::io::Result<()> {
    use std::io::{BufRead as _, Read as _};

    let mut child = tokio::process::Command::new("wsl.exe")
        .args(args)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;

    let mut stdin = child
        .stdin
        .take()
        .ok_or_else(|| std::io::Error::other("wsl.exe child did not expose a stdin handle"))?;

    // Synchronous decompressing reader over the layer file.
    let file = std::fs::File::open(layer_path)?;
    let mut buffered = std::io::BufReader::new(file);
    let compression = match CompressionType::from_media_type(media_type) {
        Some(c) => c,
        None => CompressionType::from_magic_bytes(buffered.fill_buf()?),
    };
    // `+ Send`: this reader is held across the `stdin.write_all(...).await`
    // below, so the enclosing async fn's future must be `Send` (the async-trait
    // method requires it). `BufReader`/`GzDecoder`/zstd `Decoder` over a `Send`
    // inner are all `Send`.
    let mut reader: Box<dyn std::io::Read + Send> = match compression {
        CompressionType::None => Box::new(buffered),
        CompressionType::Gzip => Box::new(flate2::read::GzDecoder::new(buffered)),
        CompressionType::Zstd => Box::new(zstd::stream::Decoder::new(buffered)?),
    };

    let mut buf = vec![0u8; 1024 * 1024];
    loop {
        let n = reader.read(&mut buf)?;
        if n == 0 {
            break;
        }
        stdin.write_all(&buf[..n]).await?;
    }
    stdin.shutdown().await?;
    drop(stdin);

    let output = child.wait_with_output().await?;
    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        return Err(std::io::Error::other(format!(
            "wsl.exe exited with {:?}: {}",
            output.status.code(),
            stderr.trim()
        )));
    }
    Ok(())
}

/// Build a conventional `AgentError::Network` describing a nonzero
/// `zlayer runtime <subcommand>` exit.
///
/// Format: `zlayer runtime <subcommand> failed (status <code>): <stderr>`.
/// Never swallows stderr so the user sees both the command and the distro's
/// own diagnostic output.
fn zlayer_runtime_error(subcommand: &str, output: &Output) -> AgentError {
    let status = output.status.code();
    let stderr = String::from_utf8_lossy(&output.stderr);
    AgentError::Network(format!(
        "zlayer runtime {subcommand} failed (status {status:?}): {}",
        stderr.trim()
    ))
}

/// Produce a youki-compatible slug for a [`ContainerId`].
///
/// `ContainerId::Display` emits `"{service}-rep-{replica}"`, which is already
/// alphanumeric-plus-dashes except for the degenerate case where a service
/// name contains characters youki rejects (whitespace, quotes, slashes). This
/// helper filters those so a malformed service spec can't break shell-out
/// argument quoting.
fn id_slug(id: &ContainerId) -> String {
    let raw = id.to_string();
    raw.chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
                c
            } else {
                '-'
            }
        })
        .collect()
}

/// Parse the gateway IP out of `ip route show default` output.
///
/// The line looks like `default via 172.20.16.1 dev eth0 ...`; we return the
/// token immediately after `via`. Returns `None` if no default route / no
/// `via` clause is present.
fn parse_default_gateway(route_output: &str) -> Option<String> {
    for line in route_output.lines() {
        let mut prev = "";
        for tok in line.split_whitespace() {
            if prev == "via" {
                return Some(tok.to_string());
            }
            prev = tok;
        }
    }
    None
}

/// Rewrite the host portion of a `host:port` `WireGuard` endpoint to `gateway`,
/// preserving the port. An empty endpoint (a roaming peer) is left as-is, and an
/// unparseable endpoint is returned unchanged. Mirrors the macOS VZ-Linux
/// runtime's helper of the same name.
fn rewrite_endpoint_host(endpoint: &str, gateway: &str) -> String {
    if endpoint.is_empty() {
        return String::new();
    }
    match endpoint.rsplit_once(':') {
        Some((_host, port)) => format!("{gateway}:{port}"),
        None => endpoint.to_string(),
    }
}

/// Single-quote a string for safe embedding inside a `sh -c "..."` command,
/// escaping any embedded single quotes as `'\''`. Used for the `WireGuard`
/// private key + the generated resolv.conf contents, neither of which normally
/// contains a quote — but the escaping keeps a hostile value from breaking out.
fn sh_single_quote(s: &str) -> String {
    format!("'{}'", s.replace('\'', "'\\''"))
}

/// Subset of `youki state` output we care about. Matches the runtime-spec
/// `State` document, which youki emits verbatim.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
struct YoukiState {
    /// `creating` | `created` | `running` | `stopped`.
    status: String,
    /// PID of the init process inside the container (absent for `creating`).
    #[serde(default)]
    pid: Option<u32>,
    /// Exit code — youki adds this on top of the spec when status is
    /// `stopped`. `None` while the container is still alive.
    #[serde(default)]
    exit_code: Option<i32>,
}

impl YoukiState {
    fn is_stopped(&self) -> bool {
        self.status.eq_ignore_ascii_case("stopped")
    }

    fn as_container_state(&self) -> ContainerState {
        match self.status.to_ascii_lowercase().as_str() {
            "creating" => ContainerState::Pending,
            "created" => ContainerState::Initializing,
            "running" => ContainerState::Running,
            "stopped" => ContainerState::Exited {
                code: self.exit_code.unwrap_or(0),
            },
            other => ContainerState::Failed {
                reason: format!("unknown youki state: {other}"),
            },
        }
    }
}

/// Parse the JSON payload emitted by `zlayer runtime state <id>`.
fn parse_youki_state(output: &Output) -> Result<YoukiState> {
    let stdout = std::str::from_utf8(&output.stdout).map_err(|e| {
        AgentError::Internal(format!("zlayer runtime state: stdout not utf-8: {e}"))
    })?;
    serde_json::from_str::<YoukiState>(stdout.trim()).map_err(|e| {
        AgentError::Internal(format!(
            "zlayer runtime state: failed to parse JSON: {e} (raw: {:?})",
            stdout.chars().take(256).collect::<String>()
        ))
    })
}

/// Best-effort parser for the JSON `youki events --stats <id>` payload.
///
/// Youki follows the `runc`-compatible shape: `{"cpu":{"usage":{"total":N}},
/// "memory":{"usage":{"usage":N,"limit":M}}}`. We tolerate missing fields by
/// defaulting to zero / `u64::MAX` so callers never see a malformed-payload
/// error for a metrics sample.
fn parse_youki_stats(output: &Output, timestamp: std::time::Instant) -> ContainerStats {
    let raw = String::from_utf8_lossy(&output.stdout);
    let v: serde_json::Value = match serde_json::from_str(raw.trim()) {
        Ok(v) => v,
        Err(_) => {
            return ContainerStats {
                cpu_usage_usec: 0,
                memory_bytes: 0,
                memory_limit: u64::MAX,
                timestamp,
            };
        }
    };
    // CPU total is in nanoseconds for runc-style stats; convert to usec.
    let cpu_ns = v
        .pointer("/cpu/usage/total")
        .and_then(serde_json::Value::as_u64)
        .unwrap_or(0);
    let cpu_usage_usec = cpu_ns / 1_000;
    let memory_bytes = v
        .pointer("/memory/usage/usage")
        .and_then(serde_json::Value::as_u64)
        .unwrap_or(0);
    let memory_limit = v
        .pointer("/memory/usage/limit")
        .and_then(serde_json::Value::as_u64)
        .unwrap_or(u64::MAX);
    ContainerStats {
        cpu_usage_usec,
        memory_bytes,
        memory_limit,
        timestamp,
    }
}

// ---------------------------------------------------------------------------
// WSL2 GPU exposure (Phase 5.D)
//
// When a service spec carries `resources.gpu`, the youki bundle running inside
// the WSL2 distro needs three things mirrored from how WSLg exposes GPU to a
// regular user shell:
//
//   1. `/dev/dxg`            — the WSL DirectX kernel interface, the actual
//                              entry point apps talk to for GPU work.
//   2. `/usr/lib/wsl/`       — the WSLg shim library tree (`libdxcore.so`,
//                              `libd3d12.so`, `libdxguid.so`).
//   3. `/usr/lib/wsl/drivers/` (NVIDIA-only) — the NVIDIA WDDM driver shim that
//                              CUDA libraries resolve through.
//
// On the host (Windows) we can't touch any of these directly; everything
// lives inside the WSL2 distro. So the probe trait below shells out to
// `wsl.exe -d <distro> -- ...` to stat `/dev/dxg` and check the lib trees.
// The pure mutation helper [`inject_wsl_gpu_mounts`] takes the probe's
// answers as inputs so it can be unit-tested without any real WSL2 host.
// ---------------------------------------------------------------------------

/// Result of probing the WSL2 distro for GPU readiness.
///
/// Populated by an impl of [`WslGpuHostProbe`] before
/// [`inject_wsl_gpu_mounts`] runs so the mutation helper stays purely
/// in-memory and testable.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct WslGpuHostState {
    /// `(major, minor)` for `/dev/dxg` inside the distro, or `None` if the
    /// device node is absent. `None` is a hard error when GPU was requested:
    /// silently falling back to CPU is surprising.
    pub dxg_devno: Option<(i64, i64)>,
    /// Whether `/usr/lib/wsl` is present inside the distro. When false the
    /// shim libraries are missing and GPU work will fail at link time even
    /// with `/dev/dxg` mounted.
    pub wsl_lib_present: bool,
    /// Whether `/usr/lib/wsl/drivers` is present inside the distro. This is
    /// the NVIDIA-specific driver shim path; absent on AMD/Intel hosts.
    pub wsl_drivers_present: bool,
}

/// Probe trait. Lets unit tests substitute a stub that doesn't need a real
/// WSL2 host with `/dev/dxg` exposed.
#[async_trait]
pub(crate) trait WslGpuHostProbe: Send + Sync {
    async fn probe(&self) -> Result<WslGpuHostState>;
}

/// Production probe that shells out via [`Wsl2DelegateRuntime::wsl`] to read
/// the actual state of the configured distro.
struct DefaultWslGpuHostProbe<'a> {
    runtime: &'a Wsl2DelegateRuntime,
}

#[async_trait]
impl WslGpuHostProbe for DefaultWslGpuHostProbe<'_> {
    async fn probe(&self) -> Result<WslGpuHostState> {
        // `stat -c '%t %T' /dev/dxg` prints major/minor as hex. We swallow
        // stat's nonzero exit (file missing) and report devno=None, which
        // `inject_wsl_gpu_mounts` translates into `WslGpuUnavailable`.
        let dxg_devno = match self.runtime.wsl("stat", &["-c", "%t %T", "/dev/dxg"]).await {
            Ok(output) if output.status.success() => {
                let raw = String::from_utf8_lossy(&output.stdout);
                parse_stat_hex_devno(raw.trim())
            }
            _ => None,
        };

        // `test -d` returns 0 when the directory exists. Both -d probes are
        // best-effort: a failure to spawn `wsl.exe` is the same as the path
        // being absent — we'd fail later anyway.
        let wsl_lib_present = self
            .runtime
            .wsl("test", &["-d", "/usr/lib/wsl"])
            .await
            .map(|o| o.status.success())
            .unwrap_or(false);
        let wsl_drivers_present = self
            .runtime
            .wsl("test", &["-d", "/usr/lib/wsl/drivers"])
            .await
            .map(|o| o.status.success())
            .unwrap_or(false);

        Ok(WslGpuHostState {
            dxg_devno,
            wsl_lib_present,
            wsl_drivers_present,
        })
    }
}

/// Parse the output of `stat -c '%t %T' <path>`: two whitespace-separated
/// hexadecimal numbers (major, then minor). Returns `None` on any parse
/// failure so callers treat malformed stat output as "device missing".
fn parse_stat_hex_devno(s: &str) -> Option<(i64, i64)> {
    let mut parts = s.split_ascii_whitespace();
    let major = i64::from_str_radix(parts.next()?, 16).ok()?;
    let minor = i64::from_str_radix(parts.next()?, 16).ok()?;
    Some((major, minor))
}

/// Inject `/dev/dxg` + the `WSLg` shim mounts into a bundle's mounts/devices
/// and prepend the `WSLg` lib paths to `LD_LIBRARY_PATH`.
///
/// Returns `Err(AgentError::WslGpuUnavailable)` when GPU was requested but
/// the host cannot deliver it (no `/dev/dxg`). Idempotent w.r.t. existing
/// user-supplied `/dev/dxg` entries: if the spec's device list already names
/// `/dev/dxg`, the helper leaves that entry alone instead of duplicating.
pub(crate) fn inject_wsl_gpu_mounts(
    mounts: &mut Vec<Mount>,
    env: &mut Vec<String>,
    devices: &mut Vec<LinuxDevice>,
    _gpu_spec: &GpuSpec,
    host: &WslGpuHostState,
) -> Result<()> {
    let (major, minor) = host
        .dxg_devno
        .ok_or_else(|| AgentError::WslGpuUnavailable {
            reason: "/dev/dxg is not exposed by the WSL2 kernel inside the configured distro; \
                 enable WSL2 GPU support (Windows 11 + a recent WSL kernel) or drop \
                 `resources.gpu` from the service spec"
                .to_string(),
        })?;

    let has_dxg_mount = mounts
        .iter()
        .any(|m| m.destination().as_path() == std::path::Path::new("/dev/dxg"));
    if !has_dxg_mount {
        let mount = MountBuilder::default()
            .destination("/dev/dxg".to_string())
            .source("/dev/dxg".to_string())
            .typ("bind".to_string())
            .options(vec!["bind".to_string(), "rw".to_string()])
            .build()
            .map_err(|e| AgentError::WslGpuUnavailable {
                reason: format!("failed to build /dev/dxg bind mount: {e}"),
            })?;
        mounts.push(mount);
    }

    let has_dxg_device = devices
        .iter()
        .any(|d| d.path().as_path() == std::path::Path::new("/dev/dxg"));
    if !has_dxg_device {
        let device = LinuxDeviceBuilder::default()
            .path("/dev/dxg")
            .typ(LinuxDeviceType::C)
            .major(major)
            .minor(minor)
            .file_mode(0o666u32)
            .uid(0u32)
            .gid(0u32)
            .build()
            .map_err(|e| AgentError::WslGpuUnavailable {
                reason: format!("failed to build /dev/dxg device node: {e}"),
            })?;
        devices.push(device);
    }

    // /usr/lib/wsl read-only shim mount.
    if host.wsl_lib_present {
        let has_wsl_lib_mount = mounts
            .iter()
            .any(|m| m.destination().as_path() == std::path::Path::new("/usr/lib/wsl"));
        if !has_wsl_lib_mount {
            let mount = MountBuilder::default()
                .destination("/usr/lib/wsl".to_string())
                .source("/usr/lib/wsl".to_string())
                .typ("bind".to_string())
                .options(vec!["bind".to_string(), "ro".to_string()])
                .build()
                .map_err(|e| AgentError::WslGpuUnavailable {
                    reason: format!("failed to build /usr/lib/wsl bind mount: {e}"),
                })?;
            mounts.push(mount);
        }
    } else {
        tracing::warn!(
            "WSL2 GPU: /usr/lib/wsl missing on host; container will see /dev/dxg but no \
             WSLg shim libraries (libdxcore.so etc.). GPU workloads will fail at dlopen."
        );
    }

    // Prepend WSLg lib paths to LD_LIBRARY_PATH (drivers ahead of lib so the
    // NVIDIA driver shim wins lookup ordering). Preserve any existing value.
    let mut new_prefix: Vec<&str> = Vec::new();
    if host.wsl_drivers_present {
        new_prefix.push("/usr/lib/wsl/drivers");
    }
    if host.wsl_lib_present {
        new_prefix.push("/usr/lib/wsl/lib");
    }
    if !new_prefix.is_empty() {
        let prefix_joined = new_prefix.join(":");
        if let Some(entry) = env.iter_mut().find(|e| e.starts_with("LD_LIBRARY_PATH=")) {
            let existing = entry.split_once('=').map_or("", |(_, v)| v).to_string();
            *entry = if existing.is_empty() {
                format!("LD_LIBRARY_PATH={prefix_joined}")
            } else {
                format!("LD_LIBRARY_PATH={prefix_joined}:{existing}")
            };
        } else {
            env.push(format!("LD_LIBRARY_PATH={prefix_joined}"));
        }
    }

    Ok(())
}

/// Mutates the already-built [`Spec`] in place to inject the WSL2 GPU
/// mounts/devices/env when `resources.gpu` is set. No-op when the spec
/// doesn't request a GPU.
async fn apply_wsl_gpu_to_spec(
    oci_spec: &mut Spec,
    service: &ServiceSpec,
    probe: &dyn WslGpuHostProbe,
) -> Result<()> {
    let Some(gpu_spec) = service.resources.gpu.as_ref() else {
        return Ok(());
    };

    let host = probe.probe().await?;

    // Pull the three slices we need to mutate out of the Spec via the getset
    // accessors. Each is wrapped in Option<Vec<...>>; we materialise empty
    // vecs as needed so the helper can push without further branching.
    let mut mounts = oci_spec.mounts().clone().unwrap_or_default();
    let mut env = oci_spec
        .process()
        .as_ref()
        .and_then(|p| p.env().clone())
        .unwrap_or_default();
    let mut devices = oci_spec
        .linux()
        .as_ref()
        .and_then(|l| l.devices().clone())
        .unwrap_or_default();

    inject_wsl_gpu_mounts(&mut mounts, &mut env, &mut devices, gpu_spec, &host)?;

    oci_spec.set_mounts(Some(mounts));
    if let Some(process) = oci_spec.process_mut().as_mut() {
        process.set_env(Some(env));
    }
    if let Some(linux) = oci_spec.linux_mut().as_mut() {
        linux.set_devices(Some(devices));
    }

    Ok(())
}

// ---------------------------------------------------------------------------
// Tests — pure-logic coverage only. Anything that shells out to wsl.exe is
// tested by the Windows-gated integration suite in Phase F-9.
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use std::net::Ipv4Addr;
    #[cfg(not(target_os = "windows"))]
    use std::os::unix::process::ExitStatusExt;
    #[cfg(target_os = "windows")]
    use std::os::windows::process::ExitStatusExt;
    use std::process::ExitStatus;
    use std::sync::Mutex as StdMutex;

    fn cid(service: &str, replica: u32) -> ContainerId {
        ContainerId::new(service, replica)
    }

    #[cfg(target_os = "windows")]
    fn make_exit_status(code: i32) -> ExitStatus {
        // On Windows, `ExitStatus::from_raw` takes the raw Windows process
        // exit code as `u32`; negative i32 values are reinterpreted
        // bit-for-bit, which is what we want for synthesising realistic
        // status payloads in tests.
        #[allow(clippy::cast_sign_loss)]
        let raw = code as u32;
        ExitStatus::from_raw(raw)
    }

    #[cfg(not(target_os = "windows"))]
    fn make_exit_status(code: i32) -> ExitStatus {
        // On Unix, `ExitStatus::from_raw` takes the raw wait-status `i32`.
        // Shift so the low byte encodes our synthesised code (Unix packs
        // the exit code in bits 8..15).
        ExitStatus::from_raw(code << 8)
    }

    fn fake_output(stdout: &str, code: i32) -> Output {
        Output {
            status: make_exit_status(code),
            stdout: stdout.as_bytes().to_vec(),
            stderr: Vec::new(),
        }
    }

    fn fake_output_err(stderr: &str, code: i32) -> Output {
        Output {
            status: make_exit_status(code),
            stdout: Vec::new(),
            stderr: stderr.as_bytes().to_vec(),
        }
    }

    /// Shared call log: one entry per `wsl.exe` invocation, captured as
    /// `(cmd, args)` so tests can assert the exact command sequence.
    type CallLog = Arc<StdMutex<Vec<(String, Vec<String>)>>>;

    /// Recording fake [`WslRunner`] used to verify the sequence of commands
    /// emitted by `push_overlay_config` / `teardown_container_netns`.
    struct RecordingRunner {
        calls: CallLog,
        /// Per-(cmd, args-join) canned responses. If a call is not present
        /// in the map we return a success code with empty output, which is
        /// the right default for the happy path.
        responses: StdMutex<HashMap<String, Output>>,
    }

    impl RecordingRunner {
        fn new() -> Self {
            Self {
                calls: Arc::new(StdMutex::new(Vec::new())),
                responses: StdMutex::new(HashMap::new()),
            }
        }

        fn calls_handle(&self) -> CallLog {
            Arc::clone(&self.calls)
        }

        fn key(cmd: &str, args: &[&str]) -> String {
            let mut k = String::from(cmd);
            for a in args {
                k.push(' ');
                k.push_str(a);
            }
            k
        }

        /// Install a canned response for the exact `cmd args...` invocation.
        fn set_response(&self, cmd: &str, args: &[&str], output: Output) {
            self.responses
                .lock()
                .expect("responses mutex poisoned")
                .insert(Self::key(cmd, args), output);
        }
    }

    #[async_trait]
    impl WslRunner for RecordingRunner {
        async fn run(&self, cmd: &str, args: &[&str]) -> anyhow::Result<Output> {
            self.calls.lock().expect("calls mutex poisoned").push((
                cmd.to_string(),
                args.iter().map(|s| (*s).to_string()).collect(),
            ));
            let key = Self::key(cmd, args);
            if let Some(out) = self
                .responses
                .lock()
                .expect("responses mutex poisoned")
                .remove(&key)
            {
                Ok(out)
            } else {
                Ok(fake_output("", 0))
            }
        }
    }

    /// Default `ResolvedConfig` for tests that don't care about the exact
    /// distro name / paths.
    fn default_resolved_config() -> ResolvedConfig {
        ResolvedConfig {
            distro: zlayer_wsl::distro::DISTRO_NAME.to_string(),
            runtime_binary: DEFAULT_RUNTIME_BINARY.to_string(),
            bundle_root: DEFAULT_BUNDLE_ROOT.to_string(),
            log_root: DEFAULT_LOG_ROOT.to_string(),
            oci_state_root: PathBuf::from(DEFAULT_OCI_STATE_ROOT),
        }
    }

    /// Build a runtime with a fully resolved config + a caller-supplied
    /// runner. Used by the J-3 netns tests so the in-memory recording
    /// runner captures the emitted `ip` commands.
    fn test_runtime_with_runner(
        config: ResolvedConfig,
        runner: Arc<dyn WslRunner>,
    ) -> Wsl2DelegateRuntime {
        Wsl2DelegateRuntime {
            config,
            pids: Arc::new(RwLock::new(HashMap::new())),
            ips: Arc::new(RwLock::new(HashMap::new())),
            bundle_roots: Arc::new(RwLock::new(HashMap::new())),
            image_cache: Arc::new(RwLock::new(HashMap::new())),
            netns: Arc::new(RwLock::new(HashMap::new())),
            runner,
            auth_context: None,
            secrets_provider: parking_lot::RwLock::new(None),
        }
    }

    /// Shorthand for the G-2/G-4/G-5 unit tests that don't exercise the
    /// J-3 runner plumbing. Uses the default (production) runner — those
    /// tests don't actually dispatch `ip` commands.
    fn test_runtime(config: ResolvedConfig) -> Wsl2DelegateRuntime {
        test_runtime_with_runner(config, Arc::new(DefaultWslRunner))
    }

    /// Shorthand for the J-3 tests that want a default-config runtime with
    /// an injected recording runner.
    fn make_runtime(runner: Arc<dyn WslRunner>) -> Wsl2DelegateRuntime {
        test_runtime_with_runner(default_resolved_config(), runner)
    }

    #[test]
    fn id_slug_sanitizes_special_chars() {
        // Normal ContainerId formats produce an already-safe slug.
        assert_eq!(id_slug(&cid("web", 0)), "web-rep-0");

        // Weird service names get their offending characters replaced with
        // dashes so youki never sees a shell metacharacter.
        let weird = cid("my svc/with:quotes", 3);
        let slug = id_slug(&weird);
        assert!(
            slug.chars()
                .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_'),
            "slug must be alnum/dash/underscore only: {slug}"
        );
        assert!(
            slug.contains("my-svc"),
            "slug should preserve alnum: {slug}"
        );
        assert!(slug.ends_with("-rep-3"), "slug should keep replica: {slug}");
    }

    #[test]
    fn parse_youki_state_running() {
        let raw = r#"{"ociVersion":"1.0.2","id":"web-rep-0","status":"running","pid":12345,"bundle":"/var/lib/zlayer/bundles/web-rep-0"}"#;
        let out = fake_output(raw, 0);
        let state = parse_youki_state(&out).expect("valid state JSON");
        assert_eq!(state.status, "running");
        assert_eq!(state.pid, Some(12345));
        assert_eq!(state.exit_code, None);
        assert!(!state.is_stopped());
        assert_eq!(state.as_container_state(), ContainerState::Running);
    }

    #[test]
    fn parse_youki_state_stopped() {
        let raw = r#"{"ociVersion":"1.0.2","id":"job-rep-0","status":"stopped","exitCode":42}"#;
        let out = fake_output(raw, 0);
        let state = parse_youki_state(&out).expect("valid state JSON");
        assert_eq!(state.status, "stopped");
        assert_eq!(state.exit_code, Some(42));
        assert!(state.is_stopped());
        assert_eq!(
            state.as_container_state(),
            ContainerState::Exited { code: 42 }
        );
    }

    #[test]
    fn parse_youki_state_creating_maps_to_pending() {
        let raw = r#"{"ociVersion":"1.0.2","id":"x","status":"creating"}"#;
        let out = fake_output(raw, 0);
        let state = parse_youki_state(&out).expect("valid state JSON");
        assert_eq!(state.as_container_state(), ContainerState::Pending);
    }

    #[test]
    fn parse_youki_state_unknown_maps_to_failed() {
        let raw = r#"{"ociVersion":"1.0.2","id":"x","status":"paused"}"#;
        let out = fake_output(raw, 0);
        let state = parse_youki_state(&out).expect("valid state JSON");
        assert!(matches!(
            state.as_container_state(),
            ContainerState::Failed { .. }
        ));
    }

    #[test]
    fn parse_youki_stats_handles_full_payload() {
        let raw = r#"{"cpu":{"usage":{"total":2500000000}},"memory":{"usage":{"usage":104857600,"limit":268435456}}}"#;
        let out = fake_output(raw, 0);
        let stats = parse_youki_stats(&out, std::time::Instant::now());
        // 2_500_000_000 ns / 1_000 = 2_500_000 usec.
        assert_eq!(stats.cpu_usage_usec, 2_500_000);
        assert_eq!(stats.memory_bytes, 104_857_600);
        assert_eq!(stats.memory_limit, 268_435_456);
    }

    #[test]
    fn parse_youki_stats_defaults_on_malformed() {
        let out = fake_output("not json", 0);
        let stats = parse_youki_stats(&out, std::time::Instant::now());
        assert_eq!(stats.cpu_usage_usec, 0);
        assert_eq!(stats.memory_bytes, 0);
        assert_eq!(stats.memory_limit, u64::MAX);
    }

    #[tokio::test]
    async fn record_container_ip_then_get_returns_it() {
        let runtime = make_runtime(Arc::new(RecordingRunner::new()));
        let id = cid("web", 2);
        let ip = IpAddr::V4(Ipv4Addr::new(10, 200, 0, 42));

        // Before recording, get returns None.
        assert_eq!(runtime.get_container_ip(&id).await.unwrap(), None);

        runtime.record_container_ip(&id, ip).await;
        assert_eq!(runtime.get_container_ip(&id).await.unwrap(), Some(ip));
    }

    // `#[ignore]`: this test originally piggy-backed on the side effect that
    // `wsl.exe -d zlayer -- mkdir -p …` fails fast on hosts that lack the
    // `zlayer` distro, which kept the call inside `create_container` and
    // satisfied the "not Unsupported" assertion in ~100 ms. Once a real
    // `zlayer` distro exists (production hosts AND any CI runner that has
    // run `setup_distro` once) the mkdir succeeds and execution proceeds
    // through the live registry pull → `wsl_stdin_pipe` tar-extract →
    // `zlayer runtime create` chain, none of which have inline timeouts.
    // The result is a multi-minute hang on a test whose only purpose was
    // a one-liner regression guard against re-adding the old
    // `AgentError::Unsupported` stub.
    //
    // The positive end-to-end case is covered by
    // `crates/zlayer-agent/tests/composite_dispatch_e2e.rs::composite_dispatches_linux_spec_to_wsl2`,
    // which exercises the same dispatch path against a real distro with
    // proper test infrastructure. The stub-regression contract is also
    // statically enforced — `create_container`'s body never constructs
    // `AgentError::Unsupported` — so leaving this `#[ignore]`'d does not
    // weaken the workspace's protection against the regression.
    //
    // The proper unwind here is workstream B8 (mock `ImagePullerLike` +
    // route `create_container`'s wsl calls through the existing
    // `WslRunner` trait) so the test can exercise the dispatch path
    // without any live I/O. Until then: ignored.
    #[ignore = "hangs on hosts with a real `zlayer` WSL distro; see comment + B8"]
    #[tokio::test]
    async fn create_container_no_longer_returns_unsupported() {
        // The G-2 contract: `create_container` is wired end-to-end. Without
        // a live WSL2 distro or a cached image, the call must still fail
        // *cleanly* (CreateFailed / PullFailed / Network) rather than with
        // the old `AgentError::Unsupported` stub. This test codifies the
        // negative assertion so a future stub-regression fails loudly.
        use zlayer_spec::DeploymentSpec;

        let runtime = test_runtime(default_resolved_config());
        let id = cid("svc", 0);
        // Test-only fixture image: our own GHCR-hosted retag of alpine:3.19.
        // Avoids docker.io rate limits + the public-internet dependency that
        // makes this test flake on hosts where the `zlayer` WSL distro exists
        // (mkdir succeeds → pull fires → 60s+ hang under constrained
        // networking). The pull still hits the wire — see
        // [`crate::runtimes::wsl2_delegate::ImagePullerLike`] (B8) for the
        // proper trait-injected stub that retires this footgun entirely.
        let yaml = r"
version: v1
deployment: wsl2-g2-test
services:
  svc:
    rtype: service
    image:
      name: ghcr.io/blackleafdigital/zlayer/test-fixtures:latest
    endpoints:
      - name: http
        protocol: http
        port: 8080
";
        let spec = serde_yaml::from_str::<DeploymentSpec>(yaml)
            .expect("valid deployment yaml")
            .services
            .remove("svc")
            .expect("service 'svc' present");

        let err = runtime.create_container(&id, &spec).await.unwrap_err();
        assert!(
            !matches!(err, AgentError::Unsupported(_)),
            "create_container must not return Unsupported after G-2 (got {err:?})",
        );
    }

    // -----------------------------------------------------------------
    // G-3: Real exec streaming — parser-level test.
    //
    // Running the actual `wsl.exe` child is Windows-only and requires a
    // live `zlayer` distro with youki, so the e2e exec-streaming test
    // lives in `composite_dispatch_e2e.rs` (`#[ignore]`'d). Here we
    // exercise the helper that pumps pipe output into `ExecEvent`s
    // against an in-process child process (`cmd.exe`) whose output is
    // fully deterministic.
    // -----------------------------------------------------------------

    /// `spawn_exec_event_stream` must turn a real child's piped stdout
    /// into one `ExecEvent::Stdout` per line and close with exactly one
    /// `ExecEvent::Exit` carrying the child's exit code. We drive
    /// `cmd.exe /c echo a & echo b & exit 7` which is portable across
    /// every Windows host and produces two deterministic stdout lines.
    #[tokio::test]
    async fn exec_stream_pump_yields_line_events_then_exit() {
        use futures_util::stream::StreamExt as _;

        let mut child = tokio::process::Command::new("cmd.exe")
            .args(["/c", "echo a & echo b & exit 7"])
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .expect("cmd.exe must be available on Windows test hosts");

        let stdout = child.stdout.take().expect("piped stdout");
        let stderr = child.stderr.take().expect("piped stderr");

        let mut stream = spawn_exec_event_stream(child, stdout, stderr);

        let mut stdout_lines: Vec<String> = Vec::new();
        let mut exit_code: Option<i32> = None;
        while let Some(ev) = stream.next().await {
            match ev {
                ExecEvent::Stdout(line) => stdout_lines.push(line.trim().to_string()),
                ExecEvent::Stderr(_) => {}
                ExecEvent::Exit(code) => {
                    exit_code = Some(code);
                    break;
                }
            }
        }

        assert_eq!(
            stdout_lines,
            vec!["a".to_string(), "b".to_string()],
            "stdout should have been split into one event per line",
        );
        assert_eq!(
            exit_code,
            Some(7),
            "terminal Exit event must carry the child's exit code",
        );
    }

    // -----------------------------------------------------------------
    // G-4: Youki log path plumbed through.
    // -----------------------------------------------------------------

    /// The log-path helper must compose the configured `log_root` with the
    /// slug derived from [`ContainerId`]. This is the exact string we hand
    /// to `youki create --log` and later tail from in `container_logs`, so
    /// a drift between the two would silently produce empty log responses.
    #[test]
    fn log_path_uses_configured_log_root() {
        let runtime = test_runtime(ResolvedConfig {
            distro: "zlayer".to_string(),
            runtime_binary: DEFAULT_RUNTIME_BINARY.to_string(),
            bundle_root: "/custom/bundles".to_string(),
            log_root: "/custom/logs".to_string(),
            oci_state_root: PathBuf::from(DEFAULT_OCI_STATE_ROOT),
        });
        let id = cid("web", 7);
        assert_eq!(
            runtime.log_path(&id),
            "/custom/logs/web-rep-7.youki.log",
            "log_path must be <log_root>/<slug>.youki.log",
        );
    }

    /// Sanity-check the default log root: tests that construct a runtime
    /// via [`default_resolved_config`] should see `/var/lib/zlayer/logs`,
    /// matching [`DEFAULT_LOG_ROOT`].
    #[test]
    fn log_path_uses_default_log_root_by_default() {
        let runtime = test_runtime(default_resolved_config());
        let id = cid("svc", 0);
        assert!(
            runtime.log_path(&id).starts_with(DEFAULT_LOG_ROOT),
            "default log_path should live under DEFAULT_LOG_ROOT ({DEFAULT_LOG_ROOT}), got {}",
            runtime.log_path(&id),
        );
    }

    // -----------------------------------------------------------------
    // G-5: Config-driven distro name + youki path discovery.
    // -----------------------------------------------------------------

    /// [`Wsl2DelegateConfig::default`] must reproduce the documented
    /// defaults so an operator upgrading without overriding any field
    /// silently keeps the same layout. Post-`zlayer runtime` migration the
    /// runtime binary defaults to `/usr/local/bin/zlayer` (rather than
    /// `which youki`) and `oci_state_root` is `/var/lib/zlayer/oci/state`.
    #[test]
    fn default_config_matches_previous_hardcoded_values() {
        let cfg = Wsl2DelegateConfig::default();
        // Honours `ZLAYER_WSL_DISTRO`; falls back to `DISTRO_NAME` when unset, so
        // this holds whether or not the test env has the override exported.
        assert_eq!(cfg.distro, zlayer_wsl::distro::configured_distro());
        assert_eq!(cfg.runtime_binary.as_deref(), Some(DEFAULT_RUNTIME_BINARY));
        assert_eq!(cfg.bundle_root, DEFAULT_BUNDLE_ROOT);
        assert_eq!(cfg.log_root, DEFAULT_LOG_ROOT);
        assert_eq!(cfg.oci_state_root, PathBuf::from(DEFAULT_OCI_STATE_ROOT));
    }

    /// A runtime built from a custom [`Wsl2DelegateConfig`] must propagate
    /// the chosen distro name + runtime binary path into every derived
    /// field so subsequent `wsl.exe` invocations target the right distro.
    /// We can't actually drive `wsl.exe` from unit tests, but we *can*
    /// assert that the runtime's `bundle_dir` / `log_path` / `config`
    /// carry the custom values.
    #[test]
    fn custom_config_propagates_into_runtime_fields() {
        let runtime = test_runtime(ResolvedConfig {
            distro: "ubuntu-lts".to_string(),
            runtime_binary: "/opt/zlayer/bin/zlayer".to_string(),
            bundle_root: "/srv/zlayer/bundles".to_string(),
            log_root: "/srv/zlayer/logs".to_string(),
            oci_state_root: PathBuf::from("/srv/zlayer/oci-state"),
        });
        let id = cid("api", 0);

        assert_eq!(runtime.config.distro, "ubuntu-lts");
        assert_eq!(runtime.config.runtime_binary, "/opt/zlayer/bin/zlayer");
        assert_eq!(
            runtime.bundle_dir(&id),
            "/srv/zlayer/bundles/api-rep-0",
            "bundle_dir should use the configured bundle_root",
        );
        assert_eq!(
            runtime.log_path(&id),
            "/srv/zlayer/logs/api-rep-0.youki.log",
            "log_path should use the configured log_root",
        );
    }

    /// `zlayer_runtime` must prefix every argv with `runtime --state-root <root>`
    /// and delegate to the configured runtime binary. We can't actually
    /// run `wsl.exe` from a unit test, but we can spawn the helper and
    /// assert the borrow/argv plumbing compiles + holds the expected
    /// shape by reaching into `wsl_exec_in` indirectly via a fake distro
    /// that will trivially fail on the test host. Instead, lock down the
    /// shape with a pure construction test: assemble the argv the same
    /// way `zlayer_runtime` does and verify it matches expectations.
    #[test]
    fn zlayer_runtime_prefixes_args_with_state_root() {
        let cfg = ResolvedConfig {
            distro: "zlayer".to_string(),
            runtime_binary: DEFAULT_RUNTIME_BINARY.to_string(),
            bundle_root: DEFAULT_BUNDLE_ROOT.to_string(),
            log_root: DEFAULT_LOG_ROOT.to_string(),
            oci_state_root: PathBuf::from("/var/lib/zlayer/oci/state"),
        };

        // Mirror the argv-building logic from `zlayer_runtime`.
        let state_root_owned = cfg.oci_state_root.to_string_lossy().into_owned();
        let user_args: &[&str] = &["create", "id", "--bundle", "/b", "--log", "/l"];
        let mut full_args: Vec<&str> = Vec::with_capacity(user_args.len() + 3);
        full_args.push("runtime");
        full_args.push("--state-root");
        full_args.push(state_root_owned.as_str());
        full_args.extend(user_args.iter().copied());

        assert_eq!(
            full_args,
            vec![
                "runtime",
                "--state-root",
                "/var/lib/zlayer/oci/state",
                "create",
                "id",
                "--bundle",
                "/b",
                "--log",
                "/l",
            ],
            "zlayer_runtime must prefix `runtime --state-root <root>` to its argv",
        );
    }

    // -----------------------------------------------------------------
    // Guest-managed overlay: WireGuard push lifecycle.
    // -----------------------------------------------------------------

    /// The WSL2 delegate must report the guest-managed attach kind so the
    /// service layer drives `push_overlay_config` post-start.
    #[test]
    fn overlay_attach_kind_is_guest_managed() {
        let runtime = test_runtime(default_resolved_config());
        assert_eq!(
            runtime.overlay_attach_kind(),
            OverlayAttachKind::GuestManaged
        );
    }

    /// `parse_default_gateway` extracts the `via` token from `ip route show
    /// default` output and yields `None` when there is no default route.
    #[test]
    fn parse_default_gateway_extracts_via() {
        assert_eq!(
            parse_default_gateway("default via 172.20.16.1 dev eth0 proto kernel"),
            Some("172.20.16.1".to_string())
        );
        assert_eq!(
            parse_default_gateway("10.0.0.0/24 dev eth0 scope link\n"),
            None
        );
        assert_eq!(parse_default_gateway(""), None);
    }

    /// `rewrite_endpoint_host` swaps the host but keeps the port, and leaves
    /// roaming (empty) endpoints untouched.
    #[test]
    fn rewrite_endpoint_host_keeps_port() {
        assert_eq!(
            rewrite_endpoint_host("10.200.0.5:51820", "172.20.16.1"),
            "172.20.16.1:51820"
        );
        assert_eq!(rewrite_endpoint_host("", "172.20.16.1"), "");
        // No colon -> returned unchanged.
        assert_eq!(rewrite_endpoint_host("hostonly", "172.20.16.1"), "hostonly");
    }

    /// `sh_single_quote` wraps in single quotes and escapes embedded quotes.
    #[test]
    fn sh_single_quote_escapes() {
        assert_eq!(sh_single_quote("abc+/=="), "'abc+/=='");
        assert_eq!(sh_single_quote("a'b"), "'a'\\''b'");
    }

    /// The guest `WireGuard` device name is deterministic and IFNAMSIZ-safe.
    #[test]
    fn wg_iface_name_is_ifnamsiz_safe() {
        let name = Wsl2DelegateRuntime::wg_iface_name("web-rep-0");
        assert_eq!(name, Wsl2DelegateRuntime::wg_iface_name("web-rep-0"));
        assert!(
            name.starts_with("zl-") && name.len() <= 15,
            "wg iface must be zl- prefixed and <= 15 chars, got {name:?}"
        );
        // A very long slug still hashes down to <= 15 chars.
        let long = Wsl2DelegateRuntime::wg_iface_name(
            "a-very-long-service-name-that-overflows-ifnamsiz-rep-12345",
        );
        assert!(long.len() <= 15, "long slug must hash to <= 15: {long:?}");
    }

    /// `push_overlay_config` must emit the full `WireGuard` bring-up sequence —
    /// create the device, set the key + peer (with the endpoint host rewritten
    /// to the distro gateway), move it into the named netns, address it, route
    /// it — and record the overlay IP + a `Configured` netns state.
    #[tokio::test]
    async fn push_overlay_config_emits_wireguard_sequence() {
        use zlayer_types::overlayd::{GuestOverlayConfig, PeerSpec};

        let runner = Arc::new(RecordingRunner::new());
        let calls = runner.calls_handle();
        // The default route lookup must return a parseable gateway.
        runner.set_response(
            "ip",
            &["route", "show", "default"],
            fake_output("default via 172.20.16.1 dev eth0\n", 0),
        );
        let runtime = make_runtime(runner.clone());

        let id = cid("web", 0);
        let cfg = GuestOverlayConfig {
            overlay_ip: IpAddr::V4(Ipv4Addr::new(10, 200, 0, 42)),
            prefix_len: 16,
            private_key: "PRIVKEYBASE64==".to_string(),
            public_key: "PUBKEYBASE64==".to_string(),
            listen_port: 51820,
            peers: vec![PeerSpec {
                public_key: "PEERPUB==".to_string(),
                endpoint: "10.200.0.1:51820".to_string(),
                allowed_ips: "10.200.0.0/16".to_string(),
                persistent_keepalive_secs: 25,
                candidates: Vec::new(),
            }],
            dns_server: Some(IpAddr::V4(Ipv4Addr::new(10, 200, 0, 1))),
            dns_domain: Some("svc.zlayer.local".to_string()),
        };

        runtime
            .push_overlay_config(&id, &cfg)
            .await
            .expect("push must succeed against the recording runner");

        let joined: Vec<String> = calls
            .lock()
            .unwrap()
            .iter()
            .map(|(c, a)| format!("{} {}", c, a.join(" ")))
            .collect();
        let wg = Wsl2DelegateRuntime::wg_iface_name("web-rep-0");
        let must_contain = [
            format!("ip link add {wg} type wireguard"),
            format!("wg set {wg} private-key"),
            // Endpoint host rewritten to the distro gateway, port preserved.
            format!("wg set {wg} peer PEERPUB== endpoint 172.20.16.1:51820 allowed-ips 10.200.0.0/16 persistent-keepalive 25"),
            format!("ip link set {wg} netns web-rep-0"),
            format!("ip netns exec web-rep-0 ip addr add 10.200.0.42/16 dev {wg}"),
            "ip netns exec web-rep-0 ip route add default dev".to_string(),
        ];
        for needle in &must_contain {
            assert!(
                joined.iter().any(|c| c.contains(needle.as_str())),
                "expected a command matching {needle:?} in {joined:?}"
            );
        }

        // IP recorded + netns marked configured.
        assert_eq!(
            runtime.get_container_ip(&id).await.unwrap(),
            Some(cfg.overlay_ip)
        );
        assert!(matches!(
            runtime.netns.read().await.get(&id),
            Some(NetnsState::Configured { .. })
        ));
    }

    /// A failure mid-push must surface as an error (no silent host fallback)
    /// and best-effort delete the partial `WireGuard` device.
    #[tokio::test]
    async fn push_overlay_config_hard_errors_on_failure() {
        use zlayer_types::overlayd::GuestOverlayConfig;

        let runner = Arc::new(RecordingRunner::new());
        runner.set_response(
            "ip",
            &["route", "show", "default"],
            fake_output("default via 172.20.16.1 dev eth0\n", 0),
        );
        let wg = Wsl2DelegateRuntime::wg_iface_name("web-rep-7");
        // Make the device creation fail.
        runner.set_response(
            "ip",
            &["link", "add", &wg, "type", "wireguard"],
            fake_output_err("RTNETLINK answers: Operation not permitted", 2),
        );
        let runtime = make_runtime(runner.clone());

        let id = cid("web", 7);
        let cfg = GuestOverlayConfig {
            overlay_ip: IpAddr::V4(Ipv4Addr::new(10, 200, 0, 99)),
            prefix_len: 16,
            private_key: "K==".to_string(),
            public_key: "P==".to_string(),
            listen_port: 51820,
            peers: Vec::new(),
            dns_server: None,
            dns_domain: None,
        };

        let err = runtime.push_overlay_config(&id, &cfg).await.unwrap_err();
        assert!(
            matches!(err, AgentError::Network(_)),
            "expected a hard Network error, got {err:?}"
        );
        // No IP should be advertised after a failed push.
        assert_eq!(runtime.get_container_ip(&id).await.unwrap(), None);
    }

    /// Teardown deletes the `WireGuard` device + named netns when one was
    /// provisioned (state present), and is a no-op otherwise.
    #[tokio::test]
    async fn teardown_container_netns_deletes_wg_and_ns() {
        let runner = Arc::new(RecordingRunner::new());
        let calls = runner.calls_handle();
        let runtime = make_runtime(runner.clone());

        let id = cid("web", 0);
        // Simulate a provisioned named netns.
        runtime
            .netns
            .write()
            .await
            .insert(id.clone(), NetnsState::Created);

        runtime.teardown_container_netns(&id).await;

        let joined: Vec<String> = calls
            .lock()
            .unwrap()
            .iter()
            .map(|(c, a)| format!("{} {}", c, a.join(" ")))
            .collect();
        let wg = Wsl2DelegateRuntime::wg_iface_name("web-rep-0");
        assert!(
            joined.iter().any(|c| c == &format!("ip link delete {wg}")),
            "teardown must delete the wg device, got {joined:?}"
        );
        assert!(
            joined.iter().any(|c| c == "ip netns delete web-rep-0"),
            "teardown must delete the netns, got {joined:?}"
        );
        assert!(runtime.netns.read().await.get(&id).is_none());

        // No recorded state => no commands at all.
        calls.lock().unwrap().clear();
        let id2 = cid("web", 9);
        runtime.teardown_container_netns(&id2).await;
        assert!(
            calls.lock().unwrap().is_empty(),
            "teardown without state must be a no-op"
        );
    }

    // -------------------------------------------------------------------
    // Phase 5.D: WSL2 GPU exposure tests.
    //
    // These exercise `inject_wsl_gpu_mounts` against a stub host state so we
    // don't need a real `/dev/dxg` on the Windows CI runner. They cover
    // the four spec'd cases plus the GPU-not-requested no-op path.
    // -------------------------------------------------------------------

    fn test_gpu_spec() -> GpuSpec {
        GpuSpec {
            count: 1,
            vendor: "nvidia".to_string(),
            mode: None,
            model: None,
            scheduling: None,
            distributed: None,
            sharding: None,
            sharing: None,
            mps_pipe_dir: None,
            mps_log_dir: None,
            time_slice_index: None,
            time_slicing_config_path: None,
        }
    }

    fn happy_host_state() -> WslGpuHostState {
        WslGpuHostState {
            dxg_devno: Some((10, 117)),
            wsl_lib_present: true,
            wsl_drivers_present: true,
        }
    }

    #[test]
    fn inject_wsl_gpu_mounts_adds_dxg_mount() {
        let mut mounts: Vec<Mount> = Vec::new();
        let mut env: Vec<String> = Vec::new();
        let mut devices: Vec<LinuxDevice> = Vec::new();
        let gpu = test_gpu_spec();
        let host = happy_host_state();

        inject_wsl_gpu_mounts(&mut mounts, &mut env, &mut devices, &gpu, &host)
            .expect("inject must succeed on happy host");

        let dxg_mounts: Vec<_> = mounts
            .iter()
            .filter(|m| m.destination().as_path() == std::path::Path::new("/dev/dxg"))
            .collect();
        assert_eq!(
            dxg_mounts.len(),
            1,
            "expected exactly one /dev/dxg mount, got {}: {:?}",
            dxg_mounts.len(),
            mounts
        );
        assert_eq!(
            dxg_mounts[0]
                .source()
                .as_ref()
                .map(std::path::PathBuf::as_path),
            Some(std::path::Path::new("/dev/dxg"))
        );
        assert_eq!(dxg_mounts[0].typ().as_deref(), Some("bind"));

        // Device cgroup entry must be present too with the probed major/minor.
        let dxg_devs: Vec<_> = devices
            .iter()
            .filter(|d| d.path().as_path() == std::path::Path::new("/dev/dxg"))
            .collect();
        assert_eq!(dxg_devs.len(), 1);
        assert_eq!(dxg_devs[0].major(), 10);
        assert_eq!(dxg_devs[0].minor(), 117);
        assert_eq!(dxg_devs[0].typ(), LinuxDeviceType::C);
    }

    #[test]
    fn inject_wsl_gpu_mounts_respects_existing_user_mount() {
        // User already declared a /dev/dxg bind mount via spec.devices; the
        // helper must not double-mount.
        let pre_existing = MountBuilder::default()
            .destination("/dev/dxg".to_string())
            .source("/dev/dxg".to_string())
            .typ("bind".to_string())
            .options(vec!["bind".to_string(), "rw".to_string()])
            .build()
            .unwrap();
        let mut mounts = vec![pre_existing];

        let pre_existing_dev = LinuxDeviceBuilder::default()
            .path("/dev/dxg")
            .typ(LinuxDeviceType::C)
            .major(10)
            .minor(117)
            .file_mode(0o666u32)
            .uid(0u32)
            .gid(0u32)
            .build()
            .unwrap();
        let mut devices = vec![pre_existing_dev];

        let mut env: Vec<String> = Vec::new();
        let gpu = test_gpu_spec();
        let host = happy_host_state();

        inject_wsl_gpu_mounts(&mut mounts, &mut env, &mut devices, &gpu, &host)
            .expect("inject must succeed on happy host");

        assert_eq!(
            mounts
                .iter()
                .filter(|m| m.destination().as_path() == std::path::Path::new("/dev/dxg"))
                .count(),
            1,
            "expected no duplicate /dev/dxg mount, mounts: {mounts:?}"
        );
        assert_eq!(
            devices
                .iter()
                .filter(|d| d.path().as_path() == std::path::Path::new("/dev/dxg"))
                .count(),
            1,
            "expected no duplicate /dev/dxg device, devices: {devices:?}"
        );
    }

    #[test]
    fn inject_wsl_gpu_mounts_prepends_ld_library_path() {
        let mut mounts: Vec<Mount> = Vec::new();
        let mut env: Vec<String> = vec!["LD_LIBRARY_PATH=/foo".to_string()];
        let mut devices: Vec<LinuxDevice> = Vec::new();
        let gpu = test_gpu_spec();
        let host = happy_host_state();

        inject_wsl_gpu_mounts(&mut mounts, &mut env, &mut devices, &gpu, &host)
            .expect("inject must succeed on happy host");

        let ld_entries: Vec<_> = env
            .iter()
            .filter(|e| e.starts_with("LD_LIBRARY_PATH="))
            .collect();
        assert_eq!(
            ld_entries.len(),
            1,
            "exactly one LD_LIBRARY_PATH entry: {env:?}"
        );
        // Drivers ahead of lib (NVIDIA shim wins lookup), original /foo preserved.
        assert_eq!(
            ld_entries[0], "LD_LIBRARY_PATH=/usr/lib/wsl/drivers:/usr/lib/wsl/lib:/foo",
            "unexpected LD_LIBRARY_PATH: {env:?}"
        );
    }

    #[test]
    fn inject_wsl_gpu_mounts_appends_ld_library_path_when_absent() {
        // No existing LD_LIBRARY_PATH — the helper should add one rather than
        // silently dropping the prefix.
        let mut mounts: Vec<Mount> = Vec::new();
        let mut env: Vec<String> = vec!["PATH=/bin".to_string()];
        let mut devices: Vec<LinuxDevice> = Vec::new();
        let gpu = test_gpu_spec();
        let host = happy_host_state();

        inject_wsl_gpu_mounts(&mut mounts, &mut env, &mut devices, &gpu, &host).unwrap();

        assert!(env.contains(&"PATH=/bin".to_string()));
        assert!(env
            .iter()
            .any(|e| e == "LD_LIBRARY_PATH=/usr/lib/wsl/drivers:/usr/lib/wsl/lib"));
    }

    #[test]
    fn inject_wsl_gpu_mounts_fails_when_dxg_missing() {
        let mut mounts: Vec<Mount> = Vec::new();
        let mut env: Vec<String> = Vec::new();
        let mut devices: Vec<LinuxDevice> = Vec::new();
        let gpu = test_gpu_spec();
        let host = WslGpuHostState {
            dxg_devno: None,
            wsl_lib_present: true,
            wsl_drivers_present: true,
        };

        let err = inject_wsl_gpu_mounts(&mut mounts, &mut env, &mut devices, &gpu, &host)
            .expect_err("expected WslGpuUnavailable when /dev/dxg missing");
        assert!(
            matches!(err, AgentError::WslGpuUnavailable { .. }),
            "expected WslGpuUnavailable, got: {err:?}"
        );
        assert!(mounts.is_empty(), "no mount should have been pushed");
        assert!(devices.is_empty(), "no device should have been pushed");
    }

    #[tokio::test]
    async fn apply_wsl_gpu_to_spec_no_op_when_gpu_not_requested() {
        struct ExplodingProbe;
        #[async_trait]
        impl WslGpuHostProbe for ExplodingProbe {
            async fn probe(&self) -> Result<WslGpuHostState> {
                panic!("probe must not run when no GPU is requested");
            }
        }

        // Build a minimal Spec via the OCI default; the apply function should
        // leave it untouched when the service spec doesn't request a GPU.
        let mut oci_spec = Spec::default();
        let mounts_before = oci_spec.mounts().clone();

        // ServiceSpec has no Default impl in zlayer-types; parse a minimal
        // fixture from YAML the same way the spec crate's own tests do.
        let yaml = r"
version: v1
deployment: gpu-noop-test
services:
  cpu-only:
    rtype: service
    image:
      name: alpine:latest
";
        let deployment: zlayer_spec::DeploymentSpec =
            serde_yaml::from_str(yaml).expect("parse fixture deployment");
        let service = deployment
            .services
            .get("cpu-only")
            .cloned()
            .expect("cpu-only service");
        assert!(
            service.resources.gpu.is_none(),
            "fixture must not request GPU"
        );

        apply_wsl_gpu_to_spec(&mut oci_spec, &service, &ExplodingProbe)
            .await
            .expect("no-op should succeed");

        assert_eq!(oci_spec.mounts(), &mounts_before);
        let has_dxg = oci_spec.mounts().as_ref().is_some_and(|ms| {
            ms.iter()
                .any(|m| m.destination().as_path() == std::path::Path::new("/dev/dxg"))
        });
        assert!(
            !has_dxg,
            "no /dev/dxg mount should appear for CPU-only spec"
        );
    }

    #[test]
    fn parse_stat_hex_devno_handles_typical_wsl_output() {
        // `stat -c '%t %T' /dev/dxg` on a real WSL2 distro returns e.g. "a 75".
        assert_eq!(parse_stat_hex_devno("a 75"), Some((10, 117)));
        assert_eq!(parse_stat_hex_devno("  a   75  "), Some((10, 117)));
        assert_eq!(parse_stat_hex_devno(""), None);
        assert_eq!(parse_stat_hex_devno("nothex zz"), None);
        assert_eq!(parse_stat_hex_devno("a"), None);
    }

    /// Named-netns provisioning tolerates an already-existing namespace so a
    /// crash-restart cycle can re-enter the create path without failing.
    #[tokio::test]
    async fn provision_named_netns_idempotent_on_existing_ns() {
        let runner = Arc::new(RecordingRunner::new());
        let runtime = make_runtime(runner.clone());

        let id = cid("web", 3);
        runner.set_response(
            "ip",
            &["netns", "add", "web-rep-3"],
            fake_output_err(
                "Cannot create namespace file \"/run/netns/web-rep-3\": File exists",
                1,
            ),
        );

        runtime
            .provision_named_netns(&id, "web-rep-3")
            .await
            .expect("pre-existing netns must be reused, not error");

        assert!(matches!(
            runtime.netns.read().await.get(&id),
            Some(NetnsState::Created)
        ));
    }
}