zlayer-agent 0.13.0

//! macOS Seatbelt sandbox runtime
//!
//! Implements the [`Runtime`] trait using macOS process isolation:
//! - `fork()` + `sandbox_init()` + `exec()` for process creation
//! - Seatbelt `.sb` profiles for mandatory access control (deny-default whitelist)
//! - APFS `clonefile()` for copy-on-write filesystem isolation
//! - Direct Metal/MPS GPU access at 100% native performance
//!
//! This module is only compiled on macOS targets (`#[cfg(target_os = "macos")]`).
//! It provides lightweight process-level isolation without requiring Docker or
//! a Linux container runtime.
//!
//! ## Architecture
//!
//! Each "container" is a native macOS process running under a generated Seatbelt
//! profile. The profile restricts filesystem, network, IPC, and device access
//! based on the [`ServiceSpec`]. The rootfs is cloned from a pulled OCI image
//! using APFS copy-on-write (nearly instantaneous, zero additional disk space
//! until files are modified).
//!
//! ## Directory Layout
//!
//! ```text
//! {data_dir}/
//!   images/
//!     {sanitized_image_name}/
//!       rootfs/           -- extracted OCI image layers
//!   containers/
//!     {service}-{replica}/
//!       rootfs/           -- APFS clone of base image rootfs
//!       config.json       -- serialized ServiceSpec
//!       sandbox.sb        -- generated Seatbelt profile
//!       stdout.log        -- captured stdout
//!       stderr.log        -- captured stderr
//!       pid               -- PID file
//!       tmp/              -- container temp directory
//! ```

use crate::cgroups_stats::ContainerStats;
use crate::error::{AgentError, Result};
use crate::runtime::{
    ContainerId, ContainerState, ImageInfo, ImageInspectInfo, LogChannel, LogChunk, LogsStream,
    LogsStreamOptions, OverlayAttachKind, PruneResult, Runtime,
};
use crate::storage_manager::StorageManager;
use crate::MacSandboxConfig;
use std::collections::{HashMap, HashSet};
use std::fmt::Write;
use std::net::IpAddr;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::sync::OnceLock;
use std::time::{Duration, Instant};
use tokio::io::{AsyncBufReadExt, AsyncSeekExt, BufReader};
use tokio::sync::mpsc;
use tokio::sync::RwLock;
use tokio_stream::wrappers::ReceiverStream;
use zlayer_observability::logs::{LogEntry, LogSource, LogStream};
use zlayer_registry::BlobCacheBackend;
use zlayer_spec::{RegistryAuth, ServiceSpec};

/// Default PATH for macOS Seatbelt containers when neither the image nor the
/// spec supplies one (scratch/CI images bake no PATH). Includes /usr/bin + /bin
/// where macOS CLT puts `env`, `bash`, `clang`. Set-if-absent so a user/image
/// PATH wins.
pub(crate) const DEFAULT_SEATBELT_PATH: &str =
    "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin";

/// Inject [`DEFAULT_SEATBELT_PATH`] into `spec.env` when the spec carries no
/// `PATH` **or** carries an empty/whitespace `PATH`. Set-if-absent-or-empty: a
/// non-empty user- or image-supplied PATH always wins. Called in
/// `create_container` so the on-disk `config.json` the executor reads carries a
/// usable PATH even for scratch images that bake none (or bake a bare `PATH=`).
/// Kept in lock-step with [`ensure_spawn_path`] so `config.json` and the live
/// process env agree.
fn ensure_default_path(spec: &mut ServiceSpec) {
    match spec.env.get("PATH") {
        // Non-empty user/image PATH wins — leave it untouched.
        Some(v) if !v.trim().is_empty() => {}
        // Absent, or present-but-empty: stamp the default.
        _ => {
            spec.env
                .insert("PATH".to_string(), DEFAULT_SEATBELT_PATH.to_string());
        }
    }
}

/// Spawn-time PATH guard for the FINAL `(KEY, VALUE)` env handed to
/// `Command::envs`. `merge_env` faithfully layers the image/spec env, so a
/// scratch/CI image (bare `PATH=`) or a spec with `PATH=""` can leave a `PATH`
/// key present but EMPTY in `env_base`. Because the spawn clears the inherited
/// environment (`env_clear`), an empty `PATH` is handed to the child verbatim,
/// and `exec`-by-name (`env`, `bash`) then fails with "not found". Inject the
/// default when PATH is absent OR present-but-empty; a non-empty user/image PATH
/// always wins. Same const + semantics as [`ensure_default_path`].
fn ensure_spawn_path(env_vars: &mut Vec<(String, String)>) {
    match env_vars.iter_mut().find(|(k, _)| k == "PATH") {
        // Present-but-empty: overwrite with the default (the actual live bug).
        Some((_, v)) if v.trim().is_empty() => *v = DEFAULT_SEATBELT_PATH.to_string(),
        // Non-empty user/image PATH wins.
        Some(_) => {}
        // Absent: append the default.
        None => env_vars.push(("PATH".to_string(), DEFAULT_SEATBELT_PATH.to_string())),
    }
}

/// Inject a provisioned toolchain (the macOS `git` keg from `zlayer-toolchain`)
/// into the FINAL `(KEY, VALUE)` spawn/exec env.
///
/// Two effects, mirrored at both spawn and exec sites:
/// - **PATH**: PREPEND the toolchain's `path_dirs` (the keg's `bin`) so the
///   provisioned git wins over the host `/usr/bin/git`. Host git EPERMs reading
///   `/etc/gitconfig` under Seatbelt (deny-default `/private/etc`); the keg git
///   reads its OWN `GIT_CONFIG_SYSTEM` instead. Must run AFTER `ensure_spawn_path`
///   so PATH is already non-empty and we cleanly prepend.
/// - **env**: insert each of the toolchain's vars set-if-absent, so a job that
///   pinned any of them keeps its own value. The source-built git keg ships only
///   `GIT_EXEC_PATH` (no `GIT_CONFIG_SYSTEM` / `DYLD_FALLBACK_LIBRARY_PATH`): the
///   built git has absolute system-library load commands and reads its own
///   compiled `<keg>/etc/gitconfig`.
fn inject_toolchain_env(
    env_vars: &mut Vec<(String, String)>,
    handle: &zlayer_toolchain::ToolchainHandle,
) {
    if !handle.path_dirs.is_empty() {
        let prefix = handle.path_dirs.join(":");
        match env_vars.iter_mut().find(|(k, _)| k == "PATH") {
            // Non-empty PATH present: prepend so the toolchain bin is searched first.
            Some((_, v)) if !v.trim().is_empty() => *v = format!("{prefix}:{v}"),
            // Present-but-empty: replace with just the toolchain dirs.
            Some((_, v)) => *v = prefix,
            // Absent: seed it.
            None => env_vars.push(("PATH".to_string(), prefix)),
        }
    }
    // Sort for deterministic ordering across the HashMap iteration.
    let mut extra: Vec<(&String, &String)> = handle.env.iter().collect();
    extra.sort_by(|a, b| a.0.cmp(b.0));
    for (k, val) in extra {
        if !env_vars.iter().any(|(ek, _)| ek == k) {
            env_vars.push((k.clone(), val.clone()));
        }
    }
}

/// Canonical readable path to the system CA bundle inside the Seatbelt sandbox.
///
/// The deny-default profile grants `/private/etc` read but NOT the `/etc` ->
/// `/private/etc` symlink traversal, so the OpenSSL/LibreSSL default of
/// `/etc/ssl/cert.pem` can't be reached. Both vars below point straight at the
/// resolved `/private/etc` path the profile grants.
const SANDBOX_CA_BUNDLE: &str = "/private/etc/ssl/cert.pem";

/// Point every sandboxed process at the system CA bundle via its canonical
/// readable path, **set-if-absent** so a job can override.
///
/// macOS's system OpenSSL/LibreSSL (and anything statically linked against it,
/// e.g. a keg `git`/`curl`) defaults its CA bundle to `/etc/ssl/cert.pem`.
/// Under the deny-default Seatbelt profile the `/etc` symlink isn't traversable,
/// so an HTTPS handshake fails with `error setting certificate verify
/// locations: CAfile: /etc/ssl/cert.pem CApath: none` — exactly the live Mac
/// `actions/checkout` clone failure. Exporting both `SSL_CERT_FILE` (the generic
/// OpenSSL override, benefiting any TLS tool) and `GIT_SSL_CAINFO` (git's
/// curl-backend override) at the resolved `/private/etc` path fixes it for all
/// sandboxed processes — not git-specific.
fn inject_ca_cert_env(env_vars: &mut Vec<(String, String)>) {
    for key in ["SSL_CERT_FILE", "GIT_SSL_CAINFO"] {
        if !env_vars.iter().any(|(k, _)| k == key) {
            env_vars.push((key.to_string(), SANDBOX_CA_BUNDLE.to_string()));
        }
    }
}

/// Point every sandboxed process at a writable per-container `HOME`,
/// **set-if-absent** so a job can override.
///
/// Unlike Linux/VZ and Windows containers, the darwin Seatbelt runtime has no
/// mount namespace and never seeded `HOME` — the child started with no home at
/// all. Anything that writes to `$HOME` then dies: `git config --global`
/// (setup-go's private-module `insteadOf` auth) fails with `fatal: $HOME not
/// set`, and `go`'s subsequent private-module git fetch needs the same writable
/// home. Point `HOME` at a granted, per-container dir (`{container_dir}/home`,
/// added to the SBPL writable dirs) so those writes land somewhere allowed.
fn inject_home_env(env_vars: &mut Vec<(String, String)>, home_dir: &Path) {
    if !env_vars.iter().any(|(k, _)| k == "HOME") {
        env_vars.push(("HOME".to_string(), home_dir.to_string_lossy().into_owned()));
    }
}

/// Host's active Xcode/CLT developer dir, resolved once via `xcode-select -p`.
///
/// Apple's `clang`/`xcrun` shims shell out to `xcode-select`, which reads the
/// host selector at `/var/select/developer_dir` — a path the deny-default
/// Seatbelt profile blocks (`clang --version` dies with "unable to read data
/// link ... No developer tools were found"). Exporting `DEVELOPER_DIR` makes
/// them skip that read. Resolved on the host (unsandboxed) and cached for the
/// process; `None` when xcode-select is missing/errors (no CLT) so callers skip.
fn host_developer_dir() -> Option<&'static str> {
    static DEV_DIR: OnceLock<Option<String>> = OnceLock::new();
    DEV_DIR
        .get_or_init(|| {
            let out = std::process::Command::new("/usr/bin/xcode-select")
                .arg("-p")
                .output()
                .ok()?;
            if !out.status.success() {
                return None;
            }
            let dir = String::from_utf8_lossy(&out.stdout).trim().to_string();
            (!dir.is_empty()).then_some(dir)
        })
        .as_deref()
}

/// If `path` lives inside a macOS `.app` bundle, return the bundle root (the
/// `…app` prefix); otherwise `None`.
///
/// Full Xcode's `clang` reaches the WHOLE bundle, not just `Contents/Developer`:
/// it stats `Contents/Info.plist` and dlopens
/// `Contents/SharedFrameworks/DVTSystemPrerequisites.framework`. So when the
/// selected developer dir is `…/Xcode.app/Contents/Developer` we must grant the
/// `…/Xcode.app` root — the selected toolchain bundle only, NOT all of
/// `/Applications` and not a maintained allowlist.
fn enclosing_app_bundle(path: &str) -> Option<&str> {
    if let Some(idx) = path.find(".app/") {
        return Some(&path[..idx + ".app".len()]);
    }
    Path::new(path)
        .extension()
        .is_some_and(|ext| ext == "app")
        .then_some(path)
}

/// Host Darwin per-user temp + cache dirs, resolved once.
struct DarwinUserDirs {
    temp: Option<PathBuf>,
    cache: Option<PathBuf>,
}

/// Resolve the Darwin per-user temp/cache dirs (`DARWIN_USER_TEMP_DIR` /
/// `DARWIN_USER_CACHE_DIR`) on the host via `getconf`, canonicalized, cached once.
///
/// `clang`/`xcrun` write their `xcrun_db` SDK cache + module cache here — located
/// via `confstr`, NOT `TMPDIR` — so without a grant every compiler invocation
/// dies with "couldn't create cache file ... Operation not permitted". `getconf`
/// hands back the `/var/folders/<hash>/{T,C}/` form; we canonicalize so the
/// `/var` -> `/private/var` symlink resolves to the real path Seatbelt's subpath
/// rules match against.
fn host_darwin_user_dirs() -> &'static DarwinUserDirs {
    static DIRS: OnceLock<DarwinUserDirs> = OnceLock::new();
    DIRS.get_or_init(|| DarwinUserDirs {
        temp: getconf_canonical("DARWIN_USER_TEMP_DIR"),
        cache: getconf_canonical("DARWIN_USER_CACHE_DIR"),
    })
}

/// Run `getconf <name>` on the host and canonicalize the result to a real path.
fn getconf_canonical(name: &str) -> Option<PathBuf> {
    let out = std::process::Command::new("/usr/bin/getconf")
        .arg(name)
        .output()
        .ok()?;
    if !out.status.success() {
        return None;
    }
    let raw = String::from_utf8_lossy(&out.stdout).trim().to_string();
    if raw.is_empty() {
        return None;
    }
    std::fs::canonicalize(&raw).ok()
}

// ---------------------------------------------------------------------------
// Seatbelt profile types and generation
// ---------------------------------------------------------------------------

/// GPU access level for the sandbox profile.
///
/// Controls which `IOKit` user client classes, Mach services, and framework
/// paths are allowed in the generated Seatbelt profile.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GpuAccess {
    /// No GPU access -- deny all `IOKit` and GPU Mach services.
    None,
    /// Full Metal compute -- shader compilation + `IOKit` GPU access.
    /// Required for custom Metal shaders, `PyTorch` MPS with JIT compilation,
    /// and any workload that calls `MTLCreateSystemDefaultDevice()`.
    MetalCompute,
    /// MPS only -- pre-compiled kernels, no `MTLCompilerService` needed.
    /// Suitable for inference-only workloads using Apple's pre-built MPS kernels.
    /// Smaller attack surface than full Metal compute.
    MpsOnly,
}

/// Network access level for the sandbox profile.
#[derive(Debug, Clone)]
pub enum NetworkAccess {
    /// No network access at all.
    None,
    /// Only specific localhost ports (for inter-service communication).
    LocalhostOnly {
        bind_ports: Vec<u16>,
        connect_ports: Vec<u16>,
        /// Overlay CIDR (e.g. "10.200.0.0/16") to allow outbound to, so a
        /// host-net-shared Seatbelt container can reach the node overlay IP
        /// (daemon) + sibling overlay IPs. `None` = no overlay egress.
        overlay_cidr: Option<String>,
        /// Host ZLayer daemon API port to additionally whitelist via the literal
        /// `127.0.0.1` remote-ip form (not just `localhost`). Set only when the
        /// sandbox is given daemon creds (`ZLAYER_API_URL`/`ZLAYER_TOKEN`), so a
        /// nested `zlayer build` can delegate to the host daemon on loopback.
        /// `None` = no daemon egress opened.
        loopback_api_port: Option<u16>,
    },
    /// Full network access (outbound + inbound + bind).
    Full,
}

/// macOS keychain / code-signing access level for the sandbox profile.
///
/// The Seatbelt sandbox is an SBPL profile layered over the *host* filesystem
/// (the rootfs is a writable subpath; there is no `pivot_root`). That means the
/// host's `securityd`, the host keychain databases, and the host code-signing
/// CLIs (`/usr/bin/codesign`, `/usr/bin/security`) are all reachable from inside
/// a sandboxed container -- but only if the profile explicitly grants the
/// `com.apple.SecurityServer` Mach service plus file + executable-mapping access
/// to the relevant paths. By default ([`KeychainAccess::None`]) all of this is
/// denied (the base profile never names `SecurityServer`), so a container cannot
/// touch the keychain. A container opts in via the `zlayer.io/keychain` label,
/// and the daemon operator must additionally enable the capability
/// (`MacSandboxConfig::keychain_access_allowed`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum KeychainAccess {
    /// No keychain access -- deny `securityd`, keychain files, and the signing
    /// CLIs. This is the default and matches the historical behavior.
    None,
    /// Allow keychain + code-signing access scoped to the given host keychain
    /// database files. Each path's parent directory is also opened read/write
    /// (securityd writes `*.keychain-db`, `-shm`, `-wal`, and lock siblings).
    Enabled {
        /// Absolute paths to the host keychain database files to permit (e.g.
        /// `/Users/zachary/Library/Keychains/zlayer-build.keychain-db`).
        keychain_paths: Vec<PathBuf>,
    },
}

/// Complete sandbox configuration used to generate a Seatbelt profile.
#[derive(Debug, Clone)]
pub struct SandboxConfig {
    /// Path to the container's cloned rootfs.
    pub rootfs_dir: PathBuf,
    /// Path to the container's workspace/state directory.
    pub workspace_dir: PathBuf,
    /// GPU access level.
    pub gpu_access: GpuAccess,
    /// Keychain / code-signing access level.
    pub keychain_access: KeychainAccess,
    /// Network access configuration.
    pub network_access: NetworkAccess,
    /// Directories the process can write to (volume mounts).
    pub writable_dirs: Vec<PathBuf>,
    /// Additional read-only directories.
    pub readonly_dirs: Vec<PathBuf>,
    /// Host toolchain-cache dir holding the provisioned `git` keg
    /// (`zlayer-toolchain`). Granted `file-read*` + `file-map-executable` so the
    /// keg git — which lives OUTSIDE the rootfs clone — can be `stat`'d, read,
    /// and exec-mapped by dyld. `None` when no toolchain was provisioned (the
    /// grant is then omitted).
    pub toolchain_cache: Option<PathBuf>,
    /// Maximum open file descriptors.
    pub max_files: u64,
    /// CPU time limit in seconds (`RLIMIT_CPU`).
    pub cpu_time_limit: Option<u64>,
    /// Memory limit in bytes (for watchdog, not kernel-enforced on macOS).
    pub memory_limit: Option<u64>,
}

/// Full Metal compute profile section.
///
/// Allows `IOKit` GPU access, Mach shader compilation services,
/// and all filesystem paths needed for Metal.framework.
///
/// `IOKit` user client class names were derived from:
/// - `ioreg -l -w0` on macOS 26 / Apple M5 (`AGXAcceleratorG17G`, `AGXDeviceUserClient`)
/// - Apple's own system sandbox profiles:
///   - `/System/Library/Sandbox/Profiles/com.apple.intelligenceplatformd.sb`
///   - `/System/Library/Sandbox/Profiles/safety-inference-extension-macos.sb`
///   - `/System/Library/Sandbox/Profiles/com.apple.intelligenceplatform.IntelligencePlatformComputeService.sb`
///
/// Key insight: On Apple Silicon (M1+), the actual `IOUserClient` class opened by
/// `MTLCreateSystemDefaultDevice()` is `AGXDeviceUserClient` -- NOT `AGXAccelerator`
/// (which is the IOService/kernel driver class name, not a user client class).
/// `AGXSharedUserClient` is needed for multi-process GPU sharing.
const METAL_COMPUTE_PROFILE_SECTION: &str = "\
; --- GPU: Full Metal Compute ---

; IOKit user clients for GPU hardware access
; Apple Silicon (M1/M2/M3/M4/M5): AGXDeviceUserClient is the actual user client
; class opened by Metal. AGXSharedUserClient handles multi-process GPU sharing.
; The IOAccel* classes are IOKit compatibility shims (still needed).
(allow iokit-open
  (iokit-user-client-class \"AGXDeviceUserClient\")
  (iokit-user-client-class \"AGXSharedUserClient\")
  (iokit-user-client-class \"IOSurfaceRootUserClient\")
  (iokit-user-client-class \"IOSurfaceAcceleratorClient\")
  (iokit-user-client-class \"IOAccelDevice\")
  (iokit-user-client-class \"IOAccelDevice2\")
  (iokit-user-client-class \"IOAccelContext\")
  (iokit-user-client-class \"IOAccelContext2\")
  (iokit-user-client-class \"IOAccelSharedUserClient\")
  (iokit-user-client-class \"IOAccelSharedUserClient2\")
  (iokit-user-client-class \"IOAccelSubmitter2\")
  (iokit-user-client-class \"RootDomainUserClient\"))

; IOKit service-level access (macOS 26+ fine-grained syntax)
; AGXAcceleratorG* prefix matches all Apple Silicon GPU generations.
(allow iokit-open-service
  (iokit-user-client-class \"IOSurfaceRoot\")
  (iokit-registry-entry-class-prefix \"AGXAcceleratorG\"))

; IOKit user-client-level access (macOS 26+ fine-grained syntax)
(allow iokit-open-user-client
  (iokit-user-client-class \"AGXDeviceUserClient\")
  (iokit-user-client-class \"AGXSharedUserClient\")
  (iokit-user-client-class \"IOSurfaceRootUserClient\")
  (iokit-user-client-class \"IOSurfaceAcceleratorClient\"))

; GPU IOKit properties (comprehensive set from Apple's safety-inference profile)
(allow iokit-get-properties
  (iokit-property \"AGCInfo\")
  (iokit-property \"AGXCliqueTracingDefaults\")
  (iokit-property \"AGXInternalPerfCounterResourcesPath\")
  (iokit-property \"AGXLimitersDirName\")
  (iokit-property \"AGXParameterBufferMaxSize\")
  (iokit-property \"AGXParameterBufferMaxSizeEverMemless\")
  (iokit-property \"AGXParameterBufferMaxSizeNeverMemless\")
  (iokit-property \"AGXTraceCodeVersion\")
  (iokit-property \"CFBundleIdentifier\")
  (iokit-property \"CFBundleIdentifierKernel\")
  (iokit-property \"chip-id\")
  (iokit-property \"CommandSubmissionEnabled\")
  (iokit-property \"CompactVRAM\")
  (iokit-property \"EnableBlitLib\")
  (iokit-property \"gpu-core-count\")
  (iokit-property \"GPUConfigurationVariable\")
  (iokit-property \"GPUDCCDisplayable\")
  (iokit-property \"GPUDebugNullClientMask\")
  (iokit-property \"GpuDebugPolicy\")
  (iokit-property \"GPURawCounterBundleName\")
  (iokit-property \"GPURawCounterPluginClassName\")
  (iokit-property \"IOClass\")
  (iokit-property \"IOClassNameOverride\")
  (iokit-property \"IOGeneralInterest\")
  (iokit-property \"IOGLBundleName\")
  (iokit-property \"IOGLESBundleName\")
  (iokit-property \"IOGLESDefaultUseMetal\")
  (iokit-property \"IOGLESMetalBundleName\")
  (iokit-property \"IOMatchCategory\")
  (iokit-property \"IOMatchedAtBoot\")
  (iokit-property \"IONameMatch\")
  (iokit-property \"IONameMatched\")
  (iokit-property \"IOPCIMatch\")
  (iokit-property \"IOPersonalityPublisher\")
  (iokit-property \"IOPowerManagement\")
  (iokit-property \"IOProbeScore\")
  (iokit-property \"IOProviderClass\")
  (iokit-property \"IORegistryEntryPropertyKeys\")
  (iokit-property \"IOReportLegend\")
  (iokit-property \"IOReportLegendPublic\")
  (iokit-property \"IOSourceVersion\")
  (iokit-property \"KDebugVersion\")
  (iokit-property \"MetalCoalesce\")
  (iokit-property \"MetalPluginClassName\")
  (iokit-property \"MetalPluginName\")
  (iokit-property \"MetalStatisticsName\")
  (iokit-property \"MetalStatisticsScriptName\")
  (iokit-property \"model\")
  (iokit-property \"PerformanceStatistics\")
  (iokit-property \"Removable\")
  (iokit-property \"SafeEjectRequested\")
  (iokit-property \"SchedulerState\")
  (iokit-property \"SCMBuildTime\")
  (iokit-property \"SCMVersionNumber\")
  (iokit-property \"soc-generation\")
  (iokit-property \"SurfaceList\")
  (iokit-property \"vendor-id\")
  (iokit-property \"device-id\")
  (iokit-property \"class-code\"))

; Mach services for Metal shader compilation and GPU memory
(allow mach-lookup
  (global-name \"com.apple.MTLCompilerService\")
  (global-name \"com.apple.CARenderServer\")
  (global-name \"com.apple.PowerManagement.control\")
  (global-name \"com.apple.gpu.process\")
  (global-name \"com.apple.gpumemd.source\")
  (global-name \"com.apple.cvmsServ\"))

; XPC services for shader compilation (Apple Silicon)
(allow mach-lookup
  (xpc-service-name \"com.apple.MTLCompilerService\")
  (xpc-service-name-prefix \"com.apple.AGXCompilerService\"))

; User preferences for Metal/OpenGL
(allow user-preference-read
  (preference-domain \"com.apple.opengl\")
  (preference-domain \"com.apple.Metal\")
  (preference-domain \"com.nvidia.OpenGL\"))

; GPU driver bundles and libraries
(allow file-read*
  (subpath \"/Library/GPUBundles\")
  (subpath \"/System/Library/Frameworks/Metal.framework\")
  (subpath \"/System/Library/Frameworks/MetalPerformanceShaders.framework\")
  (subpath \"/System/Library/Frameworks/MetalPerformanceShadersGraph.framework\")
  (subpath \"/System/Library/PrivateFrameworks/GPUCompiler.framework\"))

";

/// MPS-only profile section (subset of Metal compute).
///
/// MPS mode provides a smaller attack surface than full Metal compute by
/// restricting `IOKit` access to a minimal set and omitting `AGXCompilerService`
/// XPC services. However, `MTLCompilerService` is still required because
/// `MPSGraph` on macOS 26+ uses JIT compilation internally for kernel fusion.
///
/// Uses the same corrected `IOKit` user client classes as the full Metal profile
/// (`AGXDeviceUserClient` instead of the incorrect `AGXAccelerator`).
const MPS_ONLY_PROFILE_SECTION: &str = "\
; --- GPU: MPS Only (pre-compiled kernels, no shader compilation) ---

; IOKit user clients for GPU hardware access (minimal set)
; AGXDeviceUserClient is required -- MTLCreateSystemDefaultDevice() opens this class.
(allow iokit-open
  (iokit-user-client-class \"AGXDeviceUserClient\")
  (iokit-user-client-class \"AGXSharedUserClient\")
  (iokit-user-client-class \"IOSurfaceRootUserClient\")
  (iokit-user-client-class \"IOAccelDevice2\")
  (iokit-user-client-class \"IOAccelContext2\")
  (iokit-user-client-class \"IOAccelSharedUserClient2\")
  (iokit-user-client-class \"RootDomainUserClient\"))

; IOKit service-level access (macOS 26+ fine-grained syntax)
(allow iokit-open-service
  (iokit-user-client-class \"IOSurfaceRoot\")
  (iokit-registry-entry-class-prefix \"AGXAcceleratorG\"))

; IOKit user-client-level access (macOS 26+ fine-grained syntax)
(allow iokit-open-user-client
  (iokit-user-client-class \"AGXDeviceUserClient\")
  (iokit-user-client-class \"IOSurfaceRootUserClient\"))

; GPU IOKit properties (minimal set for MPS)
(allow iokit-get-properties
  (iokit-property \"MetalPluginClassName\")
  (iokit-property \"MetalPluginName\")
  (iokit-property \"IOClass\")
  (iokit-property \"IOGLESDefaultUseMetal\")
  (iokit-property \"IORegistryEntryPropertyKeys\")
  (iokit-property \"IOSourceVersion\")
  (iokit-property \"GPUConfigurationVariable\")
  (iokit-property \"GPURawCounterBundleName\")
  (iokit-property \"gpu-core-count\")
  (iokit-property \"model\")
  (iokit-property \"vendor-id\")
  (iokit-property \"device-id\")
  (iokit-property \"soc-generation\"))

; Mach services for MPS
; MTLCompilerService is required because MPSGraph on macOS 26+ uses JIT
; compilation internally for kernel fusion, even for pre-compiled MPS kernels.
; gpumemd.source is needed for GPU memory management.
(allow mach-lookup
  (global-name \"com.apple.MTLCompilerService\")
  (global-name \"com.apple.PowerManagement.control\")
  (global-name \"com.apple.gpumemd.source\"))

; XPC service for MTLCompilerService (required by MPSGraph JIT)
(allow mach-lookup
  (xpc-service-name \"com.apple.MTLCompilerService\"))

; User preferences
(allow user-preference-read
  (preference-domain \"com.apple.Metal\"))

; MPS framework access
(allow file-read*
  (subpath \"/Library/GPUBundles\")
  (subpath \"/System/Library/Frameworks/Metal.framework\")
  (subpath \"/System/Library/Frameworks/MetalPerformanceShaders.framework\")
  (subpath \"/System/Library/Frameworks/MetalPerformanceShadersGraph.framework\"))

";

/// Emit `(allow file-read-metadata (literal "<dir>"))` for each ancestor dir of
/// every path in `writable_roots` (from each path's parent up to and including
/// `/`), de-duplicated across the whole set.
///
/// Why: a Seatbelt `(subpath "<dir>")` grant covers `<dir>` and its descendants,
/// but NOT its ancestors. When a sandboxed process creates a writable subpath —
/// e.g. the CI executor's `os.MkdirAll("/Users/<u>/.zlayer/volumes/ws-1")` — the
/// libc/Go path walk `lstat`s every parent from `/` down. Under Seatbelt those
/// ancestor `lstat`s are DENIED, the parent looks "missing", and `MkdirAll`
/// tries `mkdir /Users` → `EEXIST` → "mkdir /Users: file exists". Granting
/// `file-read-metadata` on each ancestor lets the `lstat`/traverse succeed.
///
/// Metadata-only and tight: `literal` (not `subpath`) names exactly that one dir,
/// and `file-read-metadata` permits stat/traverse only — no directory listing,
/// no content read. Allow rules are additive under `(deny default)`, so order
/// relative to the writable grants does not matter.
///
/// NOTE: the SBPL `file-read-metadata` + `literal` ancestor syntax should be
/// smoke-tested on a real macOS host (`sandbox_compile`) — the box CI is the
/// authoritative check that the kernel accepts these rules.
fn write_ancestor_metadata_rules(profile: &mut String, writable_roots: &[PathBuf]) {
    let mut seen: HashSet<String> = HashSet::new();
    let mut rules = String::new();
    for root in writable_roots {
        let mut cur = root.parent();
        while let Some(anc) = cur {
            let s = anc.to_string_lossy().into_owned();
            if seen.insert(s.clone()) {
                let _ = writeln!(rules, "(allow file-read-metadata (literal \"{s}\"))");
            }
            if anc == Path::new("/") {
                break;
            }
            cur = anc.parent();
        }
    }
    if !rules.is_empty() {
        profile.push_str(
            "; --- Ancestor metadata (lstat/traverse parents of writable subpaths) ---\n",
        );
        profile.push_str(&rules);
        profile.push('\n');
    }
}

/// Append SBPL rules that let a sandboxed process reach a per-container Docker
/// Engine API socket at `socket_path` (a HOST path outside the rootfs).
///
/// The socket is provisioned at START time (its path embeds a monotonic
/// per-spawn counter, so it is not known when the profile is generated at CREATE
/// time), so these rules are appended to the already-generated profile just
/// before it is applied in the child. This MIRRORS how the daemon admin socket
/// (`ZLAYER_SOCKET`) is made reachable: a deny-default Seatbelt profile grants
/// `AF_UNIX` `connect()` via `file-read*`/`file-write*` on the socket node plus
/// `file-read-metadata` on each ancestor directory (the kernel `lstat`s every
/// path component when resolving the socket for `connect()`). No `network-*`
/// grant is emitted — matching the proven daemon-socket path and avoiding any
/// risk of `sandbox_init` rejecting an unfamiliar `unix-socket` remote filter
/// (which would fail-closed and break the whole container).
fn append_docker_socket_sbpl_rules(profile: &mut String, socket_path: &Path) {
    let _ = writeln!(profile, "; --- Per-container Docker Engine API socket ---");
    let _ = writeln!(
        profile,
        "(allow file-read* file-write* (subpath \"{}\"))",
        socket_path.display()
    );
    write_ancestor_metadata_rules(profile, std::slice::from_ref(&socket_path.to_path_buf()));
}

/// True once the Seatbelt container has reached a terminal state (or its record
/// is gone), so a follow-mode log tailer can deliver EOF instead of tailing
/// forever.
///
/// Unlike youki — where the workload is a grandchild and exit is read from an
/// on-disk runc status file — a Seatbelt workload is a DIRECT child of the
/// daemon, reaped via `waitpid` in `wait_container` / `container_state` / the
/// supervisor, each of which records `Exited`/`Failed` in this shared map. The
/// Docker-compat log consumer (e.g. the Forgejo runner) issues its
/// `/containers/{id}/wait` concurrently with the log follow, so that reap flips
/// the state here while we poll — no separate exit channel is needed. We only
/// READ the state (never reap) so we never steal the real exit code from
/// `wait_container`.
async fn sandbox_container_stopped(
    containers: &Arc<RwLock<HashMap<String, SandboxContainer>>>,
    dir_name: &str,
) -> bool {
    let guard = containers.read().await;
    match guard.get(dir_name) {
        Some(c) => matches!(
            c.state,
            ContainerState::Exited { .. } | ContainerState::Failed { .. }
        ),
        // Record gone (removed/torn down) -> no more output will ever arrive.
        None => true,
    }
}

/// Tail one of a Seatbelt container's captured log files (`stdout.log` /
/// `stderr.log`), emitting one [`LogChunk`] per line over `tx` until the
/// workload reaches a terminal state, then draining any final trailing line.
///
/// Mirrors the youki file-tailer (`stream_log_file`): the runtime modules don't
/// share code (each is `cfg`-gated to its own OS) so the proven shape is ported
/// rather than imported.
async fn stream_sandbox_log_file(
    path: PathBuf,
    channel: LogChannel,
    opts: LogsStreamOptions,
    tx: mpsc::Sender<Result<LogChunk>>,
    containers: Arc<RwLock<HashMap<String, SandboxContainer>>>,
    dir_name: String,
) {
    // A channel whose file never materialised (nothing written to stderr yet,
    // say) just contributes nothing; closing this sender lets the merged stream
    // finish on the other channel.
    let Ok(file) = tokio::fs::File::open(&path).await else {
        return;
    };
    let mut reader = BufReader::new(file);

    // Apply `tail`: seek so the next read begins at the start of the last N
    // lines.
    if let Some(tail) = opts.tail {
        if tail > 0 {
            if let Ok(meta) = reader.get_ref().metadata().await {
                let start = compute_tail_offset(reader.get_mut(), meta.len(), tail).await;
                if reader.seek(std::io::SeekFrom::Start(start)).await.is_err() {
                    return;
                }
            }
        }
    }

    let mut line = String::new();
    loop {
        line.clear();
        let Ok(bytes_read) = reader.read_line(&mut line).await else {
            return;
        };

        if bytes_read == 0 {
            // EOF. Without follow we're done.
            if !opts.follow {
                return;
            }
            // Follow mode: EOF is usually transient (more output is coming). But
            // once the workload has exited no more output can ever arrive, so the
            // stream must CLOSE — otherwise a consumer that does
            // `stream -> EOF -> wait-for-exit` would hang. Exit is observed via
            // the shared in-memory state the concurrent reap records.
            if !sandbox_container_stopped(&containers, &dir_name).await {
                tokio::time::sleep(Duration::from_millis(200)).await;
                continue;
            }
            // Exited: let the filesystem settle, then make one final read to
            // catch a trailing line flushed between our EOF and the status flip.
            tokio::time::sleep(Duration::from_millis(100)).await;
            line.clear();
            match reader.read_line(&mut line).await {
                Ok(0) | Err(_) => return,
                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 {
                return;
            }
        }

        let chunk = LogChunk {
            stream: channel,
            bytes: bytes::Bytes::copy_from_slice(line.as_bytes()),
            timestamp: opts.timestamps.then_some(now),
        };
        if tx.send(Ok(chunk)).await.is_err() {
            // Receiver dropped — stop tailing.
            return;
        }
    }
}

/// 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. Mirrors `tail -n N`; a file with
/// fewer than `N` lines yields `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;
        }
        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 {
                    let absolute = pos + (i as u64) + 1;
                    return absolute.min(file_len);
                }
            }
        }
    }

    0
}

/// Rewrite the daemon URL injected into a Seatbelt container to loopback.
///
/// A Seatbelt-sandboxed process runs natively in the HOST network namespace, so
/// it reaches the daemon on `127.0.0.1`. The `ContainerAuthContext` carries the
/// daemon's container-reachable URL, which is the overlay node IP (e.g.
/// `http://10.200.0.1:3669`) whenever overlay is enabled — correct for VZ-Linux
/// guests that join the mesh, but UNROUTABLE from a host-native Seatbelt process
/// (the in-container executor's `dial tcp 10.200.0.1:3669: i/o timeout`). Force
/// the host to loopback, preserving scheme + port; fall back verbatim on any
/// unexpected shape.
fn loopback_api_url(url: &str) -> String {
    let Some((scheme, authority)) = url.split_once("://") else {
        return url.to_string();
    };
    // authority is `host:port` or `[ipv6]:port` (no path — see
    // `container_reachable_api_url`); the port is whatever follows the last `:`.
    match authority.rsplit_once(':') {
        Some((_host, port)) if !port.is_empty() && port.chars().all(|c| c.is_ascii_digit()) => {
            format!("{scheme}://127.0.0.1:{port}")
        }
        _ => url.to_string(),
    }
}

/// Extract the TCP port from a daemon API URL (`scheme://host:port`), so the
/// Seatbelt egress rules can whitelist exactly the daemon's loopback port.
/// Returns `None` when the URL has no explicit numeric port (the caller then
/// falls back to the default daemon port, 3669).
fn api_url_port(url: &str) -> Option<u16> {
    let (_scheme, authority) = url.split_once("://")?;
    let (_host, port) = authority.rsplit_once(':')?;
    port.parse::<u16>().ok()
}

/// Generate a complete Seatbelt profile from a [`SandboxConfig`].
///
/// The profile follows a deny-default whitelist model: everything is denied
/// unless explicitly allowed. The profile is structured in sections:
///
/// 1. Base process rules (always needed for any process to run)
/// 2. System library access (dyld, libSystem, frameworks)
/// 3. Container rootfs access (read + write)
/// 4. Volume mount access (writable dirs)
/// 5. GPU rules (if `gpu_access != None`)
/// 6. Keychain / code-signing rules (if `keychain_access != None`)
/// 7. Network rules (based on `network_access`)
/// 8. Logging and /dev/null access
#[must_use]
#[allow(clippy::too_many_lines)]
pub fn generate_sandbox_profile(config: &SandboxConfig) -> String {
    let mut profile = String::with_capacity(4096);

    // Header
    profile.push_str("(version 1)\n");
    profile.push_str("(deny default)\n");
    profile.push('\n');

    // ===== Section 1: Base process rules =====
    profile.push_str("; --- Base process rules ---\n");
    profile.push_str("(allow process-exec)\n");
    profile.push_str("(allow process-fork)\n");
    profile.push_str("(allow signal (target same-sandbox))\n");
    profile.push_str("(allow process-info* (target self))\n");
    profile.push_str("(allow process-info-pidinfo)\n");
    profile.push_str("(allow process-info-rusage)\n");
    profile.push('\n');

    // ===== Section 2: System library and framework access =====
    profile.push_str("; --- System libraries (required for any process to run) ---\n");
    profile.push_str("(allow file-read*\n");
    profile.push_str("  (subpath \"/usr/lib\")\n");
    profile.push_str("  (subpath \"/System/Library/Frameworks\")\n");
    profile.push_str("  (subpath \"/System/Library/PrivateFrameworks\")\n");
    profile.push_str("  (subpath \"/System/Library/Extensions\")\n");
    profile.push_str("  (subpath \"/System/Library/ColorSync\")\n");
    // Host toolchain dirs: a scratch/CI image bakes no `/bin/bash`, `clang`,
    // `env`, etc., so the Seatbelt runtime MUST be able to read+map the HOST
    // binaries the workload resolves off `DEFAULT_SEATBELT_PATH`. Without these,
    // `(allow process-exec)` lets the exec syscall through but PATH lookups
    // (`command -v clang`) `stat()` files in dirs that are deny-default, so every
    // host tool reads as "not found" and shell steps die with exit 1 / 127. The
    // Xcode CLT tree is included because `/usr/bin/clang` is a shim that re-execs
    // `/Library/Developer/CommandLineTools/usr/bin/clang`.
    profile.push_str("  (subpath \"/bin\")\n");
    profile.push_str("  (subpath \"/sbin\")\n");
    profile.push_str("  (subpath \"/usr/bin\")\n");
    profile.push_str("  (subpath \"/usr/sbin\")\n");
    profile.push_str("  (subpath \"/usr/libexec\")\n");
    profile.push_str("  (subpath \"/usr/local/bin\")\n");
    profile.push_str("  (subpath \"/usr/local/sbin\")\n");
    profile.push_str("  (subpath \"/opt/homebrew\")\n");
    profile.push_str("  (subpath \"/Library/Developer/CommandLineTools\")\n");
    // System config dir (real path of the `/etc` symlink — Seatbelt resolves
    // `/etc` -> `/private/etc`). Tools resolve config here: git reads
    // `/etc/gitconfig` and an EPERM on it is FATAL (exit 128 — broke `actions/
    // checkout` and `go build` VCS stamping on every non-container Mac job),
    // curl/openssl read `/etc/ssl`, ssh reads `/etc/ssh`. Read-only; writes stay
    // denied by the deny-default.
    profile.push_str("  (subpath \"/private/etc\")\n");
    profile.push_str("  (literal \"/\")\n");
    profile.push_str("  (literal \"/dev/random\")\n");
    profile.push_str("  (literal \"/dev/urandom\"))\n");
    profile.push('\n');
    profile.push_str("; --- Executable mapping (required for dyld) ---\n");
    profile.push_str("(allow file-map-executable\n");
    profile.push_str("  (subpath \"/usr/lib\")\n");
    profile.push_str("  (subpath \"/System/Library/Frameworks\")\n");
    profile.push_str("  (subpath \"/System/Library/PrivateFrameworks\")\n");
    profile.push_str("  (subpath \"/System/Library/Extensions\")\n");
    // Same host toolchain dirs as above: dyld must map the host executables +
    // their dylibs so `bash`, `clang`, `env`, `git`, etc. can actually exec.
    profile.push_str("  (subpath \"/bin\")\n");
    profile.push_str("  (subpath \"/sbin\")\n");
    profile.push_str("  (subpath \"/usr/bin\")\n");
    profile.push_str("  (subpath \"/usr/sbin\")\n");
    profile.push_str("  (subpath \"/usr/libexec\")\n");
    profile.push_str("  (subpath \"/usr/local/bin\")\n");
    profile.push_str("  (subpath \"/usr/local/sbin\")\n");
    profile.push_str("  (subpath \"/opt/homebrew\")\n");
    profile.push_str("  (subpath \"/Library/Developer/CommandLineTools\"))\n");
    profile.push('\n');
    profile.push_str("; --- System info (hw detection, etc.) ---\n");
    profile.push_str("(allow sysctl-read)\n");
    profile.push_str("(allow system-info)\n");
    profile.push('\n');
    profile.push_str("; --- Mach basics ---\n");
    profile.push_str("(allow mach-lookup\n");
    profile.push_str("  (global-name \"com.apple.system.opendirectoryd.libinfo\"))\n");
    profile.push('\n');

    // ===== Section 3: Container rootfs access =====
    profile.push_str("; --- Container rootfs ---\n");
    let _ = writeln!(
        profile,
        "(allow file-read* file-write* (subpath \"{}\"))",
        config.rootfs_dir.display()
    );
    let _ = writeln!(
        profile,
        "(allow file-map-executable (subpath \"{}\"))",
        config.rootfs_dir.display()
    );
    profile.push('\n');

    // Workspace directory (logs, config, etc.)
    profile.push_str("; --- Workspace directory ---\n");
    let _ = writeln!(
        profile,
        "(allow file-read* file-write* (subpath \"{}\"))",
        config.workspace_dir.display()
    );
    profile.push('\n');

    // ===== Section 4: Volume mounts =====
    if !config.writable_dirs.is_empty() {
        profile.push_str("; --- Volume mounts (writable) ---\n");
        for dir in &config.writable_dirs {
            let _ = writeln!(
                profile,
                "(allow file-read* file-write* (subpath \"{}\"))",
                dir.display()
            );
        }
        profile.push('\n');
    }

    if !config.readonly_dirs.is_empty() {
        profile.push_str("; --- Volume mounts (read-only) ---\n");
        for dir in &config.readonly_dirs {
            let _ = writeln!(
                profile,
                "(allow file-read* (subpath \"{}\"))",
                dir.display()
            );
        }
        profile.push('\n');
    }

    // ===== Provisioned toolchain cache (macOS git keg) =====
    //
    // `zlayer-toolchain` installs the `git` Homebrew bottle into a keg under the
    // host toolchain-cache dir (e.g. `{data}/toolchains/git-latest-arm64/...`),
    // OUTSIDE the per-container rootfs clone. The spawn/exec env PREPENDs that
    // keg's `bin` onto PATH so the sandboxed `git` is the keg git (which reads
    // its own `GIT_CONFIG_SYSTEM`) instead of host `/usr/bin/git` (which EPERMs
    // on `/etc/gitconfig`). For that to work the keg must be readable AND
    // exec-mappable by dyld, mirroring the `/opt/homebrew` grant style. The
    // ancestor metadata for traversal is emitted alongside the writable roots
    // below.
    if let Some(tc) = &config.toolchain_cache {
        let _ = writeln!(profile, "; --- Provisioned toolchain cache (git keg) ---");
        let _ = writeln!(
            profile,
            "(allow file-read* file-map-executable (subpath \"{}\"))",
            tc.display()
        );
        profile.push('\n');
    }

    // ===== Ancestor metadata for every writable subpath =====
    //
    // Each `(allow file-read* file-write* (subpath "<dir>"))` above grants the
    // dir and its descendants but NOT its ancestors. A process creating a
    // writable subpath (e.g. the CI executor's `os.MkdirAll(workspace)`) `lstat`s
    // every parent from `/`; under Seatbelt those `lstat`s are denied unless we
    // grant metadata-read on each ancestor. Collect every writable root —
    // rootfs, workspace, each volume/writable dir, and (when keychain is opted
    // in) each keychain database's parent dir — and grant stat/traverse on their
    // ancestors. See [`write_ancestor_metadata_rules`].
    let mut writable_roots: Vec<PathBuf> =
        vec![config.rootfs_dir.clone(), config.workspace_dir.clone()];
    writable_roots.extend(config.writable_dirs.iter().cloned());
    if let KeychainAccess::Enabled { keychain_paths } = &config.keychain_access {
        for path in keychain_paths {
            if let Some(parent) = path.parent() {
                writable_roots.push(parent.to_path_buf());
            }
        }
    }
    // The toolchain-cache keg lives outside the rootfs; its own subtree is
    // covered by the read+map subpath grant above, but the kernel still `lstat`s
    // each ancestor when resolving the keg path, so add it as a metadata root.
    if let Some(tc) = &config.toolchain_cache {
        writable_roots.push(tc.clone());
    }
    write_ancestor_metadata_rules(&mut profile, &writable_roots);

    // ===== Section 5: GPU access =====
    match config.gpu_access {
        GpuAccess::MetalCompute => {
            profile.push_str(METAL_COMPUTE_PROFILE_SECTION);
        }
        GpuAccess::MpsOnly => {
            profile.push_str(MPS_ONLY_PROFILE_SECTION);
        }
        GpuAccess::None => {}
    }

    // ===== Apple toolchain (xcrun / clang) support =====
    //
    // A native `clang`/`xcrun` resolves the host toolchain through three host
    // facilities the deny-default profile otherwise blocks:
    //   - `com.apple.bsd.dirhelper`: `confstr(_CS_DARWIN_USER_{TEMP,CACHE}_DIR)`
    //     resolves the per-user temp/cache folder via this Mach service; denied,
    //     clang can't even find its cache dir ("confstr() failed ... using /tmp").
    //   - the per-user temp + cache dirs (read+write): clang/xcrun write their
    //     `xcrun_db` SDK cache + module cache there (located via `confstr`, NOT
    //     `TMPDIR`). We grant the canonical `/private/var/folders/<hash>/{T,C}`
    //     paths PLUS read on the `/var` symlink so the kernel can traverse
    //     `/var` -> `/private/var` when clang opens the `getconf` (`/var/...`) form.
    //   - `SystemVersion.plist` (read): clang reads it to determine the host OS
    //     version; denied, it errors "unable to read SDK settings for '/'".
    //   - metadata on `/Library` + `/Library/Developer`: clang `realpath`s its own
    //     binary to derive `InstalledDir` (whence it re-execs `clang -cc1` and the
    //     assembler); the CLT subpath grant covers the leaf but not these ancestor
    //     dirs, so without them `InstalledDir` comes back empty and any actual
    //     compile dies with "posix_spawn failed: No such file or directory".
    // All resolved on the host (unsandboxed) at profile-gen time; if the host has
    // no CLT/Xcode the getconf lookups return None and we simply omit the grant.
    profile.push_str("; --- Apple toolchain (xcrun/clang) support ---\n");
    profile.push_str("(allow mach-lookup (global-name \"com.apple.bsd.dirhelper\"))\n");
    profile.push_str("(allow file-read* (literal \"/var\"))\n");
    profile.push_str(
        "(allow file-read* (literal \"/System/Library/CoreServices/SystemVersion.plist\"))\n",
    );
    profile.push_str(
        "(allow file-read-metadata (literal \"/Library\") (literal \"/Library/Developer\"))\n",
    );
    // Xcode license-agreement check: a full-Xcode `clang`/`xcrun` reads the
    // "agreed GM license version" record before doing any work and aborts with
    // "You have not agreed to the Xcode license agreements" (exit 69) if it
    // can't read it — even when the host HAS accepted it (`sudo xcodebuild
    // -license accept`). The accepted-version record lives in the system
    // `com.apple.dt.Xcode.plist`; the per-user copy carries the rest of Xcode's
    // prefs. Both are deny-default under the sandbox, so grant read on each.
    profile.push_str(
        "(allow file-read* (literal \"/Library/Preferences/com.apple.dt.Xcode.plist\"))\n",
    );
    if let Ok(home) = std::env::var("HOME") {
        let _ = writeln!(
            profile,
            "(allow file-read* (literal \"{home}/Library/Preferences/com.apple.dt.Xcode.plist\"))"
        );
    }
    // Full-Xcode hosts select a developer dir OUTSIDE the CLT tree — e.g.
    // `xcode-select -p` => `/Applications/Xcode.app/Contents/Developer`. The
    // hardcoded `/Library/Developer/CommandLineTools` grants in Section 2 cover
    // CLT-only setups, but `/usr/bin/clang` is a shim that re-execs the SELECTED
    // toolchain, and every header/SDK/`clang -cc1` read then lands under the
    // chosen developer dir. Under deny-default those reads are blocked and clang
    // dies with "No developer tools were found" / SDK errors. Grant read +
    // executable-mapping on the active developer dir whenever it lives outside
    // the CLT tree (the CLT path is already granted; re-granting it would be
    // redundant), plus `file-read-metadata` on its ancestors so clang can
    // `realpath` its own binary to derive `InstalledDir`. Resolved on the host;
    // `None` (no toolchain) adds nothing.
    if let Some(dev_dir) = host_developer_dir() {
        if !Path::new(dev_dir).starts_with("/Library/Developer/CommandLineTools") {
            // Grant the enclosing `.app` bundle when the dev dir lives in one
            // (full Xcode), else the dev dir itself: `clang` reads
            // `Xcode.app/Contents/Info.plist` + dlopens
            // `Contents/SharedFrameworks/DVTSystemPrerequisites.framework`, both
            // ABOVE `Contents/Developer`, so a `…/Developer`-only grant leaves
            // them deny-default ("couldn't stat Xcode's Info.plist", framework
            // "blocked by sandbox").
            let grant_root = enclosing_app_bundle(dev_dir).unwrap_or(dev_dir);
            let _ = writeln!(
                profile,
                "(allow file-read* file-map-executable (subpath \"{grant_root}\"))"
            );
            write_ancestor_metadata_rules(&mut profile, &[PathBuf::from(grant_root)]);
        }
    }
    let darwin = host_darwin_user_dirs();
    if let Some(temp) = &darwin.temp {
        let _ = writeln!(
            profile,
            "(allow file-read* file-write* (subpath \"{}\"))",
            temp.display()
        );
    }
    if let Some(cache) = &darwin.cache {
        let _ = writeln!(
            profile,
            "(allow file-read* file-write* (subpath \"{}\"))",
            cache.display()
        );
    }
    profile.push('\n');

    // ===== Section 5b: TLS server-certificate trust evaluation =====
    //
    // TLS clients that verify server certificates through the macOS Security
    // framework -- Go's `crypto/x509` on darwin (`SecTrustEvaluate`), URLSession,
    // etc. -- do NOT honor `SSL_CERT_FILE`/a PEM bundle the way git's
    // curl/OpenSSL backend does. They call out to `trustd`/`ocspd` and read the
    // system ROOT certificate store. Under the deny-default profile those Mach
    // services and the system keychains are blocked, so every such HTTPS request
    // fails with `x509: OSStatus <n>` -- e.g. `go mod` fetching proxy.golang.org
    // dies `tls: failed to verify certificate: x509: OSStatus -26276`.
    //
    // Grant ONLY the trust-daemon Mach services, UNCONDITIONALLY. `SecTrustEvaluate`
    // XPCs to `trustd`, which reads the system root store in ITS OWN process --
    // our sandboxed process needs no keychain-file reads and no `securityd`
    // (`SecurityServer`). Verified empirically with `sandbox-exec`: a deny-default
    // profile granting only these three services (and with the system keychains
    // explicitly READ-DENIED) evaluates a real proxy.golang.org chain to TLS_OK,
    // while dropping them reproduces `x509: OSStatus -26276`. This stays orthogonal
    // to keychain SECRET access (Section 6, still opt-in), so it deliberately does
    // NOT name `SecurityServer` or any keychain path.
    profile.push_str("; --- TLS trust evaluation (server certs) ---\n");
    profile.push_str("(allow mach-lookup\n");
    profile.push_str("  (global-name \"com.apple.trustd\")\n");
    profile.push_str("  (global-name \"com.apple.trustd.agent\")\n");
    profile.push_str("  (global-name \"com.apple.ocspd\"))\n");
    profile.push('\n');

    // ===== Section 6: Keychain / code-signing access =====
    //
    // Opt-in only. The base profile never names `com.apple.SecurityServer`, so
    // by default a container cannot reach `securityd` and every keychain call
    // (codesign, `security`, SecItem*) is denied. When enabled we grant:
    //   - the Mach services `securityd`/ocspd/trustd use for file keychains and
    //     certificate trust evaluation,
    //   - read + executable-mapping of `/usr/bin` and `/usr/libexec` so the host
    //     `/usr/bin/codesign` and `/usr/bin/security` (and their xpc helpers) can
    //     be exec'd -- the sandbox is layered over the host fs (no pivot_root),
    //   - read of the system keychain stores, and
    //   - read/write of each requested keychain database file and its directory
    //     (securityd writes `-shm`/`-wal`/lock siblings next to `*.keychain-db`).
    //
    // NOTE: this Mach-service set is the grounded starting point; the empirical
    // `log stream` denial sweep on macOS 26 (see the plan's Verification step) is
    // what finalizes it. Adjust here if `sandboxd` reports a missing service.
    if let KeychainAccess::Enabled { keychain_paths } = &config.keychain_access {
        profile.push_str("; --- Keychain / code signing ---\n");
        profile.push_str("(allow mach-lookup\n");
        profile.push_str("  (global-name \"com.apple.SecurityServer\")\n");
        profile.push_str("  (global-name \"com.apple.ocspd\")\n");
        profile.push_str("  (global-name \"com.apple.trustd\")\n");
        profile.push_str("  (global-name \"com.apple.trustd.agent\")\n");
        profile.push_str("  (global-name \"com.apple.SystemConfiguration.configd\"))\n");
        profile.push_str("(allow file-read* file-map-executable\n");
        profile.push_str("  (subpath \"/usr/bin\")\n");
        profile.push_str("  (subpath \"/usr/libexec\"))\n");
        profile.push_str("(allow file-read*\n");
        profile.push_str("  (subpath \"/Library/Keychains\")\n");
        profile.push_str("  (subpath \"/System/Library/Keychains\"))\n");
        profile
            .push_str("(allow user-preference-read (preference-domain \"com.apple.security\"))\n");
        for path in keychain_paths {
            if let Some(parent) = path.parent() {
                let _ = writeln!(
                    profile,
                    "(allow file-read* file-write* (subpath \"{}\"))",
                    parent.display()
                );
            } else {
                let _ = writeln!(
                    profile,
                    "(allow file-read* file-write* (literal \"{}\"))",
                    path.display()
                );
            }
        }
        profile.push('\n');
    }

    // ===== Section 7: Network access =====
    match &config.network_access {
        NetworkAccess::None => {
            profile.push_str("; --- Network: DENIED ---\n\n");
        }
        NetworkAccess::LocalhostOnly {
            bind_ports,
            connect_ports,
            overlay_cidr,
            loopback_api_port,
        } => {
            profile.push_str("; --- Network: localhost only ---\n");
            for port in bind_ports {
                let _ = writeln!(
                    profile,
                    "(allow network-bind (local ip \"localhost:{port}\"))",
                );
            }
            for port in connect_ports {
                let _ = writeln!(
                    profile,
                    "(allow network-outbound (remote ip \"localhost:{port}\"))",
                );
            }
            // The daemon API URL injected into the sandbox uses the literal
            // `127.0.0.1` form, which an sbpl `(remote ip "localhost:port")`
            // filter does not reliably match. When daemon creds are present,
            // additionally whitelist the `127.0.0.1` form for that one port so a
            // nested `zlayer build` can reach the host daemon on loopback.
            if let Some(port) = loopback_api_port {
                let _ = writeln!(
                    profile,
                    "(allow network-outbound (remote ip \"127.0.0.1:{port}\"))",
                );
            }
            // Allow inbound on bound ports
            if !bind_ports.is_empty() {
                profile.push_str("(allow network-inbound (local ip \"localhost:*\"))\n");
            }
            if let Some(cidr) = overlay_cidr {
                // Seatbelt shares the host net namespace, so the container egresses
                // directly to the overlay (node daemon IP + sibling overlay IPs).
                //
                // NOTE: the SBPL CIDR remote-ip syntax `"<cidr>:*"` should be
                // smoke-tested on a real macOS host. If `sandbox_compile` rejects
                // it, the fallback form is `(remote ip4 (subnet "<cidr>"))`.
                let _ = writeln!(profile, "(allow network-outbound (remote ip \"{cidr}:*\"))");
                let _ = writeln!(profile, "(allow network-inbound (remote ip \"{cidr}:*\"))");
            }
            profile.push('\n');
        }
        NetworkAccess::Full => {
            profile.push_str("; --- Network: full access ---\n");
            profile.push_str("(allow network-outbound)\n");
            profile.push_str("(allow network-inbound)\n");
            profile.push_str("(allow network-bind)\n");
            profile.push_str("(allow system-socket)\n");
            profile.push('\n');
        }
    }

    // ===== Section 8: Logging, /dev/null, pseudo-tty =====
    profile.push_str("; --- I/O essentials ---\n");
    profile.push_str("(allow file-write-data\n");
    profile.push_str("  (require-all (literal \"/dev/null\") (vnode-type CHARACTER-DEVICE)))\n");
    profile.push_str("(allow file-read-data\n");
    profile.push_str("  (require-all (literal \"/dev/null\") (vnode-type CHARACTER-DEVICE)))\n");
    profile.push_str("(allow pseudo-tty)\n");
    profile.push_str("(allow file-read* file-write* file-ioctl (literal \"/dev/ptmx\"))\n");
    profile.push('\n');

    // IPC basics
    profile.push_str("; --- IPC ---\n");
    profile.push_str("(allow ipc-posix-sem)\n");
    profile.push_str("(allow ipc-posix-shm)\n");
    profile.push('\n');

    profile
}

// ---------------------------------------------------------------------------
// Seatbelt FFI
// ---------------------------------------------------------------------------

/// FFI declarations for macOS Seatbelt sandbox.
mod seatbelt_ffi {
    use std::os::raw::c_char;

    #[link(name = "System", kind = "dylib")]
    extern "C" {
        /// Apply a sandbox profile to the current process.
        ///
        /// - `profile`: SBPL string (Scheme-based sandbox profile)
        /// - `flags`: 0 for raw SBPL string, 0x0001 for named profile
        /// - `errorbuf`: receives error message on failure (free with `sandbox_free_error`)
        ///
        /// Returns 0 on success, -1 on failure.
        /// WARNING: Once applied, the sandbox CANNOT be removed or loosened.
        pub fn sandbox_init(profile: *const c_char, flags: u64, errorbuf: *mut *mut c_char) -> i32;

        /// Free an error buffer allocated by `sandbox_init`.
        pub fn sandbox_free_error(errorbuf: *mut c_char);
    }
}

/// Apply a Seatbelt profile to the current process.
///
/// This is called in the child process after `fork()` and before `exec()`.
/// Once applied, the sandbox cannot be removed or loosened.
///
/// # Errors
///
/// Returns an error if the profile string contains a null byte or if
/// `sandbox_init()` fails to apply the profile.
#[allow(unsafe_code)]
fn apply_seatbelt_profile(sbpl: &str) -> std::io::Result<()> {
    use std::ffi::CString;
    use std::ptr;

    let profile_cstr =
        CString::new(sbpl).map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))?;

    let mut error_buf: *mut std::os::raw::c_char = ptr::null_mut();

    let result = unsafe {
        seatbelt_ffi::sandbox_init(
            profile_cstr.as_ptr(),
            0, // 0 = raw SBPL string
            &raw mut error_buf,
        )
    };

    if result != 0 {
        let error_msg = if error_buf.is_null() {
            format!("sandbox_init returned error code {result}")
        } else {
            let msg = unsafe {
                std::ffi::CStr::from_ptr(error_buf)
                    .to_string_lossy()
                    .into_owned()
            };
            unsafe { seatbelt_ffi::sandbox_free_error(error_buf) };
            msg
        };
        return Err(std::io::Error::new(
            std::io::ErrorKind::PermissionDenied,
            format!("Failed to initialize sandbox: {error_msg}"),
        ));
    }

    Ok(())
}

// ---------------------------------------------------------------------------
// APFS clonefile FFI
// ---------------------------------------------------------------------------

extern "C" {
    /// Clone a file using APFS copy-on-write semantics.
    ///
    /// From `<sys/clonefile.h>`. Creates a new file at `dst` that shares
    /// storage blocks with `src`. Nearly instantaneous. Both files must
    /// be on the same APFS volume.
    fn clonefile(
        src: *const libc::c_char,
        dst: *const libc::c_char,
        flags: libc::c_int,
    ) -> libc::c_int;
}

/// Clone a single file using APFS `CoW`.
///
/// Returns `Ok(true)` if clonefile succeeded, `Ok(false)` if clonefile
/// is not supported (non-APFS volume) and the caller should fall back
/// to a regular copy.
///
/// # Errors
///
/// Returns an error if the source or destination paths contain invalid
/// characters or if `clonefile()` fails for a reason other than unsupported
/// filesystem or cross-device.
#[allow(unsafe_code)]
fn clone_file_apfs(src: &Path, dst: &Path) -> std::io::Result<bool> {
    use std::ffi::CString;

    let c_src = CString::new(src.to_str().ok_or_else(|| {
        std::io::Error::new(std::io::ErrorKind::InvalidInput, "Invalid src path")
    })?)
    .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))?;

    let c_dst = CString::new(dst.to_str().ok_or_else(|| {
        std::io::Error::new(std::io::ErrorKind::InvalidInput, "Invalid dst path")
    })?)
    .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))?;

    let ret = unsafe { clonefile(c_src.as_ptr(), c_dst.as_ptr(), 0) };

    if ret == 0 {
        Ok(true) // Clone succeeded
    } else {
        let err = std::io::Error::last_os_error();
        // ENOTSUP means the filesystem doesn't support clonefile (not APFS)
        // EXDEV means src and dst are on different volumes
        // EEXIST means the destination already exists (concurrent writer)
        if err.raw_os_error() == Some(libc::ENOTSUP)
            || err.raw_os_error() == Some(libc::EXDEV)
            || err.raw_os_error() == Some(libc::EEXIST)
        {
            Ok(false) // Fallback needed
        } else {
            Err(err)
        }
    }
}

/// Recursively clone a directory tree using APFS `clonefile` for files.
///
/// `clonefile` operates at the file level, not directory level, so we must
/// walk the directory tree, create directories in the destination, and
/// clonefile each regular file.
///
/// If APFS clonefile is not available (non-APFS volume), falls back to
/// regular file copy.
async fn clone_directory_recursive(src: &Path, dst: &Path) -> std::io::Result<()> {
    // Create destination directory
    tokio::fs::create_dir_all(dst).await?;

    let mut entries = tokio::fs::read_dir(src).await?;
    while let Some(entry) = entries.next_entry().await? {
        let entry_path = entry.path();
        let file_name = entry.file_name();
        let dest_path = dst.join(&file_name);

        let file_type = entry.file_type().await?;

        if file_type.is_dir() {
            // Recurse into subdirectories
            Box::pin(clone_directory_recursive(&entry_path, &dest_path)).await?;
        } else if file_type.is_file() {
            // Try APFS clone first, fall back to copy
            let src_clone = entry_path.clone();
            let dst_clone = dest_path.clone();

            let cloned =
                tokio::task::spawn_blocking(move || clone_file_apfs(&src_clone, &dst_clone))
                    .await
                    .map_err(std::io::Error::other)??;

            if !cloned {
                // Fallback: regular copy
                tokio::fs::copy(&entry_path, &dest_path).await?;
            }
        } else if file_type.is_symlink() {
            // Recreate symlinks
            let link_target = tokio::fs::read_link(&entry_path).await?;
            tokio::fs::symlink(&link_target, &dest_path).await?;
        }
    }

    // Preserve directory permissions
    let src_meta = tokio::fs::metadata(src).await?;
    tokio::fs::set_permissions(dst, src_meta.permissions()).await?;

    Ok(())
}

// ---------------------------------------------------------------------------
// Process stats FFI
// ---------------------------------------------------------------------------

/// Get the resident set size (RSS) of a process using `proc_pidinfo`.
///
/// On macOS, we use the `PROC_PIDTASKINFO` flavor which works for child
/// processes without special entitlements.
///
/// # Errors
///
/// Returns an error if `proc_pidinfo` fails (e.g. the process does not exist).
#[allow(unsafe_code)]
#[allow(clippy::cast_possible_truncation, clippy::cast_possible_wrap)]
fn get_process_rss(pid: u32) -> std::io::Result<u64> {
    #[repr(C)]
    #[allow(non_snake_case)]
    #[allow(clippy::struct_field_names)]
    struct ProcTaskInfo {
        pti_virtual_size: u64,
        pti_resident_size: u64,
        pti_total_user: u64,
        pti_total_system: u64,
        pti_threads_user: u64,
        pti_threads_system: u64,
        pti_policy: i32,
        pti_faults: i32,
        pti_pageins: i32,
        pti_cow_faults: i32,
        pti_messages_sent: i32,
        pti_messages_received: i32,
        pti_syscalls_mach: i32,
        pti_syscalls_unix: i32,
        pti_csw: i32,
        pti_threadnum: i32,
        pti_numrunning: i32,
        pti_priority: i32,
    }

    extern "C" {
        fn proc_pidinfo(
            pid: libc::c_int,
            flavor: libc::c_int,
            arg: u64,
            buffer: *mut libc::c_void,
            buffersize: libc::c_int,
        ) -> libc::c_int;
    }

    const PROC_PIDTASKINFO: libc::c_int = 4;

    let mut info: ProcTaskInfo = unsafe { std::mem::zeroed() };
    let size = std::mem::size_of::<ProcTaskInfo>() as libc::c_int;

    let ret = unsafe {
        proc_pidinfo(
            pid as libc::c_int,
            PROC_PIDTASKINFO,
            0,
            (&raw mut info).cast::<libc::c_void>(),
            size,
        )
    };

    if ret <= 0 {
        return Err(std::io::Error::last_os_error());
    }

    Ok(info.pti_resident_size)
}

/// Get CPU time (user + system) and memory RSS for a process.
///
/// CPU time is returned in microseconds. Memory is returned in bytes.
///
/// # Errors
///
/// Returns an error if `proc_pidinfo` fails (e.g. the process does not exist).
#[allow(unsafe_code)]
#[allow(clippy::cast_possible_truncation, clippy::cast_possible_wrap)]
fn get_process_stats(pid: u32) -> Result<(u64, u64)> {
    #[repr(C)]
    #[allow(non_snake_case)]
    #[allow(clippy::struct_field_names)]
    struct ProcTaskInfo {
        pti_virtual_size: u64,
        pti_resident_size: u64,
        pti_total_user: u64,
        pti_total_system: u64,
        pti_threads_user: u64,
        pti_threads_system: u64,
        pti_policy: i32,
        pti_faults: i32,
        pti_pageins: i32,
        pti_cow_faults: i32,
        pti_messages_sent: i32,
        pti_messages_received: i32,
        pti_syscalls_mach: i32,
        pti_syscalls_unix: i32,
        pti_csw: i32,
        pti_threadnum: i32,
        pti_numrunning: i32,
        pti_priority: i32,
    }

    extern "C" {
        fn proc_pidinfo(
            pid: libc::c_int,
            flavor: libc::c_int,
            arg: u64,
            buffer: *mut libc::c_void,
            buffersize: libc::c_int,
        ) -> libc::c_int;
    }

    const PROC_PIDTASKINFO: libc::c_int = 4;

    let mut info: ProcTaskInfo = unsafe { std::mem::zeroed() };
    let size = std::mem::size_of::<ProcTaskInfo>() as libc::c_int;

    let ret = unsafe {
        proc_pidinfo(
            pid as libc::c_int,
            PROC_PIDTASKINFO,
            0,
            (&raw mut info).cast::<libc::c_void>(),
            size,
        )
    };

    if ret <= 0 {
        return Err(AgentError::Internal(format!(
            "proc_pidinfo failed for pid {pid}: {}",
            std::io::Error::last_os_error()
        )));
    }

    // pti_total_user and pti_total_system are in Mach absolute time units (nanoseconds on Apple Silicon).
    // Convert to microseconds.
    let cpu_usec = (info.pti_total_user + info.pti_total_system) / 1000;
    let rss = info.pti_resident_size;

    Ok((cpu_usec, rss))
}

// ---------------------------------------------------------------------------
// Resource limits
// ---------------------------------------------------------------------------

/// Set resource limits for the sandboxed process.
///
/// Called in the child process after `fork()` via `pre_exec`.
///
/// # Errors
///
/// Returns an error if `setrlimit` fails.
#[allow(unsafe_code)]
fn set_resource_limits(max_files: u64, cpu_seconds: Option<u64>) -> std::io::Result<()> {
    // Limit open file descriptors
    let file_limit = libc::rlimit {
        rlim_cur: max_files,
        rlim_max: max_files,
    };
    if unsafe { libc::setrlimit(libc::RLIMIT_NOFILE, &raw const file_limit) } != 0 {
        return Err(std::io::Error::last_os_error());
    }

    // Limit CPU time if specified
    if let Some(seconds) = cpu_seconds {
        let cpu_limit = libc::rlimit {
            rlim_cur: seconds,
            rlim_max: seconds,
        };
        if unsafe { libc::setrlimit(libc::RLIMIT_CPU, &raw const cpu_limit) } != 0 {
            return Err(std::io::Error::last_os_error());
        }
    }

    Ok(())
}

// ---------------------------------------------------------------------------
// Memory watchdog
// ---------------------------------------------------------------------------

/// Memory watchdog that monitors a process's RSS and kills it if exceeded.
///
/// This is necessary because macOS does NOT enforce `RLIMIT_RSS` or `RLIMIT_AS`.
/// The watchdog polls every 2 seconds using `proc_pidinfo` (Mach API).
#[allow(unsafe_code)]
#[allow(clippy::cast_possible_wrap)]
async fn memory_watchdog(pid: u32, limit_bytes: u64) {
    let check_interval = Duration::from_secs(2);

    loop {
        tokio::time::sleep(check_interval).await;

        // Check if process is still alive
        let alive = unsafe { libc::kill(pid as i32, 0) } == 0;
        if !alive {
            tracing::debug!(pid = pid, "Memory watchdog: process exited");
            return;
        }

        // Get current RSS
        match get_process_rss(pid) {
            Ok(rss_bytes) => {
                if rss_bytes > limit_bytes {
                    tracing::warn!(
                        pid = pid,
                        rss_mb = rss_bytes / (1024 * 1024),
                        limit_mb = limit_bytes / (1024 * 1024),
                        "Memory limit exceeded, sending SIGKILL"
                    );
                    unsafe {
                        libc::kill(pid as i32, libc::SIGKILL);
                    }
                    return;
                }
            }
            Err(e) => {
                tracing::debug!(pid = pid, error = %e, "Failed to read process RSS");
            }
        }
    }
}

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

/// Build [`NetworkAccess`] from a [`ServiceSpec`]'s endpoints and network
/// configuration, honoring the workload's resolved [`NetworkIsolation`] policy.
///
/// Seatbelt rules apply to a single host process on the shared host network
/// stack — there is no per-peer (5-tuple by remote identity) firewalling the
/// way the Linux iptables ISO chain or the per-service `WireGuard` path provide.
/// So for `fenced` workloads (`isolated`/`dedicated`) we cannot grant
/// selective overlay-peer reachability; we enforce fencing as "deny broad
/// overlay reach + egress-only" instead: no overlay CIDR is whitelisted and
/// only DNS/HTTP/HTTPS + the workload's own ports are reachable. True per-peer
/// fencing happens on the Linux/VM runtime paths.
fn build_network_access(
    spec: &ServiceSpec,
    overlay_cidr: Option<String>,
    iso: &zlayer_types::overlay::NetworkIsolation,
    daemon_api_port: Option<u16>,
) -> NetworkAccess {
    // Collect ports from endpoints
    let mut bind_ports = Vec::new();
    for endpoint in &spec.endpoints {
        bind_ports.push(endpoint.target_port());
    }

    // No outbound at all (e.g. `--network none`): deny networking entirely.
    if iso.egress == zlayer_types::overlay::EgressPolicy::None {
        return NetworkAccess::None;
    }

    if iso.fenced {
        // Fenced (isolated/dedicated): no broad overlay reach. Egress is limited
        // to DNS/HTTP/HTTPS plus the workload's own ports (health checks). When
        // daemon creds are injected, also open the host daemon API port so a
        // nested `zlayer build` can delegate on loopback.
        let mut connect_ports = vec![53, 80, 443];
        connect_ports.extend_from_slice(&bind_ports);
        if let Some(port) = daemon_api_port {
            if !connect_ports.contains(&port) {
                connect_ports.push(port);
            }
        }
        return NetworkAccess::LocalhostOnly {
            bind_ports,
            connect_ports,
            overlay_cidr: None,
            loopback_api_port: daemon_api_port,
        };
    }

    // Normal (auto/shared/host): preserve prior behavior exactly.
    //
    // If no endpoints and no special network config, default to full access
    // (most services need outbound connectivity for dependencies).
    if bind_ports.is_empty() {
        return NetworkAccess::Full;
    }

    // Add common connect ports (DNS, HTTP, HTTPS) for outbound, plus all bind
    // ports as connect ports too (for health checks). When daemon creds are
    // injected, also open the host daemon API port for nested delegation.
    let mut connect_ports = vec![53, 80, 443];
    connect_ports.extend_from_slice(&bind_ports);
    if let Some(port) = daemon_api_port {
        if !connect_ports.contains(&port) {
            connect_ports.push(port);
        }
    }

    NetworkAccess::LocalhostOnly {
        bind_ports,
        connect_ports,
        overlay_cidr,
        loopback_api_port: daemon_api_port,
    }
}

/// Build [`KeychainAccess`] from a [`ServiceSpec`]'s `zlayer.io/keychain` label,
/// falling back to the operator's `--keychain-default` selection.
///
/// The caller has already confirmed the daemon permits keychain access; this
/// only translates the selected value into the set of host keychain database
/// paths to open. A workload's explicit `zlayer.io/keychain` label wins; when it
/// is absent the `operator_default` (from `--keychain-default` /
/// `ZLAYER_KEYCHAIN_DEFAULT`) is used instead, so the operator — not the
/// workload — decides which keychain a job may reach. Value grammar
/// (case-insensitive keywords), resolved against `$HOME/Library/Keychains`:
///
/// - `login` -> `login.keychain-db`
/// - `build` / `true` / `default` / `1` / `yes` / `on` -> `zlayer-build.keychain-db`
/// - `build:<name>` -> `<name>.keychain-db`
/// - one or more absolute paths, comma-separated (used verbatim)
/// - anything else / absent -> [`KeychainAccess::None`]
///
/// Multiple comma-separated entries are unioned (e.g. `login,build` grants both
/// the login and the managed build keychain).
fn build_keychain_access(spec: &ServiceSpec, operator_default: Option<&str>) -> KeychainAccess {
    // Explicit workload label wins; otherwise fall back to the operator default.
    // An explicit-but-bogus label is NOT silently widened to the default.
    let Some(raw) = spec
        .labels
        .get("zlayer.io/keychain")
        .map(String::as_str)
        .or(operator_default)
    else {
        return KeychainAccess::None;
    };
    // The daemon runs as the keychain-owning user (per-user LaunchAgent), so
    // `$HOME` resolves to that user's home where the login/build keychains live.
    let home = std::env::var("HOME").unwrap_or_default();
    let keychain_dir = format!("{home}/Library/Keychains");
    let paths = keychain_paths_from_label(raw, &keychain_dir);
    if paths.is_empty() {
        KeychainAccess::None
    } else {
        KeychainAccess::Enabled {
            keychain_paths: paths,
        }
    }
}

/// The operator-configured default keychain selection from the
/// `ZLAYER_KEYCHAIN_DEFAULT` env var (set by the `--keychain-default` flag).
///
/// This is the bridge the sandbox runtime uses to honor the flag — mirroring
/// [`keychain_access_allowed_from_env`](crate::keychain_access_allowed_from_env)
/// for `--allow-keychain-access` — so the default `--runtime auto` deployment can
/// opt in without the explicit `RuntimeConfig::MacSandbox` arm. Returns `None`
/// when unset or blank, preserving the deny-default.
#[must_use]
fn operator_keychain_default_from_env() -> Option<String> {
    std::env::var("ZLAYER_KEYCHAIN_DEFAULT")
        .ok()
        .map(|v| v.trim().to_string())
        .filter(|v| !v.is_empty())
}

/// Pure translation of a `zlayer.io/keychain` label value into the set of
/// keychain database paths, resolved against `keychain_dir`. Split out from
/// [`build_keychain_access`] so it is testable without mutating the process
/// environment. See [`build_keychain_access`] for the value grammar.
fn keychain_paths_from_label(raw: &str, keychain_dir: &str) -> Vec<PathBuf> {
    let mut paths: Vec<PathBuf> = Vec::new();
    for entry in raw.split(',') {
        let entry = entry.trim();
        if entry.is_empty() {
            continue;
        }
        let lower = entry.to_ascii_lowercase();
        let path = match lower.as_str() {
            "login" => PathBuf::from(format!("{keychain_dir}/login.keychain-db")),
            "build" | "true" | "default" | "1" | "yes" | "on" => {
                PathBuf::from(format!("{keychain_dir}/zlayer-build.keychain-db"))
            }
            _ if lower.starts_with("build:") => {
                let name = &entry["build:".len()..];
                PathBuf::from(format!("{keychain_dir}/{name}.keychain-db"))
            }
            // Absolute path -> use verbatim. A bare relative token that matches
            // no keyword is ignored (treated as a typo, not silently widened).
            _ if entry.starts_with('/') => PathBuf::from(entry),
            _ => {
                tracing::warn!(
                    label = %entry,
                    "Ignoring unrecognized zlayer.io/keychain value (expected login|build|build:<name>|/abs/path)"
                );
                continue;
            }
        };
        if !paths.contains(&path) {
            paths.push(path);
        }
    }
    paths
}

/// The container-absolute mount target of a [`StorageSpec`].
fn storage_target(storage: &zlayer_spec::StorageSpec) -> &str {
    match storage {
        zlayer_spec::StorageSpec::Named { target, .. }
        | zlayer_spec::StorageSpec::Anonymous { target, .. }
        | zlayer_spec::StorageSpec::Bind { target, .. }
        | zlayer_spec::StorageSpec::Tmpfs { target, .. }
        | zlayer_spec::StorageSpec::S3 { target, .. } => target,
    }
}

/// True for the conventional CI workspace target (`/workspace`).
fn is_workspace_target(target: &str) -> bool {
    matches!(target.trim_end_matches('/'), "/workspace" | "workspace")
}

/// Host directories a sandboxed container may write to, plus the host directory
/// exposed to the workload as its workspace.
struct SandboxStorage {
    /// Absolute host paths to grant `file-write*` in the Seatbelt profile (and
    /// create on disk before launch).
    writable_dirs: Vec<PathBuf>,
    /// Host directory the workload should treat as `/workspace`, surfaced via the
    /// injected `GITHUB_WORKSPACE` env.
    workspace_host: PathBuf,
}

/// Resolve, for the Seatbelt sandbox, the host directories the workload may write
/// to and the host directory to expose as its workspace.
///
/// Seatbelt has no mount namespace, so a container-absolute target (`/workspace`)
/// can't be remapped onto a different inode the way a real container mount can.
/// Instead each volume is backed by a real host directory — a persistent volume
/// dir / anonymous scratch dir / bind source resolved from `volume_paths` — and
/// any non-volume target (e.g. tmpfs) is rebased onto the writable rootfs clone,
/// mirroring `resolve_program`. The workload reaches its workspace through the
/// `GITHUB_WORKSPACE` env injected by [`spawn_sandboxed_process`]; a `/workspace`
/// volume maps to its backing dir, otherwise a fresh `{rootfs}/workspace` is used.
fn resolve_sandbox_storage(
    spec: &ServiceSpec,
    rootfs_dir: &Path,
    container_dir: &Path,
    volume_paths: &std::collections::HashMap<String, PathBuf>,
) -> SandboxStorage {
    let mut writable_dirs = vec![
        container_dir.join("tmp"),  // Always allow a tmp directory
        container_dir.join("home"), // Per-container writable HOME (git config, go)
    ];
    let mut workspace_host: Option<PathBuf> = None;

    for storage in &spec.storage {
        let target = storage_target(storage);
        // Host-backed volumes (named/anonymous/bind/s3) write to their backing
        // dir; everything else (tmpfs) is rebased into the writable rootfs clone.
        let host = crate::runtimes::volume_prep::resolve_host_source(storage, volume_paths)
            .map_or_else(
                || rootfs_dir.join(target.trim_start_matches('/')),
                |rm| rm.host,
            );
        if is_workspace_target(target) {
            workspace_host = Some(host.clone());
        }
        writable_dirs.push(host);
    }

    // No explicit `/workspace` volume: give the workload a writable workspace in
    // the rootfs clone so its `mkdir`/checkout still lands somewhere writable.
    let workspace_host = workspace_host.unwrap_or_else(|| rootfs_dir.join("workspace"));
    if !writable_dirs.contains(&workspace_host) {
        writable_dirs.push(workspace_host.clone());
    }

    SandboxStorage {
        writable_dirs,
        workspace_host,
    }
}

/// Parse a memory string like "512Mi" or "2Gi" into bytes.
#[must_use]
fn parse_memory_string(s: &str) -> Option<u64> {
    let s = s.trim();
    if let Some(num) = s.strip_suffix("Gi") {
        num.parse::<u64>().ok().map(|v| v * 1024 * 1024 * 1024)
    } else if let Some(num) = s.strip_suffix("Mi") {
        num.parse::<u64>().ok().map(|v| v * 1024 * 1024)
    } else if let Some(num) = s.strip_suffix("Ki") {
        num.parse::<u64>().ok().map(|v| v * 1024)
    } else {
        s.parse::<u64>().ok()
    }
}

/// Sanitize an image name for use as a filesystem directory name.
#[must_use]
fn sanitize_image_name(image: &str) -> String {
    image.replace(['/', ':', '@'], "_")
}

/// Read the `image-config.json` sidecar written next to a pulled image's rootfs
/// at `{images}/{sanitized}/image-config.json`. Returns the parsed config so the
/// image's default `Env` (and `WorkingDir`) can be layered under the spec. `None`
/// when the sidecar is absent (a pull predating it) or unparseable — the caller
/// then uses the spec-only env. Non-fatal; logged at debug.
fn read_sandbox_image_config(image_dir: &Path) -> Option<zlayer_registry::ImageConfig> {
    let sidecar = image_dir.join("image-config.json");
    let bytes = std::fs::read(&sidecar).ok()?;
    match serde_json::from_slice::<zlayer_registry::ImageConfig>(&bytes) {
        Ok(cfg) => Some(cfg),
        Err(e) => {
            tracing::debug!(
                sidecar = %sidecar.display(),
                error = %e,
                "sandbox: failed to parse image-config.json sidecar; image Env will not layer",
            );
            None
        }
    }
}

/// Read the `metadata.json` platform sidecar an image directory carries for a
/// natively-built image (`{images}/{sanitized}/metadata.json`). Returns the
/// parsed [`LocalImageMetadata`] so `inspect_image_native` can report
/// `os`/`architecture`/`digest`/`created`. `None` when the sidecar is absent
/// (a pulled image, or one built before it existed) or unparseable.
fn read_local_image_metadata(
    image_dir: &Path,
) -> Option<zlayer_types::local_image::LocalImageMetadata> {
    let path = image_dir.join(zlayer_types::local_image::LOCAL_IMAGE_METADATA_FILE);
    let bytes = std::fs::read(&path).ok()?;
    match serde_json::from_slice(&bytes) {
        Ok(meta) => Some(meta),
        Err(e) => {
            tracing::debug!(
                sidecar = %path.display(),
                error = %e,
                "sandbox: failed to parse image metadata.json sidecar",
            );
            None
        }
    }
}

/// The subset of the Seatbelt builder's `config.json` (`SandboxImageConfig`)
/// that `inspect_image_native` surfaces. Deserialize-only and tolerant of the
/// builder's extra fields (serde ignores unknown keys), so the agent need not
/// depend on `zlayer-builder` to read it.
#[derive(Default, serde::Deserialize)]
struct SandboxBuildConfig {
    #[serde(default)]
    env: Vec<String>,
    #[serde(default)]
    working_dir: String,
    #[serde(default)]
    entrypoint: Option<Vec<String>>,
    #[serde(default)]
    cmd: Option<Vec<String>>,
    #[serde(default)]
    user: Option<String>,
    #[serde(default)]
    labels: HashMap<String, String>,
}

/// Read the Seatbelt builder's `config.json` from an image directory. `None`
/// when absent or unparseable — `inspect_image_native` then omits the
/// config-derived fields.
fn read_sandbox_build_config(image_dir: &Path) -> Option<SandboxBuildConfig> {
    let path = image_dir.join("config.json");
    let bytes = std::fs::read(&path).ok()?;
    serde_json::from_slice(&bytes).ok()
}

/// Resolve the entrypoint command from a [`ServiceSpec`].
///
/// Checks `spec.command.entrypoint` and `spec.command.args` in order,
/// then falls back to searching for a shell in the rootfs.
///
/// # Errors
///
/// Returns an error if no entrypoint is specified and no shell is found.
fn resolve_entrypoint(spec: &ServiceSpec, rootfs: &Path) -> Result<(String, Vec<String>)> {
    // Resolve a program path for the macOS sandbox runtime.
    //
    // If the program is an absolute path (e.g. "/usr/local/bin/app"):
    //   1. If it exists on the host -> use the host path (macOS platform binaries
    //      must be exec'd from their original path to pass code-signing checks).
    //   2. Else if it exists inside rootfs -> use the rootfs-resolved path so
    //      `Command::new()` can find it.
    //   3. Otherwise return as-is and let exec() produce a clear error.
    let resolve_program = |prog: &str| -> String {
        if prog.starts_with('/') {
            // Prefer the host binary (code-signing / platform binary compat)
            if std::path::Path::new(prog).exists() {
                return prog.to_string();
            }
            // Fall back to rootfs copy
            let rootfs_path = rootfs.join(prog.trim_start_matches('/'));
            if rootfs_path.exists() {
                return rootfs_path.to_string_lossy().into_owned();
            }
        }
        prog.to_string()
    };

    // Use entrypoint if specified
    if let Some(ref entrypoint) = spec.command.entrypoint {
        if !entrypoint.is_empty() {
            let program = resolve_program(&entrypoint[0]);
            let mut args: Vec<String> = entrypoint[1..].to_vec();

            // Append args from spec.command.args if present
            if let Some(ref extra_args) = spec.command.args {
                args.extend(extra_args.iter().cloned());
            }

            return Ok((program, args));
        }
    }

    // Use args as command if no entrypoint
    if let Some(ref cmd_args) = spec.command.args {
        if !cmd_args.is_empty() {
            let program = resolve_program(&cmd_args[0]);
            let args = cmd_args[1..].to_vec();
            return Ok((program, args));
        }
    }

    // Fallback: try to find a shell - prefer host path for code-signing compat,
    // then check rootfs
    for shell in &["/bin/sh", "/bin/bash", "/usr/bin/sh"] {
        if std::path::Path::new(shell).exists() {
            return Ok(((*shell).to_string(), vec![]));
        }
        if rootfs.join(shell.trim_start_matches('/')).exists() {
            let resolved = rootfs.join(shell.trim_start_matches('/'));
            return Ok((resolved.to_string_lossy().into_owned(), vec![]));
        }
    }

    Err(AgentError::InvalidSpec(
        "No command specified and no shell found in rootfs".to_string(),
    ))
}

/// Parameters for spawning a sandboxed process.
struct SandboxSpawnParams {
    program: String,
    args: Vec<String>,
    sbpl_profile: String,
    rootfs_dir: PathBuf,
    /// Host directory exposed to the workload as `GITHUB_WORKSPACE` — Seatbelt
    /// has no mount namespace, so the workload reaches its workspace through this
    /// env pointer rather than an absolute `/workspace` mount.
    workspace_host: PathBuf,
    /// Base workload environment: the image config's `Env` with the spec env
    /// layered on top (spec wins), already merged. `PORT`/`GITHUB_WORKSPACE`/
    /// `PATH`/auth vars are injected on top of this at spawn (set-if-absent).
    env_base: Vec<(String, String)>,
    stdout_path: PathBuf,
    stderr_path: PathBuf,
    spec: ServiceSpec,
    sandbox_config: SandboxConfig,
    assigned_port: u16,
    auth_env: Option<(String, String, String, bool)>, // (api_url, token, socket_path, mount_socket)
    /// `DOCKER_HOST` value for the per-container Docker Engine API socket, when
    /// one was provisioned for this container (`unix://<host-socket-path>`).
    /// Seatbelt has NO mount namespace, so — unlike the Linux/youki runtime,
    /// which mounts the socket at `/var/run/docker.sock` — the sandboxed workload
    /// must point `DOCKER_HOST` at the socket's real HOST path. `None` when no
    /// docker-socket spawner is configured or the spawn failed.
    docker_host: Option<String>,
    /// Provisioned macOS init toolchains (`git` keg + `node@lts` keg) to inject
    /// into the spawn env (each PATH-prepends its `bin` + sets its vars
    /// set-if-absent). Empty when none were provisioned.
    toolchains: Vec<zlayer_toolchain::ToolchainHandle>,
}

/// Spawn a sandboxed process using `fork()` + `sandbox_init()` + `exec()`.
///
/// The sequence:
/// 1. Open log files for stdout/stderr capture.
/// 2. Build the environment, including `PORT` and `ZLAYER_PORT` set to the
///    dynamically assigned port.
/// 3. Create a `Command` with `pre_exec` that applies the Seatbelt profile
///    and resource limits.
/// 4. Spawn the child process.
/// 5. Return the child PID.
///
/// # Errors
///
/// Returns an error if log files cannot be created or the process fails to spawn.
#[allow(unsafe_code, clippy::too_many_lines)]
fn spawn_sandboxed_process(params: &SandboxSpawnParams) -> Result<u32> {
    use std::os::unix::process::CommandExt;

    let profile = params.sbpl_profile.clone();
    let max_files = params.sandbox_config.max_files;
    let cpu_time_limit = params.sandbox_config.cpu_time_limit;

    // Base env = image config Env with spec env layered on top (merged by the
    // caller). PORT/GITHUB_WORKSPACE/PATH/auth vars are injected below.
    let mut env_vars: Vec<(String, String)> = params.env_base.clone();

    // Inject the assigned port as PORT and ZLAYER_PORT.
    //
    // PORT is the de-facto standard env var for telling a web server which
    // port to listen on (Express, Flask, Actix, Axum, Rails, etc.).
    // ZLAYER_PORT is a ZLayer-specific alias for frameworks that use PORT
    // for something else.
    //
    // Only set PORT if the user's spec didn't already define it (don't
    // override explicit user configuration).
    let port_str = params.assigned_port.to_string();
    if !env_vars.iter().any(|(k, _)| k == "PORT") {
        env_vars.push(("PORT".to_string(), port_str.clone()));
    }
    env_vars.push(("ZLAYER_PORT".to_string(), port_str));

    // Seatbelt has no mount namespace: an absolute `/workspace` resolves against
    // the read-only macOS host root, not the writable rootfs clone. Point the
    // workload at its real writable workspace host dir via `GITHUB_WORKSPACE`
    // (the override every GitHub-Actions-style runner honours), unless the spec
    // already set it.
    if !env_vars.iter().any(|(k, _)| k == "GITHUB_WORKSPACE") {
        env_vars.push((
            "GITHUB_WORKSPACE".to_string(),
            params.workspace_host.to_string_lossy().into_owned(),
        ));
    }

    // `env_clear()` below wipes the inherited environment, so a workload that
    // shells out (or, like the ZArcRunner executor, probes PATH via `env`) would
    // otherwise start with no `PATH` at all. Seed the same default the other
    // runtimes fall back to when the image/spec carries none — OR carries an
    // empty `PATH` (a bare `PATH=` in the OCI config, or `PATH=""` in the spec,
    // survives `merge_env` as an empty value and would otherwise be handed to the
    // child verbatim).
    ensure_spawn_path(&mut env_vars);

    // Init toolchains (git + node@lts): PREPEND each provisioned keg's bin onto
    // PATH (so keg `git` beats host `/usr/bin/git`, which EPERMs reading
    // `/etc/gitconfig` under Seatbelt, and the keg `node` is on PATH for node20 JS
    // actions) and set each keg's vars set-if-absent. Must run AFTER
    // `ensure_spawn_path` so PATH is already non-empty when we prepend.
    for handle in &params.toolchains {
        inject_toolchain_env(&mut env_vars, handle);
    }

    // CA bundle: point OpenSSL/git at `/private/etc/ssl/cert.pem` (set-if-absent)
    // so a sandboxed HTTPS clone/fetch can verify certs without traversing the
    // `/etc` symlink the deny-default profile blocks.
    inject_ca_cert_env(&mut env_vars);

    // HOME -> the container's writable `home` dir (set-if-absent). Seatbelt never
    // seeded HOME, so `git config --global` (setup-go private-module auth) and
    // `go`'s private-module git fetch died with `fatal: $HOME not set`. The
    // workspace_dir is the per-container state dir whose `home/` subdir is both
    // created on disk and granted writable in the SBPL profile.
    inject_home_env(
        &mut env_vars,
        &params.sandbox_config.workspace_dir.join("home"),
    );

    // TMPDIR -> the container's writable tmp dir (host /tmp is denied by Seatbelt),
    // so GITHUB_OUTPUT/GITHUB_ENV/scratch land in an allowed location. The
    // workspace_dir is the per-container state dir whose `tmp/` subdir is both
    // created on disk and granted writable in the SBPL profile.
    if !env_vars.iter().any(|(k, _)| k == "TMPDIR") {
        env_vars.push((
            "TMPDIR".to_string(),
            params
                .sandbox_config
                .workspace_dir
                .join("tmp")
                .to_string_lossy()
                .into_owned(),
        ));
    }

    // DEVELOPER_DIR -> the host's active CLT/Xcode dir (resolved once via
    // `xcode-select -p`). Apple's clang/xcrun shims otherwise shell out to
    // `xcode-select`, which reads the host selector `/var/select/developer_dir` —
    // blocked by the deny-default profile, so `clang --version` dies with
    // "unable to read data link ... No developer tools were found". Setting
    // DEVELOPER_DIR makes them skip that read. Set-if-absent so a spec wins; only
    // injected when the host actually has a developer dir (no CLT -> skip).
    if !env_vars.iter().any(|(k, _)| k == "DEVELOPER_DIR") {
        if let Some(dev_dir) = host_developer_dir() {
            env_vars.push(("DEVELOPER_DIR".to_string(), dev_dir.to_string()));
        }
    }

    // Toolchain cache. RUNNER_TOOL_CACHE / AGENT_TOOLSDIRECTORY point at this
    // container's OWN APFS-CoW clone of the shared toolchain store, which lives
    // inside the per-container rootfs at `/opt/zlayer/toolchains`. Seatbelt has
    // no mount namespace, so the workload reaches it through the host path of
    // the in-rootfs clone (same pattern as `GITHUB_WORKSPACE` -> workspace_host).
    // The clone shares blocks with the store until written, then diverges per
    // container, so concurrent containers never contend on the same files. No
    // CARGO_HOME override: cargo/go/npm/pip default caches land in the
    // per-container rootfs via the workload's HOME automatically.
    let toolchain_clone = params
        .rootfs_dir
        .join("opt/zlayer/toolchains")
        .to_string_lossy()
        .into_owned();
    if !env_vars.iter().any(|(k, _)| k == "RUNNER_TOOL_CACHE") {
        env_vars.push(("RUNNER_TOOL_CACHE".to_string(), toolchain_clone.clone()));
    }
    if !env_vars.iter().any(|(k, _)| k == "AGENT_TOOLSDIRECTORY") {
        env_vars.push(("AGENT_TOOLSDIRECTORY".to_string(), toolchain_clone));
    }

    // Inject ZLayer API credentials for container-to-host communication. The
    // host socket path (`ZLAYER_SOCKET`) is exposed only when the service opts
    // into the daemon socket (`zlayer.io/daemon-socket`); otherwise the
    // container gets only its least-privilege scoped token.
    if let Some((ref api_url, ref token, ref socket, mount_socket)) = params.auth_env {
        env_vars.push(("ZLAYER_API_URL".to_string(), api_url.clone()));
        env_vars.push(("ZLAYER_TOKEN".to_string(), token.clone()));
        if mount_socket {
            env_vars.push(("ZLAYER_SOCKET".to_string(), socket.clone()));
        }
    }

    // Per-container Docker Engine API socket. A `docker`/`zlayer build` inside
    // the sandbox drives ZLayer as its build daemon over this socket, scoped to
    // the container's own token. Seatbelt has no mount namespace, so DOCKER_HOST
    // points at the socket's real HOST path (not `/var/run/docker.sock` like the
    // Linux runtime). DOCKER_BUILDKIT=0 keeps clients on the classic `/build`
    // endpoint the shim implements. Set-if-absent so a spec override wins.
    if let Some(ref docker_host) = params.docker_host {
        if !env_vars.iter().any(|(k, _)| k == "DOCKER_HOST") {
            env_vars.push(("DOCKER_HOST".to_string(), docker_host.clone()));
        }
        if !env_vars.iter().any(|(k, _)| k == "DOCKER_BUILDKIT") {
            env_vars.push(("DOCKER_BUILDKIT".to_string(), "0".to_string()));
        }
    }

    // Open log files for stdout/stderr redirection
    let stdout_file =
        std::fs::File::create(&params.stdout_path).map_err(|e| AgentError::CreateFailed {
            id: "sandbox-process".to_string(),
            reason: format!("Failed to create stdout log: {e}"),
        })?;
    let stderr_file =
        std::fs::File::create(&params.stderr_path).map_err(|e| AgentError::CreateFailed {
            id: "sandbox-process".to_string(),
            reason: format!("Failed to create stderr log: {e}"),
        })?;

    // Honour the spec's working directory, rebased onto the rootfs clone the same
    // way `resolve_program` rebases the entrypoint (there is no mount namespace,
    // so an absolute `/srv` means `{rootfs}/srv`). Falls back to the rootfs root
    // when unset or when the resolved dir does not exist in the clone.
    let cwd = params
        .spec
        .command
        .workdir
        .as_deref()
        .filter(|w| !w.is_empty())
        .map(|w| params.rootfs_dir.join(w.trim_start_matches('/')))
        .filter(|p| p.is_dir())
        .unwrap_or_else(|| params.rootfs_dir.clone());

    // Spawn the child process with pre_exec hook for sandbox application.
    // SAFETY: pre_exec runs after fork() in the child process. We only call
    // async-signal-safe-compatible operations (our FFI calls and setrlimit).
    let child = unsafe {
        std::process::Command::new(&params.program)
            .args(&params.args)
            .current_dir(&cwd)
            .stdout(stdout_file)
            .stderr(stderr_file)
            .env_clear()
            .envs(env_vars)
            .pre_exec(move || {
                // Apply Seatbelt sandbox profile (irrevocable)
                apply_seatbelt_profile(&profile)?;

                // Apply resource limits
                set_resource_limits(max_files, cpu_time_limit)?;

                Ok(())
            })
            .spawn()
    }
    .map_err(|e| AgentError::StartFailed {
        id: "sandbox-process".to_string(),
        reason: format!("Failed to spawn sandboxed process: {e}"),
    })?;

    Ok(child.id())
}

// ---------------------------------------------------------------------------
// Port allocation
// ---------------------------------------------------------------------------

/// Reserve a free TCP port on localhost by binding to port 0.
///
/// Returns `(port, listener)`. The caller **must** hold the returned
/// `TcpListener` until the child process has started and is ready to bind
/// the same port.  Dropping the listener releases the port to the OS,
/// creating a brief window for the child to bind before anything else can
/// claim it.
///
/// The flow:
///   1. Parent calls `reserve_port()` -> gets `(port, guard_listener)`
///   2. Parent spawns the child with `PORT={port}` in its environment
///   3. Parent drops `guard_listener` immediately after `spawn()` returns
///   4. Child's framework reads `PORT`, calls `bind("0.0.0.0:{port}")`
///
/// The race window (step 3->4) is on the order of microseconds (process
/// startup before the child enters `bind()`).  On a developer laptop
/// (the target for macOS sandbox), port theft in this window is not a
/// realistic concern.  For server-class isolation use the `VmRuntime`,
/// where each VM has its own network stack.
///
/// # Errors
///
/// Returns an error if binding to port 0 fails.
fn reserve_port() -> std::io::Result<(u16, std::net::TcpListener)> {
    let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
    let port = listener.local_addr()?.port();
    Ok((port, listener))
}

// ---------------------------------------------------------------------------
// Internal state
// ---------------------------------------------------------------------------

/// Metadata for a sandboxed container process.
#[derive(Debug)]
struct SandboxContainer {
    /// Process ID of the sandboxed child (0 if not yet started).
    pid: u32,
    /// Current container state.
    state: ContainerState,
    /// Path to the container's state directory.
    state_dir: PathBuf,
    /// Path to the cloned rootfs.
    rootfs_dir: PathBuf,
    /// Host directory exposed to the workload as its workspace (via
    /// `GITHUB_WORKSPACE`). A `/workspace` volume's backing dir, else
    /// `{rootfs}/workspace`.
    workspace_host: PathBuf,
    /// Path to the stdout log file.
    stdout_path: PathBuf,
    /// Path to the stderr log file.
    stderr_path: PathBuf,
    /// When the process was started.
    started_at: Option<Instant>,
    /// The original service spec (needed for `start_container`).
    spec: ServiceSpec,
    /// Generated sandbox configuration.
    sandbox_config: SandboxConfig,
    /// Provisioned macOS init toolchains (the `git` keg + the `node@lts` keg) for
    /// this container, each injected into the spawn/exec env so a sandboxed `git`
    /// is the keg git (reads its own config instead of host `/usr/bin/git`
    /// `EPERMing` on `/etc/gitconfig`) and a sandboxed `node` (for node20 JS
    /// actions) is the LTS keg node. Empty when every provision was skipped or
    /// failed (best-effort).
    toolchain_handles: Vec<zlayer_toolchain::ToolchainHandle>,
    /// Memory limit in bytes (for watchdog).
    memory_limit: Option<u64>,
    /// Handle to memory watchdog task.
    watchdog_handle: Option<tokio::task::JoinHandle<()>>,
    /// Dynamic port assigned to this container for host-network port isolation.
    ///
    /// On macOS, all sandboxed processes share the host network stack. To support
    /// multiple replicas of the same service, each replica is assigned a unique
    /// port. This port is passed to the process via the `PORT` environment variable
    /// (a convention respected by most web frameworks). The proxy uses this port
    /// instead of the spec-declared port when constructing backend addresses.
    assigned_port: u16,
    /// Guard listener that holds the assigned port until the child process starts.
    ///
    /// Dropped in `start_container()` right after spawning the child process,
    /// freeing the port for the child to bind. Holding this prevents other
    /// processes from claiming the port between `create_container()` and
    /// `start_container()`.
    port_guard: Option<std::net::TcpListener>,
    /// Overlay (`WireGuard`) `/32` allocated to this container by overlayd, set
    /// by [`Runtime::attach_overlay_ip`]. `get_container_ip` reports this in
    /// preference to loopback so cross-node mesh routing / DNS resolve to the
    /// overlay address. `None` until the container is attached to an overlay.
    overlay_ip: Option<std::net::IpAddr>,
    /// Userspace forwarders (`<overlay_ip>:<port>` -> `127.0.0.1:<delivery>`)
    /// standing up the [`OverlayAttachKind::HostProxy`] data path for this
    /// container. Aborted (and cleared) on [`Runtime::detach_overlay_ip`].
    forwarders: Vec<tokio::task::AbortHandle>,
}

// ---------------------------------------------------------------------------
// SandboxRuntime
// ---------------------------------------------------------------------------

/// Sandbox-based container runtime for macOS.
///
/// Uses Apple's Seatbelt (`sandbox_init`) to run each container as a native
/// macOS process with mandatory access control. The rootfs is APFS-cloned
/// from pulled OCI images for copy-on-write isolation.
///
/// Monotonically increasing counter used to make staging directory names
/// unique across concurrent calls within the same process. Combined with
/// PID and nanosecond timestamp to prevent collisions.
static STAGING_COUNTER: AtomicU64 = AtomicU64::new(0);

/// Toolchains provisioned into the host keg cache at sandbox init and injected
/// onto EVERY container's PATH (spawn + exec). The set generalizes the old
/// single-`git` init so adding another tool is one entry here:
/// - **`git`** — the source-built git keg; a sandboxed `git` reads its own
///   `GIT_CONFIG_SYSTEM` instead of `EPERMing` on `/etc/gitconfig` (host
///   `/usr/bin/git`'s failure under the deny-default `/private/etc`).
/// - **`node@lts`** — the newest Node.js LTS keg (version resolved LIVE from
///   nodejs.org each provision, no pinned major). node20 JS actions
///   (`setup-node`, `setup-go`, …) are run by the runner as `node dist/index.js`,
///   but the macOS Seatbelt sandbox ships no `node`; this puts the LTS `node` on
///   PATH so those actions execute. Iterated in this order at the inject sites.
const INIT_TOOLCHAINS: &[&str] = &["git", "node@lts"];

/// Per-tool process-wide memo of the provisioned [`INIT_TOOLCHAINS`], keyed by
/// the tool token. Each cell mirrors the old single-`git` memo: populated by a
/// single BACKGROUND provision task (never an inline `get_or_init`, because a
/// cold provision is a network download — git ~12s, the Node LTS archive larger
/// — and blocking `create_container` on it would stall EVERY spawn, including
/// jobs that use neither tool, and blow the e2e budgets). `create_container`
/// only NON-BLOCKING-peeks each cell: it attaches a handle when the background
/// provision has already landed, else falls back to a disk probe (below). A
/// cell's value is `Some(handle)` on success or `None` when that tool's provision
/// failed/timed out (the tool is then simply not injected). Lazily built so the
/// key set always matches `INIT_TOOLCHAINS`. One toolchain location per process
/// is assumed (the daemon has a single data dir).
#[allow(clippy::type_complexity)]
fn init_toolchain_cells(
) -> &'static HashMap<&'static str, tokio::sync::OnceCell<Option<zlayer_toolchain::ToolchainHandle>>>
{
    static CELLS: OnceLock<
        HashMap<&'static str, tokio::sync::OnceCell<Option<zlayer_toolchain::ToolchainHandle>>>,
    > = OnceLock::new();
    CELLS.get_or_init(|| {
        INIT_TOOLCHAINS
            .iter()
            .map(|&tool| (tool, tokio::sync::OnceCell::new()))
            .collect()
    })
}

/// Latches `true` the first time a `create_container` kicks off the background
/// provision of the init toolchains, so exactly one provision task per tool is
/// spawned per process.
static INIT_TOOLCHAINS_SPAWNED: AtomicBool = AtomicBool::new(false);

/// Wall-clock bound on each cold init-toolchain provision running in the
/// background. A provision is best-effort (a failure must NOT block jobs that
/// don't use the tool), and a stalled download must not run unbounded — on
/// timeout the background task caches `None` and the tool is not injected. A cold
/// git provision is ~12s; the Node LTS archive download is larger, so 120s is
/// generous slack. Paid at most ONCE per tool per process.
const INIT_TOOLCHAIN_PROVISION_TIMEOUT: Duration = Duration::from_secs(120);

/// Wall-clock bound the ENTRYPOINT spawn ([`SandboxRuntime::start_container`])
/// will wait on a still-missing init toolchain before spawning without it.
///
/// The executor image runs node20 JavaScript actions in the task-executor's OWN
/// process env (`node dist/index.js`), so the long-lived entrypoint MUST carry
/// the Node keg on PATH or every node20 action fails with `node: not found`. The
/// handle set is a SNAPSHOT taken at `create_container`; on a cold first job the
/// background Node provision (a multi-MB download) has not finished, so that
/// snapshot is Node-less and the entrypoint would otherwise spawn without it.
/// `start_container` therefore RE-RESOLVES the init handles right before spawning
/// and, when one is still missing AND its provision is still in flight, waits up
/// to this bound for the keg's `.ready` marker to land. Paid at most once per
/// daemon (the first container triggers the provision; every later container hits
/// the ready keg with zero wait). Best-effort: a tool whose provision has already
/// completed with no handle is never waited on, and a timeout spawns without the
/// keg exactly as before.
const INIT_TOOLCHAIN_START_WAIT: Duration = Duration::from_secs(120);

/// Resolve a single init toolchain's injectable handle: prefer the
/// background-provision memo cell, else fall back to a filesystem-only `.ready`
/// probe (an already-installed keg from a previous daemon process). Returns
/// `None` when neither yields a ready keg.
async fn resolve_one_init_toolchain(
    tool: &str,
    host_tc_cache: &Path,
) -> Option<zlayer_toolchain::ToolchainHandle> {
    if let Some(h) = init_toolchain_cells()
        .get(tool)
        .and_then(|cell| cell.get().cloned().flatten())
    {
        return Some(h);
    }
    zlayer_toolchain::probe_ready_toolchain(
        tool,
        zlayer_toolchain::ToolPlatform::MacOS,
        host_tc_cache,
    )
    .await
}

/// Resolve the injectable handles for every [`INIT_TOOLCHAINS`] entry.
///
/// With `wait = None` this is the original non-blocking behaviour used at
/// `create_container`: cell-or-probe once, attaching whatever keg is already on
/// disk. With `wait = Some(budget)` it additionally waits (bounded, re-probing)
/// for any handle still missing whose background provision is still in flight —
/// used by `start_container` so the long-lived executor entrypoint that runs
/// node20 actions reliably gets the Node (and git) keg even on a cold first job.
/// A tool whose provision has already completed with no handle (cell latched to
/// `Some(None)`) is dropped from the wait immediately — further waiting can never
/// help — so an offline/failed provision never stalls the spawn for the full
/// budget.
async fn resolve_init_toolchain_handles(
    host_tc_cache: &Path,
    wait: Option<Duration>,
) -> Vec<zlayer_toolchain::ToolchainHandle> {
    let mut handles = Vec::new();
    let mut pending: Vec<&'static str> = Vec::new();
    for &tool in INIT_TOOLCHAINS {
        match resolve_one_init_toolchain(tool, host_tc_cache).await {
            Some(h) => handles.push(h),
            None => pending.push(tool),
        }
    }

    let Some(budget) = wait else {
        return handles;
    };

    let deadline = Instant::now() + budget;
    while !pending.is_empty() && Instant::now() < deadline {
        // Stop waiting on any still-pending tool whose provision has finished with
        // no handle (cell latched to `Some(None)`). A pending tool's cell is
        // otherwise unset (`None` = provision in flight, or a keg may still land
        // on disk from another process), which we keep re-probing.
        pending.retain(|&tool| {
            init_toolchain_cells()
                .get(tool)
                .and_then(|cell| cell.get())
                .is_none()
        });

        let mut still = Vec::new();
        for &tool in &pending {
            if let Some(h) = resolve_one_init_toolchain(tool, host_tc_cache).await {
                handles.push(h);
            } else {
                still.push(tool);
            }
        }
        pending = still;
        if pending.is_empty() {
            break;
        }
        tokio::time::sleep(Duration::from_millis(250)).await;
    }

    handles
}

/// GPU access (Metal/MPS) runs at 100% native performance -- no
/// virtualization overhead. Each container gets a generated `.sb` profile
/// that precisely whitelists the `IOKit` classes, Mach services, and filesystem
/// paths required for its workload.
pub struct SandboxRuntime {
    /// Runtime configuration.
    config: MacSandboxConfig,
    /// Active containers keyed by directory name (e.g., "myservice-1").
    containers: Arc<RwLock<HashMap<String, SandboxContainer>>>,
    /// Pulled image rootfs paths keyed by sanitized image name.
    image_rootfs: Arc<RwLock<HashMap<String, PathBuf>>>,
    /// Persistent-volume manager. Seatbelt has no mount namespace, so named /
    /// anonymous volumes are materialised as host directories here and the
    /// container's view of them is rebased onto the rootfs clone (see
    /// [`build_writable_dirs`]). Shared volume dir with the other runtimes so a
    /// named volume (e.g. a CI workspace reused across step containers) resolves
    /// to the same backing directory everywhere.
    storage_manager: Arc<RwLock<StorageManager>>,
    /// Auth context for container-to-host API authentication.
    auth_context: Option<crate::runtime::ContainerAuthContext>,
    /// Daemon secrets provider for `$S:` env refs. Mirrors `VzLinuxRuntime`.
    secrets_provider:
        parking_lot::RwLock<Option<std::sync::Arc<dyn zlayer_secrets::SecretsProvider>>>,
}

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

impl SandboxRuntime {
    /// Create a new sandbox runtime with the given configuration.
    ///
    /// Creates the required directory hierarchy under `config.data_dir`:
    /// - `containers/` -- per-container state
    /// - `images/` -- pulled OCI image rootfs
    ///
    /// # Errors
    ///
    /// Returns an error if the required directories cannot be created.
    pub fn new(
        config: MacSandboxConfig,
        auth_context: Option<crate::runtime::ContainerAuthContext>,
    ) -> Result<Self> {
        std::fs::create_dir_all(&config.data_dir).map_err(|e| {
            AgentError::Configuration(format!(
                "Failed to create data dir {}: {e}",
                config.data_dir.display(),
            ))
        })?;
        std::fs::create_dir_all(&config.log_dir).map_err(|e| {
            AgentError::Configuration(format!(
                "Failed to create log dir {}: {e}",
                config.log_dir.display(),
            ))
        })?;
        std::fs::create_dir_all(config.data_dir.join("containers")).map_err(|e| {
            AgentError::Configuration(format!("Failed to create containers dir: {e}"))
        })?;
        std::fs::create_dir_all(config.data_dir.join("images"))
            .map_err(|e| AgentError::Configuration(format!("Failed to create images dir: {e}")))?;

        // Resolve the persistent-volume directory the SAME way every other
        // runtime does (`ZLAYER_VOLUME_DIR` override, else `{data_dir}/volumes`)
        // so a named volume resolves to one backing directory across runtimes.
        let volume_dir = std::env::var("ZLAYER_VOLUME_DIR").map_or_else(
            |_| zlayer_paths::ZLayerDirs::new(&config.data_dir).volumes(),
            PathBuf::from,
        );
        let storage_manager = StorageManager::new(&volume_dir).map_err(|e| {
            AgentError::Configuration(format!(
                "Failed to init storage manager at {}: {e}",
                volume_dir.display(),
            ))
        })?;

        tracing::info!(
            data_dir = %config.data_dir.display(),
            log_dir = %config.log_dir.display(),
            volume_dir = %volume_dir.display(),
            gpu_access = config.gpu_access,
            "macOS sandbox runtime initialized"
        );

        Ok(Self {
            config,
            containers: Arc::new(RwLock::new(HashMap::new())),
            image_rootfs: Arc::new(RwLock::new(HashMap::new())),
            storage_manager: Arc::new(RwLock::new(storage_manager)),
            auth_context,
            secrets_provider: parking_lot::RwLock::new(None),
        })
    }

    /// Resolve the spec's env map, expanding `$S:` secret refs and `$E:` host-env
    /// refs BEFORE it is layered onto the image config env by [`merge_env`].
    ///
    /// When an injected secrets provider AND a `secret_scope` are both present,
    /// `$S:` and `$E:` are both resolved via [`crate::env::resolve_env_with_secrets`].
    /// Otherwise only `$E:` is resolved (via
    /// [`crate::env::resolve_env_vars_with_warnings`]); any `$S:` value is left
    /// untouched, matching that function's semantics. Mirrors `VzLinuxRuntime`'s
    /// env build so Seatbelt resolves secrets identically to the VZ-Linux guest.
    async fn resolve_env(
        &self,
        spec: &ServiceSpec,
    ) -> Result<std::collections::HashMap<String, String>> {
        let secrets_provider = self.secrets_provider.read().clone();
        if let (Some(provider), Some(scope)) = (secrets_provider, spec.secret_scope.as_ref()) {
            crate::env::resolve_env_with_secrets(
                &spec.env,
                provider.as_ref(),
                &scope.to_storage_scope(),
            )
            .await
            .map_err(|e| {
                AgentError::InvalidSpec(format!("environment variable resolution failed: {e}"))
            })
        } else {
            let resolved = crate::env::resolve_env_vars_with_warnings(&spec.env).map_err(|e| {
                AgentError::InvalidSpec(format!("environment variable resolution failed: {e}"))
            })?;
            for w in &resolved.warnings {
                tracing::warn!("{}", w);
            }
            let mut map = std::collections::HashMap::with_capacity(resolved.vars.len());
            for var in resolved.vars {
                if let Some((k, v)) = var.split_once('=') {
                    map.insert(k.to_string(), v.to_string());
                }
            }
            Ok(map)
        }
    }

    /// Get the runtime configuration.
    #[must_use]
    pub fn config(&self) -> &MacSandboxConfig {
        &self.config
    }

    /// Generate a directory name for a container from its [`ContainerId`].
    fn container_dir_name(id: &ContainerId) -> String {
        format!("{}-{}", id.service, id.replica)
    }

    /// Get the base container state directory for a container.
    fn container_dir(&self, id: &ContainerId) -> PathBuf {
        self.config
            .data_dir
            .join("containers")
            .join(Self::container_dir_name(id))
    }

    /// Get the images base directory.
    fn images_dir(&self) -> PathBuf {
        self.config.data_dir.join("images")
    }

    /// Register a local directory as a pre-built image rootfs.
    ///
    /// This allows using local directories (e.g., host system binaries) as
    /// image sources without pulling from a registry. The directory is
    /// copied/cloned to the standard image location and tracked for use
    /// by `create_container`.
    ///
    /// Used by E2E tests to provide macOS-native binaries, and can be used
    /// by the builder to register locally-built images.
    ///
    /// # Errors
    ///
    /// Returns an error if the image directory cannot be created or the
    /// rootfs cannot be cloned.
    ///
    /// # Panics
    ///
    /// Panics if the system clock is before the Unix epoch.
    #[allow(clippy::too_many_lines)]
    pub async fn register_local_rootfs(&self, image: &str, source_dir: &Path) -> Result<()> {
        let safe_name = sanitize_image_name(image);
        let image_dir = self.images_dir().join(&safe_name);
        let rootfs_dir = image_dir.join("rootfs");

        // Fast path: already on disk
        if rootfs_dir.exists() {
            let mut images = self.image_rootfs.write().await;
            images.insert(safe_name, rootfs_dir);
            return Ok(());
        }

        // Ensure parent dir exists
        tokio::fs::create_dir_all(&image_dir)
            .await
            .map_err(|e| AgentError::PullFailed {
                image: image.to_string(),
                reason: format!("Failed to create image dir: {e}"),
            })?;

        // Clone to a unique staging directory to avoid races when multiple
        // runtime instances register the same image concurrently.
        let staging_name = format!(
            ".rootfs-staging-{}-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos(),
            STAGING_COUNTER.fetch_add(1, Ordering::Relaxed)
        );
        let staging_dir = image_dir.join(&staging_name);

        clone_directory_recursive(source_dir, &staging_dir)
            .await
            .map_err(|e| {
                let _ = std::fs::remove_dir_all(&staging_dir);
                AgentError::PullFailed {
                    image: image.to_string(),
                    reason: format!(
                        "Failed to clone local rootfs from {}: {e}",
                        source_dir.display(),
                    ),
                }
            })?;

        // Atomic rename -- only one caller wins the race
        if tokio::fs::rename(&staging_dir, &rootfs_dir).await.is_err() {
            // Race loser: clean up staging, use winner's rootfs
            let _ = tokio::fs::remove_dir_all(&staging_dir).await;
            if !rootfs_dir.exists() {
                return Err(AgentError::PullFailed {
                    image: image.to_string(),
                    reason: "Failed to finalize rootfs and no other clone succeeded".into(),
                });
            }
        }

        let mut images = self.image_rootfs.write().await;
        images.insert(safe_name, rootfs_dir.clone());

        tracing::info!(
            image = %image,
            source = %source_dir.display(),
            rootfs = %rootfs_dir.display(),
            "Registered local rootfs as image"
        );

        Ok(())
    }

    /// Collect the set of sanitized image directory names that are still
    /// referenced by a container.
    ///
    /// The union of two sources is returned, using the identical derivation as
    /// `create_container` (`sanitize_image_name(&spec.image.name.to_string())`):
    /// 1. Every in-memory container in the `containers` map.
    /// 2. Every on-disk `containers/{dir}/config.json` (containers are not
    ///    restored into memory on daemon restart, so the on-disk specs are the
    ///    authoritative in-use set after a restart).
    ///
    /// Best-effort for source (2): a missing or unparseable `config.json` is
    /// logged at `warn` and skipped (it is treated as referencing nothing). A
    /// missing `containers/` directory contributes an empty set with no warning.
    async fn referenced_image_dirs(&self) -> HashSet<String> {
        let mut referenced = HashSet::new();

        // Source 1: in-memory containers.
        {
            let containers = self.containers.read().await;
            for container in containers.values() {
                referenced.insert(sanitize_image_name(&container.spec.image.name.to_string()));
            }
        }

        // Source 2: on-disk container config.json files.
        let containers_dir = self.config.data_dir.join("containers");
        if let Ok(mut entries) = tokio::fs::read_dir(&containers_dir).await {
            while let Ok(Some(entry)) = entries.next_entry().await {
                let path = entry.path();
                if !path.is_dir() {
                    continue;
                }
                let dir_name = entry.file_name();
                let config_path = path.join("config.json");
                let bytes = match tokio::fs::read(&config_path).await {
                    Ok(bytes) => bytes,
                    Err(e) => {
                        tracing::warn!(
                            container = %dir_name.to_string_lossy(),
                            error = %e,
                            "prune: failed to read container config.json; \
                             not counting it as referencing any image"
                        );
                        continue;
                    }
                };
                match serde_json::from_slice::<ServiceSpec>(&bytes) {
                    Ok(spec) => {
                        referenced.insert(sanitize_image_name(&spec.image.name.to_string()));
                    }
                    Err(e) => {
                        tracing::warn!(
                            container = %dir_name.to_string_lossy(),
                            error = %e,
                            "prune: failed to parse container config.json; \
                             not counting it as referencing any image"
                        );
                    }
                }
            }
        } else {
            // Missing containers dir => nothing referenced from disk.
        }

        referenced
    }

    /// Sum the LOGICAL byte size of every file under `path` (recursively).
    ///
    /// This is the sum of file lengths, NOT the number of physical blocks freed.
    /// Image rootfs directories are APFS copy-on-write clones, so multiple
    /// containers/images may share the same underlying blocks; the actual disk
    /// space reclaimed by deleting `path` can therefore be less than the value
    /// returned here. It is used only to populate `PruneResult::space_reclaimed`
    /// as a best-effort estimate.
    ///
    /// Best-effort: any `read_dir`/`metadata` error is silently skipped.
    async fn dir_size_bytes(path: &Path) -> u64 {
        let mut total: u64 = 0;
        let mut stack: Vec<PathBuf> = vec![path.to_path_buf()];

        while let Some(dir) = stack.pop() {
            let Ok(mut entries) = tokio::fs::read_dir(&dir).await else {
                continue;
            };
            while let Ok(Some(entry)) = entries.next_entry().await {
                let Ok(file_type) = entry.file_type().await else {
                    continue;
                };
                if file_type.is_dir() {
                    stack.push(entry.path());
                } else if file_type.is_file() {
                    if let Ok(meta) = entry.metadata().await {
                        total = total.saturating_add(meta.len());
                    }
                }
            }
        }

        total
    }
}

// ---------------------------------------------------------------------------
// Runtime trait implementation
// ---------------------------------------------------------------------------

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

    /// Pull an image to local storage with default policy (`IfNotPresent`).
    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 policy.
    ///
    /// Uses `zlayer_registry` to pull OCI image layers and extract them to
    /// `{data_dir}/images/{sanitized_name}/rootfs/`. On macOS, OCI images
    /// from registries contain Linux binaries -- the sandbox runtime expects
    /// macOS-native binaries or cross-platform scripts.
    ///
    /// The `_auth` parameter is accepted for trait conformance (§3.10) but
    /// currently ignored: credentials flow through the existing
    /// `AuthResolver` hostname lookup. Callers that need inline auth should
    /// use the Docker runtime.
    #[allow(clippy::too_many_lines)]
    async fn pull_image_with_policy(
        &self,
        image: &str,
        policy: zlayer_spec::PullPolicy,
        _auth: Option<&RegistryAuth>,
        _source: zlayer_spec::SourcePolicy,
    ) -> Result<()> {
        let safe_name = sanitize_image_name(image);
        let image_dir = self.images_dir().join(&safe_name);
        let rootfs_dir = image_dir.join("rootfs");

        match policy {
            zlayer_spec::PullPolicy::Always | zlayer_spec::PullPolicy::Newer => {
                /* always re-pull; drift detection happens at the service layer */
            }
            zlayer_spec::PullPolicy::IfNotPresent => {
                if rootfs_dir.exists() {
                    tracing::debug!(image = %image, "Image already present, skipping pull");
                    // Backfill the ref file for pre-existing images so a later
                    // `list_images` shows the real reference. Best-effort and only
                    // when missing, so this stays a cheap no-op on the hot path.
                    let ref_path = image_dir.join("ref");
                    if matches!(ref_path.try_exists(), Ok(false)) {
                        if let Err(e) = tokio::fs::write(&ref_path, image.as_bytes()).await {
                            tracing::debug!(
                                image = %image,
                                error = %e,
                                "sandbox: failed to backfill image ref file"
                            );
                        }
                    }
                    // Ensure it is tracked
                    let mut images = self.image_rootfs.write().await;
                    images.insert(safe_name, rootfs_dir);
                    return Ok(());
                }
            }
            zlayer_spec::PullPolicy::Never => {
                if !rootfs_dir.exists() {
                    return Err(AgentError::PullFailed {
                        image: image.to_string(),
                        reason: "Image not present and pull policy is Never".to_string(),
                    });
                }
                let mut images = self.image_rootfs.write().await;
                images.insert(safe_name, rootfs_dir);
                return Ok(());
            }
        }

        tracing::info!(
            image = %image,
            "Pulling image for macOS sandbox runtime \
             (note: sandbox runtime expects macOS-native images)"
        );

        tokio::fs::create_dir_all(&rootfs_dir)
            .await
            .map_err(|e| AgentError::PullFailed {
                image: image.to_string(),
                reason: format!("Failed to create rootfs dir: {e}"),
            })?;

        // Record the ORIGINAL image reference alongside the sanitized rootfs so a
        // later `list_images` can display the real reference instead of the
        // sanitized directory name. Best-effort: a write failure never fails the
        // pull (the image is still usable, just shows its dir name).
        let ref_path = image_dir.join("ref");
        if let Err(e) = tokio::fs::write(&ref_path, image.as_bytes()).await {
            tracing::debug!(
                image = %image,
                error = %e,
                "sandbox: failed to write image ref file; list_images will fall back to dir name"
            );
        }

        // Use zlayer-registry to pull and extract OCI image layers.
        // Build a blob cache in the images directory for layer deduplication.
        let cache_path = self.images_dir().join("blobs.redb");
        let cache_type = zlayer_registry::CacheType::persistent_at(&cache_path);
        let blob_cache = cache_type
            .build()
            .await
            .map_err(|e| AgentError::PullFailed {
                image: image.to_string(),
                reason: format!("Failed to open blob cache: {e}"),
            })?;

        let puller = zlayer_registry::ImagePuller::with_cache(blob_cache);
        // Honor ~/.docker/config.json (AuthConfig default = DockerConfig) so
        // `zlayer login` creds / Docker Hub auth apply instead of anonymous.
        let auth =
            zlayer_core::AuthResolver::new(zlayer_core::AuthConfig::default()).resolve(image);

        let layers = puller
            .pull_image(image, &auth)
            .await
            .map_err(|e| AgentError::PullFailed {
                image: image.to_string(),
                reason: format!("Failed to pull image layers: {e}"),
            })?;

        // Persist the OCI image CONFIG blob into the same `blobs.redb` while we
        // still have the network. `pull_image` caches the manifest + layers but
        // NOT the config blob, and the config's `os` field is what the
        // composite's LOCAL-ONLY dispatch inspection
        // (`fetch_image_os_in_cache_only`) reads to route an image correctly on a
        // later `create_container` with NO network. Caching it here is what lets
        // a macOS-native bundle pulled through the sandbox resolve `os=darwin`
        // locally (so it never gets mis-routed to the Linux VM) even under a
        // Docker Hub rate-limit. Non-fatal: a config-blob miss only costs the
        // local OS hint.
        match puller.pull_image_config(image, &auth).await {
            Ok(config) => {
                // Persist the image config (Env/WorkingDir/...) as a sidecar next
                // to the rootfs so `start_container` can layer the image's Env
                // (e.g. a toolchain PATH) under the spec env — matching what the
                // youki / VZ-Linux runtimes do — without a network round-trip.
                match serde_json::to_string_pretty(&config) {
                    Ok(json) => {
                        let sidecar = image_dir.join("image-config.json");
                        if let Err(e) = tokio::fs::write(&sidecar, json).await {
                            tracing::warn!(
                                image = %image,
                                error = %e,
                                "sandbox: failed to write image-config.json sidecar; image Env will not layer on run",
                            );
                        }
                    }
                    Err(e) => tracing::warn!(
                        image = %image,
                        error = %e,
                        "sandbox: failed to serialize image config sidecar",
                    ),
                }
            }
            Err(e) => {
                tracing::debug!(
                    image = %image,
                    error = %e,
                    "sandbox: failed to cache OCI config blob for local OS inspection; \
                     dispatch will rely on its fallthrough",
                );
            }
        }

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

        // Extract layers to rootfs
        let mut unpacker = zlayer_registry::LayerUnpacker::new(rootfs_dir.clone());
        unpacker
            .unpack_layers(&layers)
            .await
            .map_err(|e| AgentError::PullFailed {
                image: image.to_string(),
                reason: format!("Failed to extract rootfs: {e}"),
            })?;

        // Track the rootfs path
        let mut images = self.image_rootfs.write().await;
        images.insert(safe_name, rootfs_dir.clone());

        tracing::info!(
            image = %image,
            rootfs = %rootfs_dir.display(),
            "Image pulled successfully"
        );

        Ok(())
    }

    /// Prune unused image rootfs directories and dangling blob-cache entries.
    ///
    /// An image directory under `{data_dir}/images/` is removed when no
    /// container references it (see [`Self::referenced_image_dirs`]). The
    /// `blobs.redb` dedup cache is preserved as a directory entry and never
    /// itself pruned as an image; after the rootfs sweep a secondary best-effort
    /// pass first removes the cached manifest entries (manifest, digest, and
    /// original-ref keys) belonging to each pruned image — read from that
    /// image's `ref` file before its directory was deleted — so the layer blobs
    /// those manifests pinned become unreferenced, then garbage-collects every
    /// blob no longer referenced by any remaining cached manifest. Images
    /// pulled before the `ref` file existed cannot have their manifests
    /// identified, so their manifest entries are left in place (their blobs
    /// simply aren't reclaimed this pass).
    ///
    /// `space_reclaimed` is a best-effort LOGICAL byte estimate (see
    /// [`Self::dir_size_bytes`]); APFS copy-on-write sharing means the actual
    /// disk freed may be smaller.
    ///
    /// # Errors
    ///
    /// Returns [`AgentError::Internal`] if the images directory exists but
    /// cannot be enumerated. Per-image removal failures and the blob-cache pass
    /// are best-effort: they are logged and never fail the call. A missing
    /// images directory yields an empty [`PruneResult`].
    // Linear prune pipeline (build in-use set, scan images dir, remove unused,
    // GC shared blobs); splitting the stages into helpers would scatter the
    // sequential bookkeeping without aiding readability.
    #[allow(clippy::too_many_lines)]
    async fn prune_images(&self) -> Result<PruneResult> {
        let referenced = self.referenced_image_dirs().await;

        let images_dir = self.images_dir();
        let mut entries = match tokio::fs::read_dir(&images_dir).await {
            Ok(entries) => entries,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                return Ok(PruneResult::default());
            }
            Err(e) => {
                return Err(AgentError::Internal(format!(
                    "failed to read images dir {}: {e}",
                    images_dir.display()
                )));
            }
        };

        let mut deleted: Vec<String> = Vec::new();
        let mut space_reclaimed: u64 = 0;
        // Original image references of pruned dirs, read from each `{dir}/ref`
        // before deletion, used to evict their manifest cache entries below.
        let mut pruned_refs: Vec<String> = Vec::new();

        while let Ok(Some(entry)) = entries.next_entry().await {
            let name = entry.file_name().to_string_lossy().to_string();

            // Preserve the dedup blob cache and any non-directory entries.
            if name == "blobs.redb" {
                continue;
            }
            match entry.file_type().await {
                Ok(ft) if ft.is_dir() => {}
                _ => continue,
            }

            // Skip images still referenced by a container.
            if referenced.contains(&name) {
                continue;
            }

            let path = entry.path();
            let size = Self::dir_size_bytes(&path).await;

            // Capture the ORIGINAL image reference (written by
            // `pull_image_with_policy`) before the directory is removed, so the
            // secondary cache pass can drop this image's manifest entries and
            // let its layer blobs become reclaimable. Best-effort: a missing or
            // unreadable `ref` file (images pulled before that file existed)
            // just means this image's manifests aren't GC'd this pass.
            let ref_path = path.join("ref");
            match tokio::fs::read_to_string(&ref_path).await {
                Ok(contents) => {
                    let r = contents.trim();
                    if r.is_empty() {
                        tracing::debug!(
                            image = %name,
                            "prune: empty ref file; manifest entries not reclaimable"
                        );
                    } else {
                        pruned_refs.push(r.to_string());
                    }
                }
                Err(e) => {
                    tracing::debug!(
                        image = %name,
                        error = %e,
                        "prune: no readable ref file; manifest entries not reclaimable"
                    );
                }
            }

            if let Err(e) = tokio::fs::remove_dir_all(&path).await {
                tracing::warn!(
                    image = %name,
                    error = %e,
                    "prune: failed to remove unused image rootfs; skipping"
                );
                continue;
            }

            tracing::info!(
                image = %name,
                bytes = size,
                "prune: removed unused image rootfs"
            );
            deleted.push(name);
            space_reclaimed = space_reclaimed.saturating_add(size);
        }

        // Drop pruned images from the in-memory rootfs tracking map.
        if !deleted.is_empty() {
            let mut images = self.image_rootfs.write().await;
            for name in &deleted {
                images.remove(name);
            }
        }

        // Secondary best-effort pass: garbage-collect dangling blobs from the
        // dedup cache. Never fails the prune call.
        let cache_path = images_dir.join("blobs.redb");
        if matches!(cache_path.try_exists(), Ok(true)) {
            match zlayer_registry::CacheType::persistent_at(&cache_path)
                .build()
                .await
            {
                Ok(cache) => {
                    // Evict the manifest entries of every pruned image first so
                    // the blobs they pinned become unreferenced and the GC below
                    // can reclaim them. All three keys canonicalize the ref
                    // internally, so the original reference maps to exactly the
                    // keys the pull wrote. Delete failures are non-fatal.
                    for r in &pruned_refs {
                        for key in [
                            zlayer_registry::manifest_cache_key(r),
                            zlayer_registry::manifest_digest_cache_key(r),
                            zlayer_registry::manifest_orig_cache_key(r),
                        ] {
                            if let Err(e) = cache.delete(&key).await {
                                tracing::warn!(
                                    image = %r,
                                    key = %key,
                                    error = %e,
                                    "prune: failed to evict manifest cache entry; continuing"
                                );
                            }
                        }
                    }

                    match zlayer_registry::prune_dangling_blobs(cache.as_ref().as_ref()).await {
                        Ok((blob_deleted, blob_bytes)) => {
                            deleted.extend(blob_deleted);
                            space_reclaimed = space_reclaimed.saturating_add(blob_bytes);
                        }
                        Err(e) => {
                            tracing::warn!(
                                error = %e,
                                "prune: failed to garbage-collect dangling blobs; skipping"
                            );
                        }
                    }
                }
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        "prune: failed to open blob cache for blob GC; skipping"
                    );
                }
            }
        }

        Ok(PruneResult {
            deleted,
            space_reclaimed,
        })
    }

    /// List the images stored under `{data_dir}/images/`.
    ///
    /// Each image is one subdirectory named after the sanitized reference
    /// (see [`sanitize_image_name`]); the `blobs.redb` dedup cache and any
    /// non-directory entries are skipped. For each image:
    /// - `reference` is the trimmed contents of `{dir}/ref` (the original
    ///   reference recorded by `pull_image_with_policy`) when that file is
    ///   present and non-empty, otherwise the sanitized directory name (older
    ///   images pulled before the `ref` file existed).
    /// - `size_bytes` is the LOGICAL byte sum of the extracted `rootfs/` tree
    ///   (see [`Self::dir_size_bytes`]); metadata files like `ref` are excluded
    ///   so the value reflects the image payload. APFS copy-on-write sharing
    ///   means the true on-disk footprint may be smaller, and an image
    ///   directory missing its `rootfs/` reports `Some(0)`.
    /// - `digest` is the registry digest recorded under the manifest-digest
    ///   cache key for the resolved reference, when the `blobs.redb` cache opens
    ///   and the entry is present. Any cache failure or miss leaves it `None`.
    ///
    /// The blob cache is opened once before the loop (best-effort); if it
    /// cannot be opened, every entry's `digest` is `None`. Entries are returned
    /// in directory-iteration order (the youki backend does not sort either).
    ///
    /// # Errors
    ///
    /// Returns [`AgentError::Internal`] if the images directory exists but
    /// cannot be enumerated. A missing images directory yields an empty list.
    async fn list_images(&self) -> Result<Vec<ImageInfo>> {
        let images_dir = self.images_dir();
        let mut entries = match tokio::fs::read_dir(&images_dir).await {
            Ok(entries) => entries,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                return Ok(Vec::new());
            }
            Err(e) => {
                return Err(AgentError::Internal(format!(
                    "failed to read images dir {}: {e}",
                    images_dir.display()
                )));
            }
        };

        // Open the dedup cache once before the loop. Best-effort: when it is
        // absent or fails to open, every image's digest is left as `None`.
        let cache_path = images_dir.join("blobs.redb");
        let cache: Option<Arc<Box<dyn BlobCacheBackend>>> =
            if matches!(cache_path.try_exists(), Ok(true)) {
                match zlayer_registry::CacheType::persistent_at(&cache_path)
                    .build()
                    .await
                {
                    Ok(cache) => Some(cache),
                    Err(e) => {
                        tracing::debug!(
                            error = %e,
                            "list_images: failed to open blob cache; digests will be None"
                        );
                        None
                    }
                }
            } else {
                None
            };

        let mut images = Vec::new();

        while let Ok(Some(entry)) = entries.next_entry().await {
            let name = entry.file_name().to_string_lossy().to_string();

            // Skip the dedup blob cache and any non-directory entries.
            if name == "blobs.redb" {
                continue;
            }
            match entry.file_type().await {
                Ok(ft) if ft.is_dir() => {}
                _ => continue,
            }

            let dir = entry.path();

            // Prefer the ORIGINAL image reference recorded at pull time; fall
            // back to the sanitized directory name for images pulled before the
            // `ref` file existed.
            let reference = match tokio::fs::read_to_string(dir.join("ref")).await {
                Ok(contents) => {
                    let r = contents.trim();
                    if r.is_empty() {
                        name.clone()
                    } else {
                        r.to_string()
                    }
                }
                Err(_) => name.clone(),
            };

            // LOGICAL byte sum of the extracted `rootfs/` tree only; metadata
            // files like `ref` are deliberately excluded so the reported size
            // matches the image payload (APFS CoW caveat documented on
            // `dir_size_bytes`). A directory without a `rootfs/` reports
            // `Some(0)` (read_dir error on the missing path is skipped).
            let size_bytes = Some(Self::dir_size_bytes(&dir.join("rootfs")).await);

            // Best-effort registry digest from the manifest-digest cache key.
            // The key canonicalizes the reference internally, matching what the
            // pull wrote. Any failure or miss leaves the digest `None`.
            let digest = match cache.as_ref() {
                Some(cache) => cache
                    .get(&zlayer_registry::manifest_digest_cache_key(&reference))
                    .await
                    .ok()
                    .flatten()
                    .and_then(|bytes| String::from_utf8(bytes).ok()),
                None => None,
            };

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

        Ok(images)
    }

    /// Inspect a locally-built (or pulled) image from the on-disk store.
    ///
    /// The Seatbelt runtime stores each image as a plain directory at
    /// `{data_dir}/images/{sanitize_image_name(image)}/` — a `rootfs/` tree, a
    /// `config.json` (OCI-like build config), a `ref` file (canonical tag), and,
    /// for a native build, a `metadata.json` platform sidecar. There is no
    /// registry manifest to inspect, so the default trait impl returned `501`
    /// ("not supported") and an operator had no way to inspect a Mac-native
    /// image. This reads the data dir directly:
    ///
    /// - `os` / `architecture` / `digest` / `created` come from the
    ///   `metadata.json` sidecar (the same single source of truth dispatch uses).
    /// - `repo_tags` is the `ref` file (or the requested name as a fallback).
    /// - `size` is the live byte sum of `rootfs/`.
    /// - `env` / `cmd` / `entrypoint` / `working_dir` / `user` / `labels` come
    ///   from `config.json` when present.
    ///
    /// # Errors
    ///
    /// Returns [`AgentError::NotFound`] when no image directory (with a `rootfs/`)
    /// exists for `image`.
    async fn inspect_image_native(&self, image: &str) -> Result<ImageInspectInfo> {
        let safe = sanitize_image_name(image);
        let dir = self.images_dir().join(&safe);
        let rootfs = dir.join("rootfs");
        if !matches!(rootfs.try_exists(), Ok(true)) {
            return Err(AgentError::NotFound {
                container: image.to_string(),
                reason: format!("image '{image}' not found"),
            });
        }

        // Canonical reference recorded at build/pull time; fall back to the
        // requested name for images written before the `ref` file existed.
        let reference = match tokio::fs::read_to_string(dir.join("ref")).await {
            Ok(contents) if !contents.trim().is_empty() => contents.trim().to_string(),
            _ => image.to_string(),
        };

        // Live `rootfs/` size — the sidecar leaves `size` unset at build time, so
        // this never goes stale against later layer edits.
        let size = Self::dir_size_bytes(&rootfs).await;

        let mut info = ImageInspectInfo {
            repo_tags: vec![reference],
            size: Some(size),
            ..Default::default()
        };

        // Platform sidecar (os/arch/digest/created) for a native build.
        if let Some(meta) = read_local_image_metadata(&dir) {
            info.os = Some(meta.os).filter(|s| !s.is_empty());
            info.architecture = Some(meta.architecture).filter(|s| !s.is_empty());
            info.created = meta.created;
            // A native build has no registry (`repo@sha256:`) digest; surface
            // the sidecar's content digest as the image id when one was recorded.
            info.id = meta.digest;
        }

        // OCI-like build config for the runtime metadata Docker callers expect.
        if let Some(cfg) = read_sandbox_build_config(&dir) {
            info.env = cfg.env;
            info.cmd = cfg.cmd.unwrap_or_default();
            info.entrypoint = cfg.entrypoint.unwrap_or_default();
            info.working_dir = Some(cfg.working_dir).filter(|s| !s.is_empty());
            info.user = cfg.user;
            info.labels = cfg.labels.into_iter().collect();
        }

        Ok(info)
    }

    /// Remove an image's rootfs directory and its dangling manifest cache
    /// entries.
    ///
    /// The image directory is `{data_dir}/images/{sanitize_image_name(image)}`.
    /// Before deleting it, the image's `ref` file is read so the cache-cleanup
    /// pass can evict the manifest entries for both the supplied `image`
    /// argument and the recorded original reference.
    ///
    /// When the image is still referenced by a container (see
    /// [`Self::referenced_image_dirs`]) and `force` is `false`, the removal is
    /// refused. The directory is then removed with `remove_dir_all`, the image
    /// is dropped from the in-memory rootfs tracking map, and a best-effort
    /// cache pass deletes the three manifest keys (manifest, digest, and
    /// original-ref) for each candidate reference. Layer blobs are deliberately
    /// NOT deleted here: a blob can be shared by several images, so deleting it
    /// directly could corrupt an unrelated image. Orphaned blobs left behind by
    /// the evicted manifests are reclaimed by the next [`Self::prune_images`].
    ///
    /// # Errors
    ///
    /// - [`AgentError::NotFound`] if no image directory exists for `image`.
    /// - [`AgentError::InvalidSpec`] if the image is in use by a container and
    ///   `force` is `false`.
    /// - [`AgentError::Internal`] if the image directory exists but cannot be
    ///   removed.
    ///
    /// The blob-cache cleanup is best-effort: failures are logged and never
    /// fail the call.
    async fn remove_image(&self, image: &str, force: bool) -> Result<()> {
        let safe = sanitize_image_name(image);
        let dir = self.images_dir().join(&safe);

        if !matches!(dir.try_exists(), Ok(true)) {
            return Err(AgentError::NotFound {
                container: image.to_string(),
                reason: format!("image '{image}' not found"),
            });
        }

        // Refuse to remove an image still referenced by a container unless the
        // caller forces it. An in-use conflict is a client-state issue, so map
        // it to InvalidSpec (400) rather than Internal (500).
        if !force && self.referenced_image_dirs().await.contains(&safe) {
            return Err(AgentError::InvalidSpec(format!(
                "image '{image}' is in use by a container; pass --force to remove it"
            )));
        }

        // Capture the ORIGINAL image reference (written by
        // `pull_image_with_policy`) before deleting the directory, so the cache
        // pass can evict this image's manifest entries under the recorded ref as
        // well as the supplied argument.
        let ref_file = match tokio::fs::read_to_string(dir.join("ref")).await {
            Ok(contents) => {
                let r = contents.trim();
                if r.is_empty() {
                    None
                } else {
                    Some(r.to_string())
                }
            }
            Err(_) => None,
        };

        tokio::fs::remove_dir_all(&dir).await.map_err(|e| {
            AgentError::Internal(format!("failed to remove image dir {}: {e}", dir.display()))
        })?;

        // Drop the image from the in-memory rootfs tracking map.
        {
            let mut images = self.image_rootfs.write().await;
            images.remove(&safe);
        }

        // Best-effort cache cleanup: evict the manifest entries that pinned this
        // image's layer blobs. We deliberately do NOT delete the layer blobs
        // directly — a blob can be shared across multiple images, so removing it
        // here could corrupt an unrelated image. Once the manifests are evicted,
        // the now-orphaned blobs are reclaimed by the next `prune_images` GC.
        let cache_path = self.images_dir().join("blobs.redb");
        if matches!(cache_path.try_exists(), Ok(true)) {
            match zlayer_registry::CacheType::persistent_at(&cache_path)
                .build()
                .await
            {
                Ok(cache) => {
                    // Dedupe {image arg, ref-file content}: when the ref file
                    // equals the argument we only evict once.
                    let mut candidates: Vec<String> = vec![image.to_string()];
                    if let Some(r) = ref_file {
                        if !candidates.contains(&r) {
                            candidates.push(r);
                        }
                    }
                    for r in &candidates {
                        // All three keys canonicalize the ref internally, so the
                        // reference maps to exactly the keys the pull wrote.
                        for key in [
                            zlayer_registry::manifest_cache_key(r),
                            zlayer_registry::manifest_digest_cache_key(r),
                            zlayer_registry::manifest_orig_cache_key(r),
                        ] {
                            if let Err(e) = cache.delete(&key).await {
                                tracing::warn!(
                                    image = %r,
                                    key = %key,
                                    error = %e,
                                    "remove_image: failed to evict manifest cache entry; continuing"
                                );
                            }
                        }
                    }
                }
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        "remove_image: failed to open blob cache for manifest eviction; skipping"
                    );
                }
            }
        }

        tracing::info!(image = %image, "Removed image (sandbox)");
        Ok(())
    }

    /// Create a container.
    ///
    /// 1. APFS-clones the base image rootfs to a per-container directory.
    /// 2. Generates a Seatbelt profile based on the [`ServiceSpec`] (GPU, network, filesystem).
    /// 3. Writes the profile to `{container_dir}/sandbox.sb`.
    /// 4. Writes container metadata to `{container_dir}/config.json`.
    /// 5. Stores the container as [`ContainerState::Pending`].
    #[allow(clippy::too_many_lines)]
    async fn create_container(&self, id: &ContainerId, spec: &ServiceSpec) -> Result<()> {
        // Persist a default PATH into the spec so the on-disk config.json the
        // executor reads carries it (scratch images bake none). Set-if-absent;
        // user/image PATH wins. This shadowed clone is what gets serialized to
        // config.json AND stored in the in-memory container record below.
        let spec = {
            let mut s = spec.clone();
            ensure_default_path(&mut s);
            s
        };
        let spec = &spec;

        let dir_name = Self::container_dir_name(id);
        let container_dir = self.container_dir(id);
        let rootfs_dir = container_dir.join("rootfs");

        // Clean up stale container directory if it exists
        if container_dir.exists() {
            tracing::warn!(
                container = %dir_name,
                "Stale container directory found, cleaning up"
            );
            if let Err(e) = tokio::fs::remove_dir_all(&container_dir).await {
                tracing::warn!(
                    container = %dir_name,
                    error = %e,
                    "Failed to remove stale container directory"
                );
            }
        }

        // Create container state directory
        tokio::fs::create_dir_all(&container_dir)
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: dir_name.clone(),
                reason: format!(
                    "Failed to create container dir {}: {e}",
                    container_dir.display(),
                ),
            })?;

        // Create tmp directory within the container
        tokio::fs::create_dir_all(container_dir.join("tmp"))
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: dir_name.clone(),
                reason: format!("Failed to create container tmp dir: {e}"),
            })?;

        // Create the per-container HOME directory. Seatbelt has no mount
        // namespace and never seeded `HOME`, so `git config --global` /
        // `go`'s private-module fetch died with `fatal: $HOME not set`. This
        // dir is granted writable in the SBPL profile (see
        // `resolve_sandbox_storage`) and injected as `HOME` (set-if-absent) at
        // spawn/exec time via `inject_home_env`.
        tokio::fs::create_dir_all(container_dir.join("home"))
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: dir_name.clone(),
                reason: format!("Failed to create container home dir: {e}"),
            })?;

        // Locate the base image rootfs
        let image_name_str = spec.image.name.to_string();
        let safe_image_name = sanitize_image_name(&image_name_str);
        let image_rootfs = {
            let images = self.image_rootfs.read().await;
            images.get(&safe_image_name).cloned()
        };
        let image_rootfs =
            image_rootfs.unwrap_or_else(|| self.images_dir().join(&safe_image_name).join("rootfs"));

        if !image_rootfs.exists() {
            return Err(AgentError::CreateFailed {
                id: dir_name.clone(),
                reason: format!(
                    "Image rootfs not found at {}. Run pull_image first.",
                    image_rootfs.display()
                ),
            });
        }

        // APFS-clone the base image rootfs to this container's rootfs
        tracing::debug!(
            container = %dir_name,
            src = %image_rootfs.display(),
            dst = %rootfs_dir.display(),
            "Cloning rootfs (APFS CoW)"
        );
        clone_directory_recursive(&image_rootfs, &rootfs_dir)
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: dir_name.clone(),
                reason: format!(
                    "Failed to clone rootfs from {} to {}: {e}",
                    image_rootfs.display(),
                    rootfs_dir.display(),
                ),
            })?;

        // APFS-clone the shared toolchain store into THIS container's own rootfs
        // at `/opt/zlayer/toolchains` (surfaced as RUNNER_TOOL_CACHE /
        // AGENT_TOOLSDIRECTORY). clonefile shares blocks with the store until a
        // file is written, at which point the write lands privately in this
        // container's rootfs. The clone is inside rootfs_dir, which is already
        // per-container writable + profile-granted, so it needs no extra
        // writable_dirs grant and no CARGO_HOME override.
        let tc_src = zlayer_paths::ZLayerDirs::new(&self.config.data_dir).toolchain_cache();
        if tc_src.exists() {
            let tc_dst = rootfs_dir.join("opt/zlayer/toolchains");
            if let Some(parent) = tc_dst.parent() {
                tokio::fs::create_dir_all(parent).await.ok();
            }
            clone_directory_recursive(&tc_src, &tc_dst)
                .await
                .map_err(|e| AgentError::CreateFailed {
                    id: dir_name.clone(),
                    reason: format!(
                        "Failed to clone toolchain store from {} to {}: {e}",
                        tc_src.display(),
                        tc_dst.display(),
                    ),
                })?;
        }

        // Provision the macOS `git` toolchain into the host toolchain-cache dir
        // (a keg under `{tc_src}/git-<ver>-<arch>`, idempotent via a `.ready`
        // marker). The keg git reads its OWN `GIT_CONFIG_SYSTEM`, so a sandboxed
        // `git` no longer EPERMs on `/etc/gitconfig` (host git's failure under
        // Seatbelt's deny-default `/private/etc`). BEST-EFFORT: a provision
        // failure (offline, GHCR down, non-macOS) must NOT block non-git jobs —
        // log and continue with no handle (the env injection + profile grant are
        // both then skipped). The returned handle's `path_dirs`/`env` point at
        // the host keg path; the seatbelt profile grants read+exec on `tc_src`.
        // Host-side `git` keg cache. This is DELIBERATELY a separate dir from
        // `toolchain_cache()` (the actions `RUNNER_TOOL_CACHE`/hostedtoolcache
        // store cloned into each rootfs above): the keg is a HOST-side toolchain
        // the sandboxed process reaches via an absolute PATH, NOT a tool that
        // belongs in the container's tool cache. Keeping it separate also avoids
        // a race — the background provision mutates this dir
        // (`remove_dir_all`+rebuild on a cold keg), and it must not stomp the
        // rootfs-clone source `tc_src` while a concurrent `create_container`
        // clones it.
        let host_tc_cache = self.config.data_dir.join("host-toolchains");

        // Kick off provisioning of EVERY init toolchain (git + node@lts) ONCE per
        // process, each in its own BACKGROUND task. Deliberately NOT awaited here:
        // the cold path is a network download and blocking container creation on
        // it would stall every spawn (jobs that use neither tool included). The
        // first container(s) after a cold start fall back to whatever kegs are
        // already on disk (via the probe below); once a background task lands,
        // every later container injects that keg.
        if !INIT_TOOLCHAINS_SPAWNED.swap(true, Ordering::AcqRel) {
            for &tool in INIT_TOOLCHAINS {
                let host_tc_cache_bg = host_tc_cache.clone();
                tokio::spawn(async move {
                    let provision = zlayer_toolchain::ensure_toolchain(
                        tool,
                        zlayer_toolchain::ToolPlatform::MacOS,
                        &host_tc_cache_bg,
                        None,
                    );
                    let result = match tokio::time::timeout(
                        INIT_TOOLCHAIN_PROVISION_TIMEOUT,
                        provision,
                    )
                    .await
                    {
                        Ok(Ok(handle)) => Some(handle),
                        Ok(Err(e)) => {
                            tracing::warn!(
                                tool,
                                error = %e,
                                "init toolchain provisioning failed; sandboxed jobs fall back (keg not injected)"
                            );
                            None
                        }
                        Err(_elapsed) => {
                            tracing::warn!(
                                tool,
                                timeout_secs = INIT_TOOLCHAIN_PROVISION_TIMEOUT.as_secs(),
                                "init toolchain provisioning timed out; sandboxed jobs fall back (keg not injected)"
                            );
                            None
                        }
                    };
                    if let Some(h) = &result {
                        tracing::info!(
                            tool,
                            keg = %h.install_dir.display(),
                            "init toolchain provisioned; sandboxed jobs now use the keg",
                        );
                    } else {
                        tracing::debug!(
                            tool,
                            "init toolchain provision yielded no handle; tool not injected",
                        );
                    }
                    // Cell is written exactly once per process; ignore the second-set error.
                    if let Some(cell) = init_toolchain_cells().get(tool) {
                        let _ = cell.set(result);
                    }
                });
            }
        }
        // Resolve the keg handle for THIS container for each init toolchain. The
        // background provision populates a cell only after the (cold, network)
        // install lands, so a non-blocking peek alone races: the first container
        // after a daemon start — and every container during the cold-provision
        // window — would see `None` and silently skip the keg even though it is
        // (or soon will be) on disk. When a cell isn't populated yet, fall back to
        // a cheap, filesystem-only `.ready` probe so an already-provisioned keg
        // (e.g. from a previous daemon process) is injected without waiting on the
        // in-process task. The probe returns `None` until `.ready` is stamped
        // (written last), so a partially-extracted keg is never injected.
        // Non-blocking snapshot (cell-or-probe, no wait): attach whatever init
        // kegs are already on disk. `start_container` re-resolves this — WITH a
        // bounded wait — right before spawning the entrypoint, so a keg still
        // downloading now is picked up before the long-lived task-executor (which
        // runs node20 actions in its own env) launches.
        let toolchain_handles = resolve_init_toolchain_handles(&host_tc_cache, None).await;

        // Determine GPU access level from spec
        let gpu_access = if self.config.gpu_access {
            if let Some(ref gpu) = spec.resources.gpu {
                if gpu.vendor == "apple" {
                    match gpu.mode.as_deref() {
                        Some("mps") => GpuAccess::MpsOnly,
                        _ => GpuAccess::MetalCompute,
                    }
                } else {
                    GpuAccess::None
                }
            } else {
                GpuAccess::None
            }
        } else {
            GpuAccess::None
        };

        // Determine keychain / code-signing access. Deny-by-default: a spec can
        // never reach `securityd` or the host keychain just by labeling itself.
        // The OPERATOR lifts the default one of two ways, and either one is the
        // gate here:
        //   --allow-keychain-access  -> permit; the workload picks the keychain
        //                               via the `zlayer.io/keychain` label.
        //   --keychain-default <sel> -> permit + supply a default selection the
        //                               operator chose, so the workload needs no
        //                               label at all (ZLAYER_KEYCHAIN_DEFAULT env).
        // A workload's own label still wins over the operator default.
        let operator_default = operator_keychain_default_from_env();
        let keychain_access = if self.config.keychain_access_allowed || operator_default.is_some() {
            build_keychain_access(spec, operator_default.as_deref())
        } else {
            KeychainAccess::None
        };

        // Reserve a unique port for this container BEFORE building the sandbox
        // profile, so the Seatbelt profile can allow binding on the assigned port.
        //
        // All macOS sandbox containers share the host network. If we let
        // multiple replicas of the same service bind the same spec-declared
        // port, the second one gets EADDRINUSE. We solve this by assigning
        // each container a dynamically allocated port (via OS port-0 binding)
        // and passing it through the PORT environment variable (respected by
        // most web frameworks: Express, Actix, Axum, Flask, etc.).
        //
        // The guard listener holds the port until start_container() spawns
        // the child process, preventing other processes from stealing it.
        let (assigned_port, port_guard) = reserve_port().map_err(|e| AgentError::CreateFailed {
            id: dir_name.clone(),
            reason: format!("Failed to reserve a dynamic port for sandbox container: {e}"),
        })?;

        tracing::info!(
            container = %dir_name,
            assigned_port = assigned_port,
            "Reserved dynamic port for sandbox container"
        );

        // Resolve the workload's network-isolation policy so fenced
        // (isolated/dedicated) overlay modes are actually fenced on Seatbelt.
        let mode = spec.overlay.as_ref().map(|o| o.mode).unwrap_or_default();
        let nm_none = matches!(
            spec.network_mode,
            zlayer_types::spec::types::NetworkMode::None
        );
        let nm_host = matches!(
            spec.network_mode,
            zlayer_types::spec::types::NetworkMode::Host
        );
        let iso_label = spec
            .labels
            .get(zlayer_types::overlay::ISOLATION_NETWORK_LABEL)
            .map(String::as_str);
        let iso = zlayer_types::overlay::resolve_network_isolation(
            mode, nm_none, nm_host, &dir_name, iso_label,
        );

        // Determine network access from spec endpoints, including the assigned port.
        // Only open the host daemon API port (loopback egress for a nested `zlayer
        // build` to delegate) when this container is actually given daemon creds —
        // i.e. an auth context is present (see the `auth_env`/`ZLAYER_API_URL`
        // injection below). Parse the port from the daemon's advertised API URL,
        // falling back to the default daemon port (3669).
        let daemon_api_port = self
            .auth_context
            .as_ref()
            .map(|ctx| api_url_port(&ctx.api_url).unwrap_or(3669));
        let mut network_access = build_network_access(
            spec,
            self.config.overlay_cidr.clone(),
            &iso,
            daemon_api_port,
        );

        // Ensure the Seatbelt profile allows binding on the dynamically assigned port.
        // Without this, the sandbox would deny the child's bind() call.
        match &mut network_access {
            NetworkAccess::LocalhostOnly {
                ref mut bind_ports,
                ref mut connect_ports,
                ..
            } => {
                if !bind_ports.contains(&assigned_port) {
                    bind_ports.push(assigned_port);
                }
                if !connect_ports.contains(&assigned_port) {
                    connect_ports.push(assigned_port);
                }
            }
            NetworkAccess::None => {
                // If network was fully denied but we need a port, upgrade to localhost-only
                let mut connect_ports = vec![assigned_port, 53, 80, 443];
                if let Some(port) = daemon_api_port {
                    if !connect_ports.contains(&port) {
                        connect_ports.push(port);
                    }
                }
                network_access = NetworkAccess::LocalhostOnly {
                    bind_ports: vec![assigned_port],
                    connect_ports,
                    overlay_cidr: self.config.overlay_cidr.clone(),
                    loopback_api_port: daemon_api_port,
                };
            }
            NetworkAccess::Full => {
                // Full access already allows all ports
            }
        }

        // Materialise host-backed volumes (named/anonymous/bind/s3) so each has a
        // real writable host directory, then resolve the writable-dir grants and
        // the workspace host path. Seatbelt has no mount namespace, so this is how
        // a container-absolute `/workspace` is given somewhere writable to land.
        let volume_paths = {
            let mut storage_manager = self.storage_manager.write().await;
            crate::runtimes::volume_prep::prepare_storage_volumes(
                &mut storage_manager,
                &dir_name,
                &spec.storage,
            )
            .await?
        };
        let storage = resolve_sandbox_storage(spec, &rootfs_dir, &container_dir, &volume_paths);

        // Create every writable dir on disk before the profile grants it, so the
        // workload's `mkdir`/checkout finds the path already present and writable.
        for dir in &storage.writable_dirs {
            if let Err(e) = tokio::fs::create_dir_all(dir).await {
                tracing::warn!(
                    container = %dir_name,
                    dir = %dir.display(),
                    error = %e,
                    "Failed to pre-create writable dir; workload may fail if it writes here"
                );
            }
        }
        let workspace_host = storage.workspace_host.clone();

        // Parse memory limit
        let memory_limit = spec
            .resources
            .memory
            .as_ref()
            .and_then(|m| parse_memory_string(m));

        let mut sandbox_config = SandboxConfig {
            rootfs_dir: rootfs_dir.clone(),
            workspace_dir: container_dir.clone(),
            gpu_access,
            keychain_access,
            network_access,
            writable_dirs: storage.writable_dirs,
            readonly_dirs: vec![],
            // Grant read+exec on the host-side keg cache unconditionally: every
            // init keg (git + node@lts) lives under this single dir, and
            // `start_container` may resolve+inject a keg that only became ready
            // AFTER this profile was written (e.g. the Node keg finishing its cold
            // download between create and start). Gating the grant on the
            // create-time handle snapshot would leave that later-injected keg on
            // PATH but unreadable/unexecutable under the deny-default profile.
            // Granting an absent/empty dir is harmless (a path rule, not a
            // must-exist mount).
            toolchain_cache: Some(host_tc_cache.clone()),
            max_files: 4096,
            cpu_time_limit: None,
            memory_limit,
        };

        // Allow sandbox write access to the API socket ONLY when the service
        // opted into the daemon socket (`zlayer.io/daemon-socket`); otherwise
        // the container talks to the API over TCP with its scoped token and
        // never needs the (admin-capable) socket.
        if let Some(ref auth_ctx) = self.auth_context {
            let deployment = spec.deployment.as_deref().unwrap_or(&id.service);
            if crate::auth::resolve_container_api_access(deployment, &spec.labels).mount_socket {
                sandbox_config
                    .writable_dirs
                    .push(PathBuf::from(&auth_ctx.socket_path));
            }
        }

        // Generate Seatbelt profile
        let profile = generate_sandbox_profile(&sandbox_config);

        // Write profile to disk
        let profile_path = container_dir.join("sandbox.sb");
        tokio::fs::write(&profile_path, &profile)
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: dir_name.clone(),
                reason: format!("Failed to write Seatbelt profile: {e}"),
            })?;

        // Write config to disk (for use by start_container)
        let config_json =
            serde_json::to_string_pretty(spec).map_err(|e| AgentError::CreateFailed {
                id: dir_name.clone(),
                reason: format!("Failed to serialize spec: {e}"),
            })?;
        tokio::fs::write(container_dir.join("config.json"), &config_json)
            .await
            .map_err(|e| AgentError::CreateFailed {
                id: dir_name.clone(),
                reason: format!("Failed to write config.json: {e}"),
            })?;

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

        // Register the container as pending
        let container = SandboxContainer {
            pid: 0,
            state: ContainerState::Pending,
            state_dir: container_dir,
            rootfs_dir,
            workspace_host,
            stdout_path,
            stderr_path,
            started_at: None,
            spec: spec.clone(),
            sandbox_config,
            toolchain_handles,
            memory_limit,
            watchdog_handle: None,
            assigned_port,
            port_guard: Some(port_guard),
            overlay_ip: None,
            forwarders: Vec::new(),
        };

        let mut containers = self.containers.write().await;
        containers.insert(dir_name.clone(), container);

        tracing::info!(
            container = %dir_name,
            image = %spec.image.name,
            port = assigned_port,
            "Container created (sandbox)"
        );

        Ok(())
    }

    /// Start a container.
    ///
    /// Reads the saved spec and Seatbelt profile, resolves the entrypoint,
    /// and forks a child process with `sandbox_init()` applied via `pre_exec`.
    /// Stdout/stderr are redirected to log files. If a memory limit is
    /// configured, a watchdog task is spawned to enforce it.
    ///
    /// The container's dynamically assigned port is injected as `PORT` and
    /// `ZLAYER_PORT` environment variables. The port guard listener (which
    /// was holding the port since `create_container()`) is dropped immediately
    /// after the child process spawns, freeing the port for the child to bind.
    #[allow(clippy::too_many_lines)]
    async fn start_container(&self, id: &ContainerId) -> Result<()> {
        let dir_name = Self::container_dir_name(id);

        // Extract what we need from the container state, then release the lock
        // so spawn_sandboxed_process (which is blocking) does not hold it.
        let (
            rootfs_dir,
            workspace_host,
            stdout_path,
            stderr_path,
            spec,
            sandbox_config,
            toolchain_handles,
            memory_limit,
            assigned_port,
        ) = {
            let containers = self.containers.read().await;
            let container = containers
                .get(&dir_name)
                .ok_or_else(|| AgentError::NotFound {
                    container: dir_name.clone(),
                    reason: "Container not created".to_string(),
                })?;
            (
                container.rootfs_dir.clone(),
                container.workspace_host.clone(),
                container.stdout_path.clone(),
                container.stderr_path.clone(),
                container.spec.clone(),
                container.sandbox_config.clone(),
                container.toolchain_handles.clone(),
                container.memory_limit,
                container.assigned_port,
            )
        };

        // Re-resolve the init-toolchain handles right before spawning the
        // (long-lived) entrypoint. The `toolchain_handles` snapshot stored at
        // create time can be Node-less on a cold first job: the executor image
        // runs node20 JS actions in the task-executor's OWN process env
        // (`node dist/index.js`), so if the Node keg download had not finished
        // when the container was created, the entrypoint would spawn without
        // `node` on PATH and every node20 action would fail with
        // `node: not found`. Re-resolving here — and briefly waiting on any keg
        // whose provision is still in flight — makes the entrypoint reliably
        // carry the Node + git kegs. The Seatbelt profile already grants read+exec
        // on the whole host keg cache (see `create_container`), so a keg that
        // became ready only now is still executable. Keeps the richer of the two
        // resolutions so a transient FS hiccup never DROPS a keg the snapshot had.
        let host_tc_cache = self.config.data_dir.join("host-toolchains");
        let toolchain_handles = {
            let fresh =
                resolve_init_toolchain_handles(&host_tc_cache, Some(INIT_TOOLCHAIN_START_WAIT))
                    .await;
            if fresh.len() >= toolchain_handles.len() {
                fresh
            } else {
                toolchain_handles
            }
        };

        // Reflect the freshly-resolved handles back onto the container so a later
        // `exec` (node20 actions inside a job container) also injects them.
        {
            let mut containers = self.containers.write().await;
            if let Some(container) = containers.get_mut(&dir_name) {
                container.toolchain_handles.clone_from(&toolchain_handles);
            }
        }

        // Read the generated Seatbelt profile from disk. Mutable: a per-container
        // Docker socket (provisioned below at START time, since its path is not
        // known when the profile is generated at CREATE time) appends its
        // connect-allow rules before the profile is applied in the child.
        let profile_path = self.container_dir(id).join("sandbox.sb");
        let mut profile = tokio::fs::read_to_string(&profile_path)
            .await
            .map_err(|e| AgentError::StartFailed {
                id: dir_name.clone(),
                reason: format!("Failed to read Seatbelt profile: {e}"),
            })?;

        // Resolve the command to execute
        let (program, args) = resolve_entrypoint(&spec, &rootfs_dir)?;

        tracing::info!(
            container = %dir_name,
            program = %program,
            args = ?args,
            port = assigned_port,
            "Starting sandboxed process"
        );

        // Drop the port guard right before spawning.
        //
        // The guard has been holding the port since create_container() to prevent
        // other processes from claiming it. We must release it so the child can
        // bind the same port. The window between drop and child bind() is
        // microseconds (process startup time).
        {
            let mut containers = self.containers.write().await;
            if let Some(container) = containers.get_mut(&dir_name) {
                // Drop the guard listener to free the port for the child
                container.port_guard.take();
            }
        }

        // Mint a per-container JWT so the sandboxed process can authenticate
        // back to the host ZLayer API. The token record is persisted BEFORE its
        // jti is embedded (the auth layer is fail-closed and rejects a jti with
        // no record); the persist is async, so this can't live in a `.map`
        // closure.
        //
        // `docker_host` is set alongside when a per-container Docker Engine API
        // socket is provisioned for this container (see the spawner call below).
        let mut docker_host: Option<String> = None;
        let auth_env = if let Some(ctx) = self.auth_context.as_ref() {
            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);
            let token_jti = if let Some(sink) = 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(
                &ctx.jwt_secret,
                &id.service,
                &container_id,
                access.scopes,
                access.ttl,
                token_jti,
            )
            .unwrap_or_default();

            // Per-container Docker Engine API socket: provision a UDS backed by
            // THIS container's scoped token so a `docker`/`zlayer build` inside
            // the sandbox drives ZLayer as its build daemon (and can delegate a
            // Linux build to the daemon's VZ `zlayer-buildd`, which the Seatbelt
            // sandbox cannot spawn itself). Mirrors the youki runtime, except
            // Seatbelt has NO mount namespace: DOCKER_HOST must point at the
            // socket's real HOST path, and the deny-default profile must ALLOW
            // connecting to it — appended here since the socket path is only
            // known now (not at profile-generation/create time). Uses the SAME
            // container_id/token identity as the minted auth token.
            if let Some(spawner) = ctx.docker_socket_spawner.as_ref() {
                if let Some(host_sock) = spawner.spawn(&container_id, token.clone()).await {
                    append_docker_socket_sbpl_rules(&mut profile, Path::new(&host_sock));
                    docker_host = Some(format!("unix://{host_sock}"));
                } else {
                    tracing::warn!(
                        container = %container_id,
                        "the per-container Docker socket spawner failed; container \
                         starts without a docker socket"
                    );
                }
            }

            Some((
                // Host-native Seatbelt process → reach the daemon on loopback,
                // not the overlay node IP advertised for VZ-Linux guests.
                loopback_api_url(&ctx.api_url),
                token,
                ctx.socket_path.clone(),
                access.mount_socket,
            ))
        } else {
            None
        };

        // Build the base env: the image config's Env (from the sidecar written at
        // pull) with the spec env layered on top, exactly like the youki/VZ-Linux
        // runtimes. Falls back to spec-only env when no sidecar is present.
        let image_config = read_sandbox_image_config(
            &self
                .images_dir()
                .join(sanitize_image_name(&spec.image.name.to_string())),
        );
        // Resolve $S:/$E: refs in the spec env FIRST (image config stays the base,
        // resolved spec env layers on top), then merge. Without this the Seatbelt
        // workload would receive literal `$S:<name>` strings instead of secrets.
        let env_base = {
            let mut spec_for_env = spec.clone();
            spec_for_env.env = self.resolve_env(&spec).await?;
            crate::runtimes::macos_vz_shared::merge_env(&spec_for_env, image_config.as_ref())
        };

        // Spawn the sandboxed process in a blocking task so that the fork+exec
        // does not block the tokio reactor (which would prevent timers and other
        // futures from making progress on a current_thread runtime).
        let dir_name_for_err = dir_name.clone();
        let child_pid = tokio::task::spawn_blocking(move || {
            spawn_sandboxed_process(&SandboxSpawnParams {
                program,
                args,
                sbpl_profile: profile,
                rootfs_dir,
                workspace_host,
                env_base,
                stdout_path,
                stderr_path,
                spec,
                sandbox_config,
                assigned_port,
                auth_env,
                docker_host,
                toolchains: toolchain_handles,
            })
        })
        .await
        .map_err(|e| AgentError::StartFailed {
            id: dir_name_for_err,
            reason: format!("spawn task join error: {e}"),
        })??;

        // Write PID file
        let pid_path = self.container_dir(id).join("pid");
        tokio::fs::write(&pid_path, child_pid.to_string())
            .await
            .map_err(|e| AgentError::StartFailed {
                id: dir_name.clone(),
                reason: format!("Failed to write PID file: {e}"),
            })?;

        // Update container state and optionally start memory watchdog
        let mut containers = self.containers.write().await;
        if let Some(container) = containers.get_mut(&dir_name) {
            container.pid = child_pid;
            container.state = ContainerState::Running;
            container.started_at = Some(Instant::now());

            // Start memory watchdog if a memory limit is configured
            if let Some(limit) = memory_limit {
                let pid = child_pid;
                let handle = tokio::spawn(async move {
                    memory_watchdog(pid, limit).await;
                });
                container.watchdog_handle = Some(handle);
            }
        }

        tracing::info!(
            container = %dir_name,
            pid = child_pid,
            "Sandboxed process started"
        );

        Ok(())
    }

    /// Stop a container.
    ///
    /// Sends `SIGTERM` to the process and waits up to `timeout` for graceful
    /// shutdown. If the process is still alive after the timeout, sends `SIGKILL`.
    #[allow(unsafe_code)]
    #[allow(clippy::too_many_lines)]
    #[allow(clippy::cast_possible_wrap)]
    async fn stop_container(&self, id: &ContainerId, timeout: Duration) -> Result<()> {
        let dir_name = Self::container_dir_name(id);

        // Get the PID and update state to Stopping
        let pid = {
            let mut containers = self.containers.write().await;
            let container = containers
                .get_mut(&dir_name)
                .ok_or_else(|| AgentError::NotFound {
                    container: dir_name.clone(),
                    reason: "Container not found".to_string(),
                })?;

            if container.pid == 0 {
                container.state = ContainerState::Exited { code: 0 };
                return Ok(());
            }

            // Workload already done? The exit was recorded by container_state()
            // (e.g. when the executor's WaitContainer returned) and the child
            // has been reaped. Signalling a dead pid and waiting out the grace
            // just burns the caller's deadline, so skip straight to a no-op.
            if matches!(
                container.state,
                ContainerState::Exited { .. } | ContainerState::Failed { .. }
            ) {
                return Ok(());
            }

            container.state = ContainerState::Stopping;
            container.pid
        };

        tracing::info!(
            container = %dir_name,
            pid = pid,
            timeout = ?timeout,
            "Stopping sandboxed process"
        );

        // Send SIGTERM
        unsafe {
            libc::kill(pid as i32, libc::SIGTERM);
        }

        // Wait for graceful shutdown with timeout
        let deadline = Instant::now() + timeout;
        loop {
            if Instant::now() >= deadline {
                break;
            }

            // Check if process has exited (non-blocking waitpid). A negative
            // result means ECHILD -- the child was already reaped elsewhere
            // (container_state) so it's gone; treat that as a clean exit rather
            // than waiting out the full grace on a pid that will never report.
            let mut status: libc::c_int = 0;
            let result = unsafe { libc::waitpid(pid as i32, &raw mut status, libc::WNOHANG) };

            if result != 0 {
                // Process exited (result > 0) or already reaped (result < 0)
                let exit_code = if result > 0 && libc::WIFEXITED(status) {
                    libc::WEXITSTATUS(status)
                } else {
                    -1
                };

                let mut containers = self.containers.write().await;
                if let Some(c) = containers.get_mut(&dir_name) {
                    c.state = ContainerState::Exited { code: exit_code };
                    if let Some(h) = c.watchdog_handle.take() {
                        h.abort();
                    }
                }
                tracing::info!(
                    container = %dir_name,
                    exit_code = exit_code,
                    "Container stopped gracefully"
                );
                return Ok(());
            }

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

        // Timeout reached -- send SIGKILL
        tracing::warn!(
            container = %dir_name,
            pid = pid,
            "SIGTERM timeout, sending SIGKILL"
        );
        unsafe {
            libc::kill(pid as i32, libc::SIGKILL);
        }

        // Wait for SIGKILL to take effect (non-blocking poll with timeout,
        // because the child may have already been reaped by container_state())
        let pid_for_wait = pid;
        let exit_code = tokio::task::spawn_blocking(move || {
            let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
            loop {
                let mut status: libc::c_int = 0;
                let result =
                    unsafe { libc::waitpid(pid_for_wait as i32, &raw mut status, libc::WNOHANG) };
                if result > 0 || result == -1 {
                    break; // reaped or already gone
                }
                if std::time::Instant::now() >= deadline {
                    break; // give up -- process already reaped elsewhere
                }
                std::thread::sleep(std::time::Duration::from_millis(50));
            }
            -9i32
        })
        .await
        .unwrap_or(-9);

        let mut containers = self.containers.write().await;
        if let Some(c) = containers.get_mut(&dir_name) {
            c.state = ContainerState::Exited { code: exit_code };
            if let Some(h) = c.watchdog_handle.take() {
                h.abort();
            }
        }

        tracing::info!(container = %dir_name, "Container killed (SIGKILL)");
        Ok(())
    }

    /// Remove a container.
    ///
    /// Kills the process if still running, aborts the watchdog, removes the
    /// container directory (rootfs clone, profile, logs), and removes it
    /// from internal tracking.
    #[allow(unsafe_code)]
    #[allow(clippy::cast_possible_wrap)]
    async fn remove_container(&self, id: &ContainerId) -> Result<()> {
        let dir_name = Self::container_dir_name(id);

        tracing::info!(container = %dir_name, "Removing container");

        // Tear down the per-container Docker socket (if any) and revoke the
        // container's scoped token now that it is terminally gone — mirrors the
        // youki runtime. The socket key + token jti use the SAME identity minted
        // in `start_container` (`{service}-{replica}`). 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() {
            let container_id = format!("{}-{}", id.service, id.replica);
            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;
            }
        }

        // Remove from tracking, killing the process if needed
        let removed_spec = {
            let mut containers = self.containers.write().await;
            if let Some(mut c) = containers.remove(&dir_name) {
                // Abort watchdog if running
                if let Some(h) = c.watchdog_handle.take() {
                    h.abort();
                }

                // Kill process if still running
                if c.pid > 0
                    && matches!(c.state, ContainerState::Running | ContainerState::Stopping)
                {
                    unsafe {
                        libc::kill(c.pid as i32, libc::SIGKILL);
                    }
                    // Reap the zombie (non-blocking poll, child may already be reaped)
                    let pid = c.pid;
                    let _ = tokio::task::spawn_blocking(move || {
                        let deadline =
                            std::time::Instant::now() + std::time::Duration::from_secs(3);
                        loop {
                            let mut status: libc::c_int = 0;
                            let result = unsafe {
                                libc::waitpid(pid as i32, &raw mut status, libc::WNOHANG)
                            };
                            if result > 0 || result == -1 {
                                break;
                            }
                            if std::time::Instant::now() >= deadline {
                                break;
                            }
                            std::thread::sleep(std::time::Duration::from_millis(50));
                        }
                    })
                    .await;
                }
                Some(c.spec)
            } else {
                None
            }
        };

        // Reclaim the host-side volume space this container held, keyed by its
        // container id: detach named volumes (dropping their refcount), unmount
        // S3, and delete its anonymous scratch dirs. Done before the rootfs dir
        // is removed; anonymous volumes live under the shared volume dir, not in
        // the container dir, so they would otherwise leak.
        if let Some(spec) = removed_spec {
            let mut storage_manager = self.storage_manager.write().await;
            crate::runtimes::volume_prep::cleanup_storage_volumes(
                &mut storage_manager,
                &dir_name,
                &spec.storage,
            );
        }

        // Remove container directory (rootfs, logs, config, profile)
        let container_dir = self.container_dir(id);
        if container_dir.exists() {
            tokio::fs::remove_dir_all(&container_dir)
                .await
                .map_err(|e| {
                    AgentError::Internal(format!(
                        "Failed to remove container dir {}: {e}",
                        container_dir.display(),
                    ))
                })?;
        }

        tracing::info!(container = %dir_name, "Container removed");
        Ok(())
    }

    /// Get container state.
    ///
    /// If the container is in a running state, checks whether the process
    /// is still alive via `waitpid(WNOHANG)` and updates the state accordingly.
    #[allow(unsafe_code)]
    #[allow(clippy::cast_possible_wrap)]
    async fn container_state(&self, id: &ContainerId) -> Result<ContainerState> {
        let dir_name = Self::container_dir_name(id);

        let mut containers = self.containers.write().await;
        let container = containers
            .get_mut(&dir_name)
            .ok_or_else(|| AgentError::NotFound {
                container: dir_name.clone(),
                reason: "Container not found".to_string(),
            })?;

        // If already in a terminal state, return it
        match &container.state {
            ContainerState::Exited { .. } | ContainerState::Failed { .. } => {
                return Ok(container.state.clone());
            }
            ContainerState::Pending => return Ok(ContainerState::Pending),
            _ => {}
        }

        // Check if process is still alive
        if container.pid > 0 {
            let mut status: libc::c_int = 0;
            let result =
                unsafe { libc::waitpid(container.pid as i32, &raw mut status, libc::WNOHANG) };

            match result.cmp(&0) {
                std::cmp::Ordering::Greater => {
                    // Process has exited
                    let exit_code = if libc::WIFEXITED(status) {
                        libc::WEXITSTATUS(status)
                    } else if libc::WIFSIGNALED(status) {
                        -(libc::WTERMSIG(status))
                    } else {
                        -1
                    };
                    container.state = ContainerState::Exited { code: exit_code };
                    if let Some(h) = container.watchdog_handle.take() {
                        h.abort();
                    }
                }
                std::cmp::Ordering::Equal => {
                    // Process still running
                    container.state = ContainerState::Running;
                }
                std::cmp::Ordering::Less => {
                    // Error -- process disappeared
                    container.state = ContainerState::Failed {
                        reason: "Process disappeared".to_string(),
                    };
                }
            }
        }

        Ok(container.state.clone())
    }

    /// Get container logs as structured [`LogEntry`] values.
    ///
    /// If `tail > 0`, returns only the last `tail` entries.
    async fn container_logs(&self, id: &ContainerId, tail: usize) -> Result<Vec<LogEntry>> {
        let dir_name = Self::container_dir_name(id);

        let (stdout_path, stderr_path) = {
            let containers = self.containers.read().await;
            let container = containers
                .get(&dir_name)
                .ok_or_else(|| AgentError::NotFound {
                    container: dir_name.clone(),
                    reason: "Container not found".to_string(),
                })?;
            (container.stdout_path.clone(), container.stderr_path.clone())
        };

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

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

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

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

        Ok(entries)
    }

    /// Stream a container's captured stdout/stderr as [`LogChunk`]s.
    ///
    /// The Seatbelt runtime redirects the workload's stdout/stderr to the
    /// per-container `stdout.log` / `stderr.log` files; there is no live tail
    /// device. We follow those append-only files line-by-line (per channel) and
    /// emit a chunk for each new line. Before this existed the runtime fell back
    /// to the trait default (`Unsupported`), so `CompositeRuntime::logs_stream`
    /// synthesised a one-shot snapshot of whatever had been written SO FAR — for
    /// a CI job that snapshot is taken before the container produces output, so
    /// the Forgejo job log came back EMPTY. Following the files until the
    /// workload exits delivers the real, complete output.
    ///
    /// `tail`/`since`/`until`/`timestamps`/`follow` follow Docker's
    /// `GET /containers/{id}/logs` semantics; "neither stdout nor stderr
    /// requested" means "both".
    async fn logs_stream(&self, id: &ContainerId, opts: LogsStreamOptions) -> Result<LogsStream> {
        let dir_name = Self::container_dir_name(id);

        // Resolve the captured-log paths (same files `container_logs`/`get_logs`
        // read). A missing record is a genuine 404, not an empty stream.
        let (stdout_path, stderr_path) = {
            let containers = self.containers.read().await;
            let container = containers
                .get(&dir_name)
                .ok_or_else(|| AgentError::NotFound {
                    container: dir_name.clone(),
                    reason: "Container not found".to_string(),
                })?;
            (container.stdout_path.clone(), container.stderr_path.clone())
        };

        // Docker treats "neither requested" as "both".
        let none_specified = !opts.stdout && !opts.stderr;
        let want_stdout = opts.stdout || none_specified;
        let want_stderr = opts.stderr || none_specified;

        // Bounded channel so a slow consumer back-pressures the file tailers.
        let (tx, rx) = mpsc::channel::<Result<LogChunk>>(64);

        if want_stdout {
            tokio::spawn(stream_sandbox_log_file(
                stdout_path,
                LogChannel::Stdout,
                opts.clone(),
                tx.clone(),
                Arc::clone(&self.containers),
                dir_name.clone(),
            ));
        }
        if want_stderr {
            tokio::spawn(stream_sandbox_log_file(
                stderr_path,
                LogChannel::Stderr,
                opts.clone(),
                tx.clone(),
                Arc::clone(&self.containers),
                dir_name.clone(),
            ));
        }
        // Drop the original sender so the merged stream ends once every tailer
        // has exited.
        drop(tx);

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

    /// Execute a command inside a container's sandbox.
    ///
    /// Spawns a new process with the same Seatbelt profile as the container,
    /// running in the container's rootfs directory. Captures stdout/stderr
    /// and returns `(exit_code, stdout, stderr)`.
    async fn exec(&self, id: &ContainerId, cmd: &[String]) -> Result<(i32, String, String)> {
        let dir_name = Self::container_dir_name(id);

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

        let (rootfs, profile_path, spec, toolchain_handles) = {
            let containers = self.containers.read().await;
            let container = containers
                .get(&dir_name)
                .ok_or_else(|| AgentError::NotFound {
                    container: dir_name.clone(),
                    reason: "Container not found".to_string(),
                })?;
            (
                container.rootfs_dir.clone(),
                container.state_dir.join("sandbox.sb"),
                container.spec.clone(),
                container.toolchain_handles.clone(),
            )
        };

        // Read the Seatbelt profile (same sandbox as the main process)
        let profile = tokio::fs::read_to_string(&profile_path)
            .await
            .map_err(|e| AgentError::Internal(format!("Failed to read Seatbelt profile: {e}")))?;

        tracing::debug!(
            container = %dir_name,
            cmd = ?cmd,
            "Executing command in sandbox"
        );

        // Honor the image's OCI config.Env (PATH + image vars) under the spec env,
        // matching start_container's spawn path and the VZ-Linux exec path. Without
        // this the exec'd process inherits the daemon environment, not the image's.
        let image_config = read_sandbox_image_config(
            &self
                .images_dir()
                .join(sanitize_image_name(&spec.image.name.to_string())),
        );
        // Resolve $S:/$E: refs in the spec env FIRST so the exec'd process gets
        // real secret values, then layer over the image config env. Keep
        // `ensure_spawn_path` AFTER the merge so a missing/empty PATH is seeded.
        let mut env_base = {
            let mut spec_for_env = spec.clone();
            spec_for_env.env = self.resolve_env(&spec).await?;
            crate::runtimes::macos_vz_shared::merge_env(&spec_for_env, image_config.as_ref())
        };
        ensure_spawn_path(&mut env_base);

        // Same init-toolchain injection as the spawn path (each keg PATH-prepends
        // its bin + sets its vars set-if-absent), AFTER `ensure_spawn_path`, so an
        // `exec`-ed `git` resolves the provisioned keg (not host `/usr/bin/git`)
        // and an `exec`-ed `node` (node20 JS actions) resolves the LTS keg.
        for handle in &toolchain_handles {
            inject_toolchain_env(&mut env_base, handle);
        }

        // Same CA-bundle injection as the spawn path: point OpenSSL/git at the
        // resolved `/private/etc/ssl/cert.pem` (set-if-absent) so an exec'd HTTPS
        // clone verifies certs under the deny-default profile.
        inject_ca_cert_env(&mut env_base);

        // Same HOME injection as the spawn path: point HOME at the container's
        // writable `home` dir (set-if-absent) so an exec'd `git config --global`
        // / `go` private-module fetch doesn't die with `fatal: $HOME not set`.
        // `rootfs` is `{container_dir}/rootfs`, so its parent is the container
        // dir whose granted `home/` subdir we target.
        if let Some(container_dir) = rootfs.parent() {
            inject_home_env(&mut env_base, &container_dir.join("home"));
        }

        // Use sandbox-exec to run the command in the container's rootfs.
        // We pass the profile inline via `-p` to avoid needing a file path
        // that the sandbox itself can read.
        let profile_clone = profile.clone();
        let rootfs_clone = rootfs.clone();
        let cmd_clone = cmd.to_vec();
        let env_clone = env_base;

        let output = tokio::task::spawn_blocking(move || {
            std::process::Command::new("/usr/bin/sandbox-exec")
                .arg("-p")
                .arg(&profile_clone)
                .arg("--")
                .arg(&cmd_clone[0])
                .args(&cmd_clone[1..])
                .current_dir(&rootfs_clone)
                .env_clear()
                .envs(env_clone)
                .output()
        })
        .await
        .map_err(|e| AgentError::Internal(format!("exec task join error: {e}")))?
        .map_err(|e| AgentError::Internal(format!("Failed to exec: {e}")))?;

        let exit_code = output.status.code().unwrap_or(-1);
        let stdout = String::from_utf8_lossy(&output.stdout).to_string();
        let stderr = String::from_utf8_lossy(&output.stderr).to_string();

        tracing::debug!(
            container = %dir_name,
            exit_code = exit_code,
            stdout_len = stdout.len(),
            stderr_len = stderr.len(),
            "exec completed"
        );

        Ok((exit_code, stdout, stderr))
    }

    /// Get container resource statistics.
    ///
    /// Uses macOS `proc_pidinfo` with `PROC_PIDTASKINFO` to read CPU time
    /// (user + system) and resident set size for the sandboxed process.
    async fn get_container_stats(&self, id: &ContainerId) -> Result<ContainerStats> {
        let dir_name = Self::container_dir_name(id);

        let (pid, memory_limit) = {
            let containers = self.containers.read().await;
            let container = containers
                .get(&dir_name)
                .ok_or_else(|| AgentError::NotFound {
                    container: dir_name.clone(),
                    reason: "Container not found".to_string(),
                })?;

            if container.pid == 0 {
                return Err(AgentError::Internal(
                    "Container not started -- no PID available for stats".to_string(),
                ));
            }

            (container.pid, container.memory_limit.unwrap_or(0))
        };

        // Get process stats via proc_pidinfo (blocking FFI call)
        let pid_for_stats = pid;
        let (cpu_usec, memory_bytes) =
            tokio::task::spawn_blocking(move || get_process_stats(pid_for_stats))
                .await
                .map_err(|e| AgentError::Internal(format!("stats task join error: {e}")))??;

        Ok(ContainerStats {
            cpu_usage_usec: cpu_usec,
            memory_bytes,
            memory_limit,
            timestamp: Instant::now(),
        })
    }

    /// Wait for a container to exit and return its exit code.
    ///
    /// Uses `spawn_blocking` with `waitpid` (blocking) to avoid tying up
    /// the async runtime. Updates the container state on completion.
    #[allow(unsafe_code)]
    #[allow(clippy::cast_possible_wrap)]
    async fn wait_container(&self, id: &ContainerId) -> Result<i32> {
        let dir_name = Self::container_dir_name(id);
        let pid = {
            let containers = self.containers.read().await;
            let container = containers
                .get(&dir_name)
                .ok_or_else(|| AgentError::NotFound {
                    container: dir_name.clone(),
                    reason: "Container not found".to_string(),
                })?;

            // If already exited, return immediately
            if let ContainerState::Exited { code } = &container.state {
                return Ok(*code);
            }

            if container.pid == 0 {
                return Err(AgentError::Internal(
                    "Container not started -- no PID to wait on".to_string(),
                ));
            }

            container.pid
        };

        tracing::debug!(container = %dir_name, pid = pid, "Waiting for container to exit");

        // Block on waitpid in a spawned blocking task
        let exit_code = tokio::task::spawn_blocking(move || {
            let mut status: libc::c_int = 0;
            let result = unsafe { libc::waitpid(pid as i32, &raw mut status, 0) };
            if result < 0 {
                return -1;
            }
            if libc::WIFEXITED(status) {
                libc::WEXITSTATUS(status)
            } else if libc::WIFSIGNALED(status) {
                -(libc::WTERMSIG(status))
            } else {
                -1
            }
        })
        .await
        .map_err(|e| AgentError::Internal(format!("wait task join error: {e}")))?;

        // Update state
        let mut containers = self.containers.write().await;
        if let Some(c) = containers.get_mut(&dir_name) {
            c.state = ContainerState::Exited { code: exit_code };
            if let Some(h) = c.watchdog_handle.take() {
                h.abort();
            }
        }

        tracing::info!(
            container = %dir_name,
            exit_code = exit_code,
            "Container exited"
        );

        Ok(exit_code)
    }

    /// Get container logs as structured [`LogEntry`] values.
    async fn get_logs(&self, id: &ContainerId) -> Result<Vec<LogEntry>> {
        let dir_name = Self::container_dir_name(id);

        let (stdout_path, stderr_path) = {
            let containers = self.containers.read().await;
            let container = containers
                .get(&dir_name)
                .ok_or_else(|| AgentError::NotFound {
                    container: dir_name.clone(),
                    reason: "Container not found".to_string(),
                })?;
            (container.stdout_path.clone(), container.stderr_path.clone())
        };

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

        // Read stdout
        if let Ok(content) = tokio::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 let Ok(content) = tokio::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,
                });
            }
        }

        Ok(entries)
    }

    /// Get the PID of a container's main process.
    ///
    /// Returns `Some(pid)` if the container has been started, `None` if
    /// it is still in `Pending` state (pid == 0).
    async fn get_container_pid(&self, id: &ContainerId) -> Result<Option<u32>> {
        let dir_name = Self::container_dir_name(id);

        let containers = self.containers.read().await;
        let container = containers
            .get(&dir_name)
            .ok_or_else(|| AgentError::NotFound {
                container: dir_name.clone(),
                reason: "Container not found".to_string(),
            })?;

        if container.pid > 0 {
            Ok(Some(container.pid))
        } else {
            Ok(None)
        }
    }

    /// A Seatbelt container is a native host process sharing the host network
    /// stack -- it has no netns/veth and no guest. The service layer therefore
    /// allocates an overlay `/32` and pushes it back via [`attach_overlay_ip`],
    /// and we stand up userspace forwarders binding `<overlay_ip>:<port>` ->
    /// `127.0.0.1:<delivery_port>` to carry mesh traffic to the workload.
    fn overlay_attach_kind(&self) -> OverlayAttachKind {
        OverlayAttachKind::HostProxy
    }

    /// Record the overlay `/32` allocated to this container and stand up the
    /// host-proxy forwarders that make it reachable from the mesh.
    ///
    /// Each published port is forwarded `<overlay_ip>:<container_port>` ->
    /// `127.0.0.1:<delivery_port>`. The PRIMARY published port is delivered to
    /// the workload's dynamically `assigned_port` (the one injected as `PORT` /
    /// `ZLAYER_PORT` and bound on `127.0.0.1`); SECONDARY published ports are
    /// delivered to `127.0.0.1:<container_port>` on the assumption the workload
    /// binds that literal port itself.
    ///
    /// CAVEAT: a Seatbelt container has a single `assigned_port`, which models
    /// only the primary endpoint. Multi-port workloads must bind their
    /// secondary container ports verbatim on loopback for those forwarders to
    /// reach them -- there is no per-secondary-port remap as on a real netns.
    async fn attach_overlay_ip(
        &self,
        id: &ContainerId,
        overlay_ip: std::net::IpAddr,
    ) -> Result<()> {
        use std::net::{IpAddr, Ipv4Addr, SocketAddr};

        let dir_name = Self::container_dir_name(id);
        let mut guard = self.containers.write().await;
        let Some(c) = guard.get_mut(&dir_name) else {
            return Ok(());
        };
        c.overlay_ip = Some(overlay_ip);

        // Forward each published port to the workload's local delivery. Dedup on
        // (container_port, protocol) like VZ-Linux -- `PortProtocol` is not
        // `Hash`, so key the dedup on its stable string form (TCP and UDP on the
        // same port number are distinct bindings).
        let assigned = c.assigned_port;
        let mut seen = std::collections::HashSet::new();
        let mut first = true;
        for pm in &c.spec.port_mappings {
            if !seen.insert((pm.container_port, pm.protocol.as_str())) {
                continue;
            }
            let bind = SocketAddr::new(overlay_ip, pm.container_port);
            let deliver_port = if first { assigned } else { pm.container_port };
            let target = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), deliver_port);
            c.forwarders
                .push(crate::runtimes::host_forward::spawn_overlay_forward(
                    bind,
                    target,
                    pm.protocol,
                ));
            first = false;
        }
        Ok(())
    }

    /// Tear down the host-proxy forwarders and forget the overlay `/32`.
    async fn detach_overlay_ip(&self, id: &ContainerId) -> Result<()> {
        let dir_name = Self::container_dir_name(id);
        let mut guard = self.containers.write().await;
        if let Some(c) = guard.get_mut(&dir_name) {
            for h in c.forwarders.drain(..) {
                h.abort();
            }
            c.overlay_ip = None;
        }
        Ok(())
    }

    /// Get the IP address of a container.
    ///
    /// On macOS, all sandboxed processes share the host network stack.
    /// Returns `127.0.0.1` (localhost) for all containers. Port-based
    /// differentiation is handled by the proxy manager.
    async fn get_container_ip(&self, id: &ContainerId) -> Result<Option<IpAddr>> {
        let dir_name = Self::container_dir_name(id);

        let containers = self.containers.read().await;
        let container = containers
            .get(&dir_name)
            .ok_or_else(|| AgentError::NotFound {
                container: dir_name,
                reason: "Container not found".to_string(),
            })?;

        // Prefer the overlay (`WireGuard`) `/32` once the container has been
        // attached to an overlay: cross-node mesh routing / DNS should resolve to
        // the overlay address, which the host-proxy forwarders make reachable.
        if let Some(ip) = container.overlay_ip {
            return Ok(Some(ip));
        }

        // No overlay attached: all sandboxed processes share the host network.
        // Return 127.0.0.1 -- the proxy manager routes traffic by port.
        Ok(Some(IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)))
    }

    /// Get the runtime-assigned port for a container.
    ///
    /// On macOS sandbox, each container is assigned a unique dynamic port
    /// (via OS port-0 binding) to avoid port conflicts between replicas
    /// sharing the host network. This port was passed to the process as
    /// `PORT` / `ZLAYER_PORT` environment variables.
    ///
    /// The proxy uses this port instead of the spec-declared endpoint port
    /// when constructing backend addresses, allowing multiple replicas to
    /// coexist on `127.0.0.1` with distinct ports.
    async fn get_container_port_override(&self, id: &ContainerId) -> Result<Option<u16>> {
        let dir_name = Self::container_dir_name(id);

        let containers = self.containers.read().await;
        let container = containers
            .get(&dir_name)
            .ok_or_else(|| AgentError::NotFound {
                container: dir_name,
                reason: "Container not found".to_string(),
            })?;

        Ok(Some(container.assigned_port))
    }

    /// Send a signal to the sandboxed process using `libc::kill`.
    #[allow(unsafe_code)]
    #[allow(clippy::cast_possible_wrap)]
    async fn kill_container(&self, id: &ContainerId, signal: Option<&str>) -> Result<()> {
        let canonical = crate::runtime::validate_signal(signal.unwrap_or("SIGKILL"))?;
        let dir_name = Self::container_dir_name(id);

        let pid = {
            let containers = self.containers.read().await;
            let container = containers
                .get(&dir_name)
                .ok_or_else(|| AgentError::NotFound {
                    container: dir_name.clone(),
                    reason: "Container not found".to_string(),
                })?;
            container.pid
        };

        if pid == 0 {
            return Err(AgentError::InvalidSpec(format!(
                "container '{dir_name}' is not running (no pid)"
            )));
        }

        let signum = match canonical.as_str() {
            "SIGKILL" => libc::SIGKILL,
            "SIGTERM" => libc::SIGTERM,
            "SIGINT" => libc::SIGINT,
            "SIGHUP" => libc::SIGHUP,
            "SIGUSR1" => libc::SIGUSR1,
            "SIGUSR2" => libc::SIGUSR2,
            other => {
                return Err(AgentError::InvalidSpec(format!(
                    "unsupported signal '{other}'"
                )));
            }
        };

        tracing::info!(container = %dir_name, pid = pid, signal = %canonical, "killing sandboxed process");

        let ret = unsafe { libc::kill(pid as i32, signum) };
        if ret != 0 {
            let err = std::io::Error::last_os_error();
            return Err(AgentError::Internal(format!(
                "kill({pid}, {canonical}) failed: {err}"
            )));
        }
        Ok(())
    }

    /// Tagging is not supported by the macOS sandbox runtime.
    ///
    /// Images are pulled directly into per-container rootfs directories rather
    /// than a shared content-addressed store, so there is nothing to tag.
    async fn tag_image(&self, _source: &str, _target: &str) -> Result<()> {
        Err(AgentError::Unsupported(
            "tag_image is not supported by the macOS sandbox runtime".into(),
        ))
    }
}

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

    /// Lay down a `.ready`-stamped keg (bin/ + a stub tool + `.ready`, no manifest
    /// so `build_handle_from_keg` synthesizes the handle) under `cache`, named the
    /// way `probe_ready_toolchain` searches: `{formula}-{ver}-{arch}`.
    async fn seed_ready_keg(cache: &Path, formula: &str, version: &str) {
        let arch = match std::env::consts::ARCH {
            "aarch64" => "arm64",
            other => other,
        };
        let keg = cache.join(format!("{formula}-{version}-{arch}"));
        tokio::fs::create_dir_all(keg.join("bin")).await.unwrap();
        tokio::fs::write(keg.join("bin").join("tool"), b"")
            .await
            .unwrap();
        tokio::fs::write(keg.join(".ready"), b"").await.unwrap();
    }

    /// `resolve_init_toolchain_handles(.., None)` attaches every INIT keg already
    /// `.ready` on disk (the warm path a fresh executor container hits), including
    /// the Node keg whose `bin` must land on the injected PATH set. Regression for
    /// node20 actions failing with `node: not found` when the keg IS provisioned.
    #[tokio::test]
    async fn resolve_init_toolchains_probes_ready_kegs_on_disk() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let cache = tmp.path();
        seed_ready_keg(cache, "git", "2.53.0").await;
        seed_ready_keg(cache, "node@lts", "24.18.0").await;

        let handles = resolve_init_toolchain_handles(cache, None).await;
        assert_eq!(handles.len(), 2, "both init kegs resolve from disk");

        let path_dirs: Vec<String> = handles.iter().flat_map(|h| h.path_dirs.clone()).collect();
        assert!(
            path_dirs.iter().any(|d| d.contains("node@lts-24.18.0")),
            "node keg bin must be on the injectable PATH set: {path_dirs:?}"
        );
    }

    /// The cold-first-job case: the Node keg finishes provisioning AFTER the call
    /// begins. With a bounded `wait`, `resolve_init_toolchain_handles` must keep
    /// re-probing and pick it up (rather than spawn the entrypoint Node-less).
    #[tokio::test]
    async fn resolve_init_toolchains_waits_for_keg_becoming_ready() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let cache = tmp.path().to_path_buf();
        // git is ready up front; node@lts lands mid-wait.
        seed_ready_keg(&cache, "git", "2.53.0").await;

        let cache_bg = cache.clone();
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(400)).await;
            seed_ready_keg(&cache_bg, "node@lts", "24.18.0").await;
        });

        let handles = resolve_init_toolchain_handles(&cache, Some(Duration::from_secs(5))).await;
        assert_eq!(
            handles.len(),
            2,
            "the Node keg that became ready during the wait must be picked up"
        );
    }

    /// Regression for `mkdir /workspace: read-only file system`: a `/workspace`
    /// named volume must rebase onto its host backing dir (NOT the host-absolute
    /// `/workspace`), and that dir must be both the exposed workspace and granted
    /// writable. Seatbelt has no mount namespace, so the host-absolute target
    /// would land on the read-only macOS root.
    #[test]
    fn resolve_sandbox_storage_rebases_workspace_volume_off_host_root() {
        let rootfs = PathBuf::from("/data/containers/svc-0/rootfs");
        let container_dir = PathBuf::from("/data/containers/svc-0");
        let backing = PathBuf::from("/data/volumes/ws");

        let mut spec = ServiceSpec::minimal("svc", "scratch:latest");
        spec.storage = vec![zlayer_spec::StorageSpec::Named {
            name: "ws".to_string(),
            target: "/workspace".to_string(),
            readonly: false,
            tier: zlayer_spec::StorageTier::Local,
            size: None,
        }];
        let mut volume_paths = std::collections::HashMap::new();
        volume_paths.insert("ws".to_string(), backing.clone());

        let storage = resolve_sandbox_storage(&spec, &rootfs, &container_dir, &volume_paths);

        // Workspace points at the writable backing dir, never the host root.
        assert_eq!(storage.workspace_host, backing);
        assert_ne!(storage.workspace_host, PathBuf::from("/workspace"));
        // The backing dir is granted writable; the bare host `/workspace` is not.
        assert!(storage.writable_dirs.contains(&backing));
        assert!(!storage.writable_dirs.contains(&PathBuf::from("/workspace")));
    }

    /// Regression for empty-PATH executors: a scratch image bakes no PATH, so
    /// the serialized `config.json` (= the spec) must gain the Seatbelt default
    /// PATH in `create_container`, otherwise the CI executor starts its steps
    /// with an empty PATH and can't find `env`/`bash`/`clang`. Set-if-absent:
    /// a user/image PATH must be left untouched.
    #[test]
    fn ensure_default_path_injects_when_absent_and_preserves_when_present() {
        // Empty env -> default PATH injected.
        let mut spec = ServiceSpec::minimal("svc", "scratch:latest");
        assert!(
            spec.env.is_empty(),
            "minimal spec should start with empty env"
        );
        ensure_default_path(&mut spec);
        assert_eq!(
            spec.env.get("PATH").map(String::as_str),
            Some(DEFAULT_SEATBELT_PATH),
            "default PATH must be persisted into the spec env"
        );

        // Pre-set non-empty PATH -> left unchanged (a user/image PATH wins).
        let mut spec = ServiceSpec::minimal("svc", "scratch:latest");
        spec.env.insert("PATH".to_string(), "/custom".to_string());
        ensure_default_path(&mut spec);
        assert_eq!(
            spec.env.get("PATH").map(String::as_str),
            Some("/custom"),
            "a user/image PATH must win over the default"
        );

        // Pre-set EMPTY PATH -> overwritten with the default (absent-or-empty),
        // so config.json never serializes an unusable `PATH=""`.
        let mut spec = ServiceSpec::minimal("svc", "scratch:latest");
        spec.env.insert("PATH".to_string(), String::new());
        ensure_default_path(&mut spec);
        assert_eq!(
            spec.env.get("PATH").map(String::as_str),
            Some(DEFAULT_SEATBELT_PATH),
            "an empty PATH must be replaced by the default"
        );
    }

    /// Spawn-side PATH resolution — the live process env (NOT the spec clone).
    /// The empirically-confirmed bug: `merge_env` yields a `PATH` key present but
    /// EMPTY (bare `PATH=` in a scratch image, or `PATH=""` in the spec), and the
    /// old set-if-absent guard skipped it, so `env_clear()` handed the child
    /// `PATH=""` and `exec env`/`exec bash` failed with "not found".
    #[test]
    fn ensure_spawn_path_injects_when_absent_or_empty_and_preserves_otherwise() {
        // (a) Absent -> default appended.
        let mut env_vars: Vec<(String, String)> = Vec::new();
        ensure_spawn_path(&mut env_vars);
        assert_eq!(
            env_vars
                .iter()
                .find(|(k, _)| k == "PATH")
                .map(|(_, v)| v.as_str()),
            Some(DEFAULT_SEATBELT_PATH),
            "absent PATH must be seeded with the default"
        );

        // (b) Present-but-EMPTY -> overwritten with the default (the real bug).
        let mut env_vars: Vec<(String, String)> = vec![
            ("FORGEJO_TASK_DATA".to_string(), "blob".to_string()),
            ("PATH".to_string(), String::new()),
            ("ZLAYER_ENDPOINT".to_string(), "unix:///x".to_string()),
        ];
        ensure_spawn_path(&mut env_vars);
        assert_eq!(
            env_vars
                .iter()
                .find(|(k, _)| k == "PATH")
                .map(|(_, v)| v.as_str()),
            Some(DEFAULT_SEATBELT_PATH),
            "empty PATH must be OVERWRITTEN with the default"
        );
        // No duplicate PATH key introduced.
        assert_eq!(
            env_vars.iter().filter(|(k, _)| k == "PATH").count(),
            1,
            "exactly one PATH entry must remain"
        );

        // (c) Present and NON-empty -> left unchanged (user/image PATH wins).
        let mut env_vars: Vec<(String, String)> = vec![("PATH".to_string(), "/custom".to_string())];
        ensure_spawn_path(&mut env_vars);
        assert_eq!(
            env_vars
                .iter()
                .find(|(k, _)| k == "PATH")
                .map(|(_, v)| v.as_str()),
            Some("/custom"),
            "a non-empty PATH must be left untouched"
        );
    }

    /// The `exec` env-build composition: `merge_env(spec, image)` then
    /// `ensure_spawn_path` must surface the image's OCI `config.Env` (a custom
    /// `PATH` plus an image-only var) instead of the daemon environment. Without
    /// this the Seatbelt exec path inherited the host env (the Fix-2 bug).
    #[test]
    fn exec_env_build_honors_image_config_path_and_vars() {
        let spec = ServiceSpec::minimal("svc", "scratch:latest");
        let image_config = zlayer_registry::ImageConfig {
            env: Some(vec![
                "PATH=/opt/toolbin:/usr/bin".to_string(),
                "IMAGE_ONLY=present".to_string(),
            ]),
            ..Default::default()
        };

        let mut env_base = crate::runtimes::macos_vz_shared::merge_env(&spec, Some(&image_config));
        ensure_spawn_path(&mut env_base);

        // The image's PATH wins — it is NOT overwritten by the spawn-time default.
        assert_eq!(
            env_base
                .iter()
                .find(|(k, _)| k == "PATH")
                .map(|(_, v)| v.as_str()),
            Some("/opt/toolbin:/usr/bin"),
            "the image's PATH must be preserved, not replaced by the default"
        );
        // Exactly one PATH entry (no duplicate seeded).
        assert_eq!(
            env_base.iter().filter(|(k, _)| k == "PATH").count(),
            1,
            "exactly one PATH entry must remain"
        );
        // The image-only var must reach the exec'd process.
        assert!(
            env_base.contains(&("IMAGE_ONLY".to_string(), "present".to_string())),
            "the image-only var must be carried into the exec env"
        );
    }

    /// `inject_toolchain_env` must PREPEND the keg's bin onto the (already
    /// non-empty, post-`ensure_spawn_path`) PATH so the provisioned git wins over
    /// host `/usr/bin/git`, and set the source-built handle's vars set-if-absent.
    ///
    /// The source-build redesign (`zlayer-toolchain::source_build`) ships ONLY
    /// `GIT_EXEC_PATH` — the built git has absolute system-library load commands
    /// (no `DYLD_FALLBACK_LIBRARY_PATH`) and its own compiled `<keg>/etc/gitconfig`
    /// (no `GIT_CONFIG_SYSTEM`). A job that pins `GIT_EXEC_PATH` keeps its value.
    #[test]
    fn inject_toolchain_env_prepends_path_and_sets_git_vars() {
        let keg_bin = "/data/host-toolchains/git-2.55.0-arm64/bin";
        let exec_path = "/data/host-toolchains/git-2.55.0-arm64/libexec/git-core";

        let mut env = HashMap::new();
        env.insert("GIT_EXEC_PATH".to_string(), exec_path.to_string());
        let handle = zlayer_toolchain::ToolchainHandle {
            install_dir: PathBuf::from("/data/host-toolchains/git-2.55.0-arm64"),
            path_dirs: vec![keg_bin.to_string()],
            env,
        };

        // A pre-existing non-empty PATH (the default) plus a job-pinned
        // GIT_EXEC_PATH that must be PRESERVED (set-if-absent semantics).
        let mut env_vars: Vec<(String, String)> = vec![
            ("PATH".to_string(), DEFAULT_SEATBELT_PATH.to_string()),
            (
                "GIT_EXEC_PATH".to_string(),
                "/job/pinned/git-core".to_string(),
            ),
        ];
        inject_toolchain_env(&mut env_vars, &handle);

        // PATH: keg bin prepended, original default preserved after it.
        let path = env_vars
            .iter()
            .find(|(k, _)| k == "PATH")
            .map(|(_, v)| v.as_str())
            .expect("PATH present");
        assert_eq!(
            path,
            format!("{keg_bin}:{DEFAULT_SEATBELT_PATH}"),
            "keg bin must be PREPENDED so the toolchain git is searched first"
        );
        assert_eq!(
            env_vars.iter().filter(|(k, _)| k == "PATH").count(),
            1,
            "exactly one PATH entry"
        );

        // The source-built handle carries NO DYLD / GIT_CONFIG_SYSTEM, so they
        // must never appear in the injected env.
        assert!(
            !env_vars
                .iter()
                .any(|(k, _)| k == "DYLD_FALLBACK_LIBRARY_PATH"),
            "source-built git must not inject DYLD_FALLBACK_LIBRARY_PATH"
        );
        assert!(
            !env_vars.iter().any(|(k, _)| k == "GIT_CONFIG_SYSTEM"),
            "source-built git must not inject GIT_CONFIG_SYSTEM"
        );

        // The job-pinned GIT_EXEC_PATH must NOT be overwritten (set-if-absent).
        assert_eq!(
            env_vars
                .iter()
                .find(|(k, _)| k == "GIT_EXEC_PATH")
                .map(|(_, v)| v.as_str()),
            Some("/job/pinned/git-core"),
            "a job-supplied GIT_EXEC_PATH must win over the toolchain's"
        );
        assert_eq!(
            env_vars
                .iter()
                .filter(|(k, _)| k == "GIT_EXEC_PATH")
                .count(),
            1,
            "no duplicate GIT_EXEC_PATH entry"
        );
    }

    /// `inject_home_env` must set `HOME` to the per-container home dir when it is
    /// absent (Seatbelt never seeded it → `git config --global` died with
    /// `fatal: $HOME not set`), and must leave a job-supplied `HOME` untouched
    /// (set-if-absent semantics, same idiom as `inject_ca_cert_env`).
    #[test]
    fn inject_home_env_sets_when_absent_and_preserves_existing() {
        let home_dir = Path::new("/data/containers/svc-0/home");

        // Absent: HOME must be seeded to the per-container home dir.
        let mut env_vars: Vec<(String, String)> = Vec::new();
        inject_home_env(&mut env_vars, home_dir);
        assert_eq!(
            env_vars
                .iter()
                .find(|(k, _)| k == "HOME")
                .map(|(_, v)| v.as_str()),
            Some("/data/containers/svc-0/home"),
            "HOME must be seeded to the granted per-container home dir when absent"
        );
        assert_eq!(
            env_vars.iter().filter(|(k, _)| k == "HOME").count(),
            1,
            "exactly one HOME entry"
        );

        // Present: a job-supplied HOME must win (set-if-absent).
        let mut env_vars: Vec<(String, String)> =
            vec![("HOME".to_string(), "/job/pinned/home".to_string())];
        inject_home_env(&mut env_vars, home_dir);
        assert_eq!(
            env_vars
                .iter()
                .find(|(k, _)| k == "HOME")
                .map(|(_, v)| v.as_str()),
            Some("/job/pinned/home"),
            "a job-supplied HOME must NOT be overwritten"
        );
        assert_eq!(
            env_vars.iter().filter(|(k, _)| k == "HOME").count(),
            1,
            "no duplicate HOME entry"
        );
    }

    /// A trivial in-memory [`zlayer_secrets::SecretsProvider`] for the env-resolution
    /// tests: it returns the single secret it was seeded with under one scope.
    struct FakeSecretsProvider {
        scope: String,
        name: String,
        value: String,
    }

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

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

        async fn list_secrets(
            &self,
            _scope: &str,
        ) -> std::result::Result<Vec<zlayer_secrets::SecretMetadata>, zlayer_secrets::SecretsError>
        {
            Ok(vec![])
        }

        async fn exists(
            &self,
            scope: &str,
            name: &str,
        ) -> std::result::Result<bool, zlayer_secrets::SecretsError> {
            Ok(scope == self.scope && name == self.name)
        }
    }

    /// Fix 3b: with an injected provider + a `secret_scope`, `resolve_env` expands
    /// a `$S:<name>` env value to the secret's plaintext (the Seatbelt path now
    /// resolves secrets exactly like the VZ-Linux guest).
    #[tokio::test]
    async fn resolve_env_expands_secret_ref_when_provider_and_scope_present() {
        let (rt, _tmp) = runtime();
        rt.set_secrets_provider(std::sync::Arc::new(FakeSecretsProvider {
            scope: "test-deployment".to_string(),
            name: "mysecret".to_string(),
            value: "hunter2".to_string(),
        }));

        let mut spec = ServiceSpec::minimal("svc", "scratch:latest");
        spec.secret_scope = Some(zlayer_secrets::SecretScope::Deployment(
            "test-deployment".to_string(),
        ));
        spec.env
            .insert("FOO".to_string(), "$S:mysecret".to_string());

        let resolved = rt.resolve_env(&spec).await.expect("resolve env");
        assert_eq!(
            resolved.get("FOO").map(String::as_str),
            Some("hunter2"),
            "$S: ref must resolve to the provider's secret value"
        );
    }

    /// Fix 3b no-scope fallback: with no `secret_scope`, `resolve_env` resolves
    /// `$E:` host-env refs but leaves any `$S:` value untouched (the
    /// `resolve_env_vars_with_warnings` semantics).
    #[tokio::test]
    async fn resolve_env_resolves_host_ref_and_leaves_secret_ref_when_no_scope() {
        let (rt, _tmp) = runtime();
        // No provider, no scope.
        let mut spec = ServiceSpec::minimal("svc", "scratch:latest");
        assert!(spec.secret_scope.is_none());

        // Use a host var we set ourselves so the lookup is deterministic.
        let host_key = "ZL_SANDBOX_TEST_HOME_VAR";
        std::env::set_var(host_key, "/home/zl-test");
        spec.env.insert("BAR".to_string(), format!("$E:{host_key}"));
        spec.env
            .insert("BAZ".to_string(), "$S:untouched".to_string());

        let resolved = rt.resolve_env(&spec).await.expect("resolve env");
        std::env::remove_var(host_key);

        assert_eq!(
            resolved.get("BAR").map(String::as_str),
            Some("/home/zl-test"),
            "$E: ref must resolve to the host env value"
        );
        assert_eq!(
            resolved.get("BAZ").map(String::as_str),
            Some("$S:untouched"),
            "$S: ref must be left untouched when no provider/scope is present"
        );
    }

    /// With no `/workspace` volume, the workload still gets a writable workspace
    /// inside the rootfs clone (so `mkdir`/checkout lands somewhere writable).
    #[test]
    fn resolve_sandbox_storage_defaults_workspace_into_rootfs_clone() {
        let rootfs = PathBuf::from("/data/containers/svc-0/rootfs");
        let container_dir = PathBuf::from("/data/containers/svc-0");
        let spec = ServiceSpec::minimal("svc", "scratch:latest");

        let storage = resolve_sandbox_storage(
            &spec,
            &rootfs,
            &container_dir,
            &std::collections::HashMap::new(),
        );

        assert_eq!(storage.workspace_host, rootfs.join("workspace"));
        assert!(storage.writable_dirs.contains(&rootfs.join("workspace")));
    }

    /// Build a `SandboxRuntime` rooted at a fresh temp `data_dir`.
    ///
    /// Returns `(runtime, tempdir)`; the [`tempfile::TempDir`] must be kept
    /// alive for the lifetime of the test so the directory is not removed out
    /// from under the runtime. `new()` pre-creates `images/` and `containers/`
    /// under the data dir.
    fn runtime() -> (SandboxRuntime, tempfile::TempDir) {
        let tmp = tempfile::tempdir().expect("create tempdir");
        let config = MacSandboxConfig {
            data_dir: tmp.path().join("data"),
            log_dir: tmp.path().join("logs"),
            gpu_access: false,
            keychain_access_allowed: false,
            overlay_cidr: None,
        };
        let rt = SandboxRuntime::new(config, None).expect("construct SandboxRuntime");
        (rt, tmp)
    }

    /// Minimal [`SandboxConfig`] for pure profile-generation assertions.
    fn profile_config(keychain_access: KeychainAccess) -> SandboxConfig {
        SandboxConfig {
            rootfs_dir: PathBuf::from("/tmp/zl-test/rootfs"),
            workspace_dir: PathBuf::from("/tmp/zl-test/workspace"),
            gpu_access: GpuAccess::None,
            keychain_access,
            network_access: NetworkAccess::None,
            writable_dirs: vec![],
            readonly_dirs: vec![],
            toolchain_cache: None,
            max_files: 4096,
            cpu_time_limit: None,
            memory_limit: None,
        }
    }

    /// `KeychainAccess::None` (the default) must NOT grant `securityd` or any
    /// keychain path — the base profile never names `SecurityServer`, so a
    /// container has zero keychain reach unless it explicitly opts in.
    #[test]
    fn keychain_denied_by_default() {
        let profile = generate_sandbox_profile(&profile_config(KeychainAccess::None));
        assert!(
            !profile.contains("com.apple.SecurityServer"),
            "default profile must not allow securityd:\n{profile}"
        );
        assert!(
            !profile.contains("Keychains"),
            "default profile must not open any keychain path:\n{profile}"
        );
    }

    /// `KeychainAccess::Enabled` must grant the `securityd` Mach service, the
    /// host code-signing CLIs under `/usr/bin`, and read/write of each requested
    /// keychain's directory.
    #[test]
    fn keychain_enabled_grants_securityd_and_paths() {
        let kc = PathBuf::from("/Users/zachary/Library/Keychains/zlayer-build.keychain-db");
        let profile = generate_sandbox_profile(&profile_config(KeychainAccess::Enabled {
            keychain_paths: vec![kc],
        }));
        assert!(
            profile.contains("(global-name \"com.apple.SecurityServer\")"),
            "enabled profile must allow securityd:\n{profile}"
        );
        assert!(
            profile.contains("(allow file-read* file-map-executable")
                && profile.contains("(subpath \"/usr/bin\")"),
            "enabled profile must allow exec of host /usr/bin (codesign/security):\n{profile}"
        );
        assert!(
            profile.contains("/Users/zachary/Library/Keychains"),
            "enabled profile must open the keychain directory read/write:\n{profile}"
        );
    }

    /// Regression for `fatal: unable to access '/etc/gitconfig': Operation not
    /// permitted` (git exit 128, which broke `actions/checkout` + `go build` VCS
    /// stamping on every non-container Mac job): the sandbox must allow reading
    /// the system config dir. macOS resolves `/etc` -> `/private/etc`, so the
    /// grant names the real path.
    #[test]
    fn allows_reading_system_config_dir_for_git() {
        let profile = generate_sandbox_profile(&profile_config(KeychainAccess::None));
        assert!(
            profile.contains("(subpath \"/private/etc\")"),
            "profile must allow file-read of /private/etc so git can read /etc/gitconfig:\n{profile}"
        );
    }

    /// When a `git` keg was provisioned, the profile must grant `file-read*` AND
    /// `file-map-executable` on the toolchain-cache subpath (the keg lives OUTSIDE
    /// the rootfs clone, so neither the rootfs grant nor the `/opt/homebrew` grant
    /// covers it), plus `file-read-metadata` on its ancestors so the keg path is
    /// traversable. With no keg (`toolchain_cache: None`) the grant is omitted.
    #[test]
    fn grants_read_and_exec_on_toolchain_cache_when_provisioned() {
        // Absent by default — no grant emitted.
        let none = generate_sandbox_profile(&profile_config(KeychainAccess::None));
        assert!(
            !none.contains("Provisioned toolchain cache"),
            "no toolchain grant should appear when none was provisioned:\n{none}"
        );

        // Provisioned — read + exec-map on the cache subpath, ancestors stat-able.
        let mut config = profile_config(KeychainAccess::None);
        let tc = PathBuf::from("/Users/zachary/.zlayer/toolchains");
        config.toolchain_cache = Some(tc.clone());
        let profile = generate_sandbox_profile(&config);

        assert!(
            profile.contains(&format!(
                "(allow file-read* file-map-executable (subpath \"{}\"))",
                tc.display()
            )),
            "the toolchain cache must be granted file-read* + file-map-executable:\n{profile}"
        );
        // Ancestors must be lstat/traverse-able so the keg path resolves.
        for ancestor in ["/", "/Users", "/Users/zachary", "/Users/zachary/.zlayer"] {
            assert!(
                profile.contains(&format!(
                    "(allow file-read-metadata (literal \"{ancestor}\"))"
                )),
                "ancestor `{ancestor}` of the toolchain cache must be lstat-able:\n{profile}"
            );
        }
    }

    /// LIVE Seatbelt exec proof (the bar the Homebrew-bottle git path failed):
    /// build the `git` keg FROM SOURCE, then run `<keg>/bin/git --version` under
    /// the DAEMON'S OWN `generate_sandbox_profile` output (the genuine
    /// deny-default whitelist, with the keg granted via `toolchain_cache`). The
    /// bottle git aborted here (`dyld: Symbol not found … Abort trap: 6`) because
    /// its relocated `@@HOMEBREW_PREFIX@@` `LC_LOAD_DYLIB` paths don't exist and
    /// macOS strips `DYLD_*` from the signed binary; the source-built git has
    /// absolute system-library load commands, so it must exit 0 with a version.
    ///
    /// `#[ignore]` because it builds git from source (network + CLT) and only
    /// works on macOS. Run with:
    ///   `cargo test -p zlayer-agent --lib source_built_git_runs_under_real_seatbelt_profile -- --ignored --nocapture`
    #[tokio::test]
    #[ignore = "live: builds git from source + runs it under a real Seatbelt profile (macOS + CLT)"]
    async fn source_built_git_runs_under_real_seatbelt_profile() {
        let cache = std::path::PathBuf::from(std::env::var("HOME").unwrap())
            .join(".zlayer/host-toolchains");
        let handle = zlayer_toolchain::ensure_toolchain(
            "git",
            zlayer_toolchain::ToolPlatform::MacOS,
            &cache,
            None,
        )
        .await
        .expect("git toolchain should build from source");

        let git_bin = handle.install_dir.join("bin/git");
        assert!(git_bin.exists(), "keg git must exist at <keg>/bin/git");

        // Generate the REAL container profile, granting the keg cache exactly as
        // `create_container` does (`toolchain_cache = Some(<host cache>)`).
        let mut config = profile_config(KeychainAccess::None);
        config.toolchain_cache = Some(cache.clone());
        let profile = generate_sandbox_profile(&config);

        let profile_path =
            std::env::temp_dir().join(format!("zl-srcgit-seatbelt-{}.sb", std::process::id()));
        std::fs::write(&profile_path, &profile).expect("write profile");

        // Run git under the deny-default Seatbelt profile, with the keg's
        // GIT_EXEC_PATH from the handle (no DYLD / GIT_CONFIG_SYSTEM).
        let mut cmd = std::process::Command::new("/usr/bin/sandbox-exec");
        cmd.arg("-f")
            .arg(&profile_path)
            .arg(&git_bin)
            .arg("--version");
        for (k, v) in &handle.env {
            cmd.env(k, v);
        }
        // Point HOME at the granted keg dir so git's `$HOME/.gitconfig` lookup
        // resolves to a non-existent (granted) path instead of EPERM-ing on the
        // host `/Users/<me>/.gitconfig`. In a real container HOME lives inside
        // the granted rootfs; this minimal profile_config grants only the keg.
        cmd.env("HOME", &handle.install_dir);
        cmd.env("GIT_CONFIG_GLOBAL", handle.install_dir.join(".gitconfig"));
        let out = cmd.output().expect("run sandbox-exec git --version");
        let _ = std::fs::remove_file(&profile_path);

        let stdout = String::from_utf8_lossy(&out.stdout);
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            out.status.success(),
            "keg git must EXECUTE under the real Seatbelt profile (NOT Abort trap:6).\n\
             status={:?}\nstdout={stdout}\nstderr={stderr}",
            out.status
        );
        assert!(
            stdout.contains("git version"),
            "expected a git version line, got stdout={stdout} stderr={stderr}"
        );
    }

    /// LIVE Seatbelt exec proof that the resolver GENERALIZES beyond git: build
    /// the `jq` keg FROM SOURCE (autotools + builtin oniguruma) and run
    /// `<keg>/bin/jq --version` under the daemon's real `generate_sandbox_profile`
    /// output. Also asserts `otool -L` of the built binary references NO
    /// `@@HOMEBREW@@` (the bottle-relocation failure mode). jq's handle carries
    /// no extra env, so this exercises the generic (non-git-special) path.
    ///
    /// `#[ignore]` because it builds jq from source (network + CLT), macOS only:
    ///   `cargo test -p zlayer-agent --lib source_built_jq_runs_under_real_seatbelt_profile -- --ignored --nocapture`
    #[tokio::test]
    #[ignore = "live: builds jq from source + runs it under a real Seatbelt profile (macOS + CLT)"]
    async fn source_built_jq_runs_under_real_seatbelt_profile() {
        let cache = std::path::PathBuf::from(std::env::var("HOME").unwrap())
            .join(".zlayer/host-toolchains");
        let handle = zlayer_toolchain::ensure_toolchain(
            "jq",
            zlayer_toolchain::ToolPlatform::MacOS,
            &cache,
            None,
        )
        .await
        .expect("jq toolchain should build from source");

        let jq_bin = handle.install_dir.join("bin/jq");
        assert!(jq_bin.exists(), "keg jq must exist at <keg>/bin/jq");

        // otool -L must show NO @@HOMEBREW@@ in the built binary's load commands.
        let otool = std::process::Command::new("otool")
            .arg("-L")
            .arg(&jq_bin)
            .output()
            .expect("run otool -L on keg jq");
        let libs = String::from_utf8_lossy(&otool.stdout);
        assert!(
            !libs.contains("@@HOMEBREW"),
            "source-built jq must have NO @@HOMEBREW@@ load commands; otool -L:\n{libs}"
        );

        let mut config = profile_config(KeychainAccess::None);
        config.toolchain_cache = Some(cache.clone());
        let profile = generate_sandbox_profile(&config);

        let profile_path =
            std::env::temp_dir().join(format!("zl-srcjq-seatbelt-{}.sb", std::process::id()));
        std::fs::write(&profile_path, &profile).expect("write profile");

        let mut cmd = std::process::Command::new("/usr/bin/sandbox-exec");
        cmd.arg("-f")
            .arg(&profile_path)
            .arg(&jq_bin)
            .arg("--version");
        for (k, v) in &handle.env {
            cmd.env(k, v);
        }
        cmd.env("HOME", &handle.install_dir);
        let out = cmd.output().expect("run sandbox-exec jq --version");
        let _ = std::fs::remove_file(&profile_path);

        let stdout = String::from_utf8_lossy(&out.stdout);
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            out.status.success(),
            "keg jq must EXECUTE under the real Seatbelt profile (NOT Abort trap:6).\n\
             status={:?}\nstdout={stdout}\nstderr={stderr}",
            out.status
        );
        assert!(
            stdout.contains("jq-") || stderr.contains("jq-"),
            "expected a jq version line, got stdout={stdout} stderr={stderr}"
        );
    }

    /// LIVE Seatbelt exec proof for the third previously-hardcoded target,
    /// `cmake`: build the `cmake` keg FROM SOURCE through the now-generic
    /// resolver (a self-hosting `CMake` — `bootstrap` + `CMakeLists.txt` — so it
    /// takes the `CMakeBootstrap` build-system arm, NO formula==cmake code) and
    /// run `<keg>/bin/cmake --version` under the daemon's real
    /// `generate_sandbox_profile` output. Asserts `otool -L` shows NO
    /// `@@HOMEBREW@@` and the binary execs exit-0 (NOT `Abort trap: 6`).
    ///
    /// `#[ignore]` because it bootstraps + compiles cmake from source (network +
    /// CLT, a long build), macOS only:
    ///   `cargo test -p zlayer-agent --lib source_built_cmake_runs_under_real_seatbelt_profile -- --ignored --nocapture`
    #[tokio::test]
    #[ignore = "live: builds cmake from source + runs it under a real Seatbelt profile (macOS + CLT)"]
    async fn source_built_cmake_runs_under_real_seatbelt_profile() {
        let cache = std::path::PathBuf::from(std::env::var("HOME").unwrap())
            .join(".zlayer/host-toolchains");
        let handle = zlayer_toolchain::ensure_toolchain(
            "cmake",
            zlayer_toolchain::ToolPlatform::MacOS,
            &cache,
            None,
        )
        .await
        .expect("cmake toolchain should build from source");

        let cmake_bin = handle.install_dir.join("bin/cmake");
        assert!(
            cmake_bin.exists(),
            "keg cmake must exist at <keg>/bin/cmake"
        );

        // otool -L must show NO @@HOMEBREW@@ in the built binary's load commands.
        let otool = std::process::Command::new("otool")
            .arg("-L")
            .arg(&cmake_bin)
            .output()
            .expect("run otool -L on keg cmake");
        let libs = String::from_utf8_lossy(&otool.stdout);
        assert!(
            !libs.contains("@@HOMEBREW"),
            "source-built cmake must have NO @@HOMEBREW@@ load commands; otool -L:\n{libs}"
        );

        let mut config = profile_config(KeychainAccess::None);
        config.toolchain_cache = Some(cache.clone());
        let profile = generate_sandbox_profile(&config);

        let profile_path =
            std::env::temp_dir().join(format!("zl-srccmake-seatbelt-{}.sb", std::process::id()));
        std::fs::write(&profile_path, &profile).expect("write profile");

        let mut cmd = std::process::Command::new("/usr/bin/sandbox-exec");
        cmd.arg("-f")
            .arg(&profile_path)
            .arg(&cmake_bin)
            .arg("--version");
        for (k, v) in &handle.env {
            cmd.env(k, v);
        }
        cmd.env("HOME", &handle.install_dir);
        // cmake calls `getcwd()` at startup and HARD-FAILS ("Current working
        // directory cannot be established.", exit 1) if the cwd is not readable
        // under the deny-default sandbox — unlike git/jq, which tolerate a denied
        // getcwd. The daemon always runs the sandboxed process with cwd INSIDE the
        // granted tree (`.current_dir(&rootfs_clone)` at the real exec site), so
        // this never bites in production. Mirror that here: run from the keg dir
        // (under `toolchain_cache`, hence granted) instead of inheriting cargo's
        // ungranted repo cwd — exactly like the clone proof sets `current_dir`.
        cmd.current_dir(&handle.install_dir);
        let out = cmd.output().expect("run sandbox-exec cmake --version");
        let _ = std::fs::remove_file(&profile_path);

        let stdout = String::from_utf8_lossy(&out.stdout);
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            out.status.success(),
            "keg cmake must EXECUTE under the real Seatbelt profile (NOT Abort trap:6).\n\
             status={:?}\nstdout={stdout}\nstderr={stderr}",
            out.status
        );
        assert!(
            stdout.contains("cmake version") || stderr.contains("cmake version"),
            "expected a cmake version line, got stdout={stdout} stderr={stderr}"
        );
    }

    /// LIVE Seatbelt exec proof for the **`node@lts` init toolchain**: provision
    /// the newest Node.js LTS keg through the relocation-free prebuilt fetcher
    /// (version resolved LIVE from nodejs.org — today v24.x "Krypton") and run
    /// `<keg>/bin/node --version` + `node -e "console.log(process.version)"` under
    /// the daemon's REAL `generate_sandbox_profile` output (the same profile a
    /// container gets). This is the runtime node20 JS actions (`setup-node`,
    /// `setup-go`, …) execute with inside the deny-default Seatbelt sandbox, which
    /// ships no `node`. Asserts:
    ///   * `otool -L` of the vendor node references NO `@@HOMEBREW@@`,
    ///   * `node --version` execs exit-0 (NOT `Abort trap: 6`) and prints the SAME
    ///     version the resolver picked, an even (LTS) major ≥ 20 — i.e. v24.x,
    ///   * `node -e ...` runs JS in-sandbox and prints `process.version`.
    ///
    /// `#[ignore]` because it downloads the Node LTS archive (network), macOS only:
    ///   `cargo test -p zlayer-agent --lib node_lts_runs_under_real_seatbelt_profile -- --ignored --nocapture`
    #[tokio::test]
    #[ignore = "live: downloads the Node LTS keg + runs node under a real Seatbelt profile (macOS + network)"]
    async fn node_lts_runs_under_real_seatbelt_profile() {
        let cache = std::path::PathBuf::from(std::env::var("HOME").unwrap())
            .join(".zlayer/host-toolchains");
        let handle = zlayer_toolchain::ensure_toolchain(
            "node@lts",
            zlayer_toolchain::ToolPlatform::MacOS,
            &cache,
            None,
        )
        .await
        .expect("node@lts toolchain should provision via the prebuilt fetcher");

        let node_bin = handle.install_dir.join("bin/node");
        assert!(node_bin.exists(), "keg node must exist at <keg>/bin/node");

        // The keg dir is `node@lts-<ver>-<arch>`: the resolved LTS version the
        // running node must report back. (Live, this is a v24.x string.)
        let keg_name = handle
            .install_dir
            .file_name()
            .unwrap()
            .to_string_lossy()
            .into_owned();
        let expected_ver = keg_name
            .strip_prefix("node@lts-")
            .and_then(|s| s.rsplit_once('-'))
            .map(|(v, _arch)| v.to_string())
            .expect("keg name has node@lts-<ver>-<arch> shape");
        let major: u64 = expected_ver
            .split('.')
            .next()
            .and_then(|m| m.parse().ok())
            .expect("version has a numeric major");
        assert!(
            major >= 20 && major % 2 == 0,
            "resolved LTS major must be an even line >= 20 (LTS), got {major} ({expected_ver})"
        );

        // otool -L must show NO @@HOMEBREW@@ in the vendor binary's load commands.
        let otool = std::process::Command::new("otool")
            .arg("-L")
            .arg(&node_bin)
            .output()
            .expect("run otool -L on keg node");
        let libs = String::from_utf8_lossy(&otool.stdout);
        assert!(
            !libs.contains("@@HOMEBREW"),
            "vendor node must have NO @@HOMEBREW@@ load commands; otool -L:\n{libs}"
        );

        let mut config = profile_config(KeychainAccess::None);
        config.toolchain_cache = Some(cache.clone());
        let profile = generate_sandbox_profile(&config);

        let profile_path =
            std::env::temp_dir().join(format!("zl-node-seatbelt-{}.sb", std::process::id()));
        std::fs::write(&profile_path, &profile).expect("write profile");

        // Helper: run an in-sandbox node invocation, returning (status, stdout, stderr).
        let run_node = |extra: &[&str]| {
            let mut cmd = std::process::Command::new("/usr/bin/sandbox-exec");
            cmd.arg("-f").arg(&profile_path).arg(&node_bin);
            for a in extra {
                cmd.arg(a);
            }
            for (k, v) in &handle.env {
                cmd.env(k, v);
            }
            cmd.env("HOME", &handle.install_dir);
            // node calls getcwd() at startup; run from the (granted) keg dir so a
            // denied repo cwd doesn't make it bail, mirroring the real exec site's
            // `.current_dir(&rootfs)`.
            cmd.current_dir(&handle.install_dir);
            let out = cmd.output().expect("run sandbox-exec node");
            (
                out.status,
                String::from_utf8_lossy(&out.stdout).into_owned(),
                String::from_utf8_lossy(&out.stderr).into_owned(),
            )
        };

        // 1. node --version → exit 0, prints exactly the resolved LTS version.
        let (status, stdout, stderr) = run_node(&["--version"]);
        assert!(
            status.success(),
            "keg node must EXECUTE under the real Seatbelt profile (NOT Abort trap:6).\n\
             status={status:?}\nstdout={stdout}\nstderr={stderr}"
        );
        assert_eq!(
            stdout.trim(),
            format!("v{expected_ver}"),
            "in-sandbox node --version must equal the resolved LTS version"
        );

        // 2. node -e "console.log(process.version)" → runs JS in-sandbox.
        let (status2, stdout2, stderr2) = run_node(&["-e", "console.log(process.version)"]);
        let _ = std::fs::remove_file(&profile_path);
        assert!(
            status2.success(),
            "in-sandbox `node -e` must execute JS exit-0.\n\
             status={status2:?}\nstdout={stdout2}\nstderr={stderr2}"
        );
        assert_eq!(
            stdout2.trim(),
            format!("v{expected_ver}"),
            "in-sandbox node -e process.version must equal the resolved LTS version"
        );
    }

    /// LIVE Seatbelt exec proof of the **brew-emulate fallback** (`T4`): provision
    /// a formula the generic source-build runner CANNOT build — `gron` (a Go
    /// `cargo install`-style formula whose tarball has only `go.mod`/`main.go`,
    /// no `configure`/`CMakeLists.txt`/`Makefile`, so generic build-system
    /// detection deterministically FAILS) — and run its binary under the
    /// daemon's REAL `generate_sandbox_profile` Seatbelt sandbox.
    ///
    /// `ensure_toolchain("gron")` therefore falls through
    /// `source_build::ensure_from_source` → `brew_emulate::ensure_via_brew`,
    /// which runs gron's actual Homebrew recipe (`go build`) with Homebrew
    /// installed AT the keg prefix (`<keg>/brew`) and `--build-from-source`. The
    /// proof asserts:
    ///   * the keg came from brew-emulate (a `<keg>/brew` prefix exists — the
    ///     generic path never creates one),
    ///   * `otool -L` of the built binary shows NO `@@HOMEBREW@@` load command
    ///     (the bottle-relocation failure mode the custom prefix avoids),
    ///   * the binary EXECUTES exit-0 under the deny-default Seatbelt profile
    ///     (NOT `Abort trap: 6`).
    ///
    /// `#[ignore]` because it clones Homebrew + builds gron from source (network
    /// + `go` build-dep, macOS + CLT only):
    ///   `cargo test -p zlayer-agent --lib brew_emulate_gron_runs_under_real_seatbelt_profile -- --ignored --nocapture`
    #[tokio::test]
    #[ignore = "live: brew-emulate builds gron from source + runs it under a real Seatbelt profile (macOS + CLT + network)"]
    async fn brew_emulate_gron_runs_under_real_seatbelt_profile() {
        let cache = std::path::PathBuf::from(std::env::var("HOME").unwrap())
            .join(".zlayer/host-toolchains");
        let handle = zlayer_toolchain::ensure_toolchain(
            "gron",
            zlayer_toolchain::ToolPlatform::MacOS,
            &cache,
            None,
        )
        .await
        .expect("gron should provision via the brew-emulate fallback");

        // The keg must have come from brew-emulate: a `<keg>/brew` Homebrew
        // prefix exists (the generic source-build path never creates one).
        let brew_prefix = handle.install_dir.join("brew");
        assert!(
            brew_prefix.is_dir(),
            "gron keg must be brew-emulate-built (expected a <keg>/brew prefix at {})",
            brew_prefix.display()
        );

        // Locate the gron binary via the handle's first path dir (opt/gron/bin).
        let gron_bin = std::path::Path::new(&handle.path_dirs[0]).join("gron");
        assert!(
            gron_bin.exists(),
            "keg gron must exist at {}",
            gron_bin.display()
        );

        // otool -L must show NO @@HOMEBREW@@ in the built binary's load commands.
        let otool = std::process::Command::new("otool")
            .arg("-L")
            .arg(&gron_bin)
            .output()
            .expect("run otool -L on keg gron");
        let libs = String::from_utf8_lossy(&otool.stdout);
        assert!(
            !libs.contains("@@HOMEBREW"),
            "brew-emulate gron must have NO @@HOMEBREW@@ load commands; otool -L:\n{libs}"
        );

        let mut config = profile_config(KeychainAccess::None);
        config.toolchain_cache = Some(cache.clone());
        let profile = generate_sandbox_profile(&config);

        let profile_path =
            std::env::temp_dir().join(format!("zl-brewgron-seatbelt-{}.sb", std::process::id()));
        std::fs::write(&profile_path, &profile).expect("write profile");

        let mut cmd = std::process::Command::new("/usr/bin/sandbox-exec");
        cmd.arg("-f")
            .arg(&profile_path)
            .arg(&gron_bin)
            .arg("--version");
        for (k, v) in &handle.env {
            cmd.env(k, v);
        }
        cmd.env("HOME", &handle.install_dir);
        let out = cmd.output().expect("run sandbox-exec gron --version");
        let _ = std::fs::remove_file(&profile_path);

        let stdout = String::from_utf8_lossy(&out.stdout);
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            out.status.success(),
            "brew-emulate gron must EXECUTE under the real Seatbelt profile (NOT Abort trap:6).\n\
             status={:?}\nstdout={stdout}\nstderr={stderr}",
            out.status
        );
        assert!(
            !stdout.trim().is_empty() || !stderr.trim().is_empty(),
            "expected gron --version to print something"
        );
    }

    /// LIVE `actions/checkout`-style proof: build the `git` keg FROM SOURCE and
    /// `git clone --depth 1` a small public repo INSIDE the daemon's real
    /// `generate_sandbox_profile` Seatbelt sandbox (network `Full`, keg cache
    /// granted, a writable clone dir). Exercises git's full https transport +
    /// `libexec/git-core` helpers under the deny-default profile — the path a CI
    /// checkout step runs. Must exit 0 and leave a populated `<dest>/.git`.
    ///
    /// `#[ignore]` (network + CLT + a real clone), macOS only:
    ///   `cargo test -p zlayer-agent --lib keg_git_clones_small_repo_under_real_seatbelt_profile -- --ignored --nocapture`
    #[tokio::test]
    #[ignore = "live: builds git from source + clones a repo under a real Seatbelt profile (macOS + CLT + network)"]
    async fn keg_git_clones_small_repo_under_real_seatbelt_profile() {
        let cache = std::path::PathBuf::from(std::env::var("HOME").unwrap())
            .join(".zlayer/host-toolchains");
        let handle = zlayer_toolchain::ensure_toolchain(
            "git",
            zlayer_toolchain::ToolPlatform::MacOS,
            &cache,
            None,
        )
        .await
        .expect("git toolchain should build from source");
        let git_bin = handle.install_dir.join("bin/git");

        // Writable scratch for the clone destination + git's HOME. Canonicalize
        // it (`/var/folders/...` -> `/private/var/folders/...`) so the Seatbelt
        // writable grant — which sees the resolved path — matches what git writes.
        let scratch_raw = std::env::temp_dir().join(format!("zl-keg-clone-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&scratch_raw);
        std::fs::create_dir_all(&scratch_raw).expect("create scratch");
        let scratch = std::fs::canonicalize(&scratch_raw).expect("canonicalize scratch");
        let dest = scratch.join("Hello-World");

        // Real container profile: network Full + keg cache granted + scratch writable.
        let mut config = profile_config(KeychainAccess::None);
        config.toolchain_cache = Some(cache.clone());
        config.network_access = NetworkAccess::Full;
        config.writable_dirs = vec![scratch.clone()];
        let profile = generate_sandbox_profile(&config);
        let profile_path =
            std::env::temp_dir().join(format!("zl-kegclone-seatbelt-{}.sb", std::process::id()));
        std::fs::write(&profile_path, &profile).expect("write profile");

        let mut cmd = std::process::Command::new("/usr/bin/sandbox-exec");
        cmd.arg("-f")
            .arg(&profile_path)
            .arg(&git_bin)
            .arg("clone")
            .arg("--depth")
            .arg("1")
            .arg("https://github.com/octocat/Hello-World.git")
            .arg(&dest);
        // Run inside the granted scratch dir — a deny-default profile EPERMs
        // `getcwd()`/`stat(.)` on an ungranted CWD (git's remote helpers read it).
        cmd.current_dir(&scratch);
        for (k, v) in &handle.env {
            cmd.env(k, v);
        }
        cmd.env("HOME", &scratch);
        cmd.env("GIT_CONFIG_GLOBAL", scratch.join(".gitconfig"));
        cmd.env("GIT_TERMINAL_PROMPT", "0");
        // CA bundle is supplied by PRODUCTION: drive the env exactly as the
        // runtime does (`inject_ca_cert_env`) instead of hand-setting
        // `GIT_SSL_CAINFO` here — so a green clone proves the production
        // injection is correct, not just this test's local override.
        let mut prod_env: Vec<(String, String)> = Vec::new();
        inject_ca_cert_env(&mut prod_env);
        for (k, v) in &prod_env {
            cmd.env(k, v);
        }
        let out = cmd.output().expect("run sandbox-exec git clone");
        let _ = std::fs::remove_file(&profile_path);

        let stdout = String::from_utf8_lossy(&out.stdout);
        let stderr = String::from_utf8_lossy(&out.stderr);
        let cloned_ok = dest.join(".git").is_dir();
        let _ = std::fs::remove_dir_all(&scratch);

        assert!(
            out.status.success() && cloned_ok,
            "keg git must CLONE a repo under the real Seatbelt profile.\n\
             status={:?}\ncloned_ok={cloned_ok}\nstdout={stdout}\nstderr={stderr}",
            out.status
        );
    }

    /// Regression for `mkdir /Users: file exists`: every writable subpath grant
    /// must be accompanied by `file-read-metadata` `literal` rules for each of its
    /// ancestor directories, so a sandboxed `os.MkdirAll(workspace)` can `lstat`/
    /// traverse every parent from `/` down. Without these, the ancestor `lstat`s
    /// are denied, the parent looks "missing", and `MkdirAll` dies trying to
    /// `mkdir /Users`.
    #[test]
    fn writable_subpath_grants_ancestor_metadata() {
        let mut config = profile_config(KeychainAccess::None);
        let ws = PathBuf::from("/Users/zachary/.zlayer/volumes/ws-1");
        config.writable_dirs = vec![ws.clone()];
        let profile = generate_sandbox_profile(&config);

        // The writable subpath itself is granted read/write.
        assert!(
            profile.contains(&format!(
                "(allow file-read* file-write* (subpath \"{}\"))",
                ws.display()
            )),
            "writable volume dir must be granted read/write:\n{profile}"
        );

        // Every ancestor of the writable dir, up to and including `/`, must be
        // stat/traverse-able via a tight `file-read-metadata` `literal` rule.
        for ancestor in [
            "/",
            "/Users",
            "/Users/zachary",
            "/Users/zachary/.zlayer",
            "/Users/zachary/.zlayer/volumes",
        ] {
            assert!(
                profile.contains(&format!(
                    "(allow file-read-metadata (literal \"{ancestor}\"))"
                )),
                "ancestor `{ancestor}` of a writable subpath must be lstat-able:\n{profile}"
            );
        }

        // The granted dir itself must NOT be re-emitted as a metadata literal
        // (it already has the broader subpath grant); only its ancestors are.
        assert!(
            !profile.contains(&format!(
                "(allow file-read-metadata (literal \"{}\"))",
                ws.display()
            )),
            "the writable dir itself should not get a redundant metadata literal:\n{profile}"
        );
    }

    /// `keychain_paths_from_label` maps the `zlayer.io/keychain` label grammar to
    /// the expected keychain database paths, dedupes, and drops unrecognized
    /// tokens. Pure (no env mutation) by construction.
    #[test]
    fn keychain_label_grammar() {
        let dir = "/Users/tester/Library/Keychains";

        assert_eq!(
            keychain_paths_from_label("login", dir),
            vec![PathBuf::from(
                "/Users/tester/Library/Keychains/login.keychain-db"
            )]
        );
        assert_eq!(
            keychain_paths_from_label("build:ci", dir),
            vec![PathBuf::from(
                "/Users/tester/Library/Keychains/ci.keychain-db"
            )]
        );
        // Default build keychain via keyword aliases.
        assert_eq!(
            keychain_paths_from_label("true", dir),
            vec![PathBuf::from(
                "/Users/tester/Library/Keychains/zlayer-build.keychain-db"
            )]
        );
        // Absolute path used verbatim.
        assert_eq!(
            keychain_paths_from_label("/opt/keys/ci.keychain-db", dir),
            vec![PathBuf::from("/opt/keys/ci.keychain-db")]
        );
        // Union of keywords + dedupe (login appears twice); unknown token dropped.
        assert_eq!(
            keychain_paths_from_label("login, build, login, bogus", dir),
            vec![
                PathBuf::from("/Users/tester/Library/Keychains/login.keychain-db"),
                PathBuf::from("/Users/tester/Library/Keychains/zlayer-build.keychain-db"),
            ]
        );
        // Empty / whitespace-only / unrecognized yields nothing (caller -> None).
        assert!(keychain_paths_from_label("  ", dir).is_empty());
        assert!(keychain_paths_from_label("nope", dir).is_empty());
    }

    /// Create `{images}/{dir}/rootfs/{file_name}` with `contents`.
    fn write_image_file(rt: &SandboxRuntime, dir: &str, file_name: &str, contents: &[u8]) {
        let rootfs = rt.images_dir().join(dir).join("rootfs");
        std::fs::create_dir_all(&rootfs).expect("create rootfs dir");
        std::fs::write(rootfs.join(file_name), contents).expect("write image file");
    }

    /// Write `{images}/{dir}/ref` containing `reference`.
    fn write_ref_file(rt: &SandboxRuntime, dir: &str, reference: &str) {
        let image_dir = rt.images_dir().join(dir);
        std::fs::create_dir_all(&image_dir).expect("create image dir");
        std::fs::write(image_dir.join("ref"), reference).expect("write ref file");
    }

    /// Write `{containers}/{name}/config.json` holding a minimal [`ServiceSpec`]
    /// whose image sanitizes to `image` (chosen so `sanitize == raw`).
    ///
    /// Serialized exactly like `create_container` does
    /// (`serde_json::to_string_pretty`), so the byte content matches what the
    /// daemon writes and what `referenced_image_dirs` reads back.
    fn write_container_config(rt: &SandboxRuntime, name: &str, image: &str) {
        let spec = ServiceSpec::minimal(name, image);
        // The image must round-trip to its raw form so the sanitized directory
        // name equals `image`. A bare lowercase name has no '/', ':', or '@'.
        assert_eq!(
            sanitize_image_name(&spec.image.name.to_string()),
            image,
            "test image must sanitize to its raw directory name"
        );
        let dir = rt.config.data_dir.join("containers").join(name);
        std::fs::create_dir_all(&dir).expect("create container dir");
        let json = serde_json::to_string_pretty(&spec).expect("serialize spec");
        std::fs::write(dir.join("config.json"), json).expect("write config.json");
    }

    #[tokio::test]
    async fn prune_removes_unreferenced_image_dirs() {
        let (rt, _tmp) = runtime();

        // imgA is referenced by an on-disk container; imgB is dangling.
        let imgb_contents = b"imgB-rootfs-bytes";
        write_image_file(&rt, "imga", "file", b"imgA-rootfs-bytes");
        write_image_file(&rt, "imgb", "file", imgb_contents);
        write_container_config(&rt, "svc-0", "imga");

        let result = rt.prune_images().await.expect("prune succeeds");

        assert_eq!(result.deleted, vec!["imgb".to_string()]);
        assert_eq!(result.space_reclaimed, imgb_contents.len() as u64);
        assert!(
            rt.images_dir().join("imga").is_dir(),
            "referenced image must survive prune"
        );
        assert!(
            !rt.images_dir().join("imgb").exists(),
            "unreferenced image must be removed"
        );
    }

    #[tokio::test]
    async fn prune_missing_images_dir_returns_default() {
        let (rt, _tmp) = runtime();

        // `new()` pre-creates `images/`; remove it to exercise the NotFound path.
        std::fs::remove_dir_all(rt.images_dir()).expect("remove images dir");
        assert!(!rt.images_dir().exists());

        let result = rt.prune_images().await.expect("prune succeeds");

        assert!(result.deleted.is_empty());
        assert_eq!(result.space_reclaimed, 0);
    }

    #[tokio::test]
    async fn list_images_prefers_ref_file() {
        let (rt, _tmp) = runtime();

        // One image with a `ref` file (trailing newline must be trimmed), one
        // without (falls back to the sanitized directory name).
        write_image_file(&rt, "withref", "layer", b"abc");
        write_ref_file(&rt, "withref", "alpine:latest\n");
        write_image_file(&rt, "noref", "layer", b"de");

        let mut images = rt.list_images().await.expect("list succeeds");
        images.sort_by(|a, b| a.reference.cmp(&b.reference));

        assert_eq!(images.len(), 2);

        let by_ref = |r: &str| {
            images
                .iter()
                .find(|i| i.reference == r)
                .unwrap_or_else(|| panic!("missing image with reference {r}"))
                .clone()
        };

        let withref = by_ref("alpine:latest");
        assert_eq!(withref.reference, "alpine:latest");
        assert_eq!(withref.size_bytes, Some(3));

        let noref = by_ref("noref");
        assert_eq!(noref.reference, "noref");
        assert_eq!(noref.size_bytes, Some(2));

        // No `blobs.redb` cache exists, so digests are best-effort `None`.
        assert!(withref.digest.is_none());
        assert!(noref.digest.is_none());
    }

    #[tokio::test]
    async fn inspect_image_native_reports_darwin_from_sidecar() {
        let (rt, _tmp) = runtime();

        // A Seatbelt-built image: rootfs payload, a `ref` file, the platform
        // sidecar, and the builder's `config.json`.
        write_image_file(&rt, "myapp_latest", "bin", b"hello-rootfs");
        write_ref_file(&rt, "myapp_latest", "myapp:latest");
        let image_dir = rt.images_dir().join("myapp_latest");
        let meta =
            zlayer_types::local_image::LocalImageMetadata::new("myapp:latest", "darwin", "arm64");
        std::fs::write(
            image_dir.join(zlayer_types::local_image::LOCAL_IMAGE_METADATA_FILE),
            serde_json::to_vec(&meta).expect("serialize sidecar"),
        )
        .expect("write sidecar");
        std::fs::write(
            image_dir.join("config.json"),
            br#"{"env":["FOO=bar"],"working_dir":"/app","entrypoint":["/bin/app"],"cmd":["--serve"],"user":"nobody","labels":{"k":"v"}}"#,
        )
        .expect("write config.json");

        let info = rt
            .inspect_image_native("myapp:latest")
            .await
            .expect("inspect succeeds");

        assert_eq!(info.os.as_deref(), Some("darwin"));
        assert_eq!(info.architecture.as_deref(), Some("arm64"));
        assert_eq!(info.repo_tags, vec!["myapp:latest".to_string()]);
        assert_eq!(info.size, Some(12)); // "hello-rootfs"
        assert_eq!(info.env, vec!["FOO=bar".to_string()]);
        assert_eq!(info.cmd, vec!["--serve".to_string()]);
        assert_eq!(info.entrypoint, vec!["/bin/app".to_string()]);
        assert_eq!(info.working_dir.as_deref(), Some("/app"));
        assert_eq!(info.user.as_deref(), Some("nobody"));
        assert_eq!(info.labels.get("k").map(String::as_str), Some("v"));
    }

    #[tokio::test]
    async fn inspect_image_native_missing_is_not_found() {
        let (rt, _tmp) = runtime();
        let err = rt
            .inspect_image_native("does-not-exist")
            .await
            .expect_err("missing image must error");
        assert!(
            matches!(err, AgentError::NotFound { .. }),
            "expected NotFound, got {err:?}"
        );
    }

    #[tokio::test]
    async fn remove_image_not_found() {
        let (rt, _tmp) = runtime();

        let err = rt
            .remove_image("does-not-exist", false)
            .await
            .expect_err("missing image must error");
        assert!(
            matches!(err, AgentError::NotFound { .. }),
            "expected NotFound, got {err:?}"
        );
    }

    #[tokio::test]
    async fn remove_image_in_use_requires_force() {
        let (rt, _tmp) = runtime();

        write_image_file(&rt, "imga", "file", b"imgA-rootfs-bytes");
        write_container_config(&rt, "svc-0", "imga");

        // Without force: in-use conflict -> InvalidSpec.
        let err = rt
            .remove_image("imga", false)
            .await
            .expect_err("in-use image without force must error");
        assert!(
            matches!(err, AgentError::InvalidSpec(_)),
            "expected InvalidSpec, got {err:?}"
        );
        assert!(
            rt.images_dir().join("imga").is_dir(),
            "image must remain after a refused removal"
        );

        // With force: removal succeeds and the directory is gone.
        rt.remove_image("imga", true)
            .await
            .expect("forced removal succeeds");
        assert!(
            !rt.images_dir().join("imga").exists(),
            "forced removal must delete the image dir"
        );
    }

    #[tokio::test]
    async fn dir_size_bytes_sums_recursively() {
        let (_rt, tmp) = runtime();

        let root = tmp.path().join("tree");
        let nested = root.join("a").join("b");
        std::fs::create_dir_all(&nested).expect("create nested dirs");
        std::fs::write(root.join("top.bin"), vec![0u8; 10]).expect("write top file");
        std::fs::write(root.join("a").join("mid.bin"), vec![0u8; 20]).expect("write mid file");
        std::fs::write(nested.join("leaf.bin"), vec![0u8; 30]).expect("write leaf file");

        let total = SandboxRuntime::dir_size_bytes(&root).await;
        assert_eq!(total, 60);
    }

    /// A full-Xcode host selects a developer dir OUTSIDE the CLT tree (e.g.
    /// `/Applications/Xcode.app/Contents/Developer`); the profile must grant it
    /// read + executable-mapping (plus ancestor metadata) so `/usr/bin/clang`'s
    /// re-exec into the selected toolchain isn't blocked by deny-default. On a
    /// CLT-only host there is no extra grant — just the hardcoded CLT subpath.
    #[test]
    fn profile_grants_selected_developer_dir_outside_clt() {
        let profile = generate_sandbox_profile(&profile_config(KeychainAccess::None));
        match host_developer_dir() {
            Some(dir) if !Path::new(dir).starts_with("/Library/Developer/CommandLineTools") => {
                // Full Xcode → the whole `.app` bundle root is granted (clang
                // reaches Contents/Info.plist + Contents/SharedFrameworks),
                // not just the `…/Developer` dev dir.
                let grant_root = enclosing_app_bundle(dir).unwrap_or(dir);
                assert!(
                    profile.contains(&format!(
                        "(allow file-read* file-map-executable (subpath \"{grant_root}\"))"
                    )),
                    "selected toolchain bundle must be granted read + exec-map: {grant_root}\n{profile}"
                );
                // Its parent must be stat-able so clang can realpath InstalledDir.
                let parent = Path::new(grant_root)
                    .parent()
                    .expect("grant root has a parent");
                assert!(
                    profile.contains(&format!(
                        "(allow file-read-metadata (literal \"{}\"))",
                        parent.display()
                    )),
                    "toolchain-bundle ancestors must be metadata-readable\n{profile}"
                );
            }
            _ => {
                // CLT-only / no toolchain: only the Section 2 CLT subpath grant,
                // no second non-CLT developer-dir grant.
                assert!(
                    profile.contains("(subpath \"/Library/Developer/CommandLineTools\")"),
                    "CLT subpath must always be granted\n{profile}"
                );
                assert!(
                    !profile.contains("/Applications/Xcode.app"),
                    "no Xcode.app grant expected on a CLT-only host\n{profile}"
                );
            }
        }
    }

    #[test]
    fn enclosing_app_bundle_resolves_bundle_root() {
        assert_eq!(
            enclosing_app_bundle("/Applications/Xcode.app/Contents/Developer"),
            Some("/Applications/Xcode.app")
        );
        assert_eq!(
            enclosing_app_bundle("/Applications/Xcode-beta.app"),
            Some("/Applications/Xcode-beta.app")
        );
        // CLT path is not a bundle.
        assert_eq!(
            enclosing_app_bundle("/Library/Developer/CommandLineTools"),
            None
        );
    }

    /// `compute_tail_offset` mirrors `tail -n N`: it returns the byte offset of
    /// the start of the last N lines, and `0` when the file has fewer than N.
    #[tokio::test]
    async fn compute_tail_offset_matches_tail_semantics() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let path = tmp.path().join("log.txt");
        // 5 lines, each "lineN\n".
        std::fs::write(&path, "line0\nline1\nline2\nline3\nline4\n").expect("write");
        let len = std::fs::metadata(&path).expect("stat").len();

        let mut f = tokio::fs::File::open(&path).await.expect("open");
        // Last 2 lines = "line3\nline4\n" -> starts after the 3rd-from-end '\n'.
        let off = compute_tail_offset(&mut f, len, 2).await;
        let rest = std::fs::read_to_string(&path).expect("read");
        let off = usize::try_from(off).expect("offset fits usize");
        assert_eq!(&rest[off..], "line3\nline4\n");

        // Asking for more lines than exist streams the whole file.
        let mut f2 = tokio::fs::File::open(&path).await.expect("open");
        assert_eq!(compute_tail_offset(&mut f2, len, 99).await, 0);
    }

    /// Insert a fabricated container record so the log-stream tests have
    /// `stdout.log` / `stderr.log` paths and a state to follow.
    async fn insert_log_container(
        rt: &SandboxRuntime,
        id: &ContainerId,
        state: ContainerState,
        dir: &Path,
    ) {
        let container = SandboxContainer {
            pid: 0,
            state,
            state_dir: dir.to_path_buf(),
            rootfs_dir: dir.join("rootfs"),
            workspace_host: dir.join("ws"),
            stdout_path: dir.join("stdout.log"),
            stderr_path: dir.join("stderr.log"),
            started_at: None,
            spec: ServiceSpec::minimal("svc", "scratch:latest"),
            sandbox_config: profile_config(KeychainAccess::None),
            toolchain_handles: Vec::new(),
            memory_limit: None,
            watchdog_handle: None,
            assigned_port: 0,
            port_guard: None,
            overlay_ip: None,
            forwarders: Vec::new(),
        };
        rt.containers
            .write()
            .await
            .insert(SandboxRuntime::container_dir_name(id), container);
    }

    /// A follow stream on an already-exited container must deliver the FULL
    /// captured output (the bug: the composite one-shot fallback returned an
    /// empty snapshot), then close (the workload is terminal, so EOF).
    #[tokio::test]
    async fn logs_stream_follows_to_completion_on_exited_container() {
        use futures_util::StreamExt as _;

        let (rt, tmp) = runtime();
        let id = ContainerId::new("svc", 0);
        let dir = tmp.path().join("c0");
        std::fs::create_dir_all(&dir).expect("mkdir");
        std::fs::write(dir.join("stdout.log"), "hello\nworld\n").expect("stdout");
        std::fs::write(dir.join("stderr.log"), "oops\n").expect("stderr");
        insert_log_container(&rt, &id, ContainerState::Exited { code: 0 }, &dir).await;

        let opts = LogsStreamOptions {
            follow: true,
            ..Default::default()
        };
        let mut stream = rt.logs_stream(&id, opts).await.expect("logs_stream");

        let mut out = String::new();
        let mut err = String::new();
        while let Some(chunk) = stream.next().await {
            let chunk = chunk.expect("chunk ok");
            let text = String::from_utf8_lossy(&chunk.bytes).into_owned();
            match chunk.stream {
                LogChannel::Stdout => out.push_str(&text),
                LogChannel::Stderr => err.push_str(&text),
                LogChannel::Stdin => {}
            }
        }

        assert_eq!(out, "hello\nworld\n", "stdout must stream in full");
        assert_eq!(err, "oops\n", "stderr must stream in full");
    }

    /// A missing container is a genuine `NotFound` (→ 404), not an empty stream.
    #[tokio::test]
    async fn logs_stream_missing_container_is_not_found() {
        let (rt, _tmp) = runtime();
        let id = ContainerId::new("ghost", 3);
        let res = rt.logs_stream(&id, LogsStreamOptions::default()).await;
        assert!(
            matches!(res, Err(AgentError::NotFound { .. })),
            "missing container must surface NotFound"
        );
    }
}