zlayer-agent 0.14.1

//! Youki/libcontainer runtime implementation
//!
//! Implements the Runtime trait using libcontainer (youki's container library)
//! for direct OCI container management without a daemon.

use crate::cgroups_stats::{self, ContainerStats};
use crate::error::{AgentError, Result};
use crate::runtime::{
    ArchivePutOptions, ArchiveStream, ContainerId, ContainerState, ExecExitFuture, ExecHandle,
    ExecOptions, ExecPtyStream, ImageInfo, ImageInspectInfo, LogChannel, LogChunk, LogsStream,
    LogsStreamOptions, PathStat, PruneResult, PullProgress, PullProgressStream, Runtime,
    StatsSample, StatsStream,
};
use crate::storage_manager::StorageManager;
use libcontainer::container::{Container, ContainerStatus};
use oci_client::manifest::OciImageManifest;
use std::collections::HashMap;
use std::os::fd::{AsRawFd, OwnedFd};
use std::path::{Path, PathBuf};
use std::time::Duration;
use tokio::fs;
use tokio::io::{AsyncBufReadExt, AsyncSeekExt, BufReader};
use tokio::sync::{mpsc, RwLock};
use tokio_stream::wrappers::ReceiverStream;
use tracing::instrument;
use zlayer_observability::logs::{LogEntry, LogSource, LogStream};
use zlayer_spec::{RegistryAuth, ServiceSpec};

/// Safety window for [`prune_containers`]: a bundle whose mtime is newer than
/// this is assumed to belong to an in-flight `create_container` (which writes
/// the bundle before the state directory exists) and is never reclaimed.
const BUNDLE_PRUNE_SAFETY: Duration = Duration::from_secs(120);

/// Default retention window for a removed container's structured stdout/stderr
/// logs (24h). Overridable via `ZLAYER_CONTAINER_LOG_RETENTION_SECS`.
const DEFAULT_CONTAINER_LOG_RETENTION: Duration = Duration::from_secs(24 * 60 * 60);

/// How long a removed container's structured stdout/stderr logs are kept so a
/// failed container can still be inspected after teardown. Reads
/// `ZLAYER_CONTAINER_LOG_RETENTION_SECS` (seconds), else 24h.
fn container_log_retention() -> Duration {
    std::env::var("ZLAYER_CONTAINER_LOG_RETENTION_SECS")
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .map_or(DEFAULT_CONTAINER_LOG_RETENTION, Duration::from_secs)
}

/// Decide which bundle directories are orphaned and safe to reclaim.
///
/// Pure decision function (no filesystem access) so the orphan logic is unit
/// testable without a live runtime. A bundle id is selected iff:
/// * it is **not** in `live` (no on-disk state dir and not in-memory tracked), and
/// * its mtime is at least `safety` older than `now` (so a bundle being created
///   right now — written before its state dir exists — is skipped).
///
/// Bundles with an mtime in the future (clock skew) are treated as recent and
/// skipped. The returned ids are sorted for deterministic output.
fn select_orphan_bundles(
    bundle_mtimes: &std::collections::HashMap<String, std::time::SystemTime>,
    live: &std::collections::HashSet<String>,
    now: std::time::SystemTime,
    safety: Duration,
) -> Vec<String> {
    let mut orphans: Vec<String> = bundle_mtimes
        .iter()
        .filter(|(id, _)| !live.contains(id.as_str()))
        .filter(|(_, &mtime)| {
            // Old enough: now - mtime >= safety. `duration_since` errs on a
            // future mtime, which we treat as "too recent" (skip).
            now.duration_since(mtime).is_ok_and(|age| age >= safety)
        })
        .map(|(id, _)| id.clone())
        .collect();
    orphans.sort();
    orphans
}

/// Decide which extracted overlay layers in the shared layer store are safe to
/// reclaim (the layer-store analogue of [`select_orphan_bundles`]).
///
/// Pure decision function (no filesystem access) so the GC liveness logic is
/// unit testable without a live runtime or any extracted layers. A layer
/// (keyed by its sanitized digest = layer-store subdirectory name) is selected
/// iff:
/// * it is **not** in `live` (no surviving bundle's `.lowerdirs` references it),
///   and
/// * its `.done`-sentinel mtime is at least `safety` older than `now` (so a
///   layer being first-extracted right now — published moments ago, possibly for
///   an in-flight create whose `.lowerdirs` isn't written yet — is skipped).
///
/// A layer with an mtime in the future (clock skew) is treated as recent and
/// skipped. `layer_mtimes` is expected to contain ONLY published layers (those
/// with a `.done` sentinel); the caller filters unpublished dirs out before
/// calling. The returned digests are sorted for deterministic output.
fn select_prunable_layers(
    layer_mtimes: &std::collections::HashMap<String, std::time::SystemTime>,
    live: &std::collections::HashSet<String>,
    now: std::time::SystemTime,
    safety: Duration,
) -> Vec<String> {
    let mut prunable: Vec<String> = layer_mtimes
        .iter()
        .filter(|(digest, _)| !live.contains(digest.as_str()))
        .filter(|(_, &mtime)| now.duration_since(mtime).is_ok_and(|age| age >= safety))
        .map(|(digest, _)| digest.clone())
        .collect();
    prunable.sort();
    prunable
}

/// Unmount a container's overlay rootfs at `<bundle>/rootfs` (and its
/// per-container toolchain overlay at `<bundle>/toolchains-merged`) before the
/// bundle tree is removed.
///
/// MUST run before any `remove_dir_all` of a bundle that may have an overlay
/// rootfs: removing a mounted overlay would recurse into the shared lowerdirs
/// (deleting other containers' layer-store data) or fail `EBUSY`. The wrapped
/// [`overlay_rootfs::unmount_overlay_rootfs`](super::overlay_rootfs::unmount_overlay_rootfs)
/// is idempotent — a no-op for a fallback-path (plain-copy) bundle whose
/// `rootfs` is not a mountpoint — so this is safe to call unconditionally at
/// every bundle-removal site. The per-container `upper`/`work` dirs live inside
/// the bundle and so are reclaimed by the subsequent bundle removal.
///
/// The same applies to the per-container toolchain overlay built by
/// [`YoukiRuntime::prepare_toolchain_overlay`]: its merged view at
/// `<bundle>/toolchains-merged` stacks the shared toolchain store as a read-only
/// lowerdir under a per-container upperdir, so it MUST be unmounted before the
/// bundle is removed (otherwise a `remove_dir_all` would recurse into the shared
/// store). The teardown reconstructs the merged path from `bundle_path` exactly
/// as the create side does, so nothing needs to be tracked on the container
/// record. `unmount_overlay_rootfs` is idempotent for the raw-bind fallback
/// bundle whose `toolchains-merged` is not a mountpoint. No-op on non-Linux
/// (overlay is a Linux-only feature). Errors are logged, not propagated, so
/// teardown always proceeds to the removal.
fn unmount_bundle_rootfs(bundle_path: &Path) {
    #[cfg(target_os = "linux")]
    {
        let rootfs = bundle_path.join("rootfs");
        if let Err(e) = super::overlay_rootfs::unmount_overlay_rootfs(&rootfs) {
            tracing::warn!(
                rootfs = %rootfs.display(),
                error = %e,
                "failed to unmount overlay rootfs before bundle teardown (continuing)"
            );
        }
        // Per-container toolchain overlay (RO shared lower + per-container upper).
        // Idempotent no-op when the bundle used the raw-bind fallback.
        let tc_merged = bundle_path.join("toolchains-merged");
        if let Err(e) = super::overlay_rootfs::unmount_overlay_rootfs(&tc_merged) {
            tracing::warn!(
                tc_merged = %tc_merged.display(),
                error = %e,
                "failed to unmount toolchain overlay before bundle teardown (continuing)"
            );
        }
    }
    #[cfg(not(target_os = "linux"))]
    let _ = bundle_path;
}

/// Resolve the `fuse-overlayfs` binary path ONCE, but only when the rootless
/// fuse-overlay rootfs path is actually available per the daemon capability
/// survey. Returns `None` when the kernel overlay path is available (the fuse
/// path is never chosen then), when fuse-overlay is unavailable, or off Linux —
/// so the per-container create never re-scans `PATH`.
#[cfg(target_os = "linux")]
fn resolve_fuse_overlayfs_bin() -> Option<PathBuf> {
    let caps = crate::capability::DaemonCapabilities::get();
    if !caps.fuse_overlayfs_rootfs_available {
        return None;
    }
    let path_var = std::env::var("PATH").ok()?;
    for dir in path_var.split(':').filter(|d| !d.is_empty()) {
        let candidate = std::path::Path::new(dir).join("fuse-overlayfs");
        if candidate.exists() {
            return Some(candidate);
        }
    }
    None
}

#[cfg(not(target_os = "linux"))]
fn resolve_fuse_overlayfs_bin() -> Option<PathBuf> {
    None
}

/// Configuration for `YoukiRuntime`
#[derive(Debug, Clone)]
pub struct YoukiConfig {
    /// State directory for libcontainer container state
    pub state_dir: PathBuf,
    /// Directory for unpacked image rootfs
    pub rootfs_dir: PathBuf,
    /// Directory for OCI bundles
    pub bundle_dir: PathBuf,
    /// Cache directory for image blobs
    pub cache_dir: PathBuf,
    /// Directory for persistent volumes
    pub volume_dir: PathBuf,
    /// Use systemd cgroups
    pub use_systemd: bool,
    /// Cache type configuration (if None, determined from environment)
    pub cache_type: Option<zlayer_registry::CacheType>,
    /// Base directory for structured container logs.
    ///
    /// When set, container logs are written to
    /// `{log_base_dir}/{deployment_name}/{service}/{container_id}.log`
    /// instead of the bundle directory.  The bundle `logs/` directory will
    /// contain symlinks back to the structured location so that existing
    /// code paths that read from the bundle still work.
    pub log_base_dir: Option<PathBuf>,
    /// Deployment name used in log directory hierarchy.
    ///
    /// Only meaningful when `log_base_dir` is also set.  Defaults to
    /// `"default"` if unset.
    pub deployment_name: Option<String>,
}

impl YoukiConfig {
    /// Build a `YoukiConfig` whose subdirectories are scoped to the given
    /// data directory, with the existing per-directory `ZLAYER_*_DIR` env
    /// var overrides honored as escape hatches.
    pub fn from_data_dir(data_dir: &std::path::Path) -> Self {
        let dirs = zlayer_paths::ZLayerDirs::new(data_dir);
        Self {
            state_dir: std::env::var("ZLAYER_STATE_DIR")
                .map_or_else(|_| dirs.containers(), PathBuf::from),
            rootfs_dir: std::env::var("ZLAYER_ROOTFS_DIR")
                .map_or_else(|_| dirs.rootfs(), PathBuf::from),
            bundle_dir: std::env::var("ZLAYER_BUNDLE_DIR")
                .map_or_else(|_| dirs.bundles(), PathBuf::from),
            cache_dir: std::env::var("ZLAYER_CACHE_DIR")
                .map_or_else(|_| dirs.cache(), PathBuf::from),
            volume_dir: std::env::var("ZLAYER_VOLUME_DIR")
                .map_or_else(|_| dirs.volumes(), PathBuf::from),
            use_systemd: std::env::var("ZLAYER_USE_SYSTEMD")
                .is_ok_and(|v| v == "1" || v.to_lowercase() == "true"),
            cache_type: None,
            log_base_dir: None,
            deployment_name: None,
        }
    }
}

impl Default for YoukiConfig {
    fn default() -> Self {
        Self::from_data_dir(&zlayer_paths::ZLayerDirs::default_data_dir())
    }
}

/// Monotonic counter giving each cache-prefetch a unique throwaway stage dir.
static PREFETCH_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// Container tracking information
#[derive(Debug)]
struct ContainerInfo {
    /// Image reference
    #[allow(dead_code)]
    image: String,
    /// Bundle path
    #[allow(dead_code)]
    bundle_path: PathBuf,
    /// Rootfs path
    #[allow(dead_code)]
    rootfs_path: PathBuf,
    /// Stdout log file path
    stdout_path: PathBuf,
    /// Stderr log file path
    stderr_path: PathBuf,
    /// Process ID (once running)
    #[allow(dead_code)]
    pid: Option<u32>,
    /// Most recent restart policy applied via
    /// [`Runtime::update_container_resources`]. None when never updated.
    /// Persisted across runtime restarts is not yet wired; this is the
    /// in-memory copy that the supervisor consults on container exit.
    #[allow(dead_code)]
    restart_policy: Option<crate::runtime::ContainerRestartPolicyUpdate>,
    /// Real init exit code, recorded by the per-container exit watcher once the
    /// init process terminates (`None` while running or if never captured).
    /// libcontainer carries no exit code, so this in-process slot is the only
    /// source of truth for a stopped container's real code.
    exit_code: std::sync::Arc<std::sync::Mutex<Option<i32>>>,
    /// Whether a `waitpid` exit watcher has been armed for this container's
    /// init. Set via `compare_exchange` so that exactly ONE watcher exists per
    /// live pid across `start_container` and boot re-adoption — two reapers
    /// racing on the same pid would have one lose to `ECHILD`, but more
    /// importantly the guard keeps re-adoption from double-arming a container
    /// `start_container` already covers.
    watcher_armed: std::sync::Arc<std::sync::atomic::AtomicBool>,
}

/// Youki/libcontainer-based container runtime
///
/// This runtime uses libcontainer directly to create and manage OCI containers
/// without requiring a daemon like containerd.
pub struct YoukiRuntime {
    /// Configuration
    config: YoukiConfig,
    /// Local container state tracking
    containers: RwLock<HashMap<String, ContainerInfo>>,
    /// Authentication resolver for registry pulls
    auth_resolver: zlayer_core::AuthResolver,
    /// Storage volume manager
    storage_manager: std::sync::Arc<tokio::sync::RwLock<StorageManager>>,
    /// Shared blob cache for image layers (avoids repeated opens and ensures cache persistence)
    blob_cache: std::sync::Arc<Box<dyn zlayer_registry::BlobCacheBackend>>,
    /// Local OCI registry for resolving locally-built images
    local_registry: Option<std::sync::Arc<zlayer_registry::LocalRegistry>>,
    /// Cached image configs (entrypoint, cmd, env, etc.) keyed by image reference
    image_configs: RwLock<HashMap<String, zlayer_registry::ImageConfig>>,
    /// Auth context for container-to-host API authentication.
    auth_context: Option<crate::runtime::ContainerAuthContext>,
    /// Secrets provider injected post-construction via [`Runtime::set_secrets_provider`].
    ///
    /// Used in `create_container` to resolve `$S:` env references against the
    /// service's typed `secret_scope`. Interior mutability (`parking_lot::RwLock`)
    /// lets the daemon wire the provider after the runtime is built. A
    /// fully-qualified path is used because `tokio::sync::RwLock` is imported in
    /// this module — this lock is sync-only and never held across an `.await`.
    secrets_provider:
        parking_lot::RwLock<Option<std::sync::Arc<dyn zlayer_secrets::SecretsProvider>>>,
    /// Set once boot-time exit-watcher re-adoption has run (lazily, on the
    /// first orphaned-bundle sweep). Across a daemon restart the in-memory
    /// `containers` map starts empty, so the reparented container inits have
    /// no watcher; re-adoption walks the on-disk state, arms exactly one
    /// watcher per live pid, and reaps dead ones (releasing their pinned log
    /// fds — the ENOSPC leak). Gated so it runs at most once per process.
    readopt_done: std::sync::atomic::AtomicBool,
    /// Whether this daemon can build a container rootfs as an overlayfs mount
    /// (shared read-only lowerdirs + a per-container upperdir) instead of a
    /// full per-container copy of every image layer. Cached at construction
    /// from the process-wide [`DaemonCapabilities`] survey (a single read of
    /// the memoised, startup-computed value — it cannot change for a running
    /// daemon). `create_container` branches on this; `false` keeps the legacy
    /// full-copy path verbatim.
    // Only read by the Linux-only overlay arm of `build_container_rootfs`.
    #[cfg_attr(not(target_os = "linux"), allow(dead_code))]
    overlayfs_rootfs_available: bool,
    /// Whether this daemon can build a container rootfs as a ROOTLESS
    /// `fuse-overlayfs` mount — the fallback used when the kernel overlay path
    /// ([`Self::overlayfs_rootfs_available`]) is unavailable (no root /
    /// `CAP_SYS_ADMIN`). Still gives shared-layer dedup; needs only the
    /// `fuse-overlayfs` binary + `/dev/fuse`. Cached at construction from the
    /// process-wide [`DaemonCapabilities`] survey.
    #[cfg_attr(not(target_os = "linux"), allow(dead_code))]
    fuse_overlayfs_rootfs_available: bool,
    /// Resolved `fuse-overlayfs` binary path (when the rootless fuse path is
    /// available), cached so `create_container` doesn't re-scan `PATH` per
    /// container. `None` when the fuse path is unavailable or off non-Linux.
    #[cfg_attr(not(target_os = "linux"), allow(dead_code))]
    fuse_overlayfs_bin: Option<PathBuf>,
}

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

impl YoukiRuntime {
    /// Create a new `YoukiRuntime` with the given configuration
    ///
    /// # Errors
    /// Returns an error if the required directories cannot be created.
    pub async fn new(
        config: YoukiConfig,
        auth_context: Option<crate::runtime::ContainerAuthContext>,
    ) -> Result<Self> {
        // Ensure directories exist
        for dir in [
            &config.state_dir,
            &config.rootfs_dir,
            &config.bundle_dir,
            &config.cache_dir,
        ] {
            fs::create_dir_all(dir)
                .await
                .map_err(|e| AgentError::CreateFailed {
                    id: "runtime".to_string(),
                    reason: format!("failed to create directory {}: {}", dir.display(), e),
                })?;
        }

        // Initialize storage manager
        let storage_manager =
            StorageManager::new(&config.volume_dir).map_err(|e| AgentError::CreateFailed {
                id: "runtime".to_string(),
                reason: format!("failed to create storage manager: {e}"),
            })?;

        // Initialize shared blob cache using CacheType configuration
        // If cache_type is provided, use it directly; otherwise use environment-based config
        // but override the path for Persistent variant to use config.cache_dir
        let blob_cache = if let Some(cache_type) = &config.cache_type {
            cache_type
                .build()
                .await
                .map_err(|e| AgentError::CreateFailed {
                    id: "runtime".to_string(),
                    reason: format!("failed to build blob cache: {e}"),
                })?
        } else {
            let cache_type =
                zlayer_registry::CacheType::from_env().map_err(|e| AgentError::CreateFailed {
                    id: "runtime".to_string(),
                    reason: format!("failed to read cache config from env: {e}"),
                })?;
            // Override persistent path to use config.cache_dir
            #[allow(clippy::match_wildcard_for_single_variants)]
            let cache_type = match cache_type {
                zlayer_registry::CacheType::Persistent { .. } => {
                    zlayer_registry::CacheType::persistent_at(config.cache_dir.join("blobs.redb"))
                }
                other => other,
            };
            cache_type
                .build()
                .await
                .map_err(|e| AgentError::CreateFailed {
                    id: "runtime".to_string(),
                    reason: format!("failed to build blob cache: {e}"),
                })?
        };

        let local_registry = {
            let registry_path = config.cache_dir.parent().map_or_else(
                || config.cache_dir.join("registry"),
                |data_dir| data_dir.join("registry"),
            );
            match zlayer_registry::LocalRegistry::new(registry_path).await {
                Ok(reg) => Some(std::sync::Arc::new(reg)),
                Err(e) => {
                    tracing::warn!("Failed to open local registry: {e}");
                    None
                }
            }
        };

        Ok(Self {
            config,
            containers: RwLock::new(HashMap::new()),
            auth_resolver: zlayer_core::AuthResolver::new(zlayer_core::AuthConfig::default()),
            storage_manager: std::sync::Arc::new(tokio::sync::RwLock::new(storage_manager)),
            blob_cache,
            local_registry,
            image_configs: RwLock::new(HashMap::new()),
            auth_context,
            secrets_provider: parking_lot::RwLock::new(None),
            readopt_done: std::sync::atomic::AtomicBool::new(false),
            overlayfs_rootfs_available: crate::capability::DaemonCapabilities::get()
                .overlayfs_rootfs_available,
            fuse_overlayfs_rootfs_available: crate::capability::DaemonCapabilities::get()
                .fuse_overlayfs_rootfs_available,
            fuse_overlayfs_bin: resolve_fuse_overlayfs_bin(),
        })
    }

    /// Create a new `YoukiRuntime` with default configuration
    ///
    /// # Errors
    /// Returns an error if the runtime cannot be initialized.
    pub async fn with_defaults() -> Result<Self> {
        Self::new(YoukiConfig::default(), None).await
    }

    /// Create a new `YoukiRuntime` with custom auth configuration
    ///
    /// # Errors
    /// Returns an error if the runtime cannot be initialized.
    pub async fn with_auth(
        config: YoukiConfig,
        auth_config: zlayer_core::AuthConfig,
    ) -> Result<Self> {
        // Ensure directories exist
        for dir in [
            &config.state_dir,
            &config.rootfs_dir,
            &config.bundle_dir,
            &config.cache_dir,
        ] {
            fs::create_dir_all(dir)
                .await
                .map_err(|e| AgentError::CreateFailed {
                    id: "runtime".to_string(),
                    reason: format!("failed to create directory {}: {}", dir.display(), e),
                })?;
        }

        // Initialize storage manager
        let storage_manager =
            StorageManager::new(&config.volume_dir).map_err(|e| AgentError::CreateFailed {
                id: "runtime".to_string(),
                reason: format!("failed to create storage manager: {e}"),
            })?;

        // Initialize shared blob cache using CacheType configuration
        // If cache_type is provided, use it directly; otherwise use environment-based config
        // but override the path for Persistent variant to use config.cache_dir
        let blob_cache = if let Some(cache_type) = &config.cache_type {
            cache_type
                .build()
                .await
                .map_err(|e| AgentError::CreateFailed {
                    id: "runtime".to_string(),
                    reason: format!("failed to build blob cache: {e}"),
                })?
        } else {
            let cache_type =
                zlayer_registry::CacheType::from_env().map_err(|e| AgentError::CreateFailed {
                    id: "runtime".to_string(),
                    reason: format!("failed to read cache config from env: {e}"),
                })?;
            // Override persistent path to use config.cache_dir
            #[allow(clippy::match_wildcard_for_single_variants)]
            let cache_type = match cache_type {
                zlayer_registry::CacheType::Persistent { .. } => {
                    zlayer_registry::CacheType::persistent_at(config.cache_dir.join("blobs.redb"))
                }
                other => other,
            };
            cache_type
                .build()
                .await
                .map_err(|e| AgentError::CreateFailed {
                    id: "runtime".to_string(),
                    reason: format!("failed to build blob cache: {e}"),
                })?
        };

        let local_registry = {
            let registry_path = config.cache_dir.parent().map_or_else(
                || config.cache_dir.join("registry"),
                |data_dir| data_dir.join("registry"),
            );
            match zlayer_registry::LocalRegistry::new(registry_path).await {
                Ok(reg) => Some(std::sync::Arc::new(reg)),
                Err(e) => {
                    tracing::warn!("Failed to open local registry: {e}");
                    None
                }
            }
        };

        Ok(Self {
            config,
            containers: RwLock::new(HashMap::new()),
            auth_resolver: zlayer_core::AuthResolver::new(auth_config),
            storage_manager: std::sync::Arc::new(tokio::sync::RwLock::new(storage_manager)),
            blob_cache,
            local_registry,
            image_configs: RwLock::new(HashMap::new()),
            auth_context: None,
            secrets_provider: parking_lot::RwLock::new(None),
            readopt_done: std::sync::atomic::AtomicBool::new(false),
            overlayfs_rootfs_available: crate::capability::DaemonCapabilities::get()
                .overlayfs_rootfs_available,
            fuse_overlayfs_rootfs_available: crate::capability::DaemonCapabilities::get()
                .fuse_overlayfs_rootfs_available,
            fuse_overlayfs_bin: resolve_fuse_overlayfs_bin(),
        })
    }

    /// Configure S3-backed volume sync on the internal storage manager.
    ///
    /// When set, named volumes will be automatically registered with the
    /// `LayerSyncManager` on creation and restored from S3 if a backup
    /// exists. Volumes are synced to S3 when containers are stopped via
    /// the `sync_container_volumes` trait method.
    #[cfg(feature = "s3")]
    pub async fn set_layer_sync(
        &self,
        sync: std::sync::Arc<zlayer_storage::sync::LayerSyncManager>,
        service_name: impl Into<String>,
    ) {
        let mut storage_manager = self.storage_manager.write().await;
        storage_manager.set_layer_sync(sync, service_name);
    }

    /// Get the container ID string
    #[allow(clippy::unused_self)]
    fn container_id_str(&self, id: &ContainerId) -> String {
        format!("{}-{}", id.service, id.replica)
    }

    /// Get the root path for a container's state
    fn container_root(&self, id: &ContainerId) -> PathBuf {
        self.config.state_dir.join(self.container_id_str(id))
    }

    /// Get the bundle path for a container
    fn bundle_path(&self, id: &ContainerId) -> PathBuf {
        self.config.bundle_dir.join(self.container_id_str(id))
    }

    /// Shared per-layer overlay lowerdir store, kept a sibling of the bundle
    /// directory so the two always share a filesystem (the overlay mount needs
    /// its `upperdir`/`workdir` under the bundle and its `lowerdir`s here on the
    /// same fs, and `extract_layer_once`'s atomic publish must be a same-fs
    /// `rename(2)`). Derived from `bundle_dir.parent()` rather than the raw data
    /// dir so a `ZLAYER_BUNDLE_DIR` override relocates the layer store with it.
    /// Mirrors how `local_registry` derives its path from `cache_dir.parent()`.
    fn layer_store_dir(&self) -> PathBuf {
        self.config.bundle_dir.parent().map_or_else(
            || self.config.bundle_dir.join("layers"),
            |data_dir| data_dir.join("layers"),
        )
    }

    /// Build the container's rootfs from its staged image layers, choosing the
    /// overlayfs strategy when the daemon supports it and the legacy full-copy
    /// strategy otherwise.
    ///
    /// `layers` is the `(staged_path, media_type)` list returned by
    /// [`Self::pull_image_layer_files`], in OCI base-first order. `staging_dir`
    /// is the per-bundle `.layers` directory the pull materialized into; it is
    /// always removed before returning (the staged files are hardlinks into the
    /// blob cache, so dropping the stage frees no real bytes but keeps the bundle
    /// clean) — on the overlay path the stage is pure scratch, on the fallback
    /// path the copy has already consumed it.
    ///
    /// # Errors
    ///
    /// Surfaces a [`AgentError::CreateFailed`] if extraction / mount / copy fails.
    /// On the overlay path a mount failure first unmounts + removes the
    /// per-container `upper`/`work` so a half-built bundle is not left behind,
    /// and the error is propagated (a capability-true daemon that cannot mount is
    /// a real error, NOT a silent slide into the copy path mid-create).
    // `bundle_path` is only consumed by the Linux-only overlay arm.
    #[cfg_attr(not(target_os = "linux"), allow(unused_variables))]
    async fn build_container_rootfs(
        &self,
        container_id: &str,
        bundle_path: &Path,
        rootfs_path: &Path,
        layers: &[(PathBuf, String)],
        staging_dir: &Path,
    ) -> Result<()> {
        // Selection order: kernel overlay (root / CAP_SYS_ADMIN) → rootless
        // fuse-overlayfs (binary + /dev/fuse, no privilege) → full per-container
        // copy. The first two both give shared-layer dedup; only the last
        // duplicates every layer's bytes per container. `select_rootfs_strategy`
        // is the single source of truth for the order (unit-tested).
        #[cfg(target_os = "linux")]
        {
            use super::overlay_rootfs::{select_rootfs_strategy, RootfsStrategy};

            // The fuse path additionally needs a resolved binary; treat an
            // unresolved binary as fuse-unavailable so we slide to full copy.
            let fuse_usable =
                self.fuse_overlayfs_rootfs_available && self.fuse_overlayfs_bin.is_some();
            match select_rootfs_strategy(self.overlayfs_rootfs_available, fuse_usable) {
                RootfsStrategy::KernelOverlay => {
                    let res = self
                        .build_overlay_rootfs(container_id, bundle_path, rootfs_path, layers)
                        .await;
                    // Staged layer files are hardlinks into the blob cache; drop
                    // the scratch stage regardless of outcome.
                    let _ = tokio::fs::remove_dir_all(staging_dir).await;
                    return res;
                }
                RootfsStrategy::FuseOverlay => {
                    // `fuse_usable` guarantees the binary is Some.
                    let bin = self
                        .fuse_overlayfs_bin
                        .clone()
                        .expect("fuse_usable implies a resolved binary");
                    let res = self
                        .build_fuse_overlay_rootfs(
                            container_id,
                            bundle_path,
                            rootfs_path,
                            layers,
                            &bin,
                        )
                        .await;
                    let _ = tokio::fs::remove_dir_all(staging_dir).await;
                    return res;
                }
                RootfsStrategy::FullCopy => {}
            }
        }

        tracing::info!(
            container = %container_id,
            layer_count = layers.len(),
            "building container rootfs via full per-container copy (overlayfs unavailable)"
        );
        // Extract layers to this container's own rootfs, streaming each from disk.
        let mut unpacker = zlayer_registry::LayerUnpacker::new(rootfs_path.to_path_buf());
        let unpack_result = unpacker.unpack_layers_from_files(layers).await;
        // The staged layer files are hardlinks into the blob cache (or fresh
        // downloads already ingested into it); drop the stage either way.
        let _ = tokio::fs::remove_dir_all(staging_dir).await;
        unpack_result.map_err(|e| AgentError::CreateFailed {
            id: container_id.to_string(),
            reason: format!("failed to extract rootfs: {e}"),
        })
    }

    /// Overlay strategy for [`Self::build_container_rootfs`]: extract each layer
    /// once into the shared store, then mount the stacked lowerdirs under a fresh
    /// per-container upperdir at `rootfs_path`.
    ///
    /// The per-layer dedup key (the layer-store directory name) is the staged
    /// blob's filename, which `zlayer-registry`'s `blob_staging_filename` defines
    /// as `digest.replace([':','/'], "_")` — byte-identical to
    /// [`zlayer_paths::sanitize_digest`]. So the staged filename IS the sanitized
    /// compressed digest, and passing it as the `digest` argument to
    /// [`extract_layer_once`](super::overlay_rootfs::extract_layer_once) is
    /// correct: that function re-sanitizes (idempotently — no `:`/`/` remain) to
    /// form the same key. No manifest plumbing is needed.
    #[cfg(target_os = "linux")]
    async fn build_overlay_rootfs(
        &self,
        container_id: &str,
        bundle_path: &Path,
        rootfs_path: &Path,
        layers: &[(PathBuf, String)],
    ) -> Result<()> {
        use super::overlay_rootfs::{mount_overlay_rootfs, unmount_overlay_rootfs};

        if layers.is_empty() {
            return Err(AgentError::CreateFailed {
                id: container_id.to_string(),
                reason: "overlay rootfs requires at least one image layer".to_string(),
            });
        }

        let layer_store = self.layer_store_dir();

        // Extract each layer once into the shared store (base-first), collecting
        // the read-only lowerdirs and the sanitized digests this container pins.
        let mut lower_dirs: Vec<PathBuf> = Vec::with_capacity(layers.len());
        let mut pinned_digests: Vec<String> = Vec::with_capacity(layers.len());
        for (staged_path, media_type) in layers {
            // The staged filename == sanitize_digest(compressed_digest); see the
            // doc comment. Recover it as the dedup key / `.lowerdirs` entry.
            let digest = staged_path
                .file_name()
                .and_then(|n| n.to_str())
                .ok_or_else(|| AgentError::CreateFailed {
                    id: container_id.to_string(),
                    reason: format!(
                        "staged layer path has no usable filename: {}",
                        staged_path.display()
                    ),
                })?;
            let fs_dir = super::overlay_rootfs::extract_layer_once(
                &layer_store,
                digest,
                staged_path,
                media_type,
            )
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: container_id.to_string(),
                reason: format!("failed to extract layer {digest} into layer store: {e}"),
            })?;
            pinned_digests.push(zlayer_paths::sanitize_digest(digest));
            lower_dirs.push(fs_dir);
        }

        let upper = bundle_path.join("upper");
        let work = bundle_path.join("work");

        // Mount the overlay. On failure, unmount (idempotent) + remove the
        // per-container upper/work so no half-built mount/dir is left behind.
        if let Err(e) = mount_overlay_rootfs(&lower_dirs, &upper, &work, rootfs_path) {
            let _ = unmount_overlay_rootfs(rootfs_path);
            let _ = std::fs::remove_dir_all(&upper);
            let _ = std::fs::remove_dir_all(&work);
            return Err(AgentError::CreateFailed {
                id: container_id.to_string(),
                reason: format!("failed to mount overlay rootfs: {e}"),
            });
        }

        // Record the pinned layer digests so the layer-store GC
        // (`prune_layers`) can build the live referenced-digest set from every
        // surviving bundle. One sanitized digest per line.
        let lowerdirs_file = bundle_path.join(".lowerdirs");
        if let Err(e) = tokio::fs::write(&lowerdirs_file, pinned_digests.join("\n")).await {
            // Non-fatal: a missing `.lowerdirs` only makes GC conservatively
            // skip reclaiming this container's layers (mtime guard still
            // protects fresh layers). Warn so the omission is visible.
            tracing::warn!(
                container = %container_id,
                file = %lowerdirs_file.display(),
                error = %e,
                "failed to write .lowerdirs marker (layer GC will be conservative)"
            );
        }

        tracing::info!(
            container = %container_id,
            layer_count = lower_dirs.len(),
            layer_store = %layer_store.display(),
            "built container rootfs via overlayfs (shared layers + per-container upperdir)"
        );
        Ok(())
    }

    /// Rootless `fuse-overlayfs` strategy for [`Self::build_container_rootfs`]:
    /// extract each layer once into the shared store using the ROOTLESS whiteout
    /// convention (`fs-fuse` subdir per digest), then mount the stacked
    /// lowerdirs under a fresh per-container upperdir at `rootfs_path` via
    /// `fuse-overlayfs` — no `CAP_SYS_ADMIN` / `CAP_MKNOD` needed.
    ///
    /// Mirrors [`Self::build_overlay_rootfs`] exactly except for the extract
    /// variant ([`extract_layer_once_rootless`]) and the mount call
    /// ([`mount_fuse_overlay_rootfs`]). The `.lowerdirs` GC marker is written
    /// identically (sanitized digests, one per line): `prune_layers` removes the
    /// whole `<layer_store>/<digest>` dir for an unreferenced digest, reclaiming
    /// both the `fs` and `fs-fuse` forms together, so GC needs no backend
    /// awareness.
    #[cfg(target_os = "linux")]
    async fn build_fuse_overlay_rootfs(
        &self,
        container_id: &str,
        bundle_path: &Path,
        rootfs_path: &Path,
        layers: &[(PathBuf, String)],
        fuse_overlayfs_bin: &Path,
    ) -> Result<()> {
        use super::overlay_rootfs::{
            extract_layer_once_rootless, mount_fuse_overlay_rootfs, unmount_overlay_rootfs,
        };

        if layers.is_empty() {
            return Err(AgentError::CreateFailed {
                id: container_id.to_string(),
                reason: "fuse overlay rootfs requires at least one image layer".to_string(),
            });
        }

        let layer_store = self.layer_store_dir();

        let mut lower_dirs: Vec<PathBuf> = Vec::with_capacity(layers.len());
        let mut pinned_digests: Vec<String> = Vec::with_capacity(layers.len());
        for (staged_path, media_type) in layers {
            let digest = staged_path
                .file_name()
                .and_then(|n| n.to_str())
                .ok_or_else(|| AgentError::CreateFailed {
                    id: container_id.to_string(),
                    reason: format!(
                        "staged layer path has no usable filename: {}",
                        staged_path.display()
                    ),
                })?;
            let fs_dir = extract_layer_once_rootless(&layer_store, digest, staged_path, media_type)
                .await
                .map_err(|e| AgentError::CreateFailed {
                    id: container_id.to_string(),
                    reason: format!(
                        "failed to extract layer {digest} into rootless layer store: {e}"
                    ),
                })?;
            pinned_digests.push(zlayer_paths::sanitize_digest(digest));
            lower_dirs.push(fs_dir);
        }

        let upper = bundle_path.join("upper");
        let work = bundle_path.join("work");

        if let Err(e) =
            mount_fuse_overlay_rootfs(fuse_overlayfs_bin, &lower_dirs, &upper, &work, rootfs_path)
        {
            let _ = unmount_overlay_rootfs(rootfs_path);
            let _ = std::fs::remove_dir_all(&upper);
            let _ = std::fs::remove_dir_all(&work);
            return Err(AgentError::CreateFailed {
                id: container_id.to_string(),
                reason: format!("failed to mount rootless fuse overlay rootfs: {e}"),
            });
        }

        let lowerdirs_file = bundle_path.join(".lowerdirs");
        if let Err(e) = tokio::fs::write(&lowerdirs_file, pinned_digests.join("\n")).await {
            tracing::warn!(
                container = %container_id,
                file = %lowerdirs_file.display(),
                error = %e,
                "failed to write .lowerdirs marker (layer GC will be conservative)"
            );
        }

        tracing::info!(
            container = %container_id,
            layer_count = lower_dirs.len(),
            layer_store = %layer_store.display(),
            "built container rootfs via rootless fuse-overlayfs (shared layers + per-container upperdir)"
        );
        Ok(())
    }

    /// Build a PER-CONTAINER WRITABLE overlay view of the shared toolchain store
    /// and return the host path to RW-bind-mount at `/opt/zlayer/toolchains`.
    ///
    /// The shared toolchain cache (`tc_lower`, `{data}/toolchain-cache`) is
    /// stacked as a read-only `lowerdir` under a fresh per-container `upperdir`
    /// (`<bundle>/tc-upper` + `tc-work`); the merged view at
    /// `<bundle>/toolchains-merged` is what gets bound into the container. So
    /// every container READS the shared toolchains but its WRITES land in a
    /// private upperdir (reclaimed with the bundle) instead of mutating — and
    /// corrupting — the shared store the way the previous shared-RW bind did.
    ///
    /// The overlay backend mirrors the rootfs selection exactly
    /// ([`select_rootfs_strategy`](super::overlay_rootfs::select_rootfs_strategy)):
    /// kernel `overlay` first, then rootless `fuse-overlayfs`, then — when no
    /// overlay backend is usable, or any mount attempt fails — a last-resort RAW
    /// shared bind of `tc_lower` (shared-read, the pre-overlay behaviour). The
    /// kernel arm additionally falls back to the fuse arm on a mount error before
    /// giving up to the raw bind, so a daemon that can fuse-mount still gets a
    /// per-container view even when the kernel mount fails.
    ///
    /// Returns the merged dir on the overlay paths, or `tc_lower` (the raw shared
    /// cache) on every fallback. Never errors: a toolchain cache is best-effort
    /// CI ergonomics, so a failure degrades to the raw bind rather than failing
    /// the create. The matching teardown lives in [`unmount_bundle_rootfs`],
    /// which reconstructs `<bundle>/toolchains-merged` and unmounts it.
    #[cfg(target_os = "linux")]
    fn prepare_toolchain_overlay(&self, bundle_path: &Path, tc_lower: &Path) -> PathBuf {
        use super::overlay_rootfs::{
            mount_fuse_overlay_rootfs, mount_overlay_rootfs, select_rootfs_strategy, RootfsStrategy,
        };

        // The lowerdir (and the raw-bind fallback source) must exist. Create the
        // shared store if absent so the RW bind always has a valid source.
        if let Err(e) = std::fs::create_dir_all(tc_lower) {
            tracing::warn!(
                dir = %tc_lower.display(),
                error = %e,
                "failed to create shared toolchain cache dir; binding it raw"
            );
            return tc_lower.to_path_buf();
        }

        let tc_upper = bundle_path.join("tc-upper");
        let tc_work = bundle_path.join("tc-work");
        let tc_merged = bundle_path.join("toolchains-merged");
        for d in [&tc_upper, &tc_work, &tc_merged] {
            if let Err(e) = std::fs::create_dir_all(d) {
                tracing::warn!(
                    dir = %d.display(),
                    error = %e,
                    "failed to create toolchain overlay dir; binding shared cache raw"
                );
                return tc_lower.to_path_buf();
            }
        }

        let lowers = [tc_lower.to_path_buf()];
        // Same fuse-usability gate the rootfs path uses: an unresolved binary is
        // treated as fuse-unavailable so selection slides to the raw bind.
        let fuse_usable = self.fuse_overlayfs_rootfs_available && self.fuse_overlayfs_bin.is_some();

        // Attempt the rootless fuse mount of the toolchain lower onto `tc_merged`.
        // `stage` distinguishes the kernel-fallback vs primary log lines. Returns
        // `true` on a successful mount.
        let try_fuse = |stage: &str| -> bool {
            let Some(bin) = self.fuse_overlayfs_bin.as_deref() else {
                return false;
            };
            match mount_fuse_overlay_rootfs(bin, &lowers, &tc_upper, &tc_work, &tc_merged) {
                Ok(()) => true,
                Err(e) => {
                    tracing::warn!(
                        merged = %tc_merged.display(),
                        error = %e,
                        stage,
                        "fuse-overlayfs toolchain overlay mount failed; binding shared cache raw"
                    );
                    false
                }
            }
        };

        let mounted = match select_rootfs_strategy(self.overlayfs_rootfs_available, fuse_usable) {
            RootfsStrategy::KernelOverlay => {
                match mount_overlay_rootfs(&lowers, &tc_upper, &tc_work, &tc_merged) {
                    Ok(()) => true,
                    Err(e) => {
                        tracing::warn!(
                            merged = %tc_merged.display(),
                            error = %e,
                            "kernel overlay toolchain mount failed; trying fuse fallback"
                        );
                        try_fuse("kernel-fallback")
                    }
                }
            }
            RootfsStrategy::FuseOverlay => try_fuse("primary"),
            RootfsStrategy::FullCopy => false,
        };

        if mounted {
            return tc_merged;
        }

        // Last resort: bind the shared cache directly (shared-read), as before.
        tracing::debug!(
            lower = %tc_lower.display(),
            "toolchain overlay unavailable; binding shared cache raw (shared-read fallback)"
        );
        tc_lower.to_path_buf()
    }

    /// Non-Linux: overlay is a Linux-only feature, so the toolchain cache is
    /// bound raw (shared-read) exactly as `tc_lower` points.
    #[cfg(not(target_os = "linux"))]
    fn prepare_toolchain_overlay(&self, _bundle_path: &Path, tc_lower: &Path) -> PathBuf {
        tc_lower.to_path_buf()
    }

    /// Path to the pid-file `create_container` asks `zlayer runtime create` to
    /// write the container init PID to. Lives alongside the container's state
    /// directory so it is cleaned up by `remove_container`'s state-dir reap.
    fn pidfile_path(&self, id: &ContainerId) -> PathBuf {
        self.config
            .state_dir
            .join(format!("{}.pid", self.container_id_str(id)))
    }

    /// Resolve the absolute path to the `zlayer` binary to invoke for the
    /// `zlayer runtime <verb>` subprocess. This is THIS process: containers are
    /// driven by re-executing ourselves as a fresh, single-threaded process so
    /// libcontainer's allocating init forks from a single-threaded parent (the
    /// runc/containerd model) rather than from the ~135-thread daemon, where a
    /// forked malloc-arena lock would be orphaned. Falls back to the canonical
    /// install path when `current_exe()` cannot be resolved (mirrors the
    /// precedent in `zlayer-api`'s `internal_upgrade_start`).
    #[allow(clippy::unused_self)]
    fn runtime_binary(&self) -> PathBuf {
        std::env::current_exe().unwrap_or_else(|e| {
            tracing::warn!(
                error = %e,
                "current_exe() failed; falling back to /usr/local/bin/zlayer for `zlayer runtime`"
            );
            PathBuf::from("/usr/local/bin/zlayer")
        })
    }

    /// Run `zlayer runtime --state-root <state_dir> <args...>` as a fresh
    /// subprocess and return its captured [`Output`].
    ///
    /// This is the single chokepoint every lifecycle verb funnels through.
    /// Mirroring `Wsl2DelegateRuntime::zlayer_runtime`, the canonical
    /// `runtime --state-root <root>` prefix is prepended here so callers only
    /// pass the verb and its flags. A non-zero exit is surfaced as an
    /// [`AgentError::Internal`] carrying the captured stderr; callers that need
    /// to special-case "already stopped" inspect the returned `Output`
    /// themselves by calling [`Self::run_runtime_output`].
    async fn run_runtime(&self, args: &[&str]) -> Result<std::process::Output> {
        let output = self.run_runtime_output(args).await?;
        if !output.status.success() {
            let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
            return Err(AgentError::Internal(format!(
                "`zlayer runtime {}` failed (status {:?}): {stderr}",
                args.first().copied().unwrap_or("?"),
                output.status.code(),
            )));
        }
        Ok(output)
    }

    /// Like [`Self::run_runtime`] but returns the raw [`Output`] without
    /// treating a non-zero exit as an error, so callers (e.g. `stop_container`)
    /// can inspect stderr to decide whether a non-zero exit is benign
    /// ("already stopped"). A failure to *spawn* the process is still an error.
    async fn run_runtime_output(&self, args: &[&str]) -> Result<std::process::Output> {
        let exe = self.runtime_binary();
        let full_argv = build_runtime_argv(&self.config.state_dir, args);
        let mut cmd = tokio::process::Command::new(&exe);
        cmd.args(&full_argv);
        cmd.output().await.map_err(|e| {
            AgentError::Internal(format!(
                "failed to spawn `{} runtime {}`: {e}",
                exe.display(),
                args.first().copied().unwrap_or("?"),
            ))
        })
    }

    /// Read the container init PID from its pid-file (written by `zlayer
    /// runtime create`). Returns `None` if the file is missing or unparseable.
    async fn read_pidfile(&self, id: &ContainerId) -> Option<u32> {
        let path = self.pidfile_path(id);
        let raw = fs::read_to_string(&path).await.ok()?;
        raw.trim().parse::<u32>().ok()
    }

    /// Run `zlayer runtime state <id>` and return the reported init PID, if any.
    /// The CLI emits runc-compatible JSON on stdout (`{ ..., "pid": N, ... }`);
    /// a `pid` of `0` is treated as "no live PID" → `None`.
    async fn query_runtime_pid(&self, id: &ContainerId) -> Option<u32> {
        let container_id = self.container_id_str(id);
        let output = self
            .run_runtime_output(&["state", &container_id])
            .await
            .ok()?;
        if !output.status.success() {
            return None;
        }
        let json: serde_json::Value = serde_json::from_slice(&output.stdout).ok()?;
        let pid = json.get("pid")?.as_u64()?;
        if pid == 0 {
            None
        } else {
            u32::try_from(pid).ok()
        }
    }

    /// Get log directory for a container.
    ///
    /// When `log_base_dir` is configured, returns a structured path:
    ///   `{log_base_dir}/{deployment}/{service}/`
    /// Otherwise falls back to the bundle directory:
    ///   `{bundle_dir}/{container_id}/logs/`
    fn log_dir(&self, id: &ContainerId) -> PathBuf {
        if let Some(ref base) = self.config.log_base_dir {
            let deployment = self.config.deployment_name.as_deref().unwrap_or("default");
            base.join(deployment).join(&id.service)
        } else {
            // Fall back to bundle directory to avoid conflicting with libcontainer's state directory
            self.bundle_path(id).join("logs")
        }
    }

    /// Get log file paths for a container.
    ///
    /// Returns `(stdout_path, stderr_path)`. When structured logging is
    /// enabled via `log_base_dir`, the files are named after the container
    /// ID (e.g. `myservice-rep-1.stdout.log`).
    fn log_paths(&self, id: &ContainerId) -> (PathBuf, PathBuf) {
        let log_dir = self.log_dir(id);
        if self.config.log_base_dir.is_some() {
            let container_id = self.container_id_str(id);
            (
                log_dir.join(format!("{container_id}.stdout.log")),
                log_dir.join(format!("{container_id}.stderr.log")),
            )
        } else {
            (log_dir.join("stdout.log"), log_dir.join("stderr.log"))
        }
    }

    /// Best-effort prune of retained per-container stdout/stderr logs older than
    /// [`container_log_retention`].
    ///
    /// Walks the structured log hierarchy (`{log_base_dir}/{deployment}/{service}/`)
    /// and unlinks `*.log` files whose mtime is past the retention window so the
    /// logs of failed containers we deliberately keep (see `remove_container`)
    /// don't accumulate without bound. Logs belonging to currently-tracked
    /// containers are never pruned — a long-idle but still-running container must
    /// keep its (open) log file. A no-op without `log_base_dir`; all errors are
    /// swallowed (this is housekeeping, not a critical path).
    async fn prune_container_logs(&self) {
        let Some(base) = self.config.log_base_dir.clone() else {
            return;
        };
        let retention = container_log_retention();
        let now = std::time::SystemTime::now();

        // Paths of logs owned by live containers — never reclaim these.
        let live: std::collections::HashSet<PathBuf> = {
            let containers = self.containers.read().await;
            containers
                .values()
                .flat_map(|info| [info.stdout_path.clone(), info.stderr_path.clone()])
                .collect()
        };

        // base/<deployment>/<service>/<container>.{stdout,stderr}.log
        let Ok(mut deployments) = fs::read_dir(&base).await else {
            return;
        };
        while let Ok(Some(dep)) = deployments.next_entry().await {
            let Ok(mut services) = fs::read_dir(dep.path()).await else {
                continue;
            };
            while let Ok(Some(svc)) = services.next_entry().await {
                let Ok(mut files) = fs::read_dir(svc.path()).await else {
                    continue;
                };
                while let Ok(Some(entry)) = files.next_entry().await {
                    let path = entry.path();
                    if path.extension().is_none_or(|ext| ext != "log") || live.contains(&path) {
                        continue;
                    }
                    let stale = entry
                        .metadata()
                        .await
                        .ok()
                        .and_then(|m| m.modified().ok())
                        .and_then(|mtime| now.duration_since(mtime).ok())
                        .is_some_and(|age| age >= retention);
                    if stale {
                        let _ = fs::remove_file(&path).await;
                    }
                }
            }
        }
    }

    /// Map libcontainer status to our `ContainerState`
    #[allow(clippy::unused_self)]
    fn map_status(&self, status: ContainerStatus) -> ContainerState {
        match status {
            ContainerStatus::Creating | ContainerStatus::Created => ContainerState::Pending,
            ContainerStatus::Running => ContainerState::Running,
            ContainerStatus::Stopped => ContainerState::Exited { code: 0 },
            ContainerStatus::Paused => ContainerState::Stopping,
        }
    }

    /// Prepare the container's stdout/stderr log files and return their paths.
    ///
    /// When `log_base_dir` is configured, creates the structured log
    /// directory (`/var/log/zlayer/{deployment}/{service}/`) and places
    /// symlinks in the bundle's `logs/` directory so that existing code
    /// reading from the bundle still works.
    ///
    /// The files are created (and truncated) here so readers (`container_logs`,
    /// `logs_stream`) always find them, and the paths are then handed to
    /// `zlayer runtime create` via `--stdout`/`--stderr`; that subprocess
    /// re-opens them with the same `File::create` (create-or-truncate)
    /// semantics — see `bin/zlayer/src/commands/runtime.rs::open_output_file` —
    /// so the container's stdout/stderr land in these exact files. We no longer
    /// hand raw fds across the in-process libcontainer boundary because the
    /// lifecycle now runs in a subprocess.
    async fn create_log_files(&self, id: &ContainerId) -> Result<(PathBuf, PathBuf)> {
        let log_dir = self.log_dir(id);
        fs::create_dir_all(&log_dir)
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: id.to_string(),
                reason: format!("failed to create log dir: {e}"),
            })?;

        let (stdout_path, stderr_path) = self.log_paths(id);

        // Create (truncate) stdout file so readers always find it.
        std::fs::File::create(&stdout_path).map_err(|e| AgentError::CreateFailed {
            id: id.to_string(),
            reason: format!("failed to create stdout log: {e}"),
        })?;

        // Create (truncate) stderr file so readers always find it.
        std::fs::File::create(&stderr_path).map_err(|e| AgentError::CreateFailed {
            id: id.to_string(),
            reason: format!("failed to create stderr log: {e}"),
        })?;

        // When using structured logging, also create symlinks from the bundle
        // logs directory so that code reading from bundle_path/logs/ still works.
        if self.config.log_base_dir.is_some() {
            let bundle_log_dir = self.bundle_path(id).join("logs");
            let _ = fs::create_dir_all(&bundle_log_dir).await;

            // Symlink bundle_log_dir/stdout.log -> structured stdout_path
            let bundle_stdout = bundle_log_dir.join("stdout.log");
            let _ = fs::remove_file(&bundle_stdout).await;
            if let Err(e) = tokio::fs::symlink(&stdout_path, &bundle_stdout).await {
                tracing::warn!(
                    container = %id,
                    error = %e,
                    "Failed to create stdout symlink in bundle logs dir"
                );
            }

            // Symlink bundle_log_dir/stderr.log -> structured stderr_path
            let bundle_stderr = bundle_log_dir.join("stderr.log");
            let _ = fs::remove_file(&bundle_stderr).await;
            if let Err(e) = tokio::fs::symlink(&stderr_path, &bundle_stderr).await {
                tracing::warn!(
                    container = %id,
                    error = %e,
                    "Failed to create stderr symlink in bundle logs dir"
                );
            }
        }

        Ok((stdout_path, stderr_path))
    }

    /// Clean up bundle directory for a container.
    ///
    /// A bundle contains the extracted image rootfs, which routinely includes
    /// read-only directories (e.g. Fedora's `0o555` `ca-trust`). A plain
    /// `remove_dir_all` fails on those with `EACCES`, leaving a partial tree
    /// behind that never reclaims (the disk-leak the orphan GC works around).
    /// So we make the whole tree owner-writable first — the same approach the
    /// orphan GC (`prune_containers`) uses via
    /// [`zlayer_paths::safe_fs::chmod_tree_writable`]. On a residual failure we
    /// chmod + retry once, and on a second failure log the leftover path and
    /// its on-disk byte count so the leak is visible rather than silent.
    async fn cleanup_bundle(&self, id: &ContainerId) -> Result<()> {
        let bundle_path = self.bundle_path(id);
        if !bundle_path.exists() {
            return Ok(());
        }
        let id_str = id.to_string();
        tokio::task::spawn_blocking(move || cleanup_bundle_blocking(&bundle_path, &id_str))
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: id.to_string(),
                reason: format!("cleanup_bundle join error: {e}"),
            })?
    }

    /// Pull image layers and return them for extraction
    ///
    /// Uses the shared blob cache to avoid repeated network requests for cached layers.
    /// The `policy: PullPolicy` is forwarded straight to the puller: `PullPolicy::Always`
    /// clears the manifest cache before fetching, `PullPolicy::Newer` revalidates
    /// mutable tags via HEAD, and `PullPolicy::IfNotPresent` / `PullPolicy::Never`
    /// trust the local cache without revalidating against the remote.
    ///
    /// `PullPolicy::Never` short-circuits to a local-cache-only path. The puller is
    /// invoked with the same policy so it consults the local registry and blob cache
    /// without any remote HEAD revalidation. If the image is not present locally and
    /// the puller falls through to a remote fetch that fails, the error is remapped
    /// to a Never-specific message so callers can distinguish "missing locally" from
    /// a transient network failure. With the Phase 0 import fix, locally-imported
    /// images always satisfy the local lookup and no remote round-trip occurs.
    async fn pull_image_layer_files(
        &self,
        image: &str,
        policy: zlayer_spec::PullPolicy,
        staging_dir: &Path,
    ) -> Result<Vec<(PathBuf, String)>> {
        // Use the shared blob cache instead of opening a new one each time.
        // The central constructor wires the S3 tier + default registry from
        // env; no per-image source is in scope here, so use the default.
        let mut puller = zlayer_registry::ImagePuller::from_env_for_runtime(
            self.blob_cache.clone(),
            zlayer_spec::SourcePolicy::default(),
        )
        .await;
        if let Some(ref registry) = self.local_registry {
            puller = puller.with_local_registry(registry.clone());
        }
        let auth = self.auth_resolver.resolve(image);

        if matches!(policy, zlayer_spec::PullPolicy::Never) {
            tracing::debug!(
                image = %image,
                "pull_policy=Never; serving layers from local cache only"
            );
            return puller
                .pull_image_to_files_with_policy(image, &auth, staging_dir, policy)
                .await
                .map_err(|e| AgentError::PullFailed {
                    image: image.to_string(),
                    reason: format!("pull_policy=never and image not present locally: {e}"),
                });
        }

        puller
            .pull_image_to_files_with_policy(image, &auth, staging_dir, policy)
            .await
            .map_err(|e| AgentError::PullFailed {
                image: image.to_string(),
                reason: format!("failed to pull image: {e}"),
            })
    }

    /// Prepare storage volumes for a container, returning paths for mounts.
    ///
    /// Delegates to the runtime-agnostic [`crate::runtimes::volume_prep`] helper
    /// so youki, the macOS runtimes, and the WSL2 delegate all materialise
    /// volumes identically.
    async fn prepare_storage_volumes(
        &self,
        id: &ContainerId,
        spec: &ServiceSpec,
    ) -> Result<std::collections::HashMap<String, PathBuf>> {
        let container_id = self.container_id_str(id);
        let mut storage_manager = self.storage_manager.write().await;
        crate::runtimes::volume_prep::prepare_storage_volumes(
            &mut storage_manager,
            &container_id,
            &spec.storage,
        )
        .await
    }

    /// Clean up storage volumes for a container
    ///
    /// Note: This method requires the `ServiceSpec` to know which volumes to clean up.
    /// For now, `remove_container` uses a simpler approach that only cleans up anonymous volumes.
    /// This method is available for future use when the spec is stored/available at removal time.
    #[allow(dead_code)]
    async fn cleanup_storage_volumes(&self, id: &ContainerId, spec: &ServiceSpec) -> Result<()> {
        use zlayer_spec::StorageSpec;

        let container_id = self.container_id_str(id);
        let mut storage_manager = self.storage_manager.write().await;

        // Detach named volumes
        for storage in &spec.storage {
            match storage {
                StorageSpec::Named { name, .. } => {
                    if let Err(e) = storage_manager.detach_volume(name, &container_id) {
                        tracing::warn!(
                            volume = %name,
                            container = %container_id,
                            error = %e,
                            "failed to detach volume"
                        );
                    }
                }
                StorageSpec::S3 { bucket, prefix, .. } => {
                    if let Err(e) =
                        storage_manager.unmount_s3(bucket, prefix.as_deref(), &container_id)
                    {
                        tracing::warn!(
                            bucket = %bucket,
                            container = %container_id,
                            error = %e,
                            "failed to unmount S3 bucket"
                        );
                    }
                }
                _ => {}
            }
        }

        // Clean up anonymous volumes
        if let Err(e) = storage_manager.cleanup_anonymous(&container_id) {
            tracing::warn!(
                container = %container_id,
                error = %e,
                "failed to cleanup anonymous volumes"
            );
        }

        Ok(())
    }

    /// Get a cached image config by image reference
    ///
    /// Returns the previously pulled image configuration (entrypoint, cmd, env, etc.)
    /// for the given image reference, if available.
    async fn get_image_config(&self, image: &str) -> Option<zlayer_registry::ImageConfig> {
        let configs = self.image_configs.read().await;
        configs.get(image).cloned()
    }

    /// Arm exactly one `waitpid` exit watcher for `cid`'s init `pid`.
    ///
    /// Idempotent per container: a `compare_exchange` on the container's
    /// `watcher_armed` flag ensures only the FIRST caller spawns the watcher,
    /// so `start_container` and boot re-adoption can both call this without
    /// ever stacking two reapers on the same pid. When the init terminates the
    /// watcher records the real exit code into the in-process slot AND tears
    /// down the per-container Docker socket (revoking the scoped token), so an
    /// exit-without-`remove_container` no longer leaks the UDS task/socket.
    ///
    /// Returns `true` if THIS call armed the watcher, `false` if it was already
    /// armed (or the container isn't tracked / the pid doesn't fit an `i32`).
    async fn arm_exit_watcher(&self, cid: &str, pid: u32) -> bool {
        let Ok(pid_i32) = i32::try_from(pid) else {
            return false;
        };
        // Grab the slot + the arm-guard under the read lock.
        let armed = {
            let containers = self.containers.read().await;
            let Some(info) = containers.get(cid) else {
                return false;
            };
            (
                std::sync::Arc::clone(&info.exit_code),
                std::sync::Arc::clone(&info.watcher_armed),
            )
        };
        let (slot, guard) = armed;
        // Win the race to arm: only the transition false -> true proceeds.
        if guard
            .compare_exchange(
                false,
                true,
                std::sync::atomic::Ordering::AcqRel,
                std::sync::atomic::Ordering::Acquire,
            )
            .is_err()
        {
            tracing::debug!(container = %cid, pid, "exit watcher already armed; skipping");
            return false;
        }

        let cid_owned = cid.to_string();
        // Clone the docker-socket spawner + token sink so the watcher can fire
        // their teardown on exit (the same cleanup `remove_container` does).
        let docker_spawner = self
            .auth_context
            .as_ref()
            .and_then(|ctx| ctx.docker_socket_spawner.clone());
        let token_sink = self
            .auth_context
            .as_ref()
            .and_then(|ctx| ctx.token_sink.clone());
        // Reconstruct the jti exactly as create_container minted it
        // (`container:{service}:{container_id}`). `cid` is `{service}-{replica}`
        // and the service is `cid` with the trailing `-{replica}` stripped.
        let service = cid.rsplit_once('-').map_or(cid, |(svc, _)| svc).to_string();

        tokio::spawn(async move {
            if let Some(code) = watch_exit_code(pid_i32).await {
                if let Ok(mut g) = slot.lock() {
                    *g = Some(code);
                }
                tracing::debug!(container = %cid_owned, exit_code = code, "captured container init exit code");
            }
            // Init is gone: tear down its per-container docker socket + revoke
            // its scoped token so an exit-without-remove doesn't leak them.
            if let Some(spawner) = docker_spawner {
                spawner.teardown(&cid_owned).await;
            }
            if let Some(sink) = token_sink {
                let jti = format!("container:{service}:{cid_owned}");
                sink.revoke(&jti).await;
            }
        });
        true
    }

    /// Re-adopt exit watchers for every container that survives on disk with a
    /// LIVE init across a daemon restart.
    ///
    /// On restart the in-memory `containers` map is empty, so no watcher is
    /// armed for the inits that reparented to this daemon. Those reparented
    /// inits linger holding their stdout/stderr log fds (pinning deleted log
    /// inodes — the ENOSPC leak) and, if dead, sit as unreaped zombies. This
    /// walks the on-disk state directories, reads each container's init PID
    /// (pid-file, then libcontainer state), repopulates a `ContainerInfo`, and
    /// arms exactly one watcher per live pid via [`Self::arm_exit_watcher`]
    /// (which guards against double-arming). Reaping a dead init releases its
    /// fds; a live init becomes tracked so kill-on-remove works. Containers
    /// already tracked in memory are skipped (the arm-guard double-checks).
    ///
    /// Returns the number of watchers newly armed. Safe to call repeatedly.
    /// Exposed through the `Runtime::readopt_exit_watchers` trait method.
    async fn readopt_exit_watchers_impl(&self) -> usize {
        let state_dir = self.config.state_dir.clone();
        let bundle_dir = self.config.bundle_dir.clone();

        // Enumerate (name, pid) for on-disk containers with a readable init pid
        // on a blocking thread (read_dir + libcontainer state are sync).
        let candidates: Vec<(String, u32)> = tokio::task::spawn_blocking(move || {
            let mut out: Vec<(String, u32)> = Vec::new();
            let Ok(entries) = std::fs::read_dir(&state_dir) else {
                return out;
            };
            for entry in entries.flatten() {
                let path = entry.path();
                let is_dir = std::fs::symlink_metadata(&path)
                    .is_ok_and(|md| md.is_dir() && !md.file_type().is_symlink());
                if !is_dir {
                    continue;
                }
                let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
                    continue;
                };
                if let Some(pid) = read_init_pid_for(&state_dir, &path, name) {
                    out.push((name.to_string(), pid));
                }
            }
            out
        })
        .await
        .unwrap_or_default();

        let mut armed = 0usize;
        for (name, pid) in candidates {
            // Skip a container the daemon already tracks (e.g. a concurrent
            // create racing boot); arm_exit_watcher's guard is the backstop.
            {
                let containers = self.containers.read().await;
                if containers.contains_key(&name) {
                    continue;
                }
            }
            // Repopulate a minimal ContainerInfo so the watcher has a slot to
            // record into and remove_container can later unlink the logs. The
            // service/replica are recovered from the dir name; role/node_id are
            // path-irrelevant (container_id_str uses only service+replica).
            let (service, replica) = name
                .rsplit_once('-')
                .and_then(|(svc, rep)| rep.parse::<u32>().ok().map(|r| (svc.to_string(), r)))
                .unwrap_or_else(|| (name.clone(), 0));
            let id = ContainerId::new(service, replica);
            let (stdout_path, stderr_path) = self.log_paths(&id);
            let bundle_path = bundle_dir.join(&name);
            let rootfs_path = bundle_path.join("rootfs");
            {
                let mut containers = self.containers.write().await;
                containers
                    .entry(name.clone())
                    .or_insert_with(|| ContainerInfo {
                        image: String::new(),
                        bundle_path,
                        rootfs_path,
                        stdout_path,
                        stderr_path,
                        pid: Some(pid),
                        restart_policy: None,
                        exit_code: std::sync::Arc::new(std::sync::Mutex::new(None)),
                        watcher_armed: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(
                            false,
                        )),
                    });
            }
            // Arm; the compare_exchange guard inside makes this exactly-once
            // even if two boot paths race, and returns true only for the call
            // that actually armed — so the count reflects real new watchers.
            if self.arm_exit_watcher(&name, pid).await {
                armed += 1;
            }
        }

        if armed > 0 {
            tracing::info!(
                armed,
                "re-adopted exit watchers for surviving container inits"
            );
        }
        armed
    }
}

/// Top-level OCI image config shape: the runtime-defaults live in the nested
/// `config` object; `os`/`architecture`/`created` are top-level.
#[derive(serde::Deserialize)]
struct ConfigBlob {
    #[serde(default)]
    config: Option<zlayer_registry::ImageConfig>,
    #[serde(default)]
    os: Option<String>,
    #[serde(default)]
    architecture: Option<String>,
    #[serde(default)]
    created: Option<String>,
}

/// Apply an OCI image config blob's runtime metadata onto an inspect record.
fn apply_config_blob(info: &mut ImageInspectInfo, blob: &ConfigBlob) {
    if let Some(cfg) = blob.config.clone() {
        info.env = cfg.env.unwrap_or_default();
        info.cmd = cfg.cmd.unwrap_or_default();
        info.entrypoint = cfg.entrypoint.unwrap_or_default();
        info.working_dir = cfg.working_dir.filter(|s| !s.is_empty());
        info.user = cfg.user;
        info.labels = cfg.labels.unwrap_or_default().into_iter().collect();
    }
    info.os = blob.os.clone().filter(|s| !s.is_empty());
    info.architecture = blob.architecture.clone().filter(|s| !s.is_empty());
    info.created.clone_from(&blob.created);
}

#[async_trait::async_trait]
impl Runtime for YoukiRuntime {
    /// Store the secrets provider so `create_container` can resolve `$S:` env
    /// references against each service's typed `secret_scope`.
    fn set_secrets_provider(&self, provider: std::sync::Arc<dyn zlayer_secrets::SecretsProvider>) {
        *self.secrets_provider.write() = Some(provider);
    }

    /// Hand the API's `POST /images/import` handler the SAME `LocalRegistry` +
    /// blob cache this runtime already holds open, so an import writes layers
    /// into the exact store the daemon serves from (and a later
    /// restore/recreate finds them locally). Opening a second blob cache here
    /// would fail — the ZQL/redb backend is single-process exclusive — so the
    /// only correct path is to share these `Arc`s. Returns `None` only when the
    /// local registry failed to open at construction.
    fn image_store_handles(&self) -> Option<crate::runtime::ImageStoreHandles> {
        self.local_registry
            .as_ref()
            .map(|registry| (registry.clone(), self.blob_cache.clone()))
    }

    /// Pull an image to local storage
    ///
    /// Downloads image layers from a registry and unpacks them to a rootfs.
    #[instrument(
        skip(self),
        fields(
            otel.name = "image.pull",
            container.image.name = %image,
        )
    )]
    async fn pull_image(&self, image: &str) -> Result<()> {
        self.pull_image_with_policy(
            image,
            zlayer_spec::PullPolicy::IfNotPresent,
            None,
            zlayer_spec::SourcePolicy::default(),
        )
        .await
    }

    /// Pull an image to local storage with a specific pull policy
    ///
    /// This downloads image layers to the blob cache. Layers are extracted
    /// per-container in `create_container` to avoid race conditions.
    ///
    /// Caller-supplied `auth_in` (inline spec creds, or a daemon-resolved stored
    /// credential by registry host) is HONORED: it wins over the hostname-based
    /// [`AuthResolver`], which is only the fallback when no auth is passed. This
    /// is the registry-auth passthrough — without it, an authenticated private
    /// pull resolved by the daemon handler was silently dropped here and failed.
    #[instrument(
        skip(self, auth_in, source),
        fields(
            otel.name = "image.pull",
            container.image.name = %image,
            pull_policy = ?policy,
            source_policy = ?source,
        )
    )]
    async fn pull_image_with_policy(
        &self,
        image: &str,
        policy: zlayer_spec::PullPolicy,
        auth_in: Option<&RegistryAuth>,
        source: zlayer_spec::SourcePolicy,
    ) -> Result<()> {
        // Central constructor: wires the S3 tier + default registry from env AND
        // sets the per-image source policy. Chain the local registry when present.
        let mut puller =
            zlayer_registry::ImagePuller::from_env_for_runtime(self.blob_cache.clone(), source)
                .await;
        if let Some(ref registry) = self.local_registry {
            puller = puller.with_local_registry(registry.clone());
        }
        // Honor caller-supplied auth (inline / daemon-resolved by host); fall
        // back to the hostname-based AuthResolver when none was passed.
        let auth = match auth_in {
            Some(a) => zlayer_registry::spec_auth_to_oci(Some(a)),
            None => self.auth_resolver.resolve(image),
        };

        // For Never policy, skip pulling layers from the remote, but STILL
        // fetch the image config from the local blob cache (populated by a
        // prior `zlayer import` or an earlier pull). Without the image
        // config, the bundle builder has no way to know the image's
        // entrypoint/cmd/env/workdir/user and falls back to `/bin/sh`,
        // which exits immediately and kills the container. Fetching the
        // config from cache is cheap (~1 KB) and non-fatal on miss.
        if matches!(policy, zlayer_spec::PullPolicy::Never) {
            tracing::debug!(image = %image, "pull policy is Never, skipping layer pull");
            match puller.pull_image_config(image, &auth).await {
                Ok(config) => {
                    tracing::info!(
                        image = %image,
                        has_entrypoint = config.entrypoint.is_some(),
                        has_cmd = config.cmd.is_some(),
                        "image config loaded from cache"
                    );
                    let mut configs = self.image_configs.write().await;
                    configs.insert(image.to_string(), config);
                }
                Err(e) => {
                    tracing::warn!(
                        image = %image,
                        error = %e,
                        "failed to load image config from cache under pull_policy=Never, container will use spec defaults"
                    );
                }
            }
            return Ok(());
        }

        // For IfNotPresent, check if image layers are in cache by trying to pull
        // Use the shared blob cache to avoid repeated opens and ensure persistence
        // For Always, force a round-trip to the registry by clearing the manifest cache.
        tracing::info!(image = %image, ?policy, "pulling image layers to cache");

        // Prefetch warms the shared blob cache; the actual rootfs extraction
        // happens per-container in `create_container`. Stream layers to a
        // throwaway stage so no layer is ever fully buffered in RAM (cache is
        // warmed as a side effect of `pull_image_to_files`); then drop the stage
        // (its files are just hardlinks into the cache).
        let seq = PREFETCH_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let stage = self.config.rootfs_dir.join(".prefetch").join(format!(
            "{}.{}",
            std::process::id(),
            seq
        ));
        let _ = tokio::fs::remove_dir_all(&stage).await;
        let layers = puller
            .pull_image_to_files_with_policy(image, &auth, &stage, policy)
            .await
            .map_err(|e| AgentError::PullFailed {
                image: image.to_string(),
                reason: format!("failed to pull image: {e}"),
            })?;
        let _ = tokio::fs::remove_dir_all(&stage).await;

        tracing::info!(
            image = %image,
            layer_count = layers.len(),
            "image layers cached"
        );

        // Also pull and cache the image config (entrypoint, cmd, env, etc.)
        match puller
            .pull_image_config_with_policy(image, &auth, policy)
            .await
        {
            Ok(config) => {
                tracing::info!(
                    image = %image,
                    has_entrypoint = config.entrypoint.is_some(),
                    has_cmd = config.cmd.is_some(),
                    "image config cached"
                );
                let mut configs = self.image_configs.write().await;
                configs.insert(image.to_string(), config);
            }
            Err(e) => {
                // Log but don't fail - the container can still run with spec defaults
                tracing::warn!(
                    image = %image,
                    error = %e,
                    "failed to pull image config, container will use spec defaults"
                );
            }
        }

        Ok(())
    }

    /// Create a container
    ///
    /// Creates an OCI bundle and uses libcontainer to create the container.
    /// Each container gets its own rootfs extracted from cached layers.
    #[instrument(
        skip(self, spec),
        fields(
            otel.name = "container.create",
            container.id = %self.container_id_str(id),
            service.name = %id.service,
            service.replica = %id.replica,
            container.image.name = %spec.image.name,
        )
    )]
    async fn create_container(&self, id: &ContainerId, spec: &ServiceSpec) -> Result<()> {
        let container_id = self.container_id_str(id);
        let image = spec.image.name.to_string();

        // Clean up any stale container from a previous deploy (mirrors Docker runtime behavior)
        // See docker.rs:370-402 for the equivalent pattern
        let container_root = self.container_root(id);
        if container_root.exists() {
            tracing::warn!(
                container = %container_id,
                "stale container state found from previous deploy, cleaning up before re-create"
            );
            // Try to stop — process may already be dead, ignore errors
            if let Err(e) = self.stop_container(id, Duration::from_secs(5)).await {
                tracing::debug!(
                    container = %container_id,
                    error = %e,
                    "stop_container during stale cleanup (expected if process already dead)"
                );
            }
            // Remove container — does full cleanup: libcontainer delete + state dir + bundle + volumes
            if let Err(e) = self.remove_container(id).await {
                tracing::warn!(
                    container = %container_id,
                    error = %e,
                    "remove_container failed during stale cleanup, attempting manual cleanup"
                );
                // Fall back to manual cleanup if remove_container fails
                if container_root.exists() {
                    let _ = tokio::fs::remove_dir_all(&container_root).await;
                }
                let stale_bundle = self.bundle_path(id);
                if stale_bundle.exists() {
                    // Unmount a possible overlay rootfs before removing the tree.
                    unmount_bundle_rootfs(&stale_bundle);
                    let _ = tokio::fs::remove_dir_all(&stale_bundle).await;
                }
                // Also reap any stale host-mode container cgroup so the
                // recreate starts from a clean `<root>/containers/<id>` slot
                // (a lingering empty cgroup trips libcontainer's build()).
                #[cfg(target_os = "linux")]
                if !self.config.use_systemd {
                    crate::capability::remove_host_container_cgroup(&container_id);
                }
            }
            tracing::info!(container = %container_id, "stale container cleaned up");
        }

        // Also clean up stale bundle directory if it exists but state dir didn't
        let bundle_path = self.bundle_path(id);
        if bundle_path.exists() {
            tracing::warn!(container = %container_id, "stale bundle directory found, removing");
            // Unmount a possible overlay rootfs before removing the tree.
            unmount_bundle_rootfs(&bundle_path);
            let _ = tokio::fs::remove_dir_all(&bundle_path).await;
        }

        let bundle_path = self.bundle_path(id);
        let rootfs_path = bundle_path.join("rootfs");

        tracing::info!("Creating container {} from image {}", container_id, image);

        // Create bundle directory structure
        fs::create_dir_all(&bundle_path)
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: container_id.clone(),
                reason: format!("failed to create bundle directory: {e}"),
            })?;

        // Pull image layers to files (from cache if available), streaming so no
        // layer is ever fully buffered in RAM. Honor the spec's pull policy so
        // that `Always` forces a manifest refresh, `IfNotPresent` serves from cache with
        // mutable-tag revalidation, and `Never` reuses cached blobs.
        let staging_dir = bundle_path.join(".layers");
        let layers = self
            .pull_image_layer_files(&image, spec.image.pull_policy, &staging_dir)
            .await?;

        // Build the container's rootfs. Two strategies, chosen per-create from the
        // cached overlayfs capability:
        //  * overlay (capability true): extract each image layer ONCE into the
        //    shared layer store and stack those read-only lowerdirs under a fresh
        //    per-container upperdir via an `overlay` mount at `rootfs`. N
        //    containers of one image then share a single copy of every layer and
        //    pay only the cheap mount + an (initially empty) upperdir.
        //  * fallback (capability false): the legacy full COPY — unpack every
        //    layer into this container's own `rootfs` tree.
        // Either way, `root.path="rootfs"` (written into the bundle config below)
        // points at the prepared tree, so everything downstream is unchanged.
        self.build_container_rootfs(
            &container_id,
            &bundle_path,
            &rootfs_path,
            &layers,
            &staging_dir,
        )
        .await?;

        // Log rootfs diagnostics for debugging container creation failures
        match std::fs::read_dir(&rootfs_path) {
            Ok(entries) => {
                let count = entries.count();
                tracing::info!(
                    container = %container_id,
                    rootfs = %rootfs_path.display(),
                    entry_count = count,
                    "rootfs extracted successfully"
                );
            }
            Err(e) => {
                tracing::warn!(
                    container = %container_id,
                    rootfs = %rootfs_path.display(),
                    error = %e,
                    "rootfs directory not readable after extraction"
                );
            }
        }

        // Get cached image config (entrypoint, cmd, env, workdir, user)
        let img_config = self.get_image_config(&image).await;

        // Prepare storage volumes
        let volume_paths = self.prepare_storage_volumes(id, spec).await?;

        // Resolve `NetworkMode::Container { id }` (Docker `--network
        // container:<id>`) to the target's network-namespace path so the bundle
        // JOINS that netns instead of unsharing a fresh one. The handler has
        // already rewritten the `id` to a concrete `ContainerId` Display string
        // (it owns the docker-name → ContainerId map), so here we only have to
        // parse it back and ask the runtime for the target's live PID. A missing
        // target / no PID is a hard create failure — we never silently fall back
        // to an isolated netns, because the caller explicitly asked to share.
        //
        // Mutually exclusive with `host_network`: `NetworkMode::Host` sets
        // `spec.host_network = true` and never produces `Container`, so the two
        // never both fire. We pass `None` to `with_netns_path` in every
        // non-`Container` mode.
        let netns_path = match &spec.network_mode {
            zlayer_spec::NetworkMode::Container { id } => {
                let target_cid =
                    ContainerId::parse_display(id).ok_or_else(|| AgentError::CreateFailed {
                        id: container_id.clone(),
                        reason: format!(
                            "network target container {id:?} is not a resolvable container id"
                        ),
                    })?;
                let pid = self.get_container_pid(&target_cid).await.map_err(|e| {
                    AgentError::CreateFailed {
                        id: container_id.clone(),
                        reason: format!(
                            "network target container {id} not found or not running: {e}"
                        ),
                    }
                })?;
                let pid = pid.ok_or_else(|| AgentError::CreateFailed {
                    id: container_id.clone(),
                    reason: format!("network target container {id} is not running (no pid)"),
                })?;
                Some(std::path::PathBuf::from(format!("/proc/{pid}/ns/net")))
            }
            // `Bridge { name }` / `Default` => fresh netns; the overlay attach
            // path applies the isolation-network fence via the spec label.
            _ => None,
        };

        // Shared toolchain cache, exposed at /opt/zlayer/toolchains so CI
        // workloads (setup-* actions) reuse downloaded toolchains across runs.
        // Mounted unconditionally — non-CI containers simply ignore it. Derived
        // from the data dir (bundle_dir's parent) so a `ZLAYER_BUNDLE_DIR`
        // override relocates it too, mirroring `layer_store_dir`.
        //
        // The container does NOT bind the shared store read-write directly (that
        // let one container corrupt every other's toolchains). Instead each
        // container gets a PER-CONTAINER WRITABLE overlay view: the shared store
        // is a read-only lowerdir and the container's writes land in a private
        // per-bundle upperdir (reclaimed on teardown). `prepare_toolchain_overlay`
        // mirrors the rootfs's kernel→fuse→raw-bind fallback chain and returns the
        // host path to bind (the merged dir on the overlay paths, the raw shared
        // cache as a last resort).
        let toolchain_cache_dir = self.config.bundle_dir.parent().map_or_else(
            || self.config.bundle_dir.join("toolchain-cache"),
            |data_dir| zlayer_paths::ZLayerDirs::new(data_dir).toolchain_cache(),
        );
        let toolchain_mount = self.prepare_toolchain_overlay(&bundle_path, &toolchain_cache_dir);

        // Generate OCI config.json via BundleBuilder (handles capabilities, devices,
        // resource limits, storage mounts, env resolution, and command resolution)
        let mut bundle_builder = crate::bundle::BundleBuilder::new(bundle_path.clone())
            .with_volume_paths(volume_paths)
            .with_host_network(spec.host_network)
            .with_netns_path(netns_path)
            .with_toolchain_cache(toolchain_mount);
        if let Some(config) = img_config {
            bundle_builder = bundle_builder.with_image_config(config);
        }

        // Inject the secrets provider + the service's RESOLVED env scope so the
        // bundle builder can resolve `$S:` env references. Clone the provider OUT
        // of the lock immediately so no guard is held across any later `.await`.
        // Skip when no provider is wired or the service declared no environment
        // (`secret_scope` is None) — there is nothing to resolve against.
        let secrets_provider = self.secrets_provider.read().clone();
        if let (Some(provider), Some(scope)) = (secrets_provider, spec.secret_scope.clone()) {
            bundle_builder = bundle_builder
                .with_secrets_provider(provider)
                .with_deployment_scope(scope);
        }

        // Inject auth env so the container can talk to the host API with a
        // LEAST-PRIVILEGE scoped token (read-only on its own deployment by
        // default; broaden via `zlayer.io/api-scopes`). The host admin Unix
        // socket is NOT mounted unless the service explicitly 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. If persistence
            // fails, mint without a jti so the token is still accepted (bounded
            // by its 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| crate::error::AgentError::CreateFailed {
                id: id.to_string(),
                reason: format!("Failed to mint container token: {e}"),
            })?;
            bundle_builder = bundle_builder
                .with_env("ZLAYER_API_URL".to_string(), auth_ctx.api_url.clone())
                .with_env("ZLAYER_TOKEN".to_string(), token.clone());
            if access.mount_socket {
                bundle_builder = bundle_builder
                    .with_env(
                        "ZLAYER_SOCKET".to_string(),
                        zlayer_paths::ZLayerDirs::default_socket_path(),
                    )
                    .with_socket_mount(&auth_ctx.socket_path);
            }
            // Per-container Docker Engine API socket: provision a UDS backed by
            // THIS container's scoped token and bind-mount it at
            // /var/run/docker.sock so `docker` / `docker buildx` inside the
            // container drive ZLayer as the build daemon, scoped to the
            // container's own identity. Default for every container (no longer
            // label-gated) — the inner guard degrades gracefully when no
            // spawner is configured (daemon built without docker-compat).
            if let Some(spawner) = auth_ctx.docker_socket_spawner.as_ref() {
                if let Some(host_sock) = spawner.spawn(&container_id, token).await {
                    bundle_builder = bundle_builder
                        .with_docker_socket_mount(host_sock)
                        .with_env(
                            "DOCKER_HOST".to_string(),
                            "unix:///var/run/docker.sock".to_string(),
                        )
                        .with_env("DOCKER_BUILDKIT".to_string(), "0".to_string());
                } else {
                    tracing::warn!(
                        container = %container_id,
                        "the per-container Docker socket spawner failed; container \
                         starts without a docker socket"
                    );
                }
            } else {
                tracing::warn!(
                    container = %container_id,
                    "no DockerSocketSpawner is configured (daemon built without \
                     docker-compat?); container starts without a docker socket"
                );
            }
        }

        bundle_builder.write_config(id, spec).await?;

        // Create log files (paths only — the `zlayer runtime create` subprocess
        // re-opens them via `--stdout`/`--stderr`).
        let (stdout_path, stderr_path) = self.create_log_files(id).await?;

        // Drive container creation through a FRESH `zlayer runtime create`
        // subprocess rather than calling libcontainer's `ContainerBuilder::build`
        // in-process. The daemon is a ~135-thread process; libcontainer's init
        // forks an allocating child, and a fork from a multi-threaded parent
        // orphans glibc's malloc-arena lock in the child → GP fault → create
        // fails. A fresh `zlayer runtime` process is single-threaded at fork
        // time, so the fork is safe (the runc/containerd model). The subprocess
        // writes the init PID to `--pid-file` right after `build()`, which we
        // read back below so the exit watcher can be keyed on it.
        let bundle_str = bundle_path.to_string_lossy().into_owned();
        let stdout_str = stdout_path.to_string_lossy().into_owned();
        let stderr_str = stderr_path.to_string_lossy().into_owned();
        let pidfile = self.pidfile_path(id);
        let pidfile_str = pidfile.to_string_lossy().into_owned();
        // Stale pid-file from a prior incarnation must not be read as this one.
        let _ = fs::remove_file(&pidfile).await;

        let mut args: Vec<&str> = vec![
            "create",
            &container_id,
            "--bundle",
            &bundle_str,
            "--stdout",
            &stdout_str,
            "--stderr",
            &stderr_str,
            "--pid-file",
            &pidfile_str,
        ];
        if self.config.use_systemd {
            args.push("--systemd-cgroup");
        }
        if let Err(e) = self.run_runtime(&args).await {
            // Roll back the bundle so a retry sees a clean slate, matching the
            // wsl2_delegate create path.
            return Err(AgentError::CreateFailed {
                id: container_id.clone(),
                reason: format!("zlayer runtime create failed: {e}"),
            });
        }

        // Read the init PID the subprocess persisted to the pid-file.
        let created_pid = self.read_pidfile(id).await;

        // Store container info
        {
            let mut containers = self.containers.write().await;
            containers.insert(
                container_id.clone(),
                ContainerInfo {
                    image: image.clone(),
                    bundle_path,
                    rootfs_path,
                    stdout_path,
                    stderr_path,
                    pid: created_pid,
                    restart_policy: None,
                    exit_code: std::sync::Arc::new(std::sync::Mutex::new(None)),
                    watcher_armed: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
                },
            );
        }

        tracing::info!(
            container = %container_id,
            pid = ?created_pid,
            "Container created successfully"
        );
        Ok(())
    }

    /// Start a container
    ///
    /// Starts the container's init process.
    #[instrument(
        skip(self),
        fields(
            otel.name = "container.start",
            container.id = %self.container_id_str(id),
            service.name = %id.service,
        )
    )]
    async fn start_container(&self, id: &ContainerId) -> Result<()> {
        let container_id = self.container_id_str(id);

        tracing::info!("Starting container {}", container_id);

        // Start the container via a fresh `zlayer runtime start` subprocess.
        self.run_runtime(&["start", &container_id])
            .await
            .map_err(|e| AgentError::StartFailed {
                id: id.to_string(),
                reason: format!("zlayer runtime start failed: {e}"),
            })?;

        // Resolve the init PID for the exit watcher. Prefer the PID persisted to
        // the pid-file at create (the init's PID, stable across the create→start
        // transition); fall back to the live `state` JSON if the file is gone.
        let pid = match self.read_pidfile(id).await {
            Some(p) => Some(p),
            None => self.query_runtime_pid(id).await,
        };

        // Update container info with PID
        {
            let mut containers = self.containers.write().await;
            if let Some(info) = containers.get_mut(&self.container_id_str(id)) {
                info.pid = pid;
            }
        }

        // Arm the per-container exit watcher (see `arm_exit_watcher`). Because
        // the lifecycle runs in a subprocess, the init is forked by the
        // short-lived `zlayer runtime create` process and reparents to the
        // daemon (a child subreaper — see `init_daemon`), so this daemon's
        // `waitpid` succeeds, records the real exit code, and prevents the init
        // lingering as a zombie. libcontainer carries no exit code, so the slot
        // the watcher writes is the only source of truth.
        if let Some(pid) = pid {
            self.arm_exit_watcher(&self.container_id_str(id), pid).await;
        }

        tracing::info!(
            "Container {} started with PID {:?}",
            self.container_id_str(id),
            pid
        );
        Ok(())
    }

    /// Stop a container
    ///
    /// Sends SIGTERM, waits for timeout, then sends SIGKILL if needed.
    #[instrument(
        skip(self),
        fields(
            otel.name = "container.stop",
            container.id = %self.container_id_str(id),
            service.name = %id.service,
            timeout_ms = %timeout.as_millis(),
        )
    )]
    async fn stop_container(&self, id: &ContainerId, timeout: Duration) -> Result<()> {
        let container_id = self.container_id_str(id);

        tracing::info!(
            "Stopping container {} with {:?} timeout",
            container_id,
            timeout
        );

        // Send SIGTERM first, to every process in the container. A non-zero
        // exit here means the container was already stopped / not killable;
        // that is benign for stop semantics (the old in-process path swallowed
        // the equivalent libcontainer error), so we only log it.
        let term = self
            .run_runtime_output(&["kill", &container_id, "SIGTERM", "--all"])
            .await?;
        if !term.status.success() {
            tracing::debug!(
                container = %container_id,
                stderr = %String::from_utf8_lossy(&term.stderr).trim(),
                "SIGTERM via `zlayer runtime kill` failed (container may already be stopped)"
            );
        }

        // Wait for container to stop
        let start = std::time::Instant::now();
        loop {
            if start.elapsed() > timeout {
                break;
            }

            // Check container state
            let state = self.container_state(id).await?;
            if matches!(
                state,
                ContainerState::Exited { .. } | ContainerState::Failed { .. }
            ) {
                tracing::info!("Container {} stopped gracefully", container_id);
                return Ok(());
            }

            tokio::time::sleep(Duration::from_millis(100)).await;
        }

        // Timeout exceeded - send SIGKILL
        tracing::debug!(
            "Container {} did not stop gracefully, sending SIGKILL",
            container_id
        );

        let kill = self
            .run_runtime_output(&["kill", &container_id, "SIGKILL", "--all"])
            .await?;
        if !kill.status.success() {
            tracing::warn!(
                container = %container_id,
                stderr = %String::from_utf8_lossy(&kill.stderr).trim(),
                "SIGKILL via `zlayer runtime kill` failed"
            );
        }

        tracing::info!("Container {} killed", container_id);
        Ok(())
    }

    /// Remove a container
    ///
    /// Deletes the container and cleans up its bundle and state.
    /// Cleanup always proceeds even if libcontainer operations fail.
    #[instrument(
        skip(self),
        fields(
            otel.name = "container.remove",
            container.id = %self.container_id_str(id),
            service.name = %id.service,
        )
    )]
    async fn remove_container(&self, id: &ContainerId) -> Result<()> {
        let container_id = self.container_id_str(id);

        tracing::info!("Removing container {}", container_id);

        // Tear down the per-container Docker socket (if any) and revoke the
        // container's scoped token now that it is terminally gone. The youki
        // lifecycle is the natural owner of this cleanup (the token was minted
        // in create_container); without it the token record would linger until
        // its TTL and the docker socket task/file would leak. Both calls are
        // no-ops when the feature wasn't used / the key is unknown.
        if let Some(auth_ctx) = self.auth_context.as_ref() {
            if let Some(spawner) = auth_ctx.docker_socket_spawner.as_ref() {
                spawner.teardown(&container_id).await;
            }
            if let Some(sink) = auth_ctx.token_sink.as_ref() {
                let jti = format!("container:{}:{}", id.service, container_id);
                sink.revoke(&jti).await;
            }
        }

        // Resolve the init PID BEFORE deleting so we can reap it afterwards.
        // The init (and any fd-inheriting descendants) hold the container's
        // stdout/stderr log fds; if we unlink the logs and tear down the bundle
        // while a reparented-but-unreaped init still has them open, the inode
        // stays pinned and the disk space is never reclaimed (the ENOSPC leak).
        // Prefer the pid-file, fall back to the in-memory record.
        let known_pid = if let Some(p) = self.read_pidfile(id).await {
            Some(p)
        } else {
            let containers = self.containers.read().await;
            containers.get(&container_id).and_then(|info| info.pid)
        };

        // Delete the container's runtime state via a fresh `zlayer runtime
        // delete --force` subprocess (`--force` SIGKILLs a still-running init
        // before deleting, matching the old in-process `delete(true)`). Do NOT
        // fail on a non-zero exit: the container may already be gone, and the
        // in-daemon bundle/cgroup/state cleanup below must always run.
        match self
            .run_runtime_output(&["delete", &container_id, "--force"])
            .await
        {
            Ok(output) if !output.status.success() => {
                tracing::warn!(
                    container = %container_id,
                    stderr = %String::from_utf8_lossy(&output.stderr).trim(),
                    "`zlayer runtime delete` failed (may already be removed)"
                );
            }
            Ok(_) => {}
            Err(e) => {
                tracing::warn!(
                    container = %container_id,
                    error = %e,
                    "failed to spawn `zlayer runtime delete` (may already be removed)"
                );
            }
        }

        // Explicitly reap the init now that `delete --force` has SIGKILLed it.
        // The daemon is a child subreaper, so a reparented init returns here as
        // a zombie that pins its inherited log fds until `waitpid` collects it.
        // The per-container exit watcher (`start_container`) normally does this,
        // but on the remove path we cannot assume the watcher is still armed
        // (e.g. after a daemon restart re-adoption hasn't run, or the watcher
        // already fired). A short bounded wait reaps it; `ECHILD` (already
        // reaped) returns instantly. This MUST happen before the log unlink +
        // bundle teardown below so the fds are released first.
        if let Some(pid) = known_pid.and_then(|p| i32::try_from(p).ok()) {
            // Cap the reap so a stuck-uninterruptible init can't wedge remove;
            // the prior SIGKILL means a healthy init exits within milliseconds.
            let _ = tokio::time::timeout(Duration::from_secs(5), watch_exit_code(pid)).await;
        }

        // Best-effort cgroup teardown: libcontainer's delete() should reap the
        // container's cgroup, but systemd-cgroup races (and occasional cgroup-v2
        // unified hiccups) can leave an empty subdir behind. Because the next
        // create_container rebuilds the same `<root>/containers/<id>` path, that
        // orphan trips libcontainer's build() with `could not delete` on
        // restart/scale, so it MUST be reaped at the real path.
        #[cfg(target_os = "linux")]
        {
            if self.config.use_systemd {
                // systemd cgroup driver: the leaf is a `.scope` under
                // system.slice that systemd owns; idempotent rmdir of the empty
                // scope dir if it lingers after the unit is gone.
                use std::path::Path;
                let scope = format!("/sys/fs/cgroup/system.slice/zlayer-{container_id}.scope");
                let scope_path = Path::new(&scope);
                if scope_path.is_dir() {
                    if let Err(e) = std::fs::remove_dir(scope_path) {
                        tracing::debug!(
                            cgroup = %scope_path.display(),
                            error = %e,
                            "systemd scope rmdir failed (probably already gone)"
                        );
                    }
                }
            } else {
                // cgroupfs driver (host-mode placement): reap
                // /sys/fs/cgroup/zlayer/containers/<id> depth-first at the real
                // path. The old candidates-scan only looked one level under the
                // mount and never matched this two-level path.
                crate::capability::remove_host_container_cgroup(&container_id);
            }
        }

        // ALWAYS clean up bundle regardless of libcontainer result
        if let Err(e) = self.cleanup_bundle(id).await {
            tracing::warn!("Failed to cleanup bundle for {}: {}", container_id, e);
        }

        // Container-log retention: DON'T unlink the structured stdout/stderr logs
        // in the same breath as the bundle GC — a just-failed container's logs are
        // exactly what an operator needs to read after teardown. When
        // `log_base_dir` is configured these live OUTSIDE the bundle (under the
        // structured log hierarchy), so the bundle teardown above never touched
        // them; we leave them in place and let a windowed prune reclaim them later.
        // The init was already reaped above, so the fds are released and the files
        // are plain on-disk logs now (no inode pinning / ENOSPC leak). When
        // structured logging is off the logs lived inside the bundle and already
        // went with it, so there is nothing extra to retain.
        if self.config.log_base_dir.is_some() {
            self.prune_container_logs().await;
        }

        // ALWAYS clean up state directory regardless of delete result
        let state_dir = self.container_root(id);
        if state_dir.exists() {
            if let Err(e) = fs::remove_dir_all(&state_dir).await {
                tracing::warn!("Failed to remove state dir {}: {}", state_dir.display(), e);
            }
        }

        // Remove the pid-file written by `zlayer runtime create` (it lives
        // beside the state dir, not inside it, so the reap above misses it).
        let _ = fs::remove_file(self.pidfile_path(id)).await;

        // Clean up storage volumes
        // Note: We need the spec to know what to clean up, but we don't have it here
        // For now, we'll just clean up anonymous volumes by container ID
        {
            let mut storage_manager = self.storage_manager.write().await;
            if let Err(e) = storage_manager.cleanup_anonymous(&container_id) {
                tracing::warn!(
                    container = %container_id,
                    error = %e,
                    "failed to cleanup anonymous volumes"
                );
            }
        }

        // Remove from local tracking
        {
            let mut containers = self.containers.write().await;
            containers.remove(&container_id);
        }

        tracing::info!("Container {} removed", container_id);
        Ok(())
    }

    /// Get container state
    #[instrument(
        skip(self),
        fields(
            otel.name = "container.state",
            container.id = %self.container_id_str(id),
        )
    )]
    async fn container_state(&self, id: &ContainerId) -> Result<ContainerState> {
        let container_id = self.container_id_str(id);
        let container_root = self.container_root(id);

        // Check if container root exists
        if !container_root.exists() {
            return Err(AgentError::NotFound {
                container: container_id.clone(),
                reason: "container state directory not found".to_string(),
            });
        }

        // Load container and get status
        let container_id_clone = container_id.clone();

        let status = tokio::task::spawn_blocking(move || {
            let mut container =
                Container::load(container_root).map_err(|e| AgentError::NotFound {
                    container: container_id_clone.clone(),
                    reason: format!("failed to load container: {e}"),
                })?;

            // Refresh status to get current state
            let _ = container.refresh_status();

            Ok::<ContainerStatus, AgentError>(container.status())
        })
        .await
        .map_err(|e| AgentError::NotFound {
            container: container_id.clone(),
            reason: format!("task join error: {e}"),
        })??;

        // The exit watcher records the REAL init exit code on termination;
        // prefer it as the source of truth (libcontainer only yields `Stopped`
        // with no code). The slot is `Some` only once the init has exited.
        let recorded = {
            let containers = self.containers.read().await;
            containers
                .get(&container_id)
                .and_then(|info| info.exit_code.lock().ok().and_then(|g| *g))
        };
        if let Some(code) = recorded {
            return Ok(ContainerState::Exited { code });
        }
        let mapped = self.map_status(status);
        // libcontainer says Stopped but the watcher hasn't recorded yet (it
        // reaps + stores within microseconds of exit): give it a brief grace,
        // then re-read, before falling back to map_status's best-effort 0.
        if matches!(mapped, ContainerState::Exited { .. }) {
            tokio::time::sleep(Duration::from_millis(50)).await;
            let recorded = {
                let containers = self.containers.read().await;
                containers
                    .get(&container_id)
                    .and_then(|info| info.exit_code.lock().ok().and_then(|g| *g))
            };
            if let Some(code) = recorded {
                return Ok(ContainerState::Exited { code });
            }
        }
        Ok(mapped)
    }

    /// Get container logs
    ///
    /// Reads from the container's stdout/stderr log files.
    async fn container_logs(&self, id: &ContainerId, tail: usize) -> Result<Vec<LogEntry>> {
        let container_id = self.container_id_str(id);

        // Get log paths from local state
        let (stdout_path, stderr_path) = {
            let containers = self.containers.read().await;
            match containers.get(&container_id) {
                Some(info) => (info.stdout_path.clone(), info.stderr_path.clone()),
                None => {
                    // Fall back to default paths
                    self.log_paths(id)
                }
            }
        };

        let now = chrono::Utc::now();
        let source = LogSource::Container(id.to_string());
        let mut entries = Vec::new();

        // Read stdout
        if stdout_path.exists() {
            if let Ok(content) = fs::read_to_string(&stdout_path).await {
                for line in content.lines() {
                    entries.push(LogEntry {
                        timestamp: now,
                        stream: LogStream::Stdout,
                        message: line.to_string(),
                        source: source.clone(),
                        service: None,
                        deployment: None,
                    });
                }
            }
        }

        // Read stderr
        if stderr_path.exists() {
            if let Ok(content) = fs::read_to_string(&stderr_path).await {
                for line in content.lines() {
                    entries.push(LogEntry {
                        timestamp: now,
                        stream: LogStream::Stderr,
                        message: line.to_string(),
                        source: source.clone(),
                        service: None,
                        deployment: None,
                    });
                }
            }
        }

        // Sort by timestamp (all same for legacy files, but correct for future use)
        entries.sort_by_key(|e| e.timestamp);

        // Apply tail limit
        if tail > 0 && entries.len() > tail {
            entries = entries.split_off(entries.len() - tail);
        }

        Ok(entries)
    }

    /// Execute a command in a running container
    ///
    /// Uses libcontainer's tenant builder to exec into the container's namespaces.
    #[allow(unsafe_code)]
    #[instrument(
        skip(self),
        fields(
            otel.name = "container.exec",
            container.id = %self.container_id_str(id),
            command = ?cmd,
        )
    )]
    async fn exec(&self, id: &ContainerId, cmd: &[String]) -> Result<(i32, String, String)> {
        let container_id = self.container_id_str(id);

        if cmd.is_empty() {
            return Err(AgentError::InvalidSpec(
                "exec command cannot be empty".to_string(),
            ));
        }

        tracing::debug!("Executing {:?} in container {}", cmd, container_id);

        // Create temporary files for exec output
        let exec_id = uuid::Uuid::new_v4().to_string();
        let exec_dir = self.config.state_dir.join(format!("exec-{exec_id}"));
        fs::create_dir_all(&exec_dir)
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: exec_id.clone(),
                reason: format!("failed to create exec dir: {e}"),
            })?;

        let stdout_path = exec_dir.join("stdout");
        let stderr_path = exec_dir.join("stderr");

        // Run the exec through a fresh `zlayer runtime exec` subprocess. The
        // subprocess attaches to the container's namespaces via libcontainer's
        // tenant builder, redirects the exec'd process's stdout/stderr into the
        // `--stdout`/`--stderr` files (which it creates), waits for it to exit,
        // and forwards the exit status. We read the two files back below to
        // return `(code, stdout, stderr)`.
        let stdout_str = stdout_path.to_string_lossy().into_owned();
        let stderr_str = stderr_path.to_string_lossy().into_owned();
        let mut args: Vec<&str> = vec![
            "exec",
            &container_id,
            "--stdout",
            &stdout_str,
            "--stderr",
            &stderr_str,
            "--",
        ];
        args.extend(cmd.iter().map(String::as_str));

        let output = self.run_runtime_output(&args).await?;

        // Read the exec'd process's captured output.
        let stdout_content = fs::read_to_string(&stdout_path).await.unwrap_or_default();
        let stderr_content = fs::read_to_string(&stderr_path).await.unwrap_or_default();

        // Clean up exec directory.
        let _ = fs::remove_dir_all(&exec_dir).await;

        // Derive the exec'd process's real exit code. The CLI exits 0 on a
        // zero-code exec; for a non-zero exec it exits non-zero and prints
        // `exec process exited with status <N>` on ITS stderr (the exec'd
        // process's own stderr went to the file above, not here). Recover the
        // real <N> from that message so callers see the true code; if the
        // marker is absent (e.g. the exec failed to even start), fall back to
        // the subprocess's own exit code (or -1 if signalled).
        let exit_code = if output.status.success() {
            0
        } else {
            let runner_stderr = String::from_utf8_lossy(&output.stderr);
            parse_exec_exit_status(&runner_stderr)
                .or_else(|| output.status.code())
                .unwrap_or(-1)
        };

        Ok((exit_code, stdout_content, stderr_content))
    }

    /// Start an interactive exec session against a libcontainer-managed
    /// container using the runc "console-socket" model — entirely out of
    /// process, avoiding the multithreaded-fork GP-fault hazard of calling
    /// libcontainer's tenant builder in-process from the many-threaded daemon.
    ///
    /// The daemon binds + listens a Unix socket on disk, spawns a
    /// single-threaded `zlayer runtime exec --tty --console-socket <path> <id>
    /// -- <cmd...>` subprocess, and accepts the subprocess's connection.
    /// Inside that subprocess libcontainer creates the PTY master/slave pair,
    /// connects back to the socket, and sends the PTY MASTER fd via
    /// `SCM_RIGHTS` (youki `tty.rs::setup_console`). The daemon `recvmsg`s the
    /// master fd ([`super::console_fd::recv_fd`]), marks it non-blocking, and
    /// wraps it in the [`PtyDuplex`] `AsyncFd`-backed duplex stream.
    ///
    /// The returned [`ExecHandle::resize`] channel drives a small task that
    /// services `(rows, cols)` updates by issuing `ioctl(TIOCSWINSZ)` on the
    /// master fd. The [`ExecHandle::exit`] future awaits the spawned
    /// subprocess (zombie-free, since the daemon is its direct parent) and
    /// recovers the exec'd process's real exit code from the subprocess's
    /// `exec process exited with status <N>` stderr marker.
    ///
    /// `tty` is currently always honoured as a PTY allocation; the duplex
    /// stream carries raw PTY bytes either way.
    #[allow(unsafe_code)]
    #[instrument(
        skip(self, opts),
        fields(
            otel.name = "container.exec_pty",
            container.id = %self.container_id_str(id),
            command = ?opts.command,
            tty = opts.tty,
        )
    )]
    async fn exec_pty(&self, id: &ContainerId, opts: ExecOptions) -> Result<ExecHandle> {
        let container_id = self.container_id_str(id);

        if opts.command.is_empty() {
            return Err(AgentError::InvalidSpec(
                "exec_pty command cannot be empty".to_string(),
            ));
        }

        // runc "console-socket" PTY handoff. Rather than calling libcontainer's
        // `ContainerBuilder::...as_tenant().build()` IN-PROCESS (the
        // multithreaded-fork GP-fault hazard that the subprocess refactor
        // removed everywhere else), we:
        //   1. bind+listen a Unix socket on disk (daemon state, NOT /tmp),
        //   2. spawn a single-threaded `zlayer runtime exec --tty
        //      --console-socket <path> <id> -- <cmd...>` subprocess,
        //   3. inside that subprocess libcontainer creates the PTY pair and
        //      CONNECTS to the socket, sending the PTY MASTER fd back via
        //      SCM_RIGHTS (youki tty.rs `setup_console`),
        //   4. here we accept() + recvmsg() the master fd and wrap it in the
        //      existing `PtyDuplex`.
        //
        // The socket must be listening BEFORE the subprocess connects.
        let exec_id = uuid::Uuid::new_v4().to_string();
        let exec_dir = self.config.state_dir.join(format!("exec-{exec_id}"));
        fs::create_dir_all(&exec_dir)
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: exec_id.clone(),
                reason: format!("failed to create exec dir: {e}"),
            })?;
        let console_path = exec_dir.join("console.sock");

        // Bind + listen the console socket before spawning the subprocess.
        let listener = std::os::unix::net::UnixListener::bind(&console_path).map_err(|e| {
            AgentError::Internal(format!(
                "bind console socket {}: {e}",
                console_path.display()
            ))
        })?;

        // Spawn the single-threaded runtime exec subprocess. Uses the same
        // argv-prefix builder as the other lifecycle verbs so `--state-root`
        // is threaded through. Capture stderr so an early death surfaces a
        // useful error.
        let exe = self.runtime_binary();
        let console_path_str = console_path.to_string_lossy().into_owned();
        let mut argv: Vec<&str> = vec![
            "exec",
            &container_id,
            "--tty",
            "--console-socket",
            &console_path_str,
            "--",
        ];
        argv.extend(opts.command.iter().map(String::as_str));
        let full_argv = build_runtime_argv(&self.config.state_dir, &argv);
        let mut child = tokio::process::Command::new(&exe)
            .args(&full_argv)
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::piped())
            .spawn()
            .map_err(|e| {
                let _ = std::fs::remove_dir_all(&exec_dir);
                AgentError::Internal(format!(
                    "failed to spawn `{} runtime exec --console-socket`: {e}",
                    exe.display()
                ))
            })?;

        // Accept the subprocess's connection and recv the PTY master fd on a
        // blocking task, racing the child's early death and an overall
        // timeout. `UnixListener::accept` blocks, so it runs on the blocking
        // pool; the recv uses `console_fd::recv_fd`.
        let accept_listener = listener;
        let accept_task = tokio::task::spawn_blocking(move || -> std::io::Result<OwnedFd> {
            let (conn, _addr) = accept_listener.accept()?;
            super::console_fd::recv_fd(&conn)
        });

        let master_fd = tokio::select! {
            // The subprocess died before sending the master fd — surface its
            // stderr as the error.
            child_status = child.wait() => {
                let _ = std::fs::remove_dir_all(&exec_dir);
                let status = child_status.map_err(|e| {
                    AgentError::Internal(format!("waiting on exec subprocess: {e}"))
                })?;
                let stderr = match child.stderr.take() {
                    Some(mut s) => {
                        use tokio::io::AsyncReadExt;
                        let mut buf = String::new();
                        let _ = s.read_to_string(&mut buf).await;
                        buf.trim().to_string()
                    }
                    None => String::new(),
                };
                return Err(AgentError::Internal(format!(
                    "exec subprocess exited before sending PTY master (status {:?}): {stderr}",
                    status.code()
                )));
            }
            recv = tokio::time::timeout(Duration::from_secs(30), accept_task) => {
                let joined = recv
                    .map_err(|_| {
                        let _ = std::fs::remove_dir_all(&exec_dir);
                        AgentError::Internal(
                            "timed out waiting for exec subprocess to send PTY master".into(),
                        )
                    })?
                    .map_err(|e| {
                        let _ = std::fs::remove_dir_all(&exec_dir);
                        AgentError::Internal(format!("console-socket accept task join error: {e}"))
                    })?;
                joined.map_err(|e| {
                    let _ = std::fs::remove_dir_all(&exec_dir);
                    AgentError::Internal(format!("recv PTY master over console socket: {e}"))
                })?
            }
        };

        // The console socket has served its purpose; tear down the on-disk
        // listener path (the master fd is now ours via SCM_RIGHTS).
        let _ = std::fs::remove_file(&console_path);

        // Mark the received master end non-blocking so `AsyncFd` can drive it.
        nix::fcntl::fcntl(
            &master_fd,
            nix::fcntl::FcntlArg::F_SETFL(nix::fcntl::OFlag::O_NONBLOCK),
        )
        .map_err(|e| {
            let _ = std::fs::remove_dir_all(&exec_dir);
            AgentError::Internal(format!("F_SETFL O_NONBLOCK on pty master: {e}"))
        })?;

        // Resize task: pump (rows, cols) into TIOCSWINSZ on the master.
        let (resize_tx, mut resize_rx) = tokio::sync::mpsc::channel::<(u16, u16)>(8);
        let master_raw = master_fd.as_raw_fd();
        tokio::spawn(async move {
            while let Some((rows, cols)) = resize_rx.recv().await {
                let ws = nix::pty::Winsize {
                    ws_row: rows,
                    ws_col: cols,
                    ws_xpixel: 0,
                    ws_ypixel: 0,
                };
                // SAFETY: `master_raw` remains a valid fd as long as the
                // duplex stream is alive (it owns the master `OwnedFd`).
                // `ws` is a stack-allocated `winsize` matching the layout
                // the kernel expects. The ioctl reads from the pointer; it
                // does not retain it past the call.
                let rc = unsafe { libc::ioctl(master_raw, libc::TIOCSWINSZ, &ws) };
                if rc != 0 {
                    let err = std::io::Error::last_os_error();
                    tracing::warn!(?err, "TIOCSWINSZ failed on pty master");
                }
            }
        });

        // Build the exit future: await the CHILD subprocess (the single-threaded
        // `zlayer runtime exec --tty` we spawned). This is cleaner and
        // zombie-free — we are the subprocess's direct parent, so `child.wait()`
        // reaps it — versus waitpid on the reparented in-container PID. The
        // subprocess blocks on its own `waitpid` for the exec'd process and
        // forwards the real exit code via the `exec process exited with status
        // <N>` stderr marker (`parse_exec_exit_status`); when that marker is
        // absent (exec succeeded with code 0, or failed to start) we fall back
        // to the subprocess's own exit status.
        let mut child = child;
        let exec_dir_for_exit = exec_dir.clone();
        let exit_fut: ExecExitFuture = Box::pin(async move {
            let status = child
                .wait()
                .await
                .map_err(|e| AgentError::Internal(format!("waiting on exec subprocess: {e}")))?;
            let exit_code = if status.success() {
                0
            } else {
                let stderr = match child.stderr.take() {
                    Some(mut s) => {
                        use tokio::io::AsyncReadExt;
                        let mut buf = String::new();
                        let _ = s.read_to_string(&mut buf).await;
                        buf
                    }
                    None => String::new(),
                };
                parse_exec_exit_status(&stderr)
                    .or_else(|| status.code())
                    .unwrap_or(-1)
            };
            // Best-effort cleanup of the exec scratch dir once the session ends.
            let _ = tokio::fs::remove_dir_all(&exec_dir_for_exit).await;
            Ok(exit_code)
        });

        // Wrap the master end in an AsyncFd-backed duplex stream.
        let stream: ExecPtyStream = Box::new(PtyDuplex::new(master_fd)?);

        Ok(ExecHandle {
            stream,
            resize: resize_tx,
            exit: exit_fut,
        })
    }

    /// Get container resource statistics from cgroups
    ///
    /// Reads CPU and memory statistics from the cgroups v2 filesystem.
    /// Supports both systemd and cgroupfs cgroup drivers.
    async fn get_container_stats(&self, id: &ContainerId) -> Result<ContainerStats> {
        let container_id = self.container_id_str(id);

        // Determine cgroup path based on cgroup driver
        let cgroup_path = if self.config.use_systemd {
            // systemd cgroup driver: /sys/fs/cgroup/system.slice/zlayer-{id}.scope
            PathBuf::from(format!(
                "/sys/fs/cgroup/system.slice/zlayer-{container_id}.scope"
            ))
        } else {
            // cgroupfs driver: /sys/fs/cgroup/zlayer/{id}
            PathBuf::from(format!("/sys/fs/cgroup/zlayer/{container_id}"))
        };

        tracing::debug!(
            container = %container_id,
            cgroup_path = %cgroup_path.display(),
            "reading container stats from cgroups"
        );

        cgroups_stats::read_container_stats(&cgroup_path)
            .await
            .map_err(|e| {
                AgentError::Internal(format!(
                    "failed to read cgroup stats for container {container_id}: {e}"
                ))
            })
    }

    /// Wait for a container to exit and return its exit code
    ///
    /// This polls the container state until it reaches an exited state.
    /// For libcontainer, we don't have a direct "wait" API, so we poll.
    async fn wait_container(&self, id: &ContainerId) -> Result<i32> {
        let container_id = self.container_id_str(id);
        let poll_interval = Duration::from_millis(100);
        let max_wait = Duration::from_secs(3600); // 1 hour max
        let start = std::time::Instant::now();

        tracing::debug!(
            container = %container_id,
            "waiting for container to exit"
        );

        loop {
            if start.elapsed() > max_wait {
                return Err(AgentError::Timeout { timeout: max_wait });
            }

            match self.container_state(id).await {
                Ok(ContainerState::Exited { code }) => {
                    tracing::debug!(
                        container = %container_id,
                        exit_code = code,
                        "container exited"
                    );
                    return Ok(code);
                }
                Ok(ContainerState::Failed { reason }) => {
                    tracing::warn!(
                        container = %container_id,
                        reason = %reason,
                        "container failed"
                    );
                    return Err(AgentError::Internal(format!("container failed: {reason}")));
                }
                Ok(_) => {
                    // Still running, wait and poll again
                    tokio::time::sleep(poll_interval).await;
                }
                Err(AgentError::NotFound { .. }) => {
                    // Container may have been removed - treat as exited with code 0
                    tracing::debug!(
                        container = %container_id,
                        "container not found, treating as exited"
                    );
                    return Ok(0);
                }
                Err(e) => {
                    return Err(e);
                }
            }
        }
    }

    /// Get container logs (stdout/stderr combined)
    ///
    /// Reads from the container's log files and returns as a vector of lines.
    async fn get_logs(&self, id: &ContainerId) -> Result<Vec<LogEntry>> {
        let container_id = self.container_id_str(id);

        // Get log paths
        let (stdout_path, stderr_path) = {
            let containers = self.containers.read().await;
            match containers.get(&container_id) {
                Some(info) => (info.stdout_path.clone(), info.stderr_path.clone()),
                None => self.log_paths(id),
            }
        };

        let now = chrono::Utc::now();
        let source = LogSource::Container(id.to_string());
        let mut entries = Vec::new();

        // Read stdout
        if stdout_path.exists() {
            if let Ok(content) = fs::read_to_string(&stdout_path).await {
                for line in content.lines() {
                    entries.push(LogEntry {
                        timestamp: now,
                        stream: LogStream::Stdout,
                        message: line.to_string(),
                        source: source.clone(),
                        service: None,
                        deployment: None,
                    });
                }
            }
        }

        // Read stderr
        if stderr_path.exists() {
            if let Ok(content) = fs::read_to_string(&stderr_path).await {
                for line in content.lines() {
                    entries.push(LogEntry {
                        timestamp: now,
                        stream: LogStream::Stderr,
                        message: line.to_string(),
                        source: source.clone(),
                        service: None,
                        deployment: None,
                    });
                }
            }
        }

        // Sort by timestamp
        entries.sort_by_key(|e| e.timestamp);

        Ok(entries)
    }

    /// Get the PID of a container's main process
    ///
    /// Returns:
    /// - `Ok(Some(pid))` for running containers
    /// - `Ok(None)` if the container exists but has no PID (not running or stopped)
    /// - `Err` if the container doesn't exist or there's an error loading it
    #[instrument(
        skip(self),
        fields(
            otel.name = "container.get_pid",
            container.id = %self.container_id_str(id),
        )
    )]
    async fn get_container_pid(&self, id: &ContainerId) -> Result<Option<u32>> {
        let container_id = self.container_id_str(id);
        let container_root = self.container_root(id);

        // Check if container root exists
        if !container_root.exists() {
            return Err(AgentError::NotFound {
                container: container_id.clone(),
                reason: "container state directory not found".to_string(),
            });
        }

        // Load container and get PID
        let container_id_clone = container_id.clone();

        let pid = tokio::task::spawn_blocking(move || {
            let mut container =
                Container::load(container_root).map_err(|e| AgentError::NotFound {
                    container: container_id_clone.clone(),
                    reason: format!("failed to load container: {e}"),
                })?;

            // Refresh status to get current state
            let _ = container.refresh_status();

            // Get PID - returns None if container is not running
            #[allow(clippy::cast_sign_loss)]
            let pid = container.pid().map(|p| p.as_raw() as u32);

            Ok::<Option<u32>, AgentError>(pid)
        })
        .await
        .map_err(|e| AgentError::NotFound {
            container: container_id.clone(),
            reason: format!("task join error: {e}"),
        })??;

        tracing::debug!(
            container = %container_id,
            pid = ?pid,
            "retrieved container PID"
        );

        Ok(pid)
    }

    async fn get_container_ip(&self, _id: &ContainerId) -> Result<Option<std::net::IpAddr>> {
        // Youki containers use OCI network namespaces — IP assignment comes
        // from the overlay manager, not the runtime itself.
        Ok(None)
    }

    /// Sync all named volumes to S3 before container removal.
    ///
    /// When the `s3` feature is enabled and a `LayerSyncManager` has been
    /// configured on the storage manager, this iterates all non-anonymous
    /// volumes and pushes any changes to S3. Errors are logged but do not
    /// prevent container removal.
    #[allow(unused_variables)]
    async fn sync_container_volumes(&self, id: &ContainerId) -> Result<()> {
        #[cfg(feature = "s3")]
        {
            let storage_manager = self.storage_manager.read().await;
            if storage_manager.layer_sync().is_some() {
                let container_id = self.container_id_str(id);
                tracing::info!(
                    container = %container_id,
                    "syncing volumes to S3 before container removal"
                );
                match storage_manager.sync_all_volumes().await {
                    Ok(synced) => {
                        if synced > 0 {
                            tracing::info!(
                                container = %container_id,
                                synced_count = synced,
                                "volume sync complete"
                            );
                        }
                    }
                    Err(e) => {
                        tracing::warn!(
                            container = %container_id,
                            error = %e,
                            "volume sync failed, data may not be persisted"
                        );
                    }
                }
            }
        }
        Ok(())
    }

    /// Inspect a locally-stored (built or pulled) image and return a
    /// Docker-shaped detail record.
    ///
    /// Unlike the Seatbelt runtime — which stores each image as an extracted
    /// `rootfs/` directory and reads a `config.json`/`metadata.json` sidecar —
    /// the youki runtime keeps images as OCI manifests + content-addressed blobs
    /// in the local registry (authoritative) and the pull blob cache (pulled
    /// images only). This resolves the manifest the same way [`Self::list_images`]
    /// and [`Self::remove_image`] do: prefer `local_registry.get_manifest`, then
    /// fall back to the blob cache keyed by [`zlayer_registry::manifest_cache_key`].
    /// The runtime metadata (`env`/`cmd`/`entrypoint`/`working_dir`/`user`/
    /// `labels`, plus top-level `os`/`architecture`/`created`) comes from the OCI
    /// image config blob referenced by `manifest.config.digest`.
    ///
    /// # Errors
    ///
    /// Returns [`AgentError::NotFound`] when no manifest exists for `image`, and
    /// [`AgentError::Internal`] when the manifest bytes cannot be parsed.
    async fn inspect_image_native(&self, image: &str) -> Result<ImageInspectInfo> {
        // Split `image` into the registry lookup key (`name`) + reference (tag or
        // `sha256:` digest). A `:` that follows the final `/` is a tag; a `:`
        // inside a `host:port` registry segment (which contains a later `/`) is
        // not. A digest reference (`name@sha256:...`) takes precedence.
        let (name, lookup_ref) = match image.rsplit_once('@') {
            Some((n, d)) => (n.to_string(), d.to_string()),
            None => match image.rsplit_once(':') {
                Some((n, t)) if !t.contains('/') => (n.to_string(), t.to_string()),
                _ => (image.to_string(), "latest".to_string()),
            },
        };

        // Resolve manifest bytes: local registry first (authoritative catalog of
        // built + pulled images), then the pull blob cache keyed by the full ref.
        let manifest_bytes = {
            let from_registry = match self.local_registry.as_ref() {
                Some(registry) => registry.get_manifest(&name, &lookup_ref).await.ok(),
                None => None,
            };
            match from_registry {
                Some(bytes) => Some(bytes),
                None => self
                    .blob_cache
                    .get(&zlayer_registry::manifest_cache_key(image))
                    .await
                    .ok()
                    .flatten(),
            }
        };

        let Some(manifest_bytes) = manifest_bytes else {
            return Err(AgentError::NotFound {
                container: image.to_string(),
                reason: format!("image '{image}' not found"),
            });
        };

        let manifest: OciImageManifest = serde_json::from_slice(&manifest_bytes).map_err(|e| {
            AgentError::Internal(format!("failed to parse image manifest for '{image}': {e}"))
        })?;

        // Fetch + parse the OCI image config blob (config.digest) for the runtime
        // metadata Docker callers expect. Missing/corrupt config is non-fatal —
        // the inspect record is still returned with the manifest-derived fields.
        let config_digest = manifest.config.digest.as_str();
        let config_bytes = {
            let from_registry = match self.local_registry.as_ref() {
                Some(registry) => registry.get_blob(config_digest).await.ok(),
                None => None,
            };
            match from_registry {
                Some(bytes) => Some(bytes),
                None => self.blob_cache.get(config_digest).await.ok().flatten(),
            }
        };
        let config_blob: Option<ConfigBlob> = config_bytes
            .as_deref()
            .and_then(|b| serde_json::from_slice(b).ok());

        // Manifest digest + total size: prefer the registry's own accounting;
        // fall back to the blob-cache digest sidecar + summed manifest sizes,
        // the same accounting `list_images` uses for cache-only images.
        let (digest, size) = match self.local_registry.as_ref() {
            Some(registry) => registry
                .manifest_digest_and_size(&name, &lookup_ref)
                .await
                .ok()
                .map_or((None, None), |(d, s)| (Some(d), (s > 0).then_some(s))),
            None => (None, None),
        };
        let (digest, size) = if digest.is_some() {
            (digest, size)
        } else {
            let digest = self
                .blob_cache
                .get(&zlayer_registry::manifest_digest_cache_key(image))
                .await
                .ok()
                .flatten()
                .and_then(|b| String::from_utf8(b).ok());
            let layers_size: i64 = manifest.layers.iter().map(|l| l.size).sum();
            let total = layers_size.saturating_add(manifest.config.size);
            let size = (total > 0).then(|| u64::try_from(total).ok()).flatten();
            (digest, size)
        };

        let mut info = ImageInspectInfo {
            id: digest.clone(),
            repo_tags: vec![image.to_string()],
            repo_digests: match &digest {
                Some(d) => vec![format!("{name}@{d}")],
                None => Vec::new(),
            },
            size,
            layers: manifest.layers.iter().map(|l| l.digest.clone()).collect(),
            ..Default::default()
        };

        // Runtime defaults + platform/creation metadata from the config blob
        // (macOS field-mapping shape, adapted for the `Option`-typed config).
        if let Some(blob) = config_blob.as_ref() {
            apply_config_blob(&mut info, blob);
        }

        Ok(info)
    }

    async fn list_images(&self) -> Result<Vec<ImageInfo>> {
        use std::collections::HashSet;

        let mut images: Vec<ImageInfo> = Vec::new();
        // References already emitted, so the blob-cache merge below never
        // produces a duplicate of an image already sourced from index.json.
        let mut seen: HashSet<String> = HashSet::new();

        // The authoritative catalog of locally-built/stored images is the
        // registry's index.json. Enumerate every name -> tag there and build an
        // ImageInfo from the registry's own manifest (digest + summed size).
        // The blob cache only holds images that were *pulled*, so on a node that
        // has only built images locally it can be empty even though index.json
        // lists several — listing it alone is the bug this guards against.
        if let Some(registry) = self.local_registry.as_ref() {
            if let Ok(names) = registry.list_images().await {
                for name in names {
                    // Tags recorded for this image in the index. A name with no
                    // tags (digest-only) is skipped here and may still be picked
                    // up from the blob cache below.
                    let tags = match registry.get_image_info(&name).await {
                        Ok(entry) => entry.tags,
                        Err(_) => continue,
                    };

                    for tag in tags {
                        let reference = format!("{name}:{tag}");
                        if !seen.insert(reference.clone()) {
                            continue;
                        }

                        let (digest, size_bytes) =
                            match registry.manifest_digest_and_size(&name, &tag).await {
                                Ok((d, s)) => (Some(d), (s > 0).then_some(s)),
                                Err(_) => (None, None),
                            };

                        images.push(ImageInfo {
                            reference,
                            digest,
                            size_bytes,
                        });
                    }
                }
            }
        }

        // Merge in any images present only in the blob cache (pulled images that
        // were never written into the local registry index), deduped by ref.
        let keys = self
            .blob_cache
            .keys_with_prefix("manifest:")
            .await
            .map_err(|e| {
                AgentError::Internal(format!("failed to list manifest cache keys: {e}"))
            })?;

        for key in keys {
            // Strip the "manifest:" prefix
            let reference = match key.strip_prefix("manifest:") {
                Some(r) => r.to_string(),
                None => continue,
            };

            // Load manifest body to compute size and extract digest
            let Ok(Some(manifest_bytes)) = self.blob_cache.get(&key).await else {
                continue; // cache entry disappeared — skip
            };

            let Ok(manifest) = serde_json::from_slice::<OciImageManifest>(&manifest_bytes) else {
                continue; // corrupt entry — skip
            };

            // Sum layer sizes + config size
            let layers_size: i64 = manifest.layers.iter().map(|l| l.size).sum();
            let config_size = manifest.config.size;
            let total = layers_size.saturating_add(config_size);
            let size_bytes = if total > 0 {
                u64::try_from(total).ok()
            } else {
                None
            };

            // Look up the stored registry digest (Wave 2 convention)
            let digest_key = zlayer_registry::manifest_digest_cache_key(&reference);
            let digest = self
                .blob_cache
                .get(&digest_key)
                .await
                .ok()
                .flatten()
                .and_then(|bytes| String::from_utf8(bytes).ok());

            // Surface the user's ORIGINAL image ref when one was recorded at
            // pull time; fall back to the canonical reference. The orig-key is
            // idempotent on an already-canonical ref (same invariant the
            // `manifest_digest_cache_key(&reference)` lookup above relies on).
            let display_ref = self
                .blob_cache
                .get(&zlayer_registry::manifest_orig_cache_key(&reference))
                .await
                .ok()
                .flatten()
                .and_then(|b| String::from_utf8(b).ok())
                .unwrap_or_else(|| reference.clone());

            // Skip anything already sourced from the registry index above.
            if !seen.insert(display_ref.clone()) {
                continue;
            }

            images.push(ImageInfo {
                reference: display_ref,
                digest,
                size_bytes,
            });
        }

        Ok(images)
    }

    async fn remove_image(&self, image: &str, _force: bool) -> Result<()> {
        let manifest_key = zlayer_registry::manifest_cache_key(image);
        let digest_key = zlayer_registry::manifest_digest_cache_key(image);

        // Load manifest to learn the blob digests it references
        let manifest_present = self
            .blob_cache
            .get(&manifest_key)
            .await
            .map_err(|e| AgentError::Internal(format!("failed to read manifest cache: {e}")))?
            .is_some();

        if !manifest_present {
            return Err(AgentError::NotFound {
                container: image.to_string(),
                reason: format!("image '{image}' not found in cache"),
            });
        }

        // Delete ONLY the manifest body + digest sidecar. We deliberately do NOT
        // delete the layer/config blobs directly: blobs are content-addressed and
        // shared across images, so deleting them here with no reference check
        // would corrupt an unrelated image that shares a base layer (this was a
        // real bug — the macOS runtime already avoids it). Once the manifest keys
        // are gone, reachability GC below reclaims only the blobs that no
        // surviving manifest references.
        self.blob_cache
            .delete(&manifest_key)
            .await
            .map_err(|e| AgentError::Internal(format!("failed to delete manifest: {e}")))?;
        let _ = self.blob_cache.delete(&digest_key).await;

        // Reclaim now-unreferenced blobs (shared-blob-aware: consults every
        // surviving manifest). Best-effort — a GC failure must not fail the rmi.
        if let Err(e) =
            zlayer_registry::prune_dangling_blobs(self.blob_cache.as_ref().as_ref()).await
        {
            tracing::warn!(image = %image, error = %e, "post-rmi prune of dangling blobs failed");
        }

        Ok(())
    }

    async fn prune_images(&self) -> Result<PruneResult> {
        let (deleted, space_reclaimed) =
            zlayer_registry::prune_dangling_blobs(self.blob_cache.as_ref().as_ref())
                .await
                .map_err(|e| AgentError::Internal(format!("failed to prune image blobs: {e}")))?;
        Ok(PruneResult {
            deleted,
            space_reclaimed,
        })
    }

    #[instrument(
        skip(self),
        fields(
            otel.name = "container.kill",
            container.id = %self.container_id_str(id),
            service.name = %id.service,
            signal = ?signal,
        )
    )]
    async fn kill_container(&self, id: &ContainerId, signal: Option<&str>) -> Result<()> {
        // Validate/canonicalize the signal up front, exactly as before.
        let canonical = crate::runtime::validate_signal(signal.unwrap_or("SIGKILL"))?;
        let container_id = self.container_id_str(id);

        tracing::info!(
            container = %container_id,
            signal = %canonical,
            "sending signal to container"
        );

        // Deliver via a fresh `zlayer runtime kill <id> <signal> --all`
        // subprocess. The CLI re-validates `can_kill()` and re-parses the
        // signal, so a non-zero exit (non-killable state / unknown container /
        // delivery failure) propagates as an error, matching the old in-process
        // behaviour.
        self.run_runtime(&["kill", &container_id, &canonical, "--all"])
            .await
            .map_err(|e| {
                AgentError::Internal(format!(
                    "failed to deliver signal {canonical} to '{container_id}': {e}"
                ))
            })?;

        Ok(())
    }

    /// Pause a container by freezing its cgroup via `Container::pause`.
    ///
    /// Loaded from the on-disk libcontainer state; the call itself is
    /// blocking so we hop to `spawn_blocking`. Errors map to `NotFound` when
    /// the container state directory doesn't exist, `InvalidSpec` when the
    /// container is in a non-pausable state (already paused, never started),
    /// and `Internal` for cgroup write failures.
    #[instrument(
        skip(self),
        fields(
            otel.name = "container.pause",
            container.id = %self.container_id_str(id),
            service.name = %id.service,
        )
    )]
    async fn pause_container(&self, id: &ContainerId) -> Result<()> {
        let container_id = self.container_id_str(id);
        let container_root = self.container_root(id);
        let container_id_clone = container_id.clone();
        tokio::task::spawn_blocking(move || {
            let mut container =
                Container::load(container_root).map_err(|e| AgentError::NotFound {
                    container: container_id_clone.clone(),
                    reason: format!("failed to load container: {e}"),
                })?;
            if !container.can_pause() {
                return Err(AgentError::InvalidSpec(format!(
                    "container '{container_id_clone}' is not in a pausable state ({:?})",
                    container.status()
                )));
            }
            container.pause().map_err(|e| {
                AgentError::Internal(format!(
                    "failed to pause container '{container_id_clone}': {e}"
                ))
            })?;
            Ok::<(), AgentError>(())
        })
        .await
        .map_err(|e| AgentError::Internal(format!("task join error during pause: {e}")))??;
        Ok(())
    }

    /// Resume a previously-paused container via `Container::resume`.
    ///
    /// Symmetric inverse of `pause_container`: thaws the freezer cgroup. Same
    /// error mapping conventions.
    #[instrument(
        skip(self),
        fields(
            otel.name = "container.unpause",
            container.id = %self.container_id_str(id),
            service.name = %id.service,
        )
    )]
    async fn unpause_container(&self, id: &ContainerId) -> Result<()> {
        let container_id = self.container_id_str(id);
        let container_root = self.container_root(id);
        let container_id_clone = container_id.clone();
        tokio::task::spawn_blocking(move || {
            let mut container =
                Container::load(container_root).map_err(|e| AgentError::NotFound {
                    container: container_id_clone.clone(),
                    reason: format!("failed to load container: {e}"),
                })?;
            if !container.can_resume() {
                return Err(AgentError::InvalidSpec(format!(
                    "container '{container_id_clone}' is not in a resumable state ({:?})",
                    container.status()
                )));
            }
            container.resume().map_err(|e| {
                AgentError::Internal(format!(
                    "failed to resume container '{container_id_clone}': {e}"
                ))
            })?;
            Ok::<(), AgentError>(())
        })
        .await
        .map_err(|e| AgentError::Internal(format!("task join error during unpause: {e}")))??;
        Ok(())
    }

    /// Update a running container's cgroup v2 resource limits and persist
    /// the new restart policy in the supervisor's in-memory state.
    ///
    /// This implementation writes directly to the container's cgroup v2
    /// hierarchy under `/sys/fs/cgroup/zlayer/<id>` (or
    /// `/sys/fs/cgroup/system.slice/zlayer-<id>.scope` when systemd is the
    /// driver). The fields it can apply natively on cgroup v2 are:
    ///
    /// * `cpu_shares` → `cpu.weight` (mapped from the `2..262144` shares
    ///   range to v2's `1..10000` weight range)
    /// * `memory` → `memory.max` (`0` clears the limit)
    /// * `memory_reservation` → `memory.low`
    /// * `memory_swap` → `memory.swap.max` (`-1` clears the limit)
    /// * `pids_limit` → `pids.max` (`-1` or `0` clears)
    /// * `cpuset_cpus` → `cpuset.cpus`
    /// * `cpuset_mems` → `cpuset.mems`
    /// * `cpu_period` + `cpu_quota` → `cpu.max` ("`<quota> <period>`")
    /// * `blkio_weight` → `io.bfq.weight` (best-effort; emits a warning
    ///   when the BFQ controller isn't enabled)
    ///
    /// `cpu_realtime_period` / `cpu_realtime_runtime` and `kernel_memory`
    /// have no cgroup v2 equivalent and are surfaced as warnings rather
    /// than errors.
    ///
    /// `restart_policy` is captured into the in-memory `ContainerInfo`
    /// entry so the supervisor sees the new policy when the container
    /// next exits.
    #[instrument(
        skip(self, update),
        fields(
            otel.name = "container.update",
            container.id = %self.container_id_str(id),
            service.name = %id.service,
        )
    )]
    async fn update_container_resources(
        &self,
        id: &ContainerId,
        update: &crate::runtime::ContainerResourceUpdate,
    ) -> Result<crate::runtime::ContainerUpdateOutcome> {
        let container_id = self.container_id_str(id);
        if update.is_empty() {
            return Ok(crate::runtime::ContainerUpdateOutcome::default());
        }

        // Persist the new restart policy in our in-memory tracking. We
        // do this even when the container's cgroup directory is gone
        // (e.g. the container has already exited): the supervisor reads
        // the policy on the *next* exit, so updating it for a stopped
        // container is still meaningful.
        if let Some(rp) = update.restart_policy.clone() {
            let mut containers = self.containers.write().await;
            if let Some(info) = containers.get_mut(&container_id) {
                info.restart_policy = Some(rp);
            }
        }

        let cgroup_path = if self.config.use_systemd {
            PathBuf::from(format!(
                "/sys/fs/cgroup/system.slice/zlayer-{container_id}.scope"
            ))
        } else {
            PathBuf::from(format!("/sys/fs/cgroup/zlayer/{container_id}"))
        };

        let mut warnings: Vec<String> = Vec::new();

        // If there's nothing to write to cgroup files (only restart
        // policy was set), bail out before touching the filesystem.
        let needs_cgroup_write = update.cpu_shares.is_some()
            || update.memory.is_some()
            || update.memory_reservation.is_some()
            || update.memory_swap.is_some()
            || update.pids_limit.is_some()
            || update.cpuset_cpus.is_some()
            || update.cpuset_mems.is_some()
            || update.cpu_period.is_some()
            || update.cpu_quota.is_some()
            || update.blkio_weight.is_some();

        if needs_cgroup_write && !cgroup_path.exists() {
            return Err(AgentError::NotFound {
                container: container_id.clone(),
                reason: format!(
                    "cgroup directory '{}' not found — is the container running?",
                    cgroup_path.display()
                ),
            });
        }

        if update.kernel_memory.is_some() {
            warnings
                .push("KernelMemory has no cgroup v2 equivalent and was not applied".to_string());
        }
        if update.cpu_realtime_period.is_some() || update.cpu_realtime_runtime.is_some() {
            warnings.push(
                "CpuRealtimePeriod/CpuRealtimeRuntime are not supported on cgroup v2; ignored"
                    .to_string(),
            );
        }

        // cpu_shares -> cpu.weight (cgroup v2 mapping). v1 shares are
        // 2..262144 with default 1024; v2 weight is 1..10000 with
        // default 100. Use Docker's documented mapping:
        //   weight = 1 + ((shares - 2) * 9999 / 262142)
        if let Some(shares) = update.cpu_shares {
            let shares = shares.max(2);
            #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
            let weight = 1_i64 + (shares - 2) * 9999 / 262_142;
            let weight = weight.clamp(1, 10_000);
            write_cgroup_file(
                &cgroup_path.join("cpu.weight"),
                &weight.to_string(),
                &mut warnings,
            )
            .await?;
        }

        // cpu.max takes "<quota> <period>" or "max <period>".
        if update.cpu_period.is_some() || update.cpu_quota.is_some() {
            let period = update.cpu_period.unwrap_or(100_000);
            let quota_str = match update.cpu_quota {
                Some(q) if q > 0 => q.to_string(),
                _ => "max".to_string(),
            };
            let value = format!("{quota_str} {period}");
            write_cgroup_file(&cgroup_path.join("cpu.max"), &value, &mut warnings).await?;
        }

        if let Some(memory) = update.memory {
            let value = if memory <= 0 {
                "max".to_string()
            } else {
                memory.to_string()
            };
            write_cgroup_file(&cgroup_path.join("memory.max"), &value, &mut warnings).await?;
        }

        if let Some(reservation) = update.memory_reservation {
            let value = if reservation <= 0 {
                "0".to_string()
            } else {
                reservation.to_string()
            };
            write_cgroup_file(&cgroup_path.join("memory.low"), &value, &mut warnings).await?;
        }

        if let Some(swap) = update.memory_swap {
            // Docker semantics: -1 means unlimited. Memory swap on v2
            // is the *swap-only* limit, while Docker's `MemorySwap`
            // historically meant memory+swap. Pass through the absolute
            // value with a warning so the operator knows the v2
            // semantic is different.
            warnings.push(
                "MemorySwap is interpreted as cgroup v2 memory.swap.max (swap-only); \
                 Docker's v1 semantics differ"
                    .to_string(),
            );
            let value = if swap < 0 {
                "max".to_string()
            } else {
                swap.to_string()
            };
            write_cgroup_file(&cgroup_path.join("memory.swap.max"), &value, &mut warnings).await?;
        }

        if let Some(pids) = update.pids_limit {
            let value = if pids <= 0 {
                "max".to_string()
            } else {
                pids.to_string()
            };
            write_cgroup_file(&cgroup_path.join("pids.max"), &value, &mut warnings).await?;
        }

        if let Some(cpus) = update.cpuset_cpus.as_ref() {
            write_cgroup_file(&cgroup_path.join("cpuset.cpus"), cpus, &mut warnings).await?;
        }

        if let Some(mems) = update.cpuset_mems.as_ref() {
            write_cgroup_file(&cgroup_path.join("cpuset.mems"), mems, &mut warnings).await?;
        }

        if let Some(weight) = update.blkio_weight {
            // io.bfq.weight expects 1..1000; Docker's BlkioWeight is
            // 10..1000 with default 500. Pass through verbatim and let
            // the kernel reject out-of-range values.
            write_cgroup_file(
                &cgroup_path.join("io.bfq.weight"),
                &weight.to_string(),
                &mut warnings,
            )
            .await?;
        }

        Ok(crate::runtime::ContainerUpdateOutcome { warnings })
    }

    /// List the processes running inside a container.
    ///
    /// Reads the container's main PID from libcontainer's loaded state, walks
    /// `/proc/{pid}/task/*` to enumerate tids inside the container's pid
    /// namespace, then synthesises a Docker-style top response. The columns
    /// returned are a fixed minimal subset (`UID`, `PID`, `PPID`, `STIME`,
    /// `CMD`) — `ps_args` is accepted for trait conformance but ignored,
    /// because youki has no privileged `ps`-runner per container.
    #[instrument(
        skip(self, _ps_args),
        fields(
            otel.name = "container.top",
            container.id = %self.container_id_str(id),
            service.name = %id.service,
        )
    )]
    async fn top_container(
        &self,
        id: &ContainerId,
        _ps_args: &[String],
    ) -> Result<crate::runtime::ContainerTopOutput> {
        use crate::runtime::ContainerTopOutput;

        let container_id = self.container_id_str(id);
        let container_root = self.container_root(id);
        let container_id_clone = container_id.clone();

        // Snapshot the main process PID under spawn_blocking — Container::load
        // walks the on-disk state file synchronously.
        let pid = tokio::task::spawn_blocking(move || -> Result<i32> {
            let container = Container::load(container_root).map_err(|e| AgentError::NotFound {
                container: container_id_clone.clone(),
                reason: format!("failed to load container: {e}"),
            })?;
            let pid = container.pid().ok_or_else(|| {
                AgentError::InvalidSpec(format!(
                    "container '{container_id_clone}' has no running process"
                ))
            })?;
            Ok(pid.as_raw())
        })
        .await
        .map_err(|e| AgentError::Internal(format!("task join error during top: {e}")))??;

        // Walk /proc/{pid}/task/* to enumerate threads (which double as
        // process IDs from the host's perspective). Containers running a
        // single multi-threaded process expose all its tids here.
        let task_dir = format!("/proc/{pid}/task");
        let mut entries = match tokio::fs::read_dir(&task_dir).await {
            Ok(it) => it,
            Err(e) => {
                return Err(AgentError::NotFound {
                    container: container_id.clone(),
                    reason: format!("failed to read /proc/{pid}/task: {e}"),
                });
            }
        };

        let mut processes: Vec<Vec<String>> = Vec::new();
        while let Some(entry) = entries
            .next_entry()
            .await
            .map_err(|e| AgentError::Internal(format!("failed to walk /proc tree: {e}")))?
        {
            let Ok(name) = entry.file_name().into_string() else {
                continue;
            };
            // Skip entries that aren't PIDs (defensive — /proc/.../task only
            // contains numeric directories in practice).
            if !name.chars().all(|c| c.is_ascii_digit()) {
                continue;
            }
            let row = read_proc_row(&name).await;
            processes.push(row);
        }

        Ok(ContainerTopOutput {
            titles: vec![
                "UID".to_string(),
                "PID".to_string(),
                "PPID".to_string(),
                "STIME".to_string(),
                "CMD".to_string(),
            ],
            processes,
        })
    }

    /// `changes_container` is unsupported on Youki: the runtime stores
    /// containers as a single mutable rootfs (extracted from the cached
    /// layers in `bundle_path`), with no overlayfs upper/lower split to
    /// diff against. Implementing this would require either re-extracting
    /// the original image layers and walking both trees, or cooperating
    /// with the storage driver — both out of scope for this trait method.
    /// The REST layer translates `Unsupported` into a 501 response.
    async fn changes_container(
        &self,
        _id: &ContainerId,
    ) -> Result<Vec<crate::runtime::FilesystemChangeEntry>> {
        Err(AgentError::Unsupported(
            "changes_container is not supported by the youki runtime: \
             no layered filesystem to diff against"
                .into(),
        ))
    }

    /// `port_mappings_container` is unsupported on Youki: the runtime relies
    /// on the host's network namespace for port forwarding (proxy / overlay
    /// network), not on a per-container `HostConfig.PortBindings` table. The
    /// 501 from the REST layer signals to clients that they should consult
    /// the daemon's deployment / endpoint metadata rather than a runtime
    /// inspect call.
    async fn port_mappings_container(
        &self,
        _id: &ContainerId,
    ) -> Result<Vec<crate::runtime::PortMappingEntry>> {
        Err(AgentError::Unsupported(
            "port_mappings_container is not supported by the youki runtime: \
             port publishing is managed at the proxy / overlay layer"
                .into(),
        ))
    }

    /// Reclaim orphaned bundle directories.
    ///
    /// An *orphan* is a `bundle_dir/<id>` rootfs tree whose container no longer
    /// has any state record: it has no on-disk libcontainer state directory and
    /// is not tracked in this runtime's in-memory registry. Containers that are
    /// SIGKILL'd or forgotten leak their bundle (the real cause of a multi-GB
    /// `bundles/` directory on a CI host); this sweep removes those leaked
    /// trees.
    ///
    /// The sweep is cheap: it reads the top-level entries of `bundle_dir` and
    /// `state_dir` (`read_dir`, not a recursive walk) to compute the orphan set.
    /// A safety window protects bundles being created right now — `create_container`
    /// writes the bundle before the state dir exists, so a very recent bundle may
    /// belong to an in-flight create and is skipped (see
    /// [`select_orphan_bundles`]).
    ///
    /// Symlinked top-level bundle entries are skipped entirely (lstat-based), and
    /// size measurement / removal use `zlayer_paths::safe_fs` helpers which never
    /// follow symlinks, so the sweep can never escape `bundle_dir`.
    ///
    /// Per-orphan failures are logged and skipped; the sweep never aborts on one
    /// bad entry.
    #[allow(clippy::too_many_lines)]
    async fn prune_containers(&self) -> Result<crate::runtime::ContainerPruneResult> {
        use std::os::unix::fs::MetadataExt;

        // Lazily re-adopt exit watchers for inits that survived a daemon
        // restart, ONCE per process, before the sweep reads the live set. This
        // is the boot hook: the daemon calls `prune_containers` immediately at
        // startup (and the live containers re-adoption tracks are then in the
        // in-memory set below, so the sweep never reclaims a live bundle).
        // Reaping dead inits here releases their pinned (deleted) log fds — the
        // root of the ENOSPC leak. Idempotent: the `arm_exit_watcher` guard +
        // this flag make it exactly-once.
        if self
            .readopt_done
            .compare_exchange(
                false,
                true,
                std::sync::atomic::Ordering::AcqRel,
                std::sync::atomic::Ordering::Acquire,
            )
            .is_ok()
        {
            let armed = self.readopt_exit_watchers_impl().await;
            if armed > 0 {
                tracing::info!(
                    armed,
                    "re-adopted exit watchers for surviving container inits"
                );
            }
        }

        let bundle_dir = self.config.bundle_dir.clone();
        let state_dir = self.config.state_dir.clone();

        // Live id set tracked in memory by this runtime instance.
        let live_in_memory: std::collections::HashSet<String> = {
            let containers = self.containers.read().await;
            containers.keys().cloned().collect()
        };

        // Heavy bits run on a blocking thread: read_dir + recursive size sum +
        // remove_dir_all are all synchronous filesystem work.
        let result = tokio::task::spawn_blocking(
            move || -> (crate::runtime::ContainerPruneResult, Vec<String>) {
            // 1. Top-level bundle entries (skip non-dirs and symlinks via lstat),
            //    recording each bundle's mtime for the safety window.
            let mut bundle_mtimes: std::collections::HashMap<String, std::time::SystemTime> =
                std::collections::HashMap::new();
            match std::fs::read_dir(&bundle_dir) {
                Ok(entries) => {
                    for entry in entries.flatten() {
                        let path = entry.path();
                        // lstat: never follow a symlinked bundle entry.
                        let md = match std::fs::symlink_metadata(&path) {
                            Ok(md) => md,
                            Err(e) => {
                                tracing::debug!(path = %path.display(), error = %e, "prune: lstat failed, skipping bundle entry");
                                continue;
                            }
                        };
                        if md.file_type().is_symlink() || !md.is_dir() {
                            continue;
                        }
                        let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
                            continue;
                        };
                        let mtime = md.modified().unwrap_or(std::time::UNIX_EPOCH);
                        bundle_mtimes.insert(name.to_string(), mtime);
                    }
                }
                Err(e) => {
                    tracing::warn!(
                        bundle_dir = %bundle_dir.display(),
                        error = %e,
                        "prune: failed to read bundle directory; nothing to reclaim"
                    );
                    return (crate::runtime::ContainerPruneResult::default(), Vec::new());
                }
            }

            // 2. Live id set. A bundle is "live" iff a LIVE init owns it, NOT
            //    merely "a state dir exists". A half-removed container leaves
            //    its state dir behind with a DEAD init (or none at all) — that
            //    bundle MUST be reclaimable. So a state-dir entry is added to
            //    `live` only when its recorded init PID is actually alive (the
            //    `<id>.pid` file beside the state dir, falling back to
            //    libcontainer's own state). The in-memory registry is always
            //    treated as live (the daemon still references those records,
            //    and the exit watcher will reap+remove them on exit).
            let mut live: std::collections::HashSet<String> = live_in_memory;
            match std::fs::read_dir(&state_dir) {
                Ok(entries) => {
                    for entry in entries.flatten() {
                        let path = entry.path();
                        let is_dir = std::fs::symlink_metadata(&path)
                            .is_ok_and(|md| md.is_dir() && !md.file_type().is_symlink());
                        if !is_dir {
                            continue;
                        }
                        let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
                            continue;
                        };
                        // Already live (in-memory tracked): nothing to check.
                        if live.contains(name) {
                            continue;
                        }
                        if let Some(pid) = read_init_pid_for(&state_dir, &path, name) {
                            if pid_is_alive(pid) {
                                live.insert(name.to_string());
                            }
                        }
                    }
                }
                Err(e) => {
                    // Missing/unreadable state dir => no live containers on disk;
                    // the in-memory set is still authoritative.
                    tracing::debug!(
                        state_dir = %state_dir.display(),
                        error = %e,
                        "prune: state directory not readable; using in-memory live set only"
                    );
                }
            }

            // 3. Decide which bundles are orphaned (missing from live set AND old
            //    enough to not be an in-flight create).
            let now = std::time::SystemTime::now();
            let orphans = select_orphan_bundles(&bundle_mtimes, &live, now, BUNDLE_PRUNE_SAFETY);

            // 4. Measure (best-effort) and remove each orphan.
            let mut deleted = Vec::new();
            let mut space_reclaimed: u64 = 0;
            for id in orphans {
                let path = bundle_dir.join(&id);
                // Unmount a possible overlay rootfs FIRST — before the size walk
                // (so it never traverses the mounted overlay into the shared
                // lowerdirs) and before remove (so the delete can't recurse into
                // / EBUSY on the mount). Idempotent no-op for a plain-copy bundle.
                unmount_bundle_rootfs(&path);

                // Best-effort size sum via a symlink-safe walk. Never fail on a
                // measure error — a missing byte count must not block reclaim.
                let mut bytes: u64 = 0;
                let _ = zlayer_paths::safe_fs::walk_no_follow(&path, |_p, md| {
                    if md.is_file() {
                        bytes = bytes.saturating_add(md.size());
                    }
                    Ok(())
                });

                // Some bundle dirs contain read-only directories (extracted image
                // layers); make the tree writable first so remove_dir_all can
                // delete it — same approach the rest of the agent uses.
                zlayer_paths::safe_fs::chmod_tree_writable(&path);
                match std::fs::remove_dir_all(&path) {
                    Ok(()) => {
                        space_reclaimed = space_reclaimed.saturating_add(bytes);
                        deleted.push(id);
                    }
                    Err(e) => {
                        tracing::warn!(
                            path = %path.display(),
                            error = %e,
                            "prune: failed to remove orphaned bundle; skipping"
                        );
                    }
                }
            }

            // Surviving bundle ids = every bundle dir we saw minus the ones we
            // just removed. These are the bundles whose `.lowerdirs` markers pin
            // live layers (the layer-store GC reads them to build the live set).
            let deleted_set: std::collections::HashSet<&str> =
                deleted.iter().map(String::as_str).collect();
            let surviving_bundles: Vec<String> = bundle_mtimes
                .keys()
                .filter(|id| !deleted_set.contains(id.as_str()))
                .cloned()
                .collect();

            (
                crate::runtime::ContainerPruneResult {
                    deleted,
                    space_reclaimed,
                },
                surviving_bundles,
            )
        },
        )
        .await
        .map_err(|e| AgentError::Internal(format!("prune_containers join error: {e}")))?;

        let (mut result, surviving_bundles) = result;

        tracing::info!(
            reclaimed_bundles = result.deleted.len(),
            space_reclaimed_bytes = result.space_reclaimed,
            "prune_containers: reclaimed orphaned bundle directories"
        );

        // After the bundle sweep, GC the shared layer store: any extracted layer
        // no longer referenced by a surviving bundle's `.lowerdirs` (and old
        // enough to not be mid-extract for an in-flight create) is removed. The
        // freed bytes fold into the same prune result. No-op on the fallback
        // (full-copy) path, where no layer store is ever written.
        let layer_bytes = prune_layers(
            &self.layer_store_dir(),
            &self.config.bundle_dir,
            &surviving_bundles,
        )
        .await;
        result.space_reclaimed = result.space_reclaimed.saturating_add(layer_bytes);

        Ok(result)
    }

    #[instrument(
        skip(self),
        fields(
            otel.name = "image.tag",
            source = %source,
            target = %target,
        )
    )]
    async fn tag_image(&self, source: &str, target: &str) -> Result<()> {
        if source.trim().is_empty() || target.trim().is_empty() {
            return Err(AgentError::InvalidSpec(
                "source and target must be non-empty image references".to_string(),
            ));
        }
        if source == target {
            // Nothing to do; idempotent.
            return Ok(());
        }

        // Copy the source manifest and its digest sidecar under the target
        // reference. All blobs remain shared content-addressed in the cache.
        let src_manifest_key = zlayer_registry::manifest_cache_key(source);
        let manifest_bytes = self
            .blob_cache
            .get(&src_manifest_key)
            .await
            .map_err(|e| AgentError::Internal(format!("failed to read manifest cache: {e}")))?
            .ok_or_else(|| AgentError::NotFound {
                container: source.to_string(),
                reason: format!("source image '{source}' not found in cache"),
            })?;

        let dst_manifest_key = zlayer_registry::manifest_cache_key(target);
        self.blob_cache
            .put(&dst_manifest_key, &manifest_bytes)
            .await
            .map_err(|e| AgentError::Internal(format!("failed to write manifest for tag: {e}")))?;

        // Best-effort: carry over the registry digest sidecar if present.
        let src_digest_key = zlayer_registry::manifest_digest_cache_key(source);
        if let Ok(Some(digest_bytes)) = self.blob_cache.get(&src_digest_key).await {
            let dst_digest_key = zlayer_registry::manifest_digest_cache_key(target);
            if let Err(e) = self.blob_cache.put(&dst_digest_key, &digest_bytes).await {
                tracing::warn!(
                    source = %source,
                    target = %target,
                    error = %e,
                    "failed to copy manifest-digest sidecar for tag (non-fatal)"
                );
            }
        }

        tracing::info!(source = %source, target = %target, "tagged image");
        Ok(())
    }

    /// Stream container logs by tailing the on-disk stdout/stderr files
    /// produced by [`Self::create_log_files`].
    ///
    /// Implementation notes:
    /// * Each line of the file produces one [`LogChunk`]. Youki's runtime
    ///   does not write timestamps into the log files, so chunks carry the
    ///   wall-clock time the line was read when `opts.timestamps` is set
    ///   and `None` otherwise.
    /// * `opts.tail` is honoured by counting `\n` bytes from the end of
    ///   each file before streaming begins.
    /// * `opts.follow` keeps the stream alive after EOF, polling every
    ///   200ms for new content. When `follow=false` the stream completes
    ///   after the buffered lines drain.
    /// * `opts.since`/`opts.until` filter chunks by the read-time
    ///   wallclock (see above — file format has no per-line timestamps,
    ///   so this is the best the youki backend can do).
    /// * `opts.stdout`/`opts.stderr` toggle each channel; if neither is
    ///   true (Docker's default-on shorthand), both are streamed.
    #[instrument(
        skip(self),
        fields(
            otel.name = "container.logs.stream",
            container.id = %self.container_id_str(id),
        )
    )]
    async fn logs_stream(&self, id: &ContainerId, opts: LogsStreamOptions) -> Result<LogsStream> {
        // Resolve log paths from local state, falling back to the default
        // bundle/structured paths just like `container_logs` / `get_logs`.
        let (stdout_path, stderr_path) = {
            let containers = self.containers.read().await;
            match containers.get(&self.container_id_str(id)) {
                Some(info) => (info.stdout_path.clone(), info.stderr_path.clone()),
                None => self.log_paths(id),
            }
        };

        // If neither channel was requested, default to streaming both
        // (matches Docker's behaviour when `stdout=false&stderr=false` is
        // sent — Docker treats it as "both", since requesting nothing is
        // never useful).
        let none_specified = !opts.stdout && !opts.stderr;
        let want_stdout = opts.stdout || none_specified;
        let want_stderr = opts.stderr || none_specified;

        // Resolve the container state dir once so the follow-log tailers can
        // detect when the workload has exited and close the stream (EOF)
        // instead of tailing forever.
        let container_root = self.container_root(id);

        // Use a bounded channel so a slow consumer applies natural
        // back-pressure on the file readers.
        let (tx, rx) = mpsc::channel::<Result<LogChunk>>(64);

        if want_stdout && stdout_path.exists() {
            let tx = tx.clone();
            let path = stdout_path.clone();
            let opts_cloned = opts.clone();
            let stdout_root = container_root.clone();
            tokio::spawn(async move {
                let _ =
                    stream_log_file(path, LogChannel::Stdout, opts_cloned, tx, Some(stdout_root))
                        .await;
            });
        }

        if want_stderr && stderr_path.exists() {
            let tx_err = tx.clone();
            let path = stderr_path.clone();
            let opts_cloned = opts.clone();
            let stderr_root = container_root.clone();
            tokio::spawn(async move {
                let _ = stream_log_file(
                    path,
                    LogChannel::Stderr,
                    opts_cloned,
                    tx_err,
                    Some(stderr_root),
                )
                .await;
            });
        }

        // Drop the original sender so the stream terminates once both
        // (or all available) tailers exit.
        drop(tx);

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

    /// Stream periodic resource-usage samples for a container by polling
    /// its cgroup v2 directory once per second.
    ///
    /// Reuses [`cgroups_stats::read_container_stats`] for `cpu.stat`,
    /// `memory.current`, and `memory.max`, and additionally reads
    /// `pids.current` / `pids.max` directly so the [`StatsSample`] reflects
    /// pids counters that the existing internal [`ContainerStats`] type
    /// does not carry.
    ///
    /// Network and block-IO counters are not surfaced by youki's cgroup
    /// directory at the same path (network stats live in the container's
    /// netns, blkio stats require the legacy v1 hierarchy) so they are
    /// reported as `0`. Consumers that need those numbers should use the
    /// Docker runtime, which does have them.
    #[instrument(
        skip(self),
        fields(
            otel.name = "container.stats.stream",
            container.id = %self.container_id_str(id),
        )
    )]
    async fn stats_stream(&self, id: &ContainerId) -> Result<StatsStream> {
        let container_id = self.container_id_str(id);
        let cgroup_path = if self.config.use_systemd {
            PathBuf::from(format!(
                "/sys/fs/cgroup/system.slice/zlayer-{container_id}.scope"
            ))
        } else {
            PathBuf::from(format!("/sys/fs/cgroup/zlayer/{container_id}"))
        };

        let (tx, rx) = mpsc::channel::<Result<StatsSample>>(8);

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(1));
            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
            #[allow(clippy::cast_possible_truncation)]
            let online_cpus = num_cpus::get() as u32;

            loop {
                interval.tick().await;

                let sample_result = read_stats_sample(&cgroup_path, online_cpus).await;
                let send_result = match sample_result {
                    Ok(sample) => tx.send(Ok(sample)).await,
                    Err(err) => {
                        // Surface the error and terminate the stream — a
                        // missing cgroup directory means the container is
                        // gone, retrying on every tick would just spam.
                        let _ = tx.send(Err(err)).await;
                        break;
                    }
                };

                if send_result.is_err() {
                    // Receiver dropped — stop sampling.
                    break;
                }
            }
        });

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

    /// Stream image pull progress by wrapping the synchronous
    /// [`zlayer_registry::ImagePuller::pull_image_with_policy`] code path.
    ///
    /// The puller does not expose a per-layer progress callback today, so
    /// this implementation synthesises a coarse progression:
    ///   1. `Status { status: "Pulling manifest" }` before manifest fetch.
    ///   2. One `Status { status: "Pulling layer", id: <digest> }` per
    ///      layer in the manifest, emitted before each layer is fetched.
    ///   3. A final `Done { reference, digest }` event when the pull
    ///      succeeds (or an `Err` item if it fails).
    ///
    /// Each layer event carries the layer's `total` size from the manifest
    /// so consumers can render proportional progress bars even though
    /// `current` cannot be reported until the puller gains a streaming
    /// callback.
    ///
    /// Caller-supplied `auth_in` is HONORED (same passthrough as
    /// [`Self::pull_image_with_policy`]); the hostname-based
    /// [`zlayer_core::AuthResolver`] is the fallback when none is passed.
    #[instrument(
        skip(self, auth_in),
        fields(
            otel.name = "image.pull.stream",
            container.image.name = %image,
        )
    )]
    async fn pull_image_stream(
        &self,
        image: &str,
        auth_in: Option<&RegistryAuth>,
    ) -> Result<PullProgressStream> {
        let (tx, rx) = mpsc::channel::<Result<PullProgress>>(32);

        // Build the puller eagerly (cheap clone of cache + optional local
        // registry) so the spawned task owns everything it needs. The central
        // constructor wires the S3 tier + default registry from env; the
        // streaming path has no per-image source in scope, so use the default.
        let mut puller = zlayer_registry::ImagePuller::from_env_for_runtime(
            self.blob_cache.clone(),
            zlayer_spec::SourcePolicy::default(),
        )
        .await;
        if let Some(ref registry) = self.local_registry {
            puller = puller.with_local_registry(registry.clone());
        }
        let auth = match auth_in {
            Some(a) => zlayer_registry::spec_auth_to_oci(Some(a)),
            None => self.auth_resolver.resolve(image),
        };
        let image_owned = image.to_string();

        tokio::spawn(async move {
            // Step 1: announce manifest pull.
            if tx
                .send(Ok(PullProgress::Status {
                    id: None,
                    status: "Pulling manifest".to_string(),
                    progress: None,
                    current: None,
                    total: None,
                }))
                .await
                .is_err()
            {
                return;
            }

            // Step 2: fetch the manifest so we can enumerate layers.
            // `pull_image_manifest` is exposed via the public client; the
            // higher-level pull_image will redo this internally but the
            // cost is one cached lookup and the API is the cleanest way
            // to learn about layers up-front for streaming events.
            let layers_meta: Vec<(String, u64)> =
                match puller.pull_manifest(&image_owned, &auth).await {
                    Ok((manifest, _digest)) => manifest
                        .layers
                        .iter()
                        .map(|l| {
                            let size = u64::try_from(l.size).unwrap_or(0);
                            (l.digest.clone(), size)
                        })
                        .collect(),
                    Err(e) => {
                        let _ = tx
                            .send(Err(AgentError::PullFailed {
                                image: image_owned.clone(),
                                reason: format!("failed to pull manifest: {e}"),
                            }))
                            .await;
                        return;
                    }
                };

            // Step 3: emit one Status event per layer before the actual
            // pull. The puller will retrieve cached layers near-instantly
            // and uncached ones over the network; either way, consumers
            // see one event per layer with the digest as `id`.
            for (digest, size) in &layers_meta {
                if tx
                    .send(Ok(PullProgress::Status {
                        id: Some(digest.clone()),
                        status: "Pulling fs layer".to_string(),
                        progress: None,
                        current: None,
                        total: if *size > 0 { Some(*size) } else { None },
                    }))
                    .await
                    .is_err()
                {
                    return;
                }
            }

            // Step 4: do the actual pull (uses the shared blob cache;
            // already-cached layers are no-ops).
            let policy = zlayer_spec::PullPolicy::Newer;
            match puller
                .pull_image_with_policy(&image_owned, &auth, policy)
                .await
            {
                Ok(_layers) => {
                    // Best-effort fetch of the registry digest sidecar so
                    // the `Done` event can carry a content-addressed
                    // identifier when one is available.
                    let _ = puller
                        .pull_image_config_with_policy(&image_owned, &auth, policy)
                        .await;

                    let _ = tx
                        .send(Ok(PullProgress::Done {
                            reference: image_owned.clone(),
                            digest: None,
                        }))
                        .await;
                }
                Err(e) => {
                    let _ = tx
                        .send(Err(AgentError::PullFailed {
                            image: image_owned.clone(),
                            reason: format!("failed to pull image: {e}"),
                        }))
                        .await;
                }
            }
        });

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

    /// Stream a TAR archive of a path inside the container's rootfs.
    ///
    /// The youki backend doesn't run a daemon and has no live attach API to
    /// the container's mount namespace, so we satisfy `archive_get` by
    /// walking the on-disk rootfs at `<bundle>/rootfs<container_path>` and
    /// streaming the TAR archive on the fly. This works for non-running
    /// containers and for live containers whose rootfs has not been
    /// `pivot_root`'d into a private mount namespace inaccessible from the
    /// host (the standard Youki layout keeps the bundle rootfs visible).
    #[instrument(
        skip(self),
        fields(
            otel.name = "container.archive_get",
            container.id = %self.container_id_str(id),
            archive.path = %path,
        )
    )]
    async fn archive_get(&self, id: &ContainerId, path: &str) -> Result<ArchiveStream> {
        let bundle_path = self.bundle_path(id);
        let rootfs_path = bundle_path.join("rootfs");
        if !rootfs_path.exists() {
            return Err(AgentError::NotFound {
                container: self.container_id_str(id),
                reason: format!(
                    "container rootfs '{}' does not exist on disk",
                    rootfs_path.display()
                ),
            });
        }

        let rel = path.trim_start_matches('/');
        let abs_target = if rel.is_empty() {
            rootfs_path.clone()
        } else {
            rootfs_path.join(rel)
        };

        // Reject path-traversal attempts: the canonicalized target must live
        // strictly under the rootfs.
        let canon_root = tokio::fs::canonicalize(&rootfs_path).await.map_err(|e| {
            AgentError::Internal(format!(
                "failed to canonicalize rootfs '{}': {e}",
                rootfs_path.display()
            ))
        })?;
        let canon_target = match tokio::fs::canonicalize(&abs_target).await {
            Ok(p) => p,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                return Err(AgentError::NotFound {
                    container: self.container_id_str(id),
                    reason: format!("path '{path}' not found in container rootfs"),
                });
            }
            Err(e) => {
                return Err(AgentError::Internal(format!(
                    "failed to canonicalize path '{path}': {e}"
                )));
            }
        };
        if !canon_target.starts_with(&canon_root) {
            return Err(AgentError::InvalidSpec(format!(
                "archive path '{path}' escapes container rootfs"
            )));
        }

        // Build the TAR archive on a blocking thread so we never block the
        // async runtime on filesystem I/O.
        let (tx, rx) = mpsc::channel::<Result<bytes::Bytes>>(8);
        let target_for_task = canon_target.clone();
        let path_for_task = path.to_string();
        tokio::task::spawn_blocking(move || {
            let result = build_tar_into_sender(&target_for_task, &path_for_task, &tx);
            if let Err(e) = result {
                let _ = tx.blocking_send(Err(e));
            }
        });

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

    /// Extract a TAR archive into the container at `path` by unpacking
    /// directly into `<bundle>/rootfs<path>`.
    #[instrument(
        skip(self, tar_bytes),
        fields(
            otel.name = "container.archive_put",
            container.id = %self.container_id_str(id),
            archive.path = %path,
            archive.bytes = tar_bytes.len(),
        )
    )]
    async fn archive_put(
        &self,
        id: &ContainerId,
        path: &str,
        tar_bytes: bytes::Bytes,
        opts: ArchivePutOptions,
    ) -> Result<()> {
        let bundle_path = self.bundle_path(id);
        let rootfs_path = bundle_path.join("rootfs");
        if !rootfs_path.exists() {
            return Err(AgentError::NotFound {
                container: self.container_id_str(id),
                reason: format!(
                    "container rootfs '{}' does not exist on disk",
                    rootfs_path.display()
                ),
            });
        }

        let rel = path.trim_start_matches('/');
        let abs_dest = if rel.is_empty() {
            rootfs_path.clone()
        } else {
            rootfs_path.join(rel)
        };

        // The destination must already exist and be a directory (Docker's
        // semantics).
        match tokio::fs::metadata(&abs_dest).await {
            Ok(m) if m.is_dir() => {}
            Ok(_) => {
                return Err(AgentError::InvalidSpec(format!(
                    "destination '{path}' inside container is not a directory"
                )));
            }
            Err(_) => {
                return Err(AgentError::NotFound {
                    container: self.container_id_str(id),
                    reason: format!("destination path '{path}' does not exist in container"),
                });
            }
        }

        // Validate that abs_dest stays under canonical rootfs.
        let canon_root = tokio::fs::canonicalize(&rootfs_path).await.map_err(|e| {
            AgentError::Internal(format!(
                "failed to canonicalize rootfs '{}': {e}",
                rootfs_path.display()
            ))
        })?;
        let canon_dest = tokio::fs::canonicalize(&abs_dest).await.map_err(|e| {
            AgentError::Internal(format!("failed to canonicalize dest '{path}': {e}"))
        })?;
        if !canon_dest.starts_with(&canon_root) {
            return Err(AgentError::InvalidSpec(format!(
                "archive destination '{path}' escapes container rootfs"
            )));
        }

        let dest_for_task = canon_dest.clone();
        tokio::task::spawn_blocking(move || -> Result<()> {
            unpack_tar_into(&dest_for_task, tar_bytes.as_ref(), opts)
        })
        .await
        .map_err(|e| AgentError::Internal(format!("archive_put task panicked: {e}")))??;
        Ok(())
    }

    /// Return path-stat metadata for `path` inside the container's rootfs.
    #[instrument(
        skip(self),
        fields(
            otel.name = "container.archive_head",
            container.id = %self.container_id_str(id),
            archive.path = %path,
        )
    )]
    async fn archive_head(&self, id: &ContainerId, path: &str) -> Result<PathStat> {
        let bundle_path = self.bundle_path(id);
        let rootfs_path = bundle_path.join("rootfs");
        if !rootfs_path.exists() {
            return Err(AgentError::NotFound {
                container: self.container_id_str(id),
                reason: format!(
                    "container rootfs '{}' does not exist on disk",
                    rootfs_path.display()
                ),
            });
        }

        let rel = path.trim_start_matches('/');
        let abs_target = if rel.is_empty() {
            rootfs_path.clone()
        } else {
            rootfs_path.join(rel)
        };

        // symlink_metadata so we report the link itself, not its target.
        let meta = tokio::fs::symlink_metadata(&abs_target)
            .await
            .map_err(|e| match e.kind() {
                std::io::ErrorKind::NotFound => AgentError::NotFound {
                    container: self.container_id_str(id),
                    reason: format!("path '{path}' not found in container rootfs"),
                },
                _ => AgentError::Internal(format!("failed to stat path '{path}': {e}")),
            })?;

        let name = std::path::Path::new(path)
            .file_name()
            .and_then(|s| s.to_str())
            .unwrap_or("")
            .to_string();
        #[allow(clippy::cast_possible_wrap)]
        let size = meta.len() as i64;
        #[cfg(unix)]
        let mode = {
            use std::os::unix::fs::MetadataExt;
            meta.mode()
        };
        #[cfg(not(unix))]
        let mode: u32 = 0;
        let mtime = meta
            .modified()
            .ok()
            .and_then(|t| chrono::DateTime::<chrono::Utc>::from(t).to_rfc3339().into());
        let link_target = if meta.file_type().is_symlink() {
            tokio::fs::read_link(&abs_target)
                .await
                .ok()
                .and_then(|p| p.to_str().map(String::from))
                .unwrap_or_default()
        } else {
            String::new()
        };

        Ok(PathStat {
            name,
            size,
            mode,
            mtime,
            link_target,
        })
    }
}

/// Walk `target` and stream a TAR archive into `tx` synchronously.
///
/// Used by `YoukiRuntime::archive_get` from a `spawn_blocking` task. Each
/// chunk emitted by the underlying `tar::Builder` is forwarded to the
/// channel as a `bytes::Bytes` so the async caller can pipe it straight to
/// the HTTP response body.
fn build_tar_into_sender(
    target: &Path,
    archive_path: &str,
    tx: &mpsc::Sender<Result<bytes::Bytes>>,
) -> Result<()> {
    use std::io::Write;

    /// `std::io::Write` adapter that forwards every write into a tokio mpsc
    /// channel as a `bytes::Bytes` chunk.
    struct ChannelWriter<'a> {
        tx: &'a mpsc::Sender<Result<bytes::Bytes>>,
    }
    impl Write for ChannelWriter<'_> {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            let chunk = bytes::Bytes::copy_from_slice(buf);
            self.tx
                .blocking_send(Ok(chunk))
                .map_err(|e| std::io::Error::new(std::io::ErrorKind::BrokenPipe, e))?;
            Ok(buf.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    let writer = ChannelWriter { tx };
    let mut builder = tar::Builder::new(writer);
    builder.follow_symlinks(false);

    // Determine the in-archive name (Docker uses the basename of the
    // requested path so the TAR contains entries like `foo/...`).
    let entry_name = std::path::Path::new(archive_path).file_name().map_or_else(
        || std::ffi::OsString::from("."),
        std::ffi::OsStr::to_os_string,
    );

    let meta = std::fs::symlink_metadata(target)
        .map_err(|e| AgentError::Internal(format!("failed to stat archive target: {e}")))?;
    if meta.is_dir() {
        builder
            .append_dir_all(&entry_name, target)
            .map_err(|e| AgentError::Internal(format!("failed to append dir to tar: {e}")))?;
    } else {
        let mut f = std::fs::File::open(target)
            .map_err(|e| AgentError::Internal(format!("failed to open archive target: {e}")))?;
        builder
            .append_file(&entry_name, &mut f)
            .map_err(|e| AgentError::Internal(format!("failed to append file to tar: {e}")))?;
    }
    builder
        .finish()
        .map_err(|e| AgentError::Internal(format!("failed to finalize tar: {e}")))?;
    Ok(())
}

/// Unpack a TAR archive into `dest` synchronously, honouring
/// [`ArchivePutOptions`].
///
/// `no_overwrite_dir_non_dir` rejects the case where an entry would replace
/// an existing directory with a non-directory (or vice versa) — implemented
/// via a pre-pass over the archive's entries before extracting. `copy_uid_gid`
/// is forwarded to `tar::Archive::set_preserve_ownerships` so the unpacker
/// keeps the archive's uid/gid instead of chown'ing to the calling user.
fn unpack_tar_into(dest: &Path, tar_bytes: &[u8], opts: ArchivePutOptions) -> Result<()> {
    if opts.no_overwrite_dir_non_dir {
        // Pre-pass: detect directory/non-directory replacements.
        let mut probe = tar::Archive::new(std::io::Cursor::new(tar_bytes));
        let entries = probe
            .entries()
            .map_err(|e| AgentError::Internal(format!("failed to read tar entries: {e}")))?;
        for entry in entries {
            let entry =
                entry.map_err(|e| AgentError::Internal(format!("invalid tar entry: {e}")))?;
            let p = entry
                .path()
                .map_err(|e| AgentError::Internal(format!("invalid tar path: {e}")))?
                .into_owned();
            let dest_p = dest.join(&p);
            if let Ok(existing) = std::fs::symlink_metadata(&dest_p) {
                let entry_is_dir = entry.header().entry_type().is_dir();
                if existing.is_dir() != entry_is_dir {
                    return Err(AgentError::InvalidSpec(format!(
                        "archive entry '{}' would replace a {} with a {}",
                        p.display(),
                        if existing.is_dir() {
                            "directory"
                        } else {
                            "non-directory"
                        },
                        if entry_is_dir {
                            "directory"
                        } else {
                            "non-directory"
                        }
                    )));
                }
            }
        }
    }

    let mut archive = tar::Archive::new(std::io::Cursor::new(tar_bytes));
    archive.set_preserve_permissions(true);
    archive.set_preserve_ownerships(opts.copy_uid_gid);
    archive
        .unpack(dest)
        .map_err(|e| AgentError::Internal(format!("failed to unpack archive: {e}")))?;
    Ok(())
}

/// Build a TAR archive containing exactly one entry from a host path,
/// returning the bytes. Test-only helper used by
/// `archive_helpers_reject_dir_nondir_replacements`.
#[cfg(test)]
fn build_tar_from_path_for_test(src: &Path, entry_name: &str) -> Vec<u8> {
    let mut buf = Vec::new();
    {
        let mut builder = tar::Builder::new(&mut buf);
        builder.follow_symlinks(false);
        let meta = std::fs::symlink_metadata(src).unwrap();
        if meta.is_dir() {
            builder.append_dir_all(entry_name, src).unwrap();
        } else {
            let mut f = std::fs::File::open(src).unwrap();
            builder.append_file(entry_name, &mut f).unwrap();
        }
        builder.finish().unwrap();
    }
    buf
}

/// Read one row of `top`-style data for a host PID by parsing
/// `/proc/{pid}/status` and `/proc/{pid}/cmdline`.
///
/// The columns mirror the trait's documented `top_container` shape:
/// Write `value` to a cgroup v2 control file, demoting recoverable
/// errors (missing controller, invalid value) to warnings so a single
/// unsupported field doesn't sink the whole update. Hard errors
/// (permission denied, IO errors) propagate as
/// [`AgentError::Internal`].
async fn write_cgroup_file(
    path: &std::path::Path,
    value: &str,
    warnings: &mut Vec<String>,
) -> Result<()> {
    match tokio::fs::write(path, value).await {
        Ok(()) => Ok(()),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            warnings.push(format!(
                "cgroup file '{}' not found; controller may not be enabled",
                path.display()
            ));
            Ok(())
        }
        Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => {
            warnings.push(format!(
                "cgroup write to '{}' rejected value '{}': {e}",
                path.display(),
                value
            ));
            Ok(())
        }
        Err(e) => Err(AgentError::Internal(format!(
            "failed to write '{}' to {}: {e}",
            value,
            path.display()
        ))),
    }
}

/// `[UID, PID, PPID, STIME, CMD]`. Any field that fails to read is filled
/// with the empty string so the row width stays constant — `top` clients
/// expect the matrix to be rectangular.
async fn read_proc_row(pid: &str) -> Vec<String> {
    let status_path = format!("/proc/{pid}/status");
    let mut uid = String::new();
    let mut parent_pid = String::new();
    if let Ok(text) = tokio::fs::read_to_string(&status_path).await {
        for line in text.lines() {
            if let Some(rest) = line.strip_prefix("Uid:") {
                // First field after the tabs is the real UID.
                if let Some(first) = rest.split_whitespace().next() {
                    uid = first.to_string();
                }
            } else if let Some(rest) = line.strip_prefix("PPid:") {
                parent_pid = rest.trim().to_string();
            }
        }
    }

    // STIME would normally come from `ps`'s formatter; we simulate it as
    // the truncated wallclock seen on the start of the row read. This is
    // a coarse approximation but matches Docker's documented contract:
    // youki has no internal `ps` runner so we surface the best-effort
    // string with the same shape.
    let stime = chrono::Utc::now().format("%H:%M:%S").to_string();

    let cmdline_path = format!("/proc/{pid}/cmdline");
    let cmd = match tokio::fs::read(&cmdline_path).await {
        Ok(bytes) => {
            // /proc/{pid}/cmdline uses NUL separators; replace with spaces
            // and trim the trailing NUL the kernel emits.
            let normalised: Vec<u8> = bytes
                .into_iter()
                .map(|b| if b == 0 { b' ' } else { b })
                .collect();
            let mut s = String::from_utf8_lossy(&normalised).into_owned();
            while s.ends_with(' ') {
                s.pop();
            }
            s
        }
        Err(_) => String::new(),
    };

    vec![uid, pid.to_string(), parent_pid, stime, cmd]
}

/// Returns `true` once the container at `container_root` has terminated
/// (libcontainer status `Stopped`), or if its state dir is gone. Used by the
/// follow-log tailer to close the stream when the workload exits instead of
/// tailing the log file forever.
async fn container_has_stopped(container_root: PathBuf) -> bool {
    tokio::task::spawn_blocking(move || {
        if !container_root.exists() {
            // State dir gone -> definitively not running.
            return true;
        }
        match Container::load(container_root) {
            Ok(mut c) => {
                let _ = c.refresh_status();
                matches!(c.status(), ContainerStatus::Stopped)
            }
            Err(_) => true,
        }
    })
    .await
    .unwrap_or(false)
}

/// Blocking `waitpid` on `pid` until it terminates; returns the real exit
/// code (`128 + signal` for signal death), or `None` if it could not be
/// reaped (e.g. `ECHILD` — already reaped elsewhere). The container init is
/// forked by the short-lived `zlayer runtime create` subprocess and reparents
/// to the daemon, which registers itself as a child subreaper at startup (see
/// `init_daemon`'s `PR_SET_CHILD_SUBREAPER` call); the reparented init thus
/// becomes a child of this process, so this reap succeeds and also prevents
/// the init from lingering as a zombie.
fn reap_exit_code(pid: i32) -> Option<i32> {
    use nix::sys::wait::{waitpid, WaitStatus};
    use nix::unistd::Pid;
    let p = Pid::from_raw(pid);
    loop {
        match waitpid(p, None) {
            Ok(WaitStatus::Exited(_, code)) => return Some(code),
            Ok(WaitStatus::Signaled(_, sig, _)) => return Some(128 + sig as i32),
            // ECHILD/ESRCH: not our child / already reaped.
            Err(e) if e != nix::errno::Errno::EINTR => return None,
            // Job-control stops/continues, other non-terminal statuses, and
            // EINTR are not terminal — loop and keep waiting.
            _ => {}
        }
    }
}

/// Build the full argv passed to the `zlayer` binary for a `zlayer runtime
/// <verb>` invocation: `runtime --state-root <state_root> <args...>`.
///
/// Pure (no process spawn) so the canonical prefix + flag forwarding is unit
/// testable. Mirrors `Wsl2DelegateRuntime::zlayer_runtime`'s prefix.
fn build_runtime_argv(state_root: &Path, args: &[&str]) -> Vec<std::ffi::OsString> {
    let mut full_argv: Vec<std::ffi::OsString> = Vec::with_capacity(args.len() + 3);
    full_argv.push(std::ffi::OsString::from("runtime"));
    full_argv.push(std::ffi::OsString::from("--state-root"));
    full_argv.push(state_root.as_os_str().to_os_string());
    full_argv.extend(args.iter().map(std::ffi::OsString::from));
    full_argv
}

/// Recover the exec'd process's real exit code from the `zlayer runtime exec`
/// subprocess's stderr. On a non-zero exec, the CLI prints
/// `exec process exited with status <N>` (see
/// `bin/zlayer/src/commands/runtime.rs::exec`); the binary's top-level `main`
/// collapses the anyhow error to process exit code 1, so this message is the
/// only carrier of the true code. Returns `None` when the marker is absent.
fn parse_exec_exit_status(stderr: &str) -> Option<i32> {
    const MARKER: &str = "exec process exited with status ";
    let idx = stderr.find(MARKER)?;
    let tail = &stderr[idx + MARKER.len()..];
    let digits: String = tail
        .trim_start()
        .chars()
        .take_while(|c| c.is_ascii_digit() || *c == '-')
        .collect();
    digits.parse::<i32>().ok()
}

/// Spawn-blocking wrapper around [`reap_exit_code`].
async fn watch_exit_code(pid: i32) -> Option<i32> {
    tokio::task::spawn_blocking(move || reap_exit_code(pid))
        .await
        .unwrap_or(None)
}

/// `true` if `pid` names a live process this daemon could reap (or signal).
/// Uses `kill(pid, 0)`: `Ok`/`EPERM` ⇒ the process exists; `ESRCH` ⇒ gone.
/// A zombie (reparented-but-unreaped init) still counts as "exists" here,
/// which is correct — a zombie still pins its inherited fds until reaped, so
/// its bundle must NOT be reclaimed out from under the reap.
fn pid_is_alive(pid: u32) -> bool {
    let Ok(raw) = i32::try_from(pid) else {
        return false;
    };
    match nix::sys::signal::kill(nix::unistd::Pid::from_raw(raw), None) {
        // Ok ⇒ exists; EPERM ⇒ exists but we may not signal it (still live).
        Ok(()) | Err(nix::errno::Errno::EPERM) => true,
        // ESRCH (no such process) and anything else ⇒ not a reapable live pid.
        Err(_) => false,
    }
}

/// Read the init PID for an on-disk container, blocking.
///
/// Prefers the `<state_dir>/<name>.pid` file `zlayer runtime create` wrote
/// (stable across the create→start transition); falls back to libcontainer's
/// own state at `state_root` when the pid-file is absent. Returns `None` when
/// neither yields a usable, non-zero PID. Pure filesystem read — safe to call
/// from a `spawn_blocking` closure (it never touches `self`).
fn read_init_pid_for(state_dir: &Path, state_root: &Path, name: &str) -> Option<u32> {
    // 1. Pid-file beside the state dir.
    let pidfile = state_dir.join(format!("{name}.pid"));
    if let Ok(raw) = std::fs::read_to_string(&pidfile) {
        if let Ok(pid) = raw.trim().parse::<u32>() {
            if pid != 0 {
                return Some(pid);
            }
        }
    }
    // 2. libcontainer state.
    if let Ok(mut container) = Container::load(state_root.to_path_buf()) {
        let _ = container.refresh_status();
        if let Some(pid) = container.pid() {
            let raw = pid.as_raw();
            if raw > 0 {
                if let Ok(p) = u32::try_from(raw) {
                    return Some(p);
                }
            }
        }
    }
    None
}

/// Blocking bundle teardown shared by [`YoukiRuntime::cleanup_bundle`].
///
/// Makes the tree owner-writable (so read-only extracted-layer dirs don't fail
/// the delete), removes it, and on a residual failure chmods + retries once.
/// A second failure is logged with the leftover path and its measured byte
/// count (best-effort) rather than swallowed, so the leak is observable.
fn cleanup_bundle_blocking(bundle_path: &Path, id: &str) -> Result<()> {
    if !bundle_path.exists() {
        return Ok(());
    }
    // Unmount any overlay rootfs FIRST: chmod/remove of a still-mounted overlay
    // would recurse into the shared lowerdirs (corrupting other containers'
    // layers) or fail EBUSY. Idempotent no-op for a plain-copy bundle.
    unmount_bundle_rootfs(bundle_path);
    zlayer_paths::safe_fs::chmod_tree_writable(bundle_path);
    if let Err(first) = std::fs::remove_dir_all(bundle_path) {
        // Retry once: a concurrent unpack or a freshly-(re)applied read-only
        // mode can race the first chmod sweep.
        zlayer_paths::safe_fs::chmod_tree_writable(bundle_path);
        if let Err(second) = std::fs::remove_dir_all(bundle_path) {
            if bundle_path.exists() {
                let mut bytes: u64 = 0;
                let _ = zlayer_paths::safe_fs::walk_no_follow(bundle_path, |_p, md| {
                    if md.is_file() {
                        use std::os::unix::fs::MetadataExt;
                        bytes = bytes.saturating_add(md.size());
                    }
                    Ok(())
                });
                tracing::warn!(
                    container = %id,
                    bundle = %bundle_path.display(),
                    leftover_bytes = bytes,
                    first_error = %first,
                    error = %second,
                    "cleanup_bundle: bundle remove failed after chmod+retry; leaving partial tree"
                );
                return Err(AgentError::CreateFailed {
                    id: id.to_string(),
                    reason: format!("failed to remove bundle after retry: {second}"),
                });
            }
        }
    }
    Ok(())
}

/// GC the shared overlay layer store by liveness (called after the orphan
/// bundle sweep in [`YoukiRuntime::prune_containers`]).
///
/// Builds the live referenced-digest set by reading the `.lowerdirs` marker
/// from every surviving bundle (each line is a sanitized layer digest, i.e. a
/// layer-store subdirectory name), then removes each `<layer_store>/<digest>`
/// whose `.done` sentinel exists, is NOT in the live set, and whose mtime is at
/// least [`BUNDLE_PRUNE_SAFETY`] old (so a layer mid-`extract_layer_once` for an
/// in-flight create is never reclaimed). Returns the bytes freed.
///
/// A free function (not a method) so it lives outside the `Runtime` trait impl
/// and stays directly testable. Mirrors the orphan-bundle sweep: a pure decision
/// function ([`select_prunable_layers`]) decides WHAT to remove; this does the
/// I/O on a blocking thread. A missing layer store (fallback path / nothing
/// extracted yet) is a no-op returning 0. Per-layer removal failures are logged
/// and skipped.
async fn prune_layers(layer_store: &Path, bundle_dir: &Path, surviving_bundles: &[String]) -> u64 {
    use std::os::unix::fs::MetadataExt;

    let layer_store = layer_store.to_path_buf();
    let bundle_dir = bundle_dir.to_path_buf();
    let surviving: Vec<String> = surviving_bundles.to_vec();

    tokio::task::spawn_blocking(move || -> u64 {
        if !layer_store.exists() {
            return 0;
        }

        // 1. Live referenced-digest set: union of every surviving bundle's
        //    `.lowerdirs` lines (sanitized digests).
        let mut live: std::collections::HashSet<String> = std::collections::HashSet::new();
        for bundle_id in &surviving {
            let marker = bundle_dir.join(bundle_id).join(".lowerdirs");
            if let Ok(contents) = std::fs::read_to_string(&marker) {
                for line in contents.lines() {
                    let digest = line.trim();
                    if !digest.is_empty() {
                        live.insert(digest.to_string());
                    }
                }
            }
        }

        // 2. Walk the layer store: record each PUBLISHED layer's (name, mtime).
        //    A layer is published iff its `.done` sentinel exists; a
        //    half-extracted dir (no sentinel) is left for its single-flight
        //    owner. mtime is taken from the `.done` sentinel (written last, so it
        //    reflects publish time, not extraction start).
        let mut layer_mtimes: std::collections::HashMap<String, std::time::SystemTime> =
            std::collections::HashMap::new();
        match std::fs::read_dir(&layer_store) {
            Ok(entries) => {
                for entry in entries.flatten() {
                    let path = entry.path();
                    let Ok(md) = std::fs::symlink_metadata(&path) else {
                        continue;
                    };
                    if md.file_type().is_symlink() || !md.is_dir() {
                        continue;
                    }
                    let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
                        continue;
                    };
                    let sentinel = path.join(".done");
                    let Ok(sent_md) = std::fs::metadata(&sentinel) else {
                        // Not yet published — skip (owned by an in-flight extract).
                        continue;
                    };
                    let mtime = sent_md.modified().unwrap_or(std::time::UNIX_EPOCH);
                    layer_mtimes.insert(name.to_string(), mtime);
                }
            }
            Err(e) => {
                tracing::warn!(
                    layer_store = %layer_store.display(),
                    error = %e,
                    "prune_layers: failed to read layer store; nothing to reclaim"
                );
                return 0;
            }
        }

        // 3. Decide which layers are prunable (pure).
        let now = std::time::SystemTime::now();
        let prunable = select_prunable_layers(&layer_mtimes, &live, now, BUNDLE_PRUNE_SAFETY);

        // 4. Measure + remove each prunable layer dir.
        let mut freed: u64 = 0;
        for digest in prunable {
            let path = layer_store.join(&digest);
            let mut bytes: u64 = 0;
            let _ = zlayer_paths::safe_fs::walk_no_follow(&path, |_p, md| {
                if md.is_file() {
                    bytes = bytes.saturating_add(md.size());
                }
                Ok(())
            });
            // Extracted layer dirs can hold read-only entries (0o555 dirs,
            // whiteout markers); make writable first like the bundle sweep.
            zlayer_paths::safe_fs::chmod_tree_writable(&path);
            match std::fs::remove_dir_all(&path) {
                Ok(()) => {
                    freed = freed.saturating_add(bytes);
                    tracing::debug!(layer = %digest, bytes, "prune_layers: reclaimed layer");
                }
                Err(e) => {
                    tracing::warn!(
                        path = %path.display(),
                        error = %e,
                        "prune_layers: failed to remove unreferenced layer; skipping"
                    );
                }
            }
        }

        if freed > 0 {
            tracing::info!(
                bytes_freed = freed,
                "prune_layers: reclaimed unreferenced overlay layers"
            );
        }
        freed
    })
    .await
    .unwrap_or_else(|e| {
        tracing::warn!(error = %e, "prune_layers join error; reclaimed 0 layer bytes");
        0
    })
}

/// Tail a single log file and forward each line as a [`LogChunk`] over
/// `tx`. Honours [`LogsStreamOptions::follow`] (poll-on-EOF), `tail`
/// (count `\n` bytes from end before streaming), and `since`/`until`
/// (wallclock filter applied to the moment each line is read — youki log
/// files do not carry per-line timestamps). In follow mode the stream is
/// closed once the container has exited (see `container_root`) rather than
/// tailing forever.
#[allow(clippy::too_many_lines)]
async fn stream_log_file(
    path: PathBuf,
    channel: LogChannel,
    opts: LogsStreamOptions,
    tx: mpsc::Sender<Result<LogChunk>>,
    container_root: Option<PathBuf>,
) -> Result<()> {
    let file = match tokio::fs::File::open(&path).await {
        Ok(f) => f,
        Err(e) => {
            let _ = tx
                .send(Err(AgentError::Internal(format!(
                    "failed to open log file {}: {}",
                    path.display(),
                    e
                ))))
                .await;
            return Ok(());
        }
    };

    let mut reader = BufReader::new(file);

    // Apply `tail`: seek so that the next read begins at the start of
    // the most-recent N lines. Implemented by counting newlines from the
    // end of the file.
    if let Some(tail) = opts.tail {
        if tail > 0 {
            let metadata = match reader.get_ref().metadata().await {
                Ok(m) => m,
                Err(e) => {
                    let _ = tx
                        .send(Err(AgentError::Internal(format!(
                            "failed to stat log file {}: {}",
                            path.display(),
                            e
                        ))))
                        .await;
                    return Ok(());
                }
            };
            let start = compute_tail_offset(reader.get_mut(), metadata.len(), tail).await;
            if let Err(e) = reader.seek(std::io::SeekFrom::Start(start)).await {
                let _ = tx
                    .send(Err(AgentError::Internal(format!(
                        "failed to seek log file {}: {}",
                        path.display(),
                        e
                    ))))
                    .await;
                return Ok(());
            }
        }
    }

    let mut line = String::new();
    loop {
        line.clear();
        let bytes_read = match reader.read_line(&mut line).await {
            Ok(n) => n,
            Err(e) => {
                let _ = tx
                    .send(Err(AgentError::Internal(format!(
                        "failed to read log file {}: {}",
                        path.display(),
                        e
                    ))))
                    .await;
                return Ok(());
            }
        };

        if bytes_read == 0 {
            // EOF. Without follow, we're done.
            if !opts.follow {
                return Ok(());
            }
            // Follow mode: EOF is normally transient (more output is coming).
            // But once the container has EXITED, no more output will ever
            // arrive, so the stream must CLOSE (deliver EOF) rather than tail
            // forever — otherwise a client blocked on
            // `stream -> EOF -> wait-for-exit` never finalizes and only an
            // out-of-band reconcile unblocks it seconds later, cancelling the
            // workload's own final reporting mid-flight.
            let exited = match container_root.as_ref() {
                Some(root) => container_has_stopped(root.clone()).await,
                None => false,
            };
            if !exited {
                tokio::time::sleep(Duration::from_millis(200)).await;
                continue;
            }
            // Container has exited. Let the filesystem settle briefly, then make
            // one final read: a trailing line flushed between our EOF and the
            // status flip is delivered through the normal path below; if still
            // EOF, close the stream.
            tokio::time::sleep(Duration::from_millis(100)).await;
            line.clear();
            match reader.read_line(&mut line).await {
                // Still EOF, or a read error after exit: close the stream.
                Ok(0) | Err(_) => return Ok(()),
                // A trailing line was flushed: fall through to deliver `line`
                // via the normal chunk-build/send path below.
                Ok(_) => {}
            }
        }

        let now = chrono::Utc::now();
        let now_secs = now.timestamp();

        if let Some(since) = opts.since {
            if now_secs < since {
                continue;
            }
        }
        if let Some(until) = opts.until {
            if now_secs > until {
                // Past the cutoff — stop streaming this channel.
                return Ok(());
            }
        }

        let chunk = LogChunk {
            stream: channel,
            bytes: bytes::Bytes::copy_from_slice(line.as_bytes()),
            timestamp: if opts.timestamps { Some(now) } else { None },
        };

        if tx.send(Ok(chunk)).await.is_err() {
            // Receiver dropped — stop tailing.
            return Ok(());
        }
    }
}

/// Compute the byte offset of the start of the last `tail` lines in a
/// file of size `file_len`. Reads backwards in 4 KiB chunks until enough
/// newlines have been seen or the file start is reached.
///
/// "The last N lines" is defined as the bytes following the
/// `(N+1)`th-from-end newline, mirroring `tail -n N`. If the file has
/// fewer than `N` lines, returns `0` (stream the whole file).
async fn compute_tail_offset(file: &mut tokio::fs::File, file_len: u64, tail: u64) -> u64 {
    const CHUNK_USIZE: usize = 4096;

    if file_len == 0 || tail == 0 {
        return 0;
    }

    let chunk: u64 = CHUNK_USIZE as u64;
    let target = tail.saturating_add(1); // the newline *before* the first wanted line
    let mut pos = file_len;
    let mut newlines: u64 = 0;
    let mut buf = vec![0u8; CHUNK_USIZE];

    while pos > 0 {
        let read_len = std::cmp::min(chunk, pos);
        pos -= read_len;
        if file.seek(std::io::SeekFrom::Start(pos)).await.is_err() {
            return 0;
        }
        // `read_len` is bounded by `CHUNK_USIZE`, so the cast is safe on
        // every target ZLayer supports (Linux x86_64 / aarch64).
        let slice_len = usize::try_from(read_len).unwrap_or(CHUNK_USIZE);
        let buf_slice = &mut buf[..slice_len];
        if tokio::io::AsyncReadExt::read_exact(file, buf_slice)
            .await
            .is_err()
        {
            return 0;
        }
        for (i, byte) in buf_slice.iter().enumerate().rev() {
            if *byte == b'\n' {
                newlines += 1;
                if newlines == target {
                    // `i` is the index of the (tail+1)-th newline from
                    // the end *within the current chunk*. The first
                    // wanted byte sits immediately after it.
                    let absolute = pos + (i as u64) + 1;
                    return absolute.min(file_len);
                }
            }
        }
    }

    0
}

/// Read a single [`StatsSample`] from `cgroup_path`. Wraps the existing
/// [`cgroups_stats::read_container_stats`] (cpu + memory) and supplements
/// it with `pids.current` / `pids.max` read directly from sysfs.
async fn read_stats_sample(cgroup_path: &Path, online_cpus: u32) -> Result<StatsSample> {
    let stats = cgroups_stats::read_container_stats(cgroup_path)
        .await
        .map_err(|e| {
            AgentError::Internal(format!(
                "failed to read cgroup stats at {}: {}",
                cgroup_path.display(),
                e
            ))
        })?;

    let pids_current = read_u64_file(cgroup_path.join("pids.current"))
        .await
        .unwrap_or(0);
    let pids_limit = read_pids_limit(cgroup_path.join("pids.max")).await;

    let mem_limit_bytes = if stats.memory_limit == u64::MAX {
        0
    } else {
        stats.memory_limit
    };

    Ok(StatsSample {
        cpu_total_ns: stats.cpu_usage_usec.saturating_mul(1_000),
        cpu_system_ns: 0,
        online_cpus,
        mem_used_bytes: stats.memory_bytes,
        mem_limit_bytes,
        net_rx_bytes: 0,
        net_tx_bytes: 0,
        blkio_read_bytes: 0,
        blkio_write_bytes: 0,
        pids_current,
        pids_limit,
        timestamp: chrono::Utc::now(),
    })
}

/// Read a small text file containing a single decimal integer. Used for
/// `pids.current`. Returns `None` when the file is missing or unreadable
/// so the caller can substitute a sentinel (`0` for unknown counters).
async fn read_u64_file(path: PathBuf) -> Option<u64> {
    let content = tokio::fs::read_to_string(&path).await.ok()?;
    content.trim().parse::<u64>().ok()
}

/// Read `pids.max`, which is either a decimal integer or the literal
/// `"max"` (cgroup v2 sentinel for "no limit"). Returns `None` for
/// `"max"` or any read/parse error so the caller can leave
/// [`StatsSample::pids_limit`] unset.
async fn read_pids_limit(path: PathBuf) -> Option<u64> {
    let content = tokio::fs::read_to_string(&path).await.ok()?;
    let trimmed = content.trim();
    if trimmed == "max" {
        return None;
    }
    trimmed.parse::<u64>().ok()
}

/// Async duplex wrapper around a non-blocking PTY master fd, used by
/// [`YoukiRuntime::exec_pty`] to expose the master end as an
/// `AsyncRead + AsyncWrite + Send + Unpin` stream that fits the
/// [`ExecPtyStream`] trait object.
///
/// The fd is owned: dropping `PtyDuplex` closes the master, which causes the
/// kernel to send `SIGHUP` to the slave's controlling process group and tears
/// the session down cleanly.
struct PtyDuplex {
    inner: tokio::io::unix::AsyncFd<OwnedFd>,
}

impl PtyDuplex {
    fn new(fd: OwnedFd) -> Result<Self> {
        let inner = tokio::io::unix::AsyncFd::new(fd)
            .map_err(|e| AgentError::Internal(format!("AsyncFd::new on pty master: {e}")))?;
        Ok(Self { inner })
    }
}

impl tokio::io::AsyncRead for PtyDuplex {
    #[allow(unsafe_code)]
    fn poll_read(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        let this = self.get_mut();
        loop {
            let mut guard = std::task::ready!(this.inner.poll_read_ready(cx))?;
            // SAFETY: `read(2)` only writes into the buffer, never reads from
            // its uninitialised tail. We pass the unfilled portion as a raw
            // pointer + length and tell `ReadBuf` how many bytes were
            // actually written before exposing them as initialised. The
            // pointer is valid for the duration of the `read` call because
            // `buf` is borrowed mutably for the entire `poll_read` body.
            let unfilled = unsafe {
                std::slice::from_raw_parts_mut(
                    buf.unfilled_mut().as_mut_ptr().cast::<libc::c_void>(),
                    buf.remaining(),
                )
            };
            let fd = guard.get_ref().as_raw_fd();
            // SAFETY: `fd` is a valid PTY master fd owned by `self.inner`.
            // `unfilled` points into a unique mutable borrow of `buf`. The
            // syscall touches at most `unfilled.len()` bytes.
            let rc = unsafe { libc::read(fd, unfilled.as_mut_ptr(), unfilled.len()) };
            if rc < 0 {
                let err = std::io::Error::last_os_error();
                if err.kind() == std::io::ErrorKind::WouldBlock {
                    guard.clear_ready();
                    continue;
                }
                // Linux PTY masters return EIO once the slave hangs up; treat
                // that as a clean EOF so callers see end-of-stream rather
                // than a confusing error.
                if err.raw_os_error() == Some(libc::EIO) {
                    return std::task::Poll::Ready(Ok(()));
                }
                return std::task::Poll::Ready(Err(err));
            }
            // We checked `rc < 0` above, so the cast is well-defined.
            #[allow(clippy::cast_sign_loss)]
            let n = rc as usize;
            // SAFETY: the kernel just wrote `n` bytes into the unfilled
            // tail; mark them as initialised + filled.
            unsafe {
                buf.assume_init(n);
            }
            buf.advance(n);
            return std::task::Poll::Ready(Ok(()));
        }
    }
}

impl tokio::io::AsyncWrite for PtyDuplex {
    #[allow(unsafe_code)]
    fn poll_write(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        bufdata: &[u8],
    ) -> std::task::Poll<std::io::Result<usize>> {
        let this = self.get_mut();
        loop {
            let mut guard = std::task::ready!(this.inner.poll_write_ready(cx))?;
            let fd = guard.get_ref().as_raw_fd();
            // SAFETY: `fd` is owned by `self.inner` and remains valid for
            // the call. `bufdata` is a borrowed slice valid for `bufdata.len()`
            // bytes; `write(2)` only reads from it.
            let rc = unsafe { libc::write(fd, bufdata.as_ptr().cast(), bufdata.len()) };
            if rc < 0 {
                let err = std::io::Error::last_os_error();
                if err.kind() == std::io::ErrorKind::WouldBlock {
                    guard.clear_ready();
                    continue;
                }
                return std::task::Poll::Ready(Err(err));
            }
            // We checked `rc < 0` above; safe to cast.
            #[allow(clippy::cast_sign_loss)]
            let n = rc as usize;
            return std::task::Poll::Ready(Ok(n));
        }
    }

    fn poll_flush(
        self: std::pin::Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        // PTYs have no userspace buffer; the kernel handles framing.
        std::task::Poll::Ready(Ok(()))
    }

    fn poll_shutdown(
        self: std::pin::Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        // Closing the master happens on drop; no half-close on PTYs.
        std::task::Poll::Ready(Ok(()))
    }
}

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

    #[test]
    fn test_youki_config_default() {
        let config = YoukiConfig::default();
        let dirs = zlayer_paths::ZLayerDirs::system_default();

        assert_eq!(config.state_dir, dirs.containers());
        assert_eq!(config.rootfs_dir, dirs.rootfs());
        assert_eq!(config.bundle_dir, dirs.bundles());
        assert_eq!(config.cache_dir, dirs.cache());
        assert_eq!(config.volume_dir, dirs.volumes());
        assert!(!config.use_systemd);
        assert!(config.cache_type.is_none());
        assert!(config.log_base_dir.is_none());
        assert!(config.deployment_name.is_none());
    }

    #[test]
    fn test_select_orphan_bundles() {
        use std::collections::{HashMap, HashSet};
        use std::time::{Duration, SystemTime};

        let now = SystemTime::now();
        let old = now - Duration::from_secs(3600); // 1h old, well past the window
        let recent = now - Duration::from_secs(5); // inside the 120s safety window

        let mut bundle_mtimes: HashMap<String, SystemTime> = HashMap::new();
        // Orphan: no state dir, old enough -> SHOULD be reclaimed.
        bundle_mtimes.insert("orphan-old".to_string(), old);
        // Live: has a state dir -> SHOULD be kept regardless of age.
        bundle_mtimes.insert("live-1".to_string(), old);
        // Orphan but very recent -> in-flight create, SHOULD be skipped.
        bundle_mtimes.insert("orphan-recent".to_string(), recent);

        // Only "live-1" has a state directory on disk.
        let mut live: HashSet<String> = HashSet::new();
        live.insert("live-1".to_string());

        let selected = select_orphan_bundles(&bundle_mtimes, &live, now, Duration::from_secs(120));

        assert_eq!(
            selected,
            vec!["orphan-old".to_string()],
            "only the live-missing, old-enough bundle should be selected"
        );
    }

    #[test]
    fn test_select_orphan_bundles_in_memory_live() {
        use std::collections::{HashMap, HashSet};
        use std::time::{Duration, SystemTime};

        let now = SystemTime::now();
        let old = now - Duration::from_secs(3600);

        let mut bundle_mtimes: HashMap<String, SystemTime> = HashMap::new();
        bundle_mtimes.insert("tracked".to_string(), old);

        // No on-disk state dir, but tracked in memory -> not an orphan.
        let mut live: HashSet<String> = HashSet::new();
        live.insert("tracked".to_string());

        let selected = select_orphan_bundles(&bundle_mtimes, &live, now, Duration::from_secs(120));
        assert!(
            selected.is_empty(),
            "an in-memory tracked bundle must not be reclaimed"
        );
    }

    /// Pure decision fn for the layer-store GC: given a live referenced-digest
    /// set + per-layer mtimes, returns exactly the unreferenced, old-enough
    /// layers; a referenced layer (and a too-recent unreferenced one) is spared.
    #[test]
    fn test_select_prunable_layers() {
        use std::collections::{HashMap, HashSet};
        use std::time::{Duration, SystemTime};

        let now = SystemTime::now();
        let old = now - Duration::from_secs(3600); // well past the safety window
        let recent = now - Duration::from_secs(5); // inside the 120s window

        let mut layer_mtimes: HashMap<String, SystemTime> = HashMap::new();
        // Unreferenced + old  -> prune.
        layer_mtimes.insert("sha256_unref_old".to_string(), old);
        // Referenced (by a live bundle) + old -> SPARE.
        layer_mtimes.insert("sha256_referenced".to_string(), old);
        // Unreferenced but freshly published -> SPARE (in-flight create guard).
        layer_mtimes.insert("sha256_unref_recent".to_string(), recent);

        let mut live: HashSet<String> = HashSet::new();
        live.insert("sha256_referenced".to_string());

        let prunable = select_prunable_layers(&layer_mtimes, &live, now, Duration::from_secs(120));
        assert_eq!(
            prunable,
            vec!["sha256_unref_old".to_string()],
            "only the unreferenced, old-enough layer may be reclaimed"
        );
    }

    /// `prune_layers` end-to-end over a fake layer store + surviving bundles
    /// (pure filesystem; no mount/root needed). A layer pinned by a surviving
    /// bundle's `.lowerdirs` survives; an unreferenced, old, published layer is
    /// reclaimed and its bytes counted; an unpublished (no `.done`) dir is left
    /// alone.
    #[tokio::test]
    async fn prune_layers_reclaims_only_unreferenced_published_old_layers() {
        use std::time::{Duration, SystemTime};

        let tmp = tempfile::tempdir().unwrap();
        let layer_store = tmp.path().join("layers");
        let bundle_dir = tmp.path().join("bundles");
        std::fs::create_dir_all(&layer_store).unwrap();
        std::fs::create_dir_all(&bundle_dir).unwrap();

        // Helper: build `<store>/<digest>/fs` with `content` bytes, then publish
        // (write `.done`) and back-date the sentinel mtime to `age_secs` ago.
        let make_layer = |digest: &str, content: &[u8], published: bool, age_secs: u64| {
            let dir = layer_store.join(digest);
            std::fs::create_dir_all(dir.join("fs")).unwrap();
            std::fs::write(dir.join("fs").join("data"), content).unwrap();
            if published {
                let sentinel = dir.join(".done");
                std::fs::write(&sentinel, b"").unwrap();
                let mtime = SystemTime::now() - Duration::from_secs(age_secs);
                // `File::set_modified` (stable since 1.75) back-dates the
                // sentinel mtime without pulling in a `filetime` dependency.
                let f = std::fs::OpenOptions::new()
                    .write(true)
                    .open(&sentinel)
                    .unwrap();
                f.set_modified(mtime).unwrap();
            }
        };

        // referenced + old: pinned by the surviving bundle below -> SPARE.
        make_layer("sha256_referenced", b"REFERENCED-LAYER-BYTES", true, 3600);
        // unreferenced + old + published -> RECLAIM (count its bytes).
        let unref_bytes = b"UNREFERENCED-OLD-LAYER-BYTES-XXXX";
        make_layer("sha256_unref_old", unref_bytes, true, 3600);
        // unreferenced + old but UNPUBLISHED (no .done) -> SPARE (in-flight).
        make_layer("sha256_unpublished", b"PARTIAL", false, 3600);

        // One surviving bundle pinning `sha256_referenced` via `.lowerdirs`.
        let bundle = bundle_dir.join("svc-0");
        std::fs::create_dir_all(&bundle).unwrap();
        std::fs::write(bundle.join(".lowerdirs"), "sha256_referenced\n").unwrap();

        let freed = prune_layers(&layer_store, &bundle_dir, &["svc-0".to_string()]).await;

        assert!(
            !layer_store.join("sha256_unref_old").exists(),
            "unreferenced old published layer must be reclaimed"
        );
        assert!(
            layer_store.join("sha256_referenced").exists(),
            "a layer pinned by a surviving bundle's .lowerdirs must survive"
        );
        assert!(
            layer_store.join("sha256_unpublished").exists(),
            "an unpublished (no .done) layer must not be touched"
        );
        assert_eq!(
            freed,
            unref_bytes.len() as u64,
            "freed bytes must equal the reclaimed layer's file bytes"
        );
    }

    /// Missing layer store (the fallback / full-copy path, where no store is
    /// ever written) -> `prune_layers` is a no-op returning 0.
    #[tokio::test]
    async fn prune_layers_noop_when_layer_store_absent() {
        let tmp = tempfile::tempdir().unwrap();
        let layer_store = tmp.path().join("layers"); // never created
        let bundle_dir = tmp.path().join("bundles");
        std::fs::create_dir_all(&bundle_dir).unwrap();
        let freed = prune_layers(&layer_store, &bundle_dir, &[]).await;
        assert_eq!(freed, 0, "absent layer store reclaims nothing");
    }

    /// Root-gated: skip (and say why) when not uid 0 — `extract_layer_once`'s
    /// mknod whiteout markers and the `overlay` mount need `CAP_MKNOD` /
    /// `CAP_SYS_ADMIN`.
    #[cfg(target_os = "linux")]
    fn require_root_or_skip_overlay(test: &str) -> bool {
        if zlayer_paths::is_root() {
            return true;
        }
        eprintln!("skipping {test}: requires root (CAP_MKNOD / CAP_SYS_ADMIN)");
        false
    }

    /// Build a plain (uncompressed) layer tar in memory.
    #[cfg(target_os = "linux")]
    fn tar_layer_bytes(files: &[(&str, &[u8])]) -> Vec<u8> {
        let mut builder = tar::Builder::new(Vec::new());
        for (path, content) in files {
            let mut h = tar::Header::new_gnu();
            h.set_path(path).unwrap();
            h.set_size(content.len() as u64);
            h.set_mode(0o644);
            h.set_cksum();
            builder.append(&h, *content).unwrap();
        }
        builder.into_inner().unwrap()
    }

    /// Uncompressed-tar OCI layer media type.
    #[cfg(target_os = "linux")]
    const TAR_MT: &str = "application/vnd.oci.image.layer.v1.tar";

    /// Reproduces the exact overlay create path (the functions
    /// `build_overlay_rootfs` calls) and teardown (`cleanup_bundle_blocking`,
    /// which prepends `unmount_bundle_rootfs`): two containers of ONE image share
    /// one extracted copy of each layer; after one container is torn down the
    /// shared layer survives because the other still pins it.
    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn overlay_create_and_teardown_shares_layers() {
        use super::super::overlay_rootfs::{extract_layer_once, mount_overlay_rootfs};

        if !require_root_or_skip_overlay("overlay_create_and_teardown_shares_layers") {
            return;
        }
        let tmp = tempfile::tempdir().unwrap();
        let layer_store = tmp.path().join("layers");
        let bundle_dir = tmp.path().join("bundles");
        std::fs::create_dir_all(&bundle_dir).unwrap();

        // One image: two layers. Stage them as `blob_staging_filename`-style
        // files (filename == sanitized digest) like the real pull does.
        let l0 = tar_layer_bytes(&[("base.txt", b"base")]);
        let l1 = tar_layer_bytes(&[("top.txt", b"top")]);
        let stage = tmp.path().join("stage");
        std::fs::create_dir_all(&stage).unwrap();
        let d0 = zlayer_paths::sanitize_digest("sha256:layer0");
        let d1 = zlayer_paths::sanitize_digest("sha256:layer1");
        std::fs::write(stage.join(&d0), &l0).unwrap();
        std::fs::write(stage.join(&d1), &l1).unwrap();

        // Extract layers ONCE into the shared store.
        let fs0 = extract_layer_once(&layer_store, &d0, &stage.join(&d0), TAR_MT)
            .await
            .unwrap();
        let fs1 = extract_layer_once(&layer_store, &d1, &stage.join(&d1), TAR_MT)
            .await
            .unwrap();
        let lowers = vec![fs0, fs1];

        // Mount two containers' overlay rootfs from the SAME lowers, writing each
        // bundle's `.lowerdirs` marker (exactly what `build_overlay_rootfs` does).
        let mounts_before = std::fs::read_to_string("/proc/mounts").unwrap_or_default();
        let mut container_rootfs = Vec::new();
        for id in ["svc-0", "svc-1"] {
            let bundle = bundle_dir.join(id);
            std::fs::create_dir_all(&bundle).unwrap();
            let rootfs = bundle.join("rootfs");
            let upper = bundle.join("upper");
            let work = bundle.join("work");
            mount_overlay_rootfs(&lowers, &upper, &work, &rootfs).unwrap();
            std::fs::write(bundle.join(".lowerdirs"), format!("{d0}\n{d1}\n")).unwrap();
            // Both image files visible through the merged view.
            assert_eq!(std::fs::read(rootfs.join("base.txt")).unwrap(), b"base");
            assert_eq!(std::fs::read(rootfs.join("top.txt")).unwrap(), b"top");
            container_rootfs.push(rootfs);
        }
        let _ = mounts_before;

        // svc-0's rootfs is an overlay mount in /proc/mounts.
        let mounts = std::fs::read_to_string("/proc/mounts").unwrap();
        let svc0_rootfs = container_rootfs[0].display().to_string();
        assert!(
            mounts.lines().any(|l| {
                let mut f = l.split_whitespace();
                f.next();
                f.next() == Some(svc0_rootfs.as_str())
            }),
            "svc-0 rootfs must be an overlay mountpoint"
        );

        // Tear down svc-0 via the real bundle-teardown path (unmount prepended).
        cleanup_bundle_blocking(&bundle_dir.join("svc-0"), "svc-0").unwrap();

        // svc-0 mount + bundle gone.
        let mounts = std::fs::read_to_string("/proc/mounts").unwrap();
        assert!(
            !mounts.lines().any(|l| l.contains(&svc0_rootfs)),
            "svc-0 overlay mount must be gone after teardown"
        );
        assert!(
            !bundle_dir.join("svc-0").exists(),
            "svc-0 bundle must be removed"
        );

        // The shared layer dirs SURVIVE (svc-1 still references them).
        assert!(
            layer_store.join(&d0).join(".done").exists(),
            "shared layer 0 must survive while svc-1 still references it"
        );
        assert!(
            layer_store.join(&d1).join(".done").exists(),
            "shared layer 1 must survive while svc-1 still references it"
        );
        // svc-1 still mounted + working.
        assert_eq!(
            std::fs::read(container_rootfs[1].join("base.txt")).unwrap(),
            b"base",
            "svc-1 overlay must still be intact after svc-0 teardown"
        );

        // Cleanup svc-1 too (idempotent unmount).
        cleanup_bundle_blocking(&bundle_dir.join("svc-1"), "svc-1").unwrap();
    }

    /// Footprint: extract one image's layers once, create N=3 containers on the
    /// overlay path, and assert the layer store holds ~1x the layer bytes while
    /// each per-container `upper` is ~empty — total ≈ 1x image + N x upper, NOT
    /// N x image.
    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn overlay_footprint_is_one_image_plus_n_uppers() {
        use super::super::overlay_rootfs::{extract_layer_once, mount_overlay_rootfs};

        if !require_root_or_skip_overlay("overlay_footprint_is_one_image_plus_n_uppers") {
            return;
        }
        let tmp = tempfile::tempdir().unwrap();
        let layer_store = tmp.path().join("layers");
        let bundle_dir = tmp.path().join("bundles");
        std::fs::create_dir_all(&bundle_dir).unwrap();

        // A non-trivial layer payload so per-image vs per-container is measurable.
        let payload = vec![b'Z'; 256 * 1024]; // 256 KiB
        let layer = tar_layer_bytes(&[("big.bin", &payload)]);
        let digest = zlayer_paths::sanitize_digest("sha256:bigimage");
        let stage = tmp.path().join("stage");
        std::fs::create_dir_all(&stage).unwrap();
        std::fs::write(stage.join(&digest), &layer).unwrap();

        let fs0 = extract_layer_once(&layer_store, &digest, &stage.join(&digest), TAR_MT)
            .await
            .unwrap();
        let lowers = vec![fs0];

        // Measure the shared layer store once (it is extracted exactly once).
        let store_bytes = dir_bytes(&layer_store);
        assert!(
            store_bytes >= payload.len() as u64,
            "layer store must hold the extracted layer ({} >= {})",
            store_bytes,
            payload.len()
        );

        // Create N=3 containers; each upper starts empty.
        let n = 3usize;
        let mut uppers = Vec::new();
        for i in 0..n {
            let bundle = bundle_dir.join(format!("svc-{i}"));
            std::fs::create_dir_all(&bundle).unwrap();
            let upper = bundle.join("upper");
            let work = bundle.join("work");
            let rootfs = bundle.join("rootfs");
            mount_overlay_rootfs(&lowers, &upper, &work, &rootfs).unwrap();
            uppers.push(upper);
        }

        // The layer store did NOT grow per-container (still exactly one copy).
        assert_eq!(
            dir_bytes(&layer_store),
            store_bytes,
            "creating N containers must NOT duplicate the layer in the store"
        );
        // Each per-container upper is essentially empty (no writes yet).
        for upper in &uppers {
            let u = dir_bytes(upper);
            assert!(
                u < 64 * 1024,
                "a fresh per-container upper must be ~empty, got {u} bytes"
            );
        }

        // Total store footprint << N x image (the whole point of overlay).
        assert!(
            dir_bytes(&layer_store) < (n as u64) * (payload.len() as u64),
            "overlay footprint must be ~1x image, not Nx"
        );

        for i in 0..n {
            cleanup_bundle_blocking(&bundle_dir.join(format!("svc-{i}")), &format!("svc-{i}"))
                .unwrap();
        }
    }

    /// Fallback path: with `overlayfs_rootfs_available = false`,
    /// `build_container_rootfs` must take the legacy full-COPY path and produce a
    /// working rootfs (every layer's files materialized directly into the
    /// container's own `rootfs`, no overlay mount, no layer store). Builds a real
    /// `YoukiRuntime` over an in-memory blob cache in a tempdir (no disk cache, no
    /// network) and forces the capability bit off.
    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn fallback_full_copy_builds_working_rootfs() {
        let tmp = tempfile::tempdir().unwrap();
        let data_dir = tmp.path();
        let mut config = YoukiConfig::from_data_dir(data_dir);
        // In-memory blob cache => construction touches no disk DB / network.
        config.cache_type = Some(zlayer_registry::CacheType::Memory);

        let mut rt = YoukiRuntime::new(config, None).await.expect("runtime");
        // Force the legacy full-copy path regardless of the host's real
        // capabilities: BOTH overlay backends must be off, else on a box with
        // `fuse-overlayfs` installed the rootless fuse path would be selected
        // (a real overlay mount), not the copy path this test exercises.
        rt.overlayfs_rootfs_available = false;
        rt.fuse_overlayfs_rootfs_available = false;

        // Stage two layers as `blob_staging_filename`-style files.
        let l0 = tar_layer_bytes(&[("a.txt", b"alpha"), ("dir/b.txt", b"beta")]);
        let l1 = tar_layer_bytes(&[("c.txt", b"gamma")]);
        let stage = tmp.path().join("stage");
        std::fs::create_dir_all(&stage).unwrap();
        let d0 = zlayer_paths::sanitize_digest("sha256:fb0");
        let d1 = zlayer_paths::sanitize_digest("sha256:fb1");
        std::fs::write(stage.join(&d0), &l0).unwrap();
        std::fs::write(stage.join(&d1), &l1).unwrap();
        let layers = vec![
            (stage.join(&d0), TAR_MT.to_string()),
            (stage.join(&d1), TAR_MT.to_string()),
        ];

        let bundle = tmp.path().join("bundles").join("svc-0");
        std::fs::create_dir_all(&bundle).unwrap();
        let rootfs = bundle.join("rootfs");

        rt.build_container_rootfs("svc-0", &bundle, &rootfs, &layers, &stage)
            .await
            .expect("fallback rootfs build");

        // Files from BOTH layers were copied directly into the container rootfs.
        assert_eq!(std::fs::read(rootfs.join("a.txt")).unwrap(), b"alpha");
        assert_eq!(std::fs::read(rootfs.join("dir/b.txt")).unwrap(), b"beta");
        assert_eq!(std::fs::read(rootfs.join("c.txt")).unwrap(), b"gamma");
        // No overlay artifacts: rootfs is a plain dir, not a mountpoint, and no
        // layer store was populated.
        let mounts = std::fs::read_to_string("/proc/mounts").unwrap_or_default();
        let rootfs_str = rootfs.display().to_string();
        assert!(
            !mounts.lines().any(|l| l.contains(&rootfs_str)),
            "fallback rootfs must NOT be an overlay mountpoint"
        );
        assert!(
            !rt.layer_store_dir().exists() || dir_bytes(&rt.layer_store_dir()) == 0,
            "fallback path must not populate the shared layer store"
        );
    }

    /// ROOTLESS fuse path through the full `build_container_rootfs` wiring: with
    /// the kernel overlay bit OFF but the fuse bit ON (and a resolved binary),
    /// `build_container_rootfs` must select the fuse backend — producing a
    /// `fuse.*` mount at the container rootfs, a shared `fs-fuse` layer-store
    /// form, and a working merged view. Skips when `fuse-overlayfs` is absent.
    /// No root required.
    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn fuse_path_builds_rootfs_through_build_container_rootfs() {
        // Resolve fuse-overlayfs or skip.
        let path_var = std::env::var("PATH").unwrap_or_default();
        let bin = path_var
            .split(':')
            .filter(|d| !d.is_empty())
            .map(|d| std::path::Path::new(d).join("fuse-overlayfs"))
            .find(|p| p.exists());
        let Some(bin) = bin else {
            eprintln!(
                "skipping fuse_path_builds_rootfs_through_build_container_rootfs: \
                 fuse-overlayfs not installed"
            );
            return;
        };
        if std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open("/dev/fuse")
            .is_err()
        {
            eprintln!("skipping: /dev/fuse not openable");
            return;
        }

        let tmp = tempfile::tempdir().unwrap();
        let mut config = YoukiConfig::from_data_dir(tmp.path());
        config.cache_type = Some(zlayer_registry::CacheType::Memory);
        let mut rt = YoukiRuntime::new(config, None).await.expect("runtime");
        // Force the rootless fuse path: kernel overlay OFF, fuse ON with a
        // resolved binary (the survey may not have run a probe in this test env).
        rt.overlayfs_rootfs_available = false;
        rt.fuse_overlayfs_rootfs_available = true;
        rt.fuse_overlayfs_bin = Some(bin);

        let l0 = tar_layer_bytes(&[("a.txt", b"alpha"), ("dir/b.txt", b"beta")]);
        let l1 = tar_layer_bytes(&[("c.txt", b"gamma")]);
        let stage = tmp.path().join("stage");
        std::fs::create_dir_all(&stage).unwrap();
        let d0 = zlayer_paths::sanitize_digest("sha256:fz0");
        let d1 = zlayer_paths::sanitize_digest("sha256:fz1");
        std::fs::write(stage.join(&d0), &l0).unwrap();
        std::fs::write(stage.join(&d1), &l1).unwrap();
        let layers = vec![
            (stage.join(&d0), TAR_MT.to_string()),
            (stage.join(&d1), TAR_MT.to_string()),
        ];

        let bundle = tmp.path().join("bundles").join("svc-0");
        std::fs::create_dir_all(&bundle).unwrap();
        let rootfs = bundle.join("rootfs");

        rt.build_container_rootfs("svc-0", &bundle, &rootfs, &layers, &stage)
            .await
            .expect("fuse rootfs build");

        // Merged view sees both layers.
        assert_eq!(std::fs::read(rootfs.join("a.txt")).unwrap(), b"alpha");
        assert_eq!(std::fs::read(rootfs.join("dir/b.txt")).unwrap(), b"beta");
        assert_eq!(std::fs::read(rootfs.join("c.txt")).unwrap(), b"gamma");

        // The rootfs IS a fuse mount (the rootless backend was selected).
        let mounts = std::fs::read_to_string("/proc/mounts").unwrap_or_default();
        let rootfs_str = rootfs.display().to_string();
        assert!(
            mounts
                .lines()
                .any(|l| l.contains(&rootfs_str) && l.contains("fuse")),
            "rootfs must be a fuse.* mount when the rootless fuse backend is selected"
        );

        // The shared layer store holds the ROOTLESS `fs-fuse` form, not `fs`.
        let store = rt.layer_store_dir();
        assert!(
            store.join(&d0).join("fs-fuse").exists(),
            "fuse extract must populate the fs-fuse layer-store form"
        );
        assert!(
            !store.join(&d0).join("fs").exists(),
            "fuse extract must NOT populate the kernel `fs` form"
        );

        // Teardown unmounts the fuse mount cleanly.
        cleanup_bundle_blocking(&bundle, "svc-0").unwrap();
        let mounts_after = std::fs::read_to_string("/proc/mounts").unwrap_or_default();
        assert!(
            !mounts_after.lines().any(|l| l.contains(&rootfs_str)),
            "teardown must unmount the fuse rootfs"
        );
    }

    /// Sum of regular-file bytes under `dir` (symlink-safe), 0 if absent.
    #[cfg(target_os = "linux")]
    fn dir_bytes(dir: &std::path::Path) -> u64 {
        use std::os::unix::fs::MetadataExt;
        let mut bytes = 0u64;
        let _ = zlayer_paths::safe_fs::walk_no_follow(dir, |_p, md| {
            if md.is_file() {
                bytes = bytes.saturating_add(md.size());
            }
            Ok(())
        });
        bytes
    }

    #[test]
    fn test_container_id_str() {
        let id = ContainerId::new("myservice".to_string(), 1);

        let expected = "myservice-1";
        assert_eq!(format!("{}-{}", id.service, id.replica), expected);
    }

    #[test]
    fn test_rootfs_path_sanitization() {
        // Test that image names are sanitized for filesystem paths
        let images = vec![
            (
                "docker.io/library/nginx:latest",
                "docker.io_library_nginx_latest",
            ),
            ("ghcr.io/owner/repo:v1.0", "ghcr.io_owner_repo_v1.0"),
            (
                "registry.example.com/image@sha256:abc123",
                "registry.example.com_image_sha256_abc123",
            ),
        ];

        for (image, expected_suffix) in images {
            let safe_name = image.replace(['/', ':', '@'], "_");
            assert_eq!(safe_name, expected_suffix);
        }
    }

    #[test]
    fn test_map_status() {
        // Test status mapping without runtime instance
        let mappings = vec![
            (ContainerStatus::Creating, "Pending"),
            (ContainerStatus::Created, "Pending"),
            (ContainerStatus::Running, "Running"),
            (ContainerStatus::Stopped, "Exited"),
            (ContainerStatus::Paused, "Stopping"),
        ];

        for (status, expected) in mappings {
            let state = match status {
                ContainerStatus::Creating | ContainerStatus::Created => ContainerState::Pending,
                ContainerStatus::Running => ContainerState::Running,
                ContainerStatus::Stopped => ContainerState::Exited { code: 0 },
                ContainerStatus::Paused => ContainerState::Stopping,
            };

            let state_str = match state {
                ContainerState::Pending => "Pending",
                ContainerState::Running => "Running",
                ContainerState::Exited { .. } => "Exited",
                ContainerState::Stopping => "Stopping",
                _ => "Other",
            };

            assert_eq!(state_str, expected);
        }
    }

    #[test]
    #[allow(unsafe_code)]
    fn reap_exit_code_captures_nonzero_exit() {
        use nix::unistd::{fork, ForkResult};
        match unsafe { fork() }.expect("fork") {
            ForkResult::Child => {
                // Child: exit 7 immediately. Use libc::_exit to avoid running
                // any atexit/teardown in the test harness.
                unsafe { libc::_exit(7) };
            }
            ForkResult::Parent { child } => {
                assert_eq!(reap_exit_code(child.as_raw()), Some(7));
            }
        }
    }

    #[test]
    #[allow(unsafe_code)]
    fn reap_exit_code_captures_signal_death() {
        use nix::unistd::{fork, ForkResult};
        match unsafe { fork() }.expect("fork") {
            ForkResult::Child => unsafe {
                libc::raise(libc::SIGKILL);
                libc::_exit(0); // unreachable
            },
            ForkResult::Parent { child } => {
                assert_eq!(reap_exit_code(child.as_raw()), Some(137));
            }
        }
    }

    #[test]
    fn build_runtime_argv_prepends_state_root_and_verb() {
        let argv = build_runtime_argv(
            Path::new("/var/lib/zlayer/containers"),
            &["create", "svc-0", "--bundle", "/b", "--systemd-cgroup"],
        );
        let argv: Vec<String> = argv
            .into_iter()
            .map(|s| s.to_string_lossy().into_owned())
            .collect();
        assert_eq!(
            argv,
            vec![
                "runtime".to_string(),
                "--state-root".to_string(),
                "/var/lib/zlayer/containers".to_string(),
                "create".to_string(),
                "svc-0".to_string(),
                "--bundle".to_string(),
                "/b".to_string(),
                "--systemd-cgroup".to_string(),
            ]
        );
    }

    #[test]
    fn build_runtime_argv_kill_signal_order_matches_cli() {
        // `kill <id> <signal> --all` — id then signal (positional), matching
        // the Wave-1 RuntimeCommand::Kill contract.
        let argv = build_runtime_argv(Path::new("/s"), &["kill", "svc-0", "SIGTERM", "--all"]);
        let argv: Vec<String> = argv
            .into_iter()
            .map(|s| s.to_string_lossy().into_owned())
            .collect();
        assert_eq!(
            argv,
            vec![
                "runtime",
                "--state-root",
                "/s",
                "kill",
                "svc-0",
                "SIGTERM",
                "--all"
            ]
        );
    }

    #[test]
    fn parse_exec_exit_status_recovers_real_code() {
        assert_eq!(
            parse_exec_exit_status("Error: exec process exited with status 42"),
            Some(42)
        );
        assert_eq!(
            parse_exec_exit_status("noise\nexec process exited with status 1\nmore"),
            Some(1)
        );
        assert_eq!(parse_exec_exit_status("some unrelated error"), None);
        assert_eq!(parse_exec_exit_status(""), None);
    }

    /// `read_pidfile` parses the PID `zlayer runtime create` writes; the
    /// pid-file lives beside (not inside) the state dir at `<state>/<id>.pid`.
    #[tokio::test]
    async fn read_pidfile_round_trips() {
        let temp_base = ZLayerDirs::system_default()
            .tmp()
            .join(format!("youki_pidfile_test_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&temp_base);

        let config = YoukiConfig {
            state_dir: temp_base.join("state"),
            rootfs_dir: temp_base.join("rootfs"),
            bundle_dir: temp_base.join("bundles"),
            cache_dir: temp_base.join("cache"),
            volume_dir: temp_base.join("volumes"),
            use_systemd: false,
            cache_type: None,
            log_base_dir: None,
            deployment_name: None,
        };
        let runtime = YoukiRuntime::new(config, None)
            .await
            .expect("build runtime");

        let id = ContainerId::new("svc".to_string(), 1);

        // No pid-file yet -> None.
        assert_eq!(runtime.read_pidfile(&id).await, None);

        // Write a pid-file at the path create_container would use, then read it.
        let pidfile = runtime.pidfile_path(&id);
        assert_eq!(
            pidfile,
            runtime.config.state_dir.join("svc-1.pid"),
            "pid-file must sit beside the state dir, not inside it"
        );
        tokio::fs::write(&pidfile, "  12345\n")
            .await
            .expect("write pid-file");
        assert_eq!(runtime.read_pidfile(&id).await, Some(12345));

        // Garbage -> None.
        tokio::fs::write(&pidfile, "not-a-pid")
            .await
            .expect("rewrite");
        assert_eq!(runtime.read_pidfile(&id).await, None);

        let _ = std::fs::remove_dir_all(&temp_base);
    }

    /// `cleanup_bundle_blocking` must delete a bundle tree that contains
    /// read-only directories (extracted image layers are routinely `0o555`),
    /// which a plain `remove_dir_all` fails on with `EACCES`. This is the
    /// disk-leak path — a partial tree that never reclaims.
    #[test]
    fn cleanup_bundle_blocking_removes_readonly_tree() {
        use std::os::unix::fs::PermissionsExt;

        let base = ZLayerDirs::system_default()
            .tmp()
            .join(format!("youki_cleanup_ro_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&base);

        let bundle = base.join("svc-1");
        let ro_dir = bundle.join("rootfs").join("etc").join("ca-trust");
        std::fs::create_dir_all(&ro_dir).expect("mkdir tree");
        std::fs::write(ro_dir.join("anchor.pem"), b"x").expect("write file");
        // Make the leaf directory read-only (no write bit) — this is what makes
        // a naive remove_dir_all fail.
        std::fs::set_permissions(&ro_dir, std::fs::Permissions::from_mode(0o555))
            .expect("chmod ro");

        // Sanity: a plain remove_dir_all WOULD fail on this tree (proves the
        // test exercises the real failure the chmod sweep fixes). We don't
        // assert it (env may run as root, which ignores the mode), just clean.
        cleanup_bundle_blocking(&bundle, "svc-1").expect("cleanup must succeed");
        assert!(!bundle.exists(), "bundle tree must be gone after cleanup");

        // Idempotent on a missing path.
        cleanup_bundle_blocking(&bundle, "svc-1").expect("cleanup of missing path is ok");

        let _ = std::fs::remove_dir_all(&base);
    }

    /// `pid_is_alive` reports the running test process as alive and a reaped
    /// child as dead.
    #[test]
    #[allow(unsafe_code)]
    fn pid_is_alive_detects_live_and_dead() {
        use nix::unistd::{fork, ForkResult};

        // Our own pid is alive.
        assert!(pid_is_alive(std::process::id()), "own pid must be alive");

        // Fork a child that exits immediately, reap it, then it is dead.
        match unsafe { fork() }.expect("fork") {
            ForkResult::Child => unsafe { libc::_exit(0) },
            ForkResult::Parent { child } => {
                // Reap so the pid is fully gone (not a zombie that lingers).
                let _ = reap_exit_code(child.as_raw());
                let dead = u32::try_from(child.as_raw()).unwrap();
                assert!(!pid_is_alive(dead), "reaped child pid must be dead");
            }
        }
    }

    /// `read_init_pid_for` prefers the `<state>/<name>.pid` file and returns
    /// `None` when neither the pid-file nor libcontainer state yields a pid.
    #[test]
    fn read_init_pid_for_prefers_pidfile() {
        let base = ZLayerDirs::system_default()
            .tmp()
            .join(format!("youki_initpid_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&base);
        let state_dir = base.join("state");
        let state_root = state_dir.join("svc-3");
        std::fs::create_dir_all(&state_root).expect("mkdir state root");

        // No pid-file and no libcontainer state -> None.
        assert_eq!(read_init_pid_for(&state_dir, &state_root, "svc-3"), None);

        // Pid-file present -> that pid.
        std::fs::write(state_dir.join("svc-3.pid"), "4321\n").expect("write pid");
        assert_eq!(
            read_init_pid_for(&state_dir, &state_root, "svc-3"),
            Some(4321)
        );

        // A zero pid in the file is treated as "no pid".
        std::fs::write(state_dir.join("svc-3.pid"), "0").expect("write zero pid");
        assert_eq!(read_init_pid_for(&state_dir, &state_root, "svc-3"), None);

        let _ = std::fs::remove_dir_all(&base);
    }

    /// End-to-end orphan reclaim via `prune_containers`: a bundle whose state
    /// dir lingers but whose init pid is DEAD must be reclaimed (the
    /// half-removed-container leak), while a bundle whose init pid is ALIVE
    /// must be spared even though it is not tracked in memory.
    #[tokio::test]
    #[allow(unsafe_code)]
    async fn prune_reclaims_dead_init_spares_live_init() {
        use nix::sys::wait::waitpid;
        use nix::unistd::{fork, ForkResult};

        let base = ZLayerDirs::system_default()
            .tmp()
            .join(format!("youki_prune_live_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&base);

        let config = YoukiConfig {
            state_dir: base.join("state"),
            rootfs_dir: base.join("rootfs"),
            bundle_dir: base.join("bundles"),
            cache_dir: base.join("cache"),
            volume_dir: base.join("volumes"),
            use_systemd: false,
            cache_type: None,
            log_base_dir: None,
            deployment_name: None,
        };
        let runtime = YoukiRuntime::new(config, None)
            .await
            .expect("build runtime");
        let state_dir = runtime.config.state_dir.clone();
        let bundle_dir = runtime.config.bundle_dir.clone();

        // Fork a long-lived child to stand in for a LIVE container init. It
        // sleeps; we kill+reap it at the end so the test leaves no process.
        let live_child = match unsafe { fork() }.expect("fork") {
            ForkResult::Child => {
                // Sleep well past the test; killed below.
                std::thread::sleep(std::time::Duration::from_secs(30));
                unsafe { libc::_exit(0) };
            }
            ForkResult::Parent { child } => child,
        };
        let live_pid = u32::try_from(live_child.as_raw()).unwrap();

        // Build on-disk layout. BOTH get a lingering state dir + bundle dir
        // (the half-removed shape). `live-1` carries a pid-file with the live
        // child's pid; `dead-1` carries NO pid-file and NO libcontainer state,
        // so `read_init_pid_for` yields None — i.e. no LIVE init owns it. (We
        // deliberately avoid "reap a child and reuse its pid" because a busy
        // test binary can recycle that pid into a live process, making the
        // dead case flaky; the no-recoverable-pid path is the canonical
        // half-removed scenario and is deterministic.)
        std::fs::create_dir_all(state_dir.join("live-1")).expect("mkdir state");
        std::fs::create_dir_all(bundle_dir.join("live-1").join("rootfs")).expect("mkdir bundle");
        std::fs::write(state_dir.join("live-1.pid"), live_pid.to_string()).expect("write pidfile");

        std::fs::create_dir_all(state_dir.join("dead-1")).expect("mkdir state");
        std::fs::create_dir_all(bundle_dir.join("dead-1").join("rootfs")).expect("mkdir bundle");
        // No pid-file for dead-1 on purpose.

        // Back-date both bundle mtimes well past the BUNDLE_PRUNE_SAFETY window
        // so they are eligible (the safety guard only skips just-created ones).
        let old = std::time::SystemTime::now() - std::time::Duration::from_secs(3600);
        for name in ["live-1", "dead-1"] {
            let f = std::fs::File::open(bundle_dir.join(name)).expect("open bundle dir");
            f.set_modified(old).expect("set mtime");
        }

        let result = runtime.prune_containers().await.expect("prune");

        assert!(
            result.deleted.contains(&"dead-1".to_string()),
            "bundle with a DEAD init must be reclaimed; got {:?}",
            result.deleted
        );
        assert!(
            !result.deleted.contains(&"live-1".to_string()),
            "bundle with a LIVE init must be spared; got {:?}",
            result.deleted
        );
        assert!(
            !bundle_dir.join("dead-1").exists(),
            "dead bundle dir must be gone"
        );
        assert!(
            bundle_dir.join("live-1").exists(),
            "live bundle dir must remain"
        );

        // Clean up the live child.
        let _ = nix::sys::signal::kill(live_child, nix::sys::signal::Signal::SIGKILL);
        let _ = waitpid(live_child, None);
        let _ = std::fs::remove_dir_all(&base);
    }

    /// Re-adoption arms exactly ONE watcher per live pid: a second
    /// `readopt_exit_watchers_impl` over the same on-disk state must arm 0
    /// (the `watcher_armed` guard prevents double-arming).
    #[tokio::test]
    #[allow(unsafe_code)]
    async fn readopt_arms_exactly_one_watcher() {
        use nix::sys::wait::waitpid;
        use nix::unistd::{fork, ForkResult};

        let base = ZLayerDirs::system_default()
            .tmp()
            .join(format!("youki_readopt_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&base);

        let config = YoukiConfig {
            state_dir: base.join("state"),
            rootfs_dir: base.join("rootfs"),
            bundle_dir: base.join("bundles"),
            cache_dir: base.join("cache"),
            volume_dir: base.join("volumes"),
            use_systemd: false,
            cache_type: None,
            log_base_dir: None,
            deployment_name: None,
        };
        let runtime = YoukiRuntime::new(config, None)
            .await
            .expect("build runtime");
        let state_dir = runtime.config.state_dir.clone();

        // A live child init standing in for a surviving container.
        let child = match unsafe { fork() }.expect("fork") {
            ForkResult::Child => {
                std::thread::sleep(std::time::Duration::from_secs(30));
                unsafe { libc::_exit(0) };
            }
            ForkResult::Parent { child } => child,
        };
        let pid = u32::try_from(child.as_raw()).unwrap();

        std::fs::create_dir_all(state_dir.join("svc-1")).expect("mkdir state");
        std::fs::write(state_dir.join("svc-1.pid"), pid.to_string()).expect("write pid");

        // First re-adoption arms exactly one watcher and tracks the container.
        let armed_first = runtime.readopt_exit_watchers_impl().await;
        assert_eq!(armed_first, 1, "first re-adoption must arm exactly one");
        {
            let containers = runtime.containers.read().await;
            let info = containers.get("svc-1").expect("container tracked");
            assert_eq!(info.pid, Some(pid));
            assert!(
                info.watcher_armed
                    .load(std::sync::atomic::Ordering::Acquire),
                "watcher_armed must be set after arming"
            );
        }

        // Second re-adoption must arm ZERO (already tracked + already armed).
        let armed_second = runtime.readopt_exit_watchers_impl().await;
        assert_eq!(
            armed_second, 0,
            "second re-adoption must not double-arm a live pid"
        );

        // Clean up the live child.
        let _ = nix::sys::signal::kill(child, nix::sys::signal::Signal::SIGKILL);
        let _ = waitpid(child, None);
        let _ = std::fs::remove_dir_all(&base);
    }

    #[test]
    fn test_log_paths() {
        let config = YoukiConfig::default();
        let dirs = zlayer_paths::ZLayerDirs::system_default();
        let id = ContainerId::new("testservice".to_string(), 2);

        let container_id = format!("{}-{}", id.service, id.replica);
        let state_dir = config.state_dir.join(&container_id);
        let stdout = state_dir.join("stdout.log");
        let stderr = state_dir.join("stderr.log");

        assert_eq!(stdout, dirs.containers().join("testservice-2/stdout.log"));
        assert_eq!(stderr, dirs.containers().join("testservice-2/stderr.log"));
    }

    #[test]
    fn test_youki_config_clone() {
        let config = YoukiConfig {
            state_dir: PathBuf::from("/custom/state"),
            rootfs_dir: PathBuf::from("/custom/rootfs"),
            bundle_dir: PathBuf::from("/custom/bundles"),
            cache_dir: PathBuf::from("/custom/cache"),
            volume_dir: PathBuf::from("/custom/volumes"),
            use_systemd: true,
            cache_type: Some(zlayer_registry::CacheType::memory()),
            log_base_dir: Some(PathBuf::from("/var/log/zlayer")),
            deployment_name: Some("myapp".to_string()),
        };

        let cloned = config.clone();

        assert_eq!(cloned.state_dir, config.state_dir);
        assert_eq!(cloned.rootfs_dir, config.rootfs_dir);
        assert_eq!(cloned.bundle_dir, config.bundle_dir);
        assert_eq!(cloned.cache_dir, config.cache_dir);
        assert_eq!(cloned.volume_dir, config.volume_dir);
        assert_eq!(cloned.use_systemd, config.use_systemd);
        assert!(cloned.cache_type.is_some());
        assert_eq!(cloned.log_base_dir, config.log_base_dir);
        assert_eq!(cloned.deployment_name, config.deployment_name);
    }

    /// Test that `YoukiRuntime::new()` creates directories
    #[tokio::test]
    async fn test_youki_runtime_directory_creation() {
        // Use a unique temp directory based on test run
        let temp_base = ZLayerDirs::system_default()
            .tmp()
            .join(format!("youki_test_{}", std::process::id()));

        let config = YoukiConfig {
            state_dir: temp_base.join("state"),
            rootfs_dir: temp_base.join("rootfs"),
            bundle_dir: temp_base.join("bundles"),
            cache_dir: temp_base.join("cache"),
            volume_dir: temp_base.join("volumes"),
            use_systemd: false,
            cache_type: None,
            log_base_dir: None,
            deployment_name: None,
        };

        // Clean up any previous test run
        let _ = std::fs::remove_dir_all(&temp_base);

        // This should succeed and create all directories
        let result = YoukiRuntime::new(config.clone(), None).await;

        assert!(
            result.is_ok(),
            "Failed to create runtime: {:?}",
            result.err()
        );

        // Verify directories were created
        assert!(config.state_dir.exists());
        assert!(config.rootfs_dir.exists());
        assert!(config.bundle_dir.exists());
        assert!(config.cache_dir.exists());
        assert!(config.volume_dir.exists());

        // Clean up
        let _ = std::fs::remove_dir_all(&temp_base);
    }

    /// Regression: an image present only in the local registry's index.json
    /// (never written to the blob cache, e.g. a locally-built image) MUST be
    /// returned by `list_images()`. Previously `list_images()` scanned only the
    /// blob cache for `manifest:`-prefixed keys, so `zlayer image ls` returned
    /// nothing for build-only nodes even though index.json listed images.
    #[tokio::test]
    async fn test_list_images_reads_registry_index() {
        let temp_base = ZLayerDirs::system_default()
            .tmp()
            .join(format!("youki_listimg_test_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&temp_base);

        let config = YoukiConfig {
            state_dir: temp_base.join("state"),
            rootfs_dir: temp_base.join("rootfs"),
            bundle_dir: temp_base.join("bundles"),
            cache_dir: temp_base.join("cache"),
            volume_dir: temp_base.join("volumes"),
            use_systemd: false,
            // In-memory blob cache => guaranteed NO `manifest:` keys, isolating
            // the index.json path under test.
            cache_type: Some(zlayer_registry::CacheType::memory()),
            log_base_dir: None,
            deployment_name: None,
        };

        let runtime = YoukiRuntime::new(config, None)
            .await
            .expect("runtime should build");

        // Write one image straight into the runtime's own local registry index,
        // bypassing the blob cache entirely.
        let registry = runtime
            .local_registry
            .as_ref()
            .expect("local registry should be present");

        // Minimal but valid OCI image manifest with known sizes.
        let config_blob = br#"{"architecture":"amd64","os":"linux"}"#;
        let layer_blob = b"layer-bytes-1234567890";
        let config_digest = registry.put_blob(config_blob).await.unwrap();
        let layer_digest = registry.put_blob(layer_blob).await.unwrap();

        let config_size = i64::try_from(config_blob.len()).unwrap();
        let layer_size = i64::try_from(layer_blob.len()).unwrap();
        let manifest = serde_json::json!({
            "schemaVersion": 2,
            "mediaType": "application/vnd.oci.image.manifest.v1+json",
            "config": {
                "mediaType": "application/vnd.oci.image.config.v1+json",
                "digest": config_digest,
                "size": config_size,
            },
            "layers": [{
                "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
                "digest": layer_digest,
                "size": layer_size,
            }],
        });
        let manifest_bytes = serde_json::to_vec(&manifest).unwrap();
        registry
            .put_manifest("myapp", "v1", &manifest_bytes)
            .await
            .unwrap();

        let images = runtime.list_images().await.expect("list_images should ok");

        let found = images
            .iter()
            .find(|i| i.reference == "myapp:v1")
            .expect("image from index.json must be listed even with empty blob cache");

        assert!(
            found.digest.is_some(),
            "digest should be resolved from the registry manifest"
        );
        let expected_size = u64::try_from(config_size + layer_size).unwrap();
        assert_eq!(
            found.size_bytes,
            Some(expected_size),
            "size should be sum of config + layer descriptor sizes"
        );

        let _ = std::fs::remove_dir_all(&temp_base);
    }

    /// `image_store_handles()` must return the runtime's ALREADY-OPEN
    /// `LocalRegistry` + blob cache so the `POST /images/import` handler can
    /// reuse them instead of opening a second (exclusively-locked) blob cache.
    ///
    /// This is the regression guard for the 501-on-import bug: `serve.rs` used
    /// to open a *second* on-disk blob cache for the import attach, which failed
    /// with "database is locked by another process", left the registry-backed
    /// import path unwired, and made imports fall through to the youki runtime's
    /// `import_image` (Unsupported -> 501). We assert both handles come back
    /// `Some` and are the SAME `Arc`s the runtime holds (pointer-equal), and
    /// that an import written through the shared registry is visible via the
    /// runtime's own registry — proving they reference one store.
    #[tokio::test]
    async fn image_store_handles_returns_shared_open_handles() {
        let temp_base = ZLayerDirs::system_default()
            .tmp()
            .join(format!("youki_storehandles_test_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&temp_base);

        let config = YoukiConfig {
            state_dir: temp_base.join("state"),
            rootfs_dir: temp_base.join("rootfs"),
            bundle_dir: temp_base.join("bundles"),
            cache_dir: temp_base.join("cache"),
            volume_dir: temp_base.join("volumes"),
            use_systemd: false,
            cache_type: Some(zlayer_registry::CacheType::memory()),
            log_base_dir: None,
            deployment_name: None,
        };

        let runtime = YoukiRuntime::new(config, None)
            .await
            .expect("runtime should build");

        let (registry, cache) = runtime
            .image_store_handles()
            .expect("youki runtime with a local registry must expose store handles");

        // Same underlying allocations as the runtime's own fields — not a fresh
        // open. This is what guarantees the single-process-exclusive blob cache
        // is shared rather than re-opened.
        let own_registry = runtime
            .local_registry
            .as_ref()
            .expect("local registry present");
        assert!(
            std::sync::Arc::ptr_eq(&registry, own_registry),
            "returned registry must be the runtime's own Arc"
        );
        assert!(
            std::sync::Arc::ptr_eq(&cache, &runtime.blob_cache),
            "returned blob cache must be the runtime's own Arc"
        );

        // A blob/manifest written through the shared handle is visible via the
        // runtime's own registry: they reference one store.
        let config_blob = br#"{"architecture":"amd64","os":"linux"}"#;
        let layer_blob = b"shared-store-layer";
        let config_digest = registry.put_blob(config_blob).await.unwrap();
        let layer_digest = registry.put_blob(layer_blob).await.unwrap();
        let manifest = serde_json::json!({
            "schemaVersion": 2,
            "mediaType": "application/vnd.oci.image.manifest.v1+json",
            "config": {
                "mediaType": "application/vnd.oci.image.config.v1+json",
                "digest": config_digest,
                "size": i64::try_from(config_blob.len()).unwrap(),
            },
            "layers": [{
                "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
                "digest": layer_digest,
                "size": i64::try_from(layer_blob.len()).unwrap(),
            }],
        });
        let manifest_bytes = serde_json::to_vec(&manifest).unwrap();
        registry
            .put_manifest("shared", "v1", &manifest_bytes)
            .await
            .unwrap();

        own_registry
            .get_manifest("shared", "v1")
            .await
            .expect("manifest written via shared handle must be visible to the runtime's registry");

        let _ = std::fs::remove_dir_all(&temp_base);
    }

    /// Tail-from-end logic: write a file with a known number of lines,
    /// ask `compute_tail_offset` for the start of the last 2, and verify
    /// the offset lands on the third-to-last line's start.
    #[tokio::test]
    async fn youki_tail_offset_returns_last_n_lines() {
        let dir = ZLayerDirs::system_default().tmp().join(format!(
            "zlayer_tail_test_{}_{}",
            std::process::id(),
            chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0)
        ));
        tokio::fs::create_dir_all(&dir).await.unwrap();
        let path = dir.join("log");
        // 5 lines, each ending in '\n', distinguishable bodies.
        tokio::fs::write(&path, b"a\nbb\nccc\ndddd\neeeee\n")
            .await
            .unwrap();

        let mut file = tokio::fs::File::open(&path).await.unwrap();
        let len = file.metadata().await.unwrap().len();
        // Last 2 lines are "dddd\n" + "eeeee\n", combined 11 bytes;
        // offset should be `len - 11`.
        let offset = compute_tail_offset(&mut file, len, 2).await;
        assert_eq!(offset, len - 11, "expected last-2-lines offset");

        // Tail >= total line count must yield 0 (whole file).
        let mut file = tokio::fs::File::open(&path).await.unwrap();
        assert_eq!(compute_tail_offset(&mut file, len, 100).await, 0);

        let _ = tokio::fs::remove_dir_all(&dir).await;
    }

    /// `stream_log_file` with `follow=false` reads all existing lines and
    /// then terminates cleanly at EOF. Exercises the non-follow path
    /// without needing a real container.
    #[tokio::test]
    async fn youki_logs_stream_reads_static_file_without_follow() {
        let dir = ZLayerDirs::system_default().tmp().join(format!(
            "zlayer_logs_static_{}_{}",
            std::process::id(),
            chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0)
        ));
        tokio::fs::create_dir_all(&dir).await.unwrap();
        let path = dir.join("stdout.log");
        tokio::fs::write(&path, b"hello\nworld\n").await.unwrap();

        let (tx, mut rx) = mpsc::channel::<Result<LogChunk>>(8);
        let opts = LogsStreamOptions {
            follow: false,
            tail: None,
            since: None,
            until: None,
            timestamps: true,
            stdout: true,
            stderr: false,
        };
        stream_log_file(path, LogChannel::Stdout, opts, tx, None)
            .await
            .unwrap();

        let mut received = Vec::new();
        while let Some(item) = rx.recv().await {
            let chunk = item.unwrap();
            received.push(chunk);
        }
        assert_eq!(
            received.len(),
            2,
            "expected 2 chunks, got {}",
            received.len()
        );
        assert_eq!(received[0].stream, LogChannel::Stdout);
        assert!(received[0].timestamp.is_some());
        assert_eq!(received[0].bytes.as_ref(), b"hello\n");
        assert_eq!(received[1].bytes.as_ref(), b"world\n");

        let _ = tokio::fs::remove_dir_all(&dir).await;
    }

    /// `stream_log_file` with `follow=true` must CLOSE the stream once the
    /// container has exited rather than tailing the log file forever. We pass
    /// a non-existent `container_root`, so `container_has_stopped` returns
    /// `true` immediately on first EOF, exercising the exit-close path: the
    /// existing lines are delivered, then the stream ends (`recv` -> `None`).
    #[tokio::test]
    async fn youki_logs_stream_follow_closes_on_exit() {
        let dir = ZLayerDirs::system_default().tmp().join(format!(
            "zlayer_logs_follow_exit_{}_{}",
            std::process::id(),
            chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0)
        ));
        tokio::fs::create_dir_all(&dir).await.unwrap();
        let path = dir.join("stdout.log");
        tokio::fs::write(&path, b"hello\nworld\n").await.unwrap();

        // A container_root that does not exist -> `container_has_stopped`
        // returns `true`, so follow mode closes after draining.
        let nonexistent_root = dir.join("no_such_container_state");
        assert!(!nonexistent_root.exists());

        let (tx, mut rx) = mpsc::channel::<Result<LogChunk>>(8);
        let opts = LogsStreamOptions {
            follow: true,
            tail: None,
            since: None,
            until: None,
            timestamps: true,
            stdout: true,
            stderr: false,
        };

        let result = tokio::time::timeout(Duration::from_secs(5), async move {
            stream_log_file(path, LogChannel::Stdout, opts, tx, Some(nonexistent_root))
                .await
                .unwrap();

            let mut received = Vec::new();
            while let Some(item) = rx.recv().await {
                received.push(item.unwrap());
            }
            received
        })
        .await;

        let received = result.expect("follow stream did not close on container exit (timed out)");
        assert_eq!(
            received.len(),
            2,
            "expected 2 chunks before close, got {}",
            received.len()
        );
        assert_eq!(received[0].bytes.as_ref(), b"hello\n");
        assert_eq!(received[1].bytes.as_ref(), b"world\n");

        let _ = tokio::fs::remove_dir_all(&dir).await;
    }

    /// `read_stats_sample` requires a real cgroup v2 directory layout
    /// that we cannot easily synthesise on a non-root test host, so this
    /// test is `#[ignore]`-d. Run with
    /// `cargo test -p zlayer-agent youki_stats_sample_reads_cgroup -- --ignored`
    /// inside a real container or as root with a fake cgroup path.
    #[tokio::test]
    #[ignore = "requires a real cgroup v2 hierarchy"]
    async fn youki_stats_sample_reads_cgroup() {
        // Try the host's own root cgroup as a smoke target — every cgroup
        // v2 system has /sys/fs/cgroup/cpu.stat and memory.current at the
        // root. pids.current/pids.max may be missing at the root, which
        // is fine — read_stats_sample treats them as 0/None.
        let path = Path::new("/sys/fs/cgroup");
        let sample = read_stats_sample(path, 1).await.unwrap();
        // CPU monotonically increases; memory.current is non-negative.
        assert!(sample.cpu_total_ns < u64::MAX);
        assert!(sample.online_cpus >= 1);
    }

    /// `(rows, cols)` from the resize channel converts to `nix::pty::Winsize`
    /// with `ws_row` and `ws_col` populated and the pixel fields zeroed.
    /// Pure shape conversion — no fd touched.
    #[test]
    fn youki_pty_resize_winsize_shape() {
        let (rows, cols): (u16, u16) = (24, 80);
        let ws = nix::pty::Winsize {
            ws_row: rows,
            ws_col: cols,
            ws_xpixel: 0,
            ws_ypixel: 0,
        };
        assert_eq!(ws.ws_row, 24);
        assert_eq!(ws.ws_col, 80);
        assert_eq!(ws.ws_xpixel, 0);
        assert_eq!(ws.ws_ypixel, 0);

        // Maximum values still fit cleanly through the channel and the
        // ioctl payload (winsize is u16 across all four fields).
        let ws_max = nix::pty::Winsize {
            ws_row: u16::MAX,
            ws_col: u16::MAX,
            ws_xpixel: 0,
            ws_ypixel: 0,
        };
        assert_eq!(ws_max.ws_row, u16::MAX);
        assert_eq!(ws_max.ws_col, u16::MAX);
    }

    /// `exec_pty` rejects an empty command vector with `InvalidSpec` before
    /// touching libcontainer or allocating a PTY. Exercises the input
    /// validation path without needing a running container.
    #[tokio::test]
    async fn youki_exec_pty_rejects_empty_command() {
        let temp_base = ZLayerDirs::system_default()
            .tmp()
            .join(format!("youki_exec_pty_empty_{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&temp_base);

        let config = YoukiConfig {
            state_dir: temp_base.join("state"),
            rootfs_dir: temp_base.join("rootfs"),
            bundle_dir: temp_base.join("bundles"),
            cache_dir: temp_base.join("cache"),
            volume_dir: temp_base.join("volumes"),
            use_systemd: false,
            cache_type: None,
            log_base_dir: None,
            deployment_name: None,
        };

        let runtime = YoukiRuntime::new(config, None).await.unwrap();
        let id = ContainerId::new("missing".to_string(), 0);

        let result = runtime
            .exec_pty(
                &id,
                ExecOptions {
                    command: Vec::new(),
                    tty: true,
                    ..ExecOptions::default()
                },
            )
            .await;

        assert!(matches!(result, Err(AgentError::InvalidSpec(_))));

        let _ = std::fs::remove_dir_all(&temp_base);
    }

    /// `PtyDuplex::new` accepts a non-blocking PTY master fd produced by
    /// `nix::pty::openpty` and exposes it as `AsyncRead + AsyncWrite`. This
    /// is purely a wrapper smoke test — no container, no exec.
    ///
    /// Marked `#[ignore]` because real PTY allocation requires
    /// `/dev/ptmx`, which CI sandboxes occasionally restrict; run with
    /// `cargo test -p zlayer-agent --features youki-runtime youki_pty_duplex_wraps_master -- --ignored`.
    #[tokio::test]
    #[ignore = "requires /dev/ptmx access"]
    async fn youki_pty_duplex_wraps_master() {
        let pair = nix::pty::openpty(None, None).expect("openpty");
        nix::fcntl::fcntl(
            &pair.master,
            nix::fcntl::FcntlArg::F_SETFL(nix::fcntl::OFlag::O_NONBLOCK),
        )
        .expect("F_SETFL O_NONBLOCK");

        let _duplex = PtyDuplex::new(pair.master).expect("PtyDuplex::new");
        // Slave is dropped here, which makes the master EOF on next read.
        drop(pair.slave);
    }

    /// Sanity-check the `unpack_tar_into` + `build_tar_into_sender`
    /// helpers used by the youki archive endpoints: a TAR archive built
    /// from a host directory must round-trip back to the same file tree
    /// when unpacked elsewhere.
    #[tokio::test]
    async fn archive_helpers_round_trip_a_directory_tree() {
        // Build a small tree.
        let src_dir = ZLayerDirs::system_default()
            .scratch_dir("youki-archive-test-")
            .unwrap();
        let nested = src_dir.path().join("a/b");
        std::fs::create_dir_all(&nested).unwrap();
        std::fs::write(nested.join("c.txt"), b"deep file").unwrap();
        std::fs::write(src_dir.path().join("top.txt"), b"top file").unwrap();

        // Drive `build_tar_into_sender` through a Tokio mpsc and collect bytes.
        let (tx, mut rx) = mpsc::channel::<Result<bytes::Bytes>>(8);
        let target = src_dir.path().to_path_buf();
        let handle =
            tokio::task::spawn_blocking(move || super::build_tar_into_sender(&target, "root", &tx));
        let mut buf = Vec::new();
        while let Some(item) = rx.recv().await {
            let chunk = item.expect("tar chunk");
            buf.extend_from_slice(&chunk);
        }
        handle.await.unwrap().unwrap();

        // Unpack into a fresh dir; the entry should land under `root/`.
        let dest_dir = ZLayerDirs::system_default()
            .scratch_dir("youki-archive-test-")
            .unwrap();
        super::unpack_tar_into(
            dest_dir.path(),
            &buf,
            crate::runtime::ArchivePutOptions::default(),
        )
        .unwrap();
        assert!(dest_dir.path().join("root/top.txt").exists());
        assert!(dest_dir.path().join("root/a/b/c.txt").exists());
        assert_eq!(
            std::fs::read(dest_dir.path().join("root/a/b/c.txt")).unwrap(),
            b"deep file",
        );
    }

    /// `unpack_tar_into` with `no_overwrite_dir_non_dir = true` must
    /// reject an archive entry that would replace an existing directory
    /// with a non-directory (or vice versa).
    #[tokio::test]
    async fn archive_helpers_reject_dir_nondir_replacements() {
        let dest_dir = ZLayerDirs::system_default()
            .scratch_dir("youki-archive-test-")
            .unwrap();
        // Pre-create a directory at `target`.
        let target = dest_dir.path().join("target");
        std::fs::create_dir_all(&target).unwrap();

        // Build an archive whose only entry is a *file* named `target`.
        let src_file_dir = ZLayerDirs::system_default()
            .scratch_dir("youki-archive-test-")
            .unwrap();
        let src_file = src_file_dir.path().join("target");
        std::fs::write(&src_file, b"i am a file").unwrap();
        let bytes = super::build_tar_from_path_for_test(&src_file, "target");
        let opts = crate::runtime::ArchivePutOptions {
            no_overwrite_dir_non_dir: true,
            copy_uid_gid: false,
        };
        let err = super::unpack_tar_into(dest_dir.path(), &bytes, opts).unwrap_err();
        assert!(
            matches!(err, AgentError::InvalidSpec(_)),
            "expected InvalidSpec, got {err:?}"
        );
    }

    #[test]
    fn from_data_dir_scopes_all_paths() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let cfg = YoukiConfig::from_data_dir(tmp.path());
        assert!(cfg.cache_dir.starts_with(tmp.path()));
        assert!(cfg.state_dir.starts_with(tmp.path()));
        assert!(cfg.rootfs_dir.starts_with(tmp.path()));
        assert!(cfg.bundle_dir.starts_with(tmp.path()));
        assert!(cfg.volume_dir.starts_with(tmp.path()));
    }
}