cellos-host-firecracker 0.5.1

//! Firecracker-backed host backend (L2-06).
//!
//! # Architecture
//!
//! `FirecrackerCellBackend` implements the [`CellBackend`] trait by managing
//! one Firecracker VMM process per cell:
//!
//! 1. **`create`** — spawns `firecracker --api-sock <socket>`, waits for the
//!    socket to appear, then calls the Firecracker Management API to configure
//!    the machine (vCPUs, memory, kernel, rootfs) and boot it.
//! 2. **`destroy`** — sends a graceful `SendCtrlAltDel` action, waits for the
//!    process to exit, and cleans up the socket file.
//!
//! # Cell command execution
//!
//! The Firecracker path now runs `spec.run.argv` inside the guest via the
//! `cellos-init` PID-1 binary and a vsock exit-code bridge:
//!
//! 1. The host encodes `spec.run.argv` into the kernel boot args as
//!    `cellos.argv=<base64-json>`.
//! 2. The VM is configured with a vsock device and the host starts a matching
//!    Unix-socket listener.
//! 3. `cellos-init` reads `/proc/cmdline`, forks and execs the workload inside
//!    the guest, then writes the 4-byte little-endian exit code back to the
//!    host over vsock before powering off the VM.
//! 4. The supervisor calls `CellBackend::wait_for_in_vm_exit()` and skips the
//!    host-side subprocess path when this backend reports an in-VM exit code.
//!
//! The host-side subprocess fallback still exists for backends that do not
//! override `wait_for_in_vm_exit()`, but it is no longer the execution path for
//! `FirecrackerCellBackend`.

pub mod api_client;
pub mod pool;

use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

#[cfg(target_os = "linux")]
use std::collections::HashMap;

use async_trait::async_trait;

#[cfg(target_os = "linux")]
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
#[cfg(target_os = "linux")]
use base64::Engine;
#[cfg(target_os = "linux")]
use tokio::io::AsyncReadExt;
#[cfg(target_os = "linux")]
use tokio::io::AsyncWriteExt;
#[cfg(target_os = "linux")]
use tokio::net::UnixListener;
#[cfg(target_os = "linux")]
use tokio::process::Child;
#[cfg(target_os = "linux")]
use tokio::sync::Mutex;
#[cfg(target_os = "linux")]
use tracing::instrument;
#[cfg(target_os = "linux")]
use uuid::Uuid;

use cellos_core::ports::{CellBackend, CellHandle, TeardownReport};
#[cfg(target_os = "linux")]
use cellos_core::EgressRule;
use cellos_core::{CellosError, ExecutionCellDocument};

#[cfg(target_os = "linux")]
use api_client::{
    BootSource, Drive, FirecrackerApiClient, InstanceAction, InstanceActionType, MachineConfig,
    NetworkInterface, VsockDevice,
};

/// How long to wait for the Firecracker socket to appear after process start.
#[cfg(target_os = "linux")]
const SOCKET_READY_TIMEOUT: Duration = Duration::from_secs(10);
/// How long to wait for graceful VM shutdown before SIGKILL.
///
/// FC-21: this is the *fallback* applied when `spec.run.limits.gracefulShutdownSeconds`
/// is absent. Per-spec overrides land in [`VmRecord::graceful_shutdown_timeout`]
/// at `create()` time and are read back in `destroy()`.
#[cfg(target_os = "linux")]
const GRACEFUL_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);

/// Resolve the graceful-shutdown timeout for a spec.
///
/// Reads `spec.run.limits.graceful_shutdown_seconds` (FC-21). When absent or
/// `None`, returns [`GRACEFUL_SHUTDOWN_TIMEOUT`]. Schema validation already
/// bounds the value to `[1, 120]` seconds, so we trust whatever made it past
/// admission here.
#[cfg(target_os = "linux")]
fn resolve_graceful_shutdown_timeout(spec: &cellos_core::ExecutionCellSpec) -> Duration {
    spec.run
        .as_ref()
        .and_then(|r| r.limits.as_ref())
        .and_then(|l| l.graceful_shutdown_seconds)
        .map(Duration::from_secs)
        .unwrap_or(GRACEFUL_SHUTDOWN_TIMEOUT)
}

/// Default vCPUs for a cell VM (override via spec.run.limits in future).
#[cfg(target_os = "linux")]
const DEFAULT_VCPU_COUNT: u32 = 1;
/// Default memory in MiB for a cell VM.
#[cfg(target_os = "linux")]
const DEFAULT_MEM_SIZE_MIB: u32 = 128;

/// Derive vCPU count from spec run limits.
///
/// Maps `spec.run.limits.cpu_max` (quota_micros / period_micros) to a vCPU
/// count by rounding UP. Clamped to [1, 32] (Firecracker's supported range).
/// Fractional CPU allocations get 1 vCPU minimum — underprovisioning a build
/// runner causes more latency regression than the 1-vCPU overhead.
///
/// Returns `DEFAULT_VCPU_COUNT` when no limits are declared.
#[cfg(target_os = "linux")]
fn derive_vcpu_count(spec: &cellos_core::ExecutionCellSpec) -> u32 {
    let Some(cpu_max) = spec
        .run
        .as_ref()
        .and_then(|r| r.limits.as_ref())
        .and_then(|l| l.cpu_max.as_ref())
    else {
        return DEFAULT_VCPU_COUNT;
    };
    let period = cpu_max.period_micros.unwrap_or(100_000).max(1);
    let vcpus = cpu_max.quota_micros.div_ceil(period) as u32;
    vcpus.clamp(1, 32)
}

/// vsock port that `cellos-init` inside the VM connects to after the cell
/// command exits.  Must match the constant in `cellos-init/src/main.rs`.
pub const VSOCK_EXIT_PORT: u32 = 9000;

/// FC-18: size in bytes of the per-cell HMAC-SHA256 authentication key.
/// `pub(crate)` — tests reach this via the doc-hidden `__fc18` shim.
pub(crate) const EXIT_HMAC_KEY_LEN: usize = 32;

/// FC-18: size in bytes of the HMAC-SHA256 tag the guest appends to its
/// exit-code report. Must match `compute_exit_hmac` in `cellos-init`.
pub(crate) const EXIT_HMAC_TAG_LEN: usize = 32;

/// FC-18: full authenticated exit-code wire format size — `4` bytes of
/// little-endian `i32` followed by `EXIT_HMAC_TAG_LEN` bytes of MAC.
#[cfg(target_os = "linux")]
const EXIT_AUTHED_FRAME_LEN: usize = 4 + EXIT_HMAC_TAG_LEN;

/// FC-18: generate a fresh 32-byte HMAC key for one cell by reading from
/// `/dev/urandom`. Linux-only because Firecracker is Linux-only and the
/// surrounding crate is `cfg(target_os = "linux")`-gated; no generic random
/// source crate is in the workspace dependency tree, and we explicitly do
/// not want to pull `getrandom` into cellos-host-firecracker for this single
/// call site.
///
/// Returns `Err` only if `/dev/urandom` is unavailable or short-reads — both
/// are catastrophic on a Linux host (the kernel guarantees /dev/urandom is
/// always readable post-boot) so the caller should treat this as a fatal
/// host configuration error and refuse to launch the cell.
#[cfg(target_os = "linux")]
fn generate_exit_hmac_key() -> Result<[u8; EXIT_HMAC_KEY_LEN], CellosError> {
    use std::io::Read;
    let mut key = [0u8; EXIT_HMAC_KEY_LEN];
    let mut f = std::fs::File::open("/dev/urandom")
        .map_err(|e| CellosError::Host(format!("open /dev/urandom: {e}")))?;
    f.read_exact(&mut key)
        .map_err(|e| CellosError::Host(format!("read /dev/urandom: {e}")))?;
    Ok(key)
}

/// FC-18: pure-Rust verification helper. Recompute the HMAC-SHA256 tag the
/// guest should have produced (`HMAC(key, exit_code_bytes ‖ cell_id_bytes)`)
/// and compare it in constant time against the bytes the supervisor read
/// off the wire.
///
/// `exit_code_bytes` is the raw 4-byte little-endian buffer the supervisor
/// already received — passing the bytes directly (rather than the parsed
/// `i32`) keeps this function trivially testable against arbitrary
/// adversarial inputs without hard-coding the byte order in two places.
///
/// Returns `true` iff the recomputed tag matches the received tag. Note:
/// uses `hmac::Mac::verify_slice` which performs a constant-time compare —
/// avoid replacing with naive `==` on the byte arrays, which leaks timing
/// information about the prefix match.
///
/// `pub(crate)`: tests reach this via the doc-hidden `__fc18` shim.
pub(crate) fn verify_exit_hmac(
    key: &[u8],
    exit_code_bytes: &[u8; 4],
    cell_id: &str,
    received_tag: &[u8],
) -> bool {
    use hmac::{digest::KeyInit, Hmac, Mac};
    use sha2::Sha256;
    type HmacSha256 = Hmac<Sha256>;

    if received_tag.len() != EXIT_HMAC_TAG_LEN {
        return false;
    }
    let mut mac = match HmacSha256::new_from_slice(key) {
        Ok(m) => m,
        Err(_) => return false,
    };
    mac.update(exit_code_bytes);
    mac.update(cell_id.as_bytes());
    mac.verify_slice(received_tag).is_ok()
}

/// Guest CID assigned to every cell VM.  The host CID (2) is used by
/// `cellos-init` to connect back; 3 is the conventional first guest CID.
#[cfg(target_os = "linux")]
const VSOCK_GUEST_CID: u32 = 3;

/// MAC address handed to the guest virtio-net device.  Each cell runs in its
/// own L2 segment (single-tap, point-to-point with the host bridge), so reusing
/// one MAC across cells is safe — there is no shared broadcast domain.
#[cfg(target_os = "linux")]
const GUEST_NIC_MAC: &str = "AA:FC:00:00:00:01";

/// Linker prefix for per-cell host TAP interfaces. Combined with an 8-character
/// sanitized cell-id slug to stay within the 15-byte `IFNAMSIZ` kernel limit
/// (`cfc-` = 4, slug = 8, total = 12).
#[cfg(target_os = "linux")]
const TAP_NAME_PREFIX: &str = "cfc-";

/// True when host TAP / nftables enforcement is supported. The runtime helpers
/// for [`create_tap_device`], [`apply_network_policy`], etc. only function on
/// Linux; on every other OS they short-circuit to a clear error so a developer
/// running tests on macOS sees the same code paths but does not need root.
#[cfg(target_os = "linux")]
const NETWORK_DEFAULT_ENABLED: bool = true;
#[cfg(not(target_os = "linux"))]
const NETWORK_DEFAULT_ENABLED: bool = false;

// ── Config ───────────────────────────────────────────────────────────────────

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FirecrackerConfig {
    pub binary_path: PathBuf,
    pub kernel_image_path: PathBuf,
    pub rootfs_image_path: PathBuf,
    pub jailer_binary_path: Option<PathBuf>,
    pub chroot_base_dir: PathBuf,
    /// Directory for Firecracker API socket files; defaults to `/tmp`.
    /// Ignored when the jailer is active — the jailer places the API socket
    /// inside the chroot at `<chroot_base>/<fc_filename>/<cell_id>/root/run/firecracker.socket`.
    pub socket_dir: PathBuf,
    /// Numeric uid the jailer drops to before exec'ing firecracker. Defaults to 10002.
    /// This user must exist on the host and must own no sensitive files.
    pub jailer_uid: u32,
    /// Numeric gid the jailer drops to before exec'ing firecracker. Defaults to 10002.
    pub jailer_gid: u32,
    /// Directory where per-cell writable scratch ext4 images are created.
    /// When set, rootfs is mounted read-only and a writable scratch drive is
    /// attached as a second virtio-blk device. When None (default), rootfs is
    /// mounted read-write (v0.2 behaviour — not safe for concurrent cells sharing
    /// the same rootfs image).
    pub scratch_dir: Option<PathBuf>,
    /// Path to the artifact manifest file produced by the build pipeline.
    /// When set, `create()` verifies the SHA256 digest of each declared
    /// artifact (kernel, rootfs, optionally firecracker) before booting the
    /// VM. When `None` AND `allow_no_manifest` is `true`, verification is
    /// skipped with a loud warning — supported for development only.
    /// Otherwise, `from_env()` rejects the configuration outright.
    pub manifest_path: Option<PathBuf>,
    /// When `true` (the default), `create()` refuses to proceed unless the
    /// jailer binary is configured. Operators may opt out for development
    /// by setting `CELLOS_FIRECRACKER_ALLOW_NO_JAILER=1`, which logs a loud
    /// warning and downgrades this flag to `false`.
    pub require_jailer: bool,
    /// When `true`, `create()` will permit booting a VM without a manifest
    /// (skipping pre-boot artifact digest verification). Defaults to `false`
    /// — a missing `CELLOS_FIRECRACKER_MANIFEST` is a hard error. Operators
    /// may opt out for development by setting BOTH
    /// `CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST=1` AND the second escape-hatch
    /// flag `CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY=1`; only the
    /// combination flips this flag to `true` and emits a loud `WARN` log
    /// containing the literal string `MANIFEST VERIFICATION DISABLED`.
    ///
    /// The two-flag handshake is deliberate. A single env var can be set in a
    /// shared base image, a Helm chart copied between environments, or an
    /// `.env` file leaking from dev to prod by mistake. Requiring a paired
    /// `_REALLY` flag forces the operator to make the trade-off explicit on
    /// the same line, in the same operation, so the dev opt-out cannot drift
    /// into a production deployment unnoticed.
    ///
    /// It is an error to set both `CELLOS_FIRECRACKER_MANIFEST` and the
    /// two-flag opt-out — that combination is inconsistent and `from_env()`
    /// rejects it. Setting `CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST=1` without
    /// the paired `_REALLY` flag (or vice-versa) is also rejected with a
    /// hard error so misconfigured deployments fail closed.
    pub allow_no_manifest: bool,
    /// When true, `create()` provisions a per-cell TAP interface, attaches it
    /// to the VM as virtio-net, and installs an nftables ruleset that drops
    /// all egress except destinations declared in `spec.authority.egressRules`.
    /// Linux-only: defaults to `true` on Linux and `false` on every other OS
    /// (the `ip` and `nft` commands and TAP devices do not exist there).
    /// Override with `CELLOS_FIRECRACKER_ENABLE_NETWORK=0|1`.
    pub enable_network: bool,
    /// When `true`, [`wait_for_command_exit`] uses a short bounded timeout
    /// instead of waiting indefinitely for the in-VM vsock exit code.
    ///
    /// Motivation: a kernel built with the wrong vsock symbol
    /// (`CONFIG_VIRTIO_VSOCK` is a typo for `CONFIG_VIRTIO_VSOCKETS`) silently
    /// has no vsock device. The supervisor then waits forever for a 4-byte
    /// exit code that no one will ever send, and the only signal the operator
    /// gets is a hung process. Set
    /// `CELLOS_FIRECRACKER_ALLOW_NO_VSOCK=1` (and optionally
    /// `CELLOS_FIRECRACKER_NO_VSOCK_TIMEOUT_SECS=<n>` — default 5) to fail
    /// fast and surface the misconfiguration in seconds.
    ///
    /// The terminal state for the cell will be `forced` (no authenticated
    /// in-VM exit was received), which is the correct audit signal.
    pub allow_no_vsock: bool,
    /// Wait budget when `allow_no_vsock` is true. Ignored otherwise.
    /// Default: 5 seconds. Override via
    /// `CELLOS_FIRECRACKER_NO_VSOCK_TIMEOUT_SECS`.
    pub no_vsock_timeout: Duration,
    /// When `true`, passes `--no-seccomp` to the Firecracker process.
    ///
    /// Firecracker's seccomp BPF filters are compiled with x86-64 syscall
    /// numbers. Under arm64 emulation (Rosetta/QEMU in Colima) the BPF
    /// program is rejected by the kernel with EINVAL because the syscall
    /// table doesn't match. Set `CELLOS_FIRECRACKER_NO_SECCOMP=1` to bypass
    /// seccomp for emulated development environments.
    ///
    /// **Never set this in production.** Seccomp is a critical attack-surface
    /// reduction — bypassing it removes a significant isolation layer.
    pub no_seccomp: bool,
}

impl FirecrackerConfig {
    pub fn from_env() -> Result<Self, CellosError> {
        Self::from_lookup(|key| std::env::var(key).ok())
    }

    pub(crate) fn from_lookup<F>(lookup: F) -> Result<Self, CellosError>
    where
        F: Fn(&str) -> Option<String>,
    {
        let cfg = Self {
            binary_path: required_absolute_path(
                &lookup,
                "CELLOS_FIRECRACKER_BINARY",
                "firecracker VMM binary",
            )?,
            kernel_image_path: required_absolute_path(
                &lookup,
                "CELLOS_FIRECRACKER_KERNEL_IMAGE",
                "Firecracker kernel image",
            )?,
            rootfs_image_path: required_absolute_path(
                &lookup,
                "CELLOS_FIRECRACKER_ROOTFS_IMAGE",
                "Firecracker rootfs image",
            )?,
            jailer_binary_path: optional_absolute_path(
                &lookup,
                "CELLOS_FIRECRACKER_JAILER_BINARY",
                "Firecracker jailer binary",
            )?,
            chroot_base_dir: optional_absolute_path(
                &lookup,
                "CELLOS_FIRECRACKER_CHROOT_BASE",
                "Firecracker chroot base directory",
            )?
            .unwrap_or_else(|| PathBuf::from("/var/lib/cellos/firecracker")),
            socket_dir: optional_absolute_path(
                &lookup,
                "CELLOS_FIRECRACKER_SOCKET_DIR",
                "Firecracker socket directory",
            )?
            .unwrap_or_else(|| PathBuf::from("/tmp")),
            jailer_uid: lookup("CELLOS_FIRECRACKER_JAILER_UID")
                .and_then(|v| v.parse().ok())
                .unwrap_or(10002),
            jailer_gid: lookup("CELLOS_FIRECRACKER_JAILER_GID")
                .and_then(|v| v.parse().ok())
                .unwrap_or(10002),
            scratch_dir: optional_absolute_path(
                &lookup,
                "CELLOS_FIRECRACKER_SCRATCH_DIR",
                "Firecracker scratch image directory",
            )?,
            manifest_path: optional_absolute_path(
                &lookup,
                "CELLOS_FIRECRACKER_MANIFEST",
                "Firecracker artifact manifest file",
            )?,
            require_jailer: {
                // Default to require_jailer=true. Operators may opt out by
                // setting CELLOS_FIRECRACKER_ALLOW_NO_JAILER=1; this is a
                // production safety knob and we log loudly when it is used.
                let allow_no_jailer = lookup("CELLOS_FIRECRACKER_ALLOW_NO_JAILER")
                    .map(|v| v.trim() == "1")
                    .unwrap_or(false);
                if allow_no_jailer {
                    tracing::warn!(
                        "CELLOS_FIRECRACKER_ALLOW_NO_JAILER=1 is set — running Firecracker WITHOUT the jailer. \
                         This is unsafe for production and should only be used for local development."
                    );
                    false
                } else {
                    // Explicit override via CELLOS_FIRECRACKER_REQUIRE_JAILER, otherwise default true.
                    lookup("CELLOS_FIRECRACKER_REQUIRE_JAILER")
                        .map(|v| {
                            let t = v.trim();
                            // Accept common truthy values; anything else flips to false.
                            !matches!(t, "0" | "false" | "FALSE" | "no" | "NO")
                        })
                        .unwrap_or(true)
                }
            },
            allow_no_manifest: {
                // Two-flag handshake (FC-05 / SEAM-23 hardening): the
                // bypass requires BOTH `CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST=1`
                // AND the paired escape-hatch `CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY=1`.
                // A single env var can leak from a dev `.env` into a
                // production rollout; requiring a second, explicitly named
                // flag forces the operator to make the trade-off on the
                // same line. We compute the boolean here and validate the
                // combination explicitly — when only one of the two flags
                // is set, return a *specific* error ("you set the first
                // flag but not the second") instead of falling through to
                // the generic "manifest is mandatory" message.
                let primary = lookup("CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST")
                    .map(|v| v.trim() == "1")
                    .unwrap_or(false);
                let secondary = lookup("CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY")
                    .map(|v| v.trim() == "1")
                    .unwrap_or(false);
                if primary && !secondary {
                    return Err(CellosError::Host(
                        "firecracker init: CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST=1 \
                         requires the paired escape-hatch flag \
                         CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY=1 to take \
                         effect. Without both flags set, manifest verification \
                         remains mandatory (production posture). The two-flag \
                         handshake exists so a dev `.env` cannot accidentally \
                         disable digest verification in production — set both \
                         on the same line, on purpose, or set neither and \
                         provide CELLOS_FIRECRACKER_MANIFEST instead."
                            .into(),
                    ));
                }
                if secondary && !primary {
                    return Err(CellosError::Host(
                        "firecracker init: CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY=1 \
                         is set but CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST=1 is not. \
                         The escape-hatch is two-flag by design: set both to opt \
                         out of pre-boot artifact digest verification (development \
                         only), or unset both for the production posture."
                            .into(),
                    ));
                }
                primary && secondary
            },
            enable_network: parse_enable_network(&lookup),
            allow_no_vsock: lookup("CELLOS_FIRECRACKER_ALLOW_NO_VSOCK")
                .map(|v| v.trim() == "1")
                .unwrap_or(false),
            no_vsock_timeout: lookup("CELLOS_FIRECRACKER_NO_VSOCK_TIMEOUT_SECS")
                .and_then(|v| v.trim().parse::<u64>().ok())
                .map(Duration::from_secs)
                .unwrap_or_else(|| Duration::from_secs(5)),
            no_seccomp: lookup("CELLOS_FIRECRACKER_NO_SECCOMP")
                .map(|v| v.trim() == "1")
                .unwrap_or(false),
        };

        // FC-41: reject jailer_uid/gid == 0. Without this gate an operator
        // can set `CELLOS_FIRECRACKER_JAILER_UID=0` and silently run as root.
        // Error carries "FC-41" for audit searchability.
        if cfg.jailer_uid == 0 {
            return Err(CellosError::Host(
                "FirecrackerConfig: jailer_uid must be non-zero (running jailer as root defeats the privilege boundary) [FC-41]"
                    .into(),
            ));
        }
        if cfg.jailer_gid == 0 {
            return Err(CellosError::Host(
                "FirecrackerConfig: jailer_gid must be non-zero (running jailer in root group defeats the privilege boundary) [FC-41]"
                    .into(),
            ));
        }

        if cfg.allow_no_vsock {
            tracing::warn!(
                timeout_secs = cfg.no_vsock_timeout.as_secs(),
                "CELLOS_FIRECRACKER_ALLOW_NO_VSOCK=1 is set — vsock exit-code wait \
                 will time out after the configured budget instead of blocking. \
                 Cell terminal state will be `forced` (no authenticated in-VM exit). \
                 This is intended for development against kernels without vsock \
                 support; production deployments MUST keep this off."
            );
        }

        if cfg.no_seccomp {
            tracing::warn!(
                "CELLOS_FIRECRACKER_NO_SECCOMP=1 is set — Firecracker will start \
                 with --no-seccomp. Seccomp syscall filtering is DISABLED. This is \
                 only safe for emulated development environments (e.g. arm64 Rosetta) \
                 where the x86-64 BPF filters are rejected by the host kernel. \
                 NEVER set this in production."
            );
        }

        // Manifest enforcement. Mirrors the jailer guard above: production
        // deployments MUST verify kernel/rootfs/firecracker artifact digests
        // before boot. Operators may opt out for development by setting BOTH
        // CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST=1 AND
        // CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY=1 (the two-flag
        // handshake validated when `cfg.allow_no_manifest` was assembled
        // above); we log loudly when that combination is used. Path-required
        // errors (binary/kernel/rootfs missing) take precedence over the
        // manifest gate so callers see the most specific misconfiguration
        // first.
        match (cfg.manifest_path.is_some(), cfg.allow_no_manifest) {
            (true, true) => {
                // Inconsistent config — fail closed rather than silently
                // picking one interpretation.
                return Err(CellosError::Host(
                    "firecracker init: CELLOS_FIRECRACKER_MANIFEST is set AND \
                     the two-flag opt-out (CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST=1 \
                     plus CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY=1) is also \
                     set — these are mutually exclusive. Unset both opt-out flags \
                     to perform manifest verification, or unset \
                     CELLOS_FIRECRACKER_MANIFEST to run in dev mode without it."
                        .into(),
                ));
            }
            (false, false) => {
                return Err(CellosError::Host(
                    "firecracker init: CELLOS_FIRECRACKER_MANIFEST is not set \
                     — pre-boot artifact digest verification is mandatory by \
                     default. Set CELLOS_FIRECRACKER_MANIFEST to a v1 \
                     manifest path, or set BOTH \
                     CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST=1 AND \
                     CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY=1 to opt \
                     out (development only — the second flag is a deliberate \
                     speed-bump to keep dev opt-outs from leaking into prod)."
                        .into(),
                ));
            }
            (false, true) => {
                tracing::warn!(
                    "MANIFEST VERIFICATION DISABLED — CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST=1 \
                     and CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY=1 are both set. \
                     Booting Firecracker cells WITHOUT pre-boot artifact digest verification. \
                     This is unsafe for production and should only be used for local development."
                );
            }
            (true, false) => {
                // Standard production posture: manifest set, opt-out unset.
            }
        }

        Ok(cfg)
    }
}

/// Parse `CELLOS_FIRECRACKER_ENABLE_NETWORK`.  Accepts `1`/`true`/`yes`/`on`
/// (case-insensitive) as true and `0`/`false`/`no`/`off` as false; any other
/// value falls back to the platform default.  On non-Linux hosts where the
/// default is `false`, log a warning the first time the value is read so the
/// operator knows TAP/nftables are not active in their dev environment.
fn parse_enable_network<F>(lookup: &F) -> bool
where
    F: Fn(&str) -> Option<String>,
{
    let raw = lookup("CELLOS_FIRECRACKER_ENABLE_NETWORK");
    let parsed = raw
        .as_deref()
        .and_then(|v| match v.trim().to_ascii_lowercase().as_str() {
            "1" | "true" | "yes" | "on" => Some(true),
            "0" | "false" | "no" | "off" | "" => Some(false),
            _ => None,
        });
    let enabled = parsed.unwrap_or(NETWORK_DEFAULT_ENABLED);
    #[cfg(not(target_os = "linux"))]
    {
        if enabled {
            tracing::warn!(
                "CELLOS_FIRECRACKER_ENABLE_NETWORK requested on non-Linux host — \
                 TAP and nftables enforcement are Linux-only and will fail at runtime"
            );
        } else if raw.is_none() {
            tracing::warn!(
                "firecracker network enforcement disabled by default on non-Linux host \
                 (set CELLOS_FIRECRACKER_ENABLE_NETWORK=1 to override; runtime calls will error)"
            );
        }
    }
    enabled
}

// ── Live VM record ────────────────────────────────────────────────────────────

#[cfg(target_os = "linux")]
struct VmRecord {
    socket_path: PathBuf,
    /// Base path used for the vsock UDS (`<vsock_uds_path>_<port>`).
    vsock_uds_path: PathBuf,
    /// The spawned Firecracker process (kept alive for the lifetime of the cell).
    child: Child,
    /// Receives the `cellos-init` exit code reported over vsock.
    /// `None` until the guest connects and writes the 4-byte value; `Some(code)`
    /// once it is received.
    exit_rx: tokio::sync::watch::Receiver<Option<i32>>,
    /// When the jailer is in use, the chroot tree the jailer created for this
    /// cell (e.g. `<chroot_base>/firecracker/<cell_id>`). `destroy()` removes
    /// this tree on best-effort basis. `None` when the jailer is not in use.
    chroot_cell_dir: Option<PathBuf>,
    /// Path to the per-cell scratch ext4 image (when scratch_dir is configured).
    /// Set to `Some(path)` if a scratch image was created in `create()`; cleaned
    /// up in `destroy()`.
    scratch_image_path: Option<PathBuf>,
    /// Name of the host TAP interface created for this cell, when network
    /// enforcement is enabled.  `None` when `enable_network` is false.
    /// Used by `destroy()` to tear down the TAP and nftables table.
    tap_iface: Option<String>,
    /// Per-cell graceful-shutdown timeout (FC-21). Captured at `create()` from
    /// `spec.run.limits.graceful_shutdown_seconds`, falling back to
    /// [`GRACEFUL_SHUTDOWN_TIMEOUT`]. Read in `destroy()` when waiting for the
    /// VM to exit gracefully before SIGKILL.
    graceful_shutdown_timeout: Duration,
}

// ── Backend ───────────────────────────────────────────────────────────────────

/// Firecracker-backed [`CellBackend`] (L2-06).
///
/// On non-Linux hosts the live-VM table collapses to an unused unit field —
/// the backend still constructs (so the supervisor's composition root keeps
/// compiling) but every [`CellBackend`] method short-circuits to an
/// `Unsupported`-shaped [`CellosError::Host`]. This shapes Windows/macOS
/// `cargo check` builds without dragging Linux-only kernel surface
/// (`tokio::net::UnixStream`, TAP, nftables) into the cross-platform side
/// of the workspace.
pub struct FirecrackerCellBackend {
    config: FirecrackerConfig,
    #[cfg(target_os = "linux")]
    running_vms: Arc<Mutex<HashMap<String, VmRecord>>>,
    /// L2-06-2 warm pool of pre-booted/snapshotted VMs. `checkout` is consulted
    /// at the head of `create()` so a populated pool slot bypasses the
    /// ~125ms cold-boot path in favour of a ~10ms `PUT /snapshot/load`.
    /// Constructed with [`pool::pool_size_from_env()`] slots — when the
    /// `CELLOS_FIRECRACKER_POOL_SIZE` env var is unset/zero, the pool is a
    /// zero-slot no-op and `checkout` always yields `None`, making the
    /// wiring an inert pass-through on the cold-boot path.
    #[cfg(target_os = "linux")]
    pool: Arc<Mutex<pool::FirecrackerPool>>,
    /// Optional CellOS event sink. When set, the backend emits a
    /// `dev.cellos.events.cell.firecracker.v1.pool_checkout` CloudEvent
    /// after every warm-pool checkout attempt in `create()` (best-effort —
    /// emit failures are logged but do not fail VM creation). Left `None`
    /// in tests and in compositions that do not opt in.
    event_sink: Option<Arc<dyn cellos_core::ports::EventSink>>,
}

impl FirecrackerCellBackend {
    #[cfg(target_os = "linux")]
    pub fn new(config: FirecrackerConfig) -> Self {
        let pool_size = pool::pool_size_from_env();
        Self {
            config,
            running_vms: Arc::new(Mutex::new(HashMap::new())),
            pool: Arc::new(Mutex::new(pool::FirecrackerPool::new(pool_size))),
            event_sink: None,
        }
    }

    /// Non-Linux constructor for the compile-time stub. The backend cannot be
    /// driven on Windows/macOS — Firecracker is Linux-only — so any
    /// [`CellBackend`] method returns a clear error. The constructor itself
    /// succeeds so that the supervisor's composition root, which gates
    /// runtime selection on the `CELLOS_CELL_BACKEND=firecracker` env var,
    /// still type-checks on every host.
    #[cfg(not(target_os = "linux"))]
    pub fn new(config: FirecrackerConfig) -> Self {
        Self {
            config,
            event_sink: None,
        }
    }

    pub fn from_env() -> Result<Self, CellosError> {
        Ok(Self::new(FirecrackerConfig::from_env()?))
    }

    /// Attach a CellOS [`EventSink`](cellos_core::ports::EventSink) for
    /// best-effort emission of warm-pool checkout CloudEvents.
    ///
    /// When set, every `create()` call emits one
    /// `dev.cellos.events.cell.firecracker.v1.pool_checkout` event after
    /// consulting the warm pool, recording whether the boot took the
    /// snapshot fast path (`poolHit`) and the pre-checkout `Available` slot
    /// count. Emission failures are logged at `warn` and never abort VM
    /// creation — the audit event must not become a critical-path
    /// dependency.
    pub fn with_event_sink(mut self, event_sink: Arc<dyn cellos_core::ports::EventSink>) -> Self {
        self.event_sink = Some(event_sink);
        self
    }

    pub fn config(&self) -> &FirecrackerConfig {
        &self.config
    }

    /// Number of warm-pool slots configured (any state). Returns the value of
    /// `CELLOS_FIRECRACKER_POOL_SIZE` resolved at backend construction. Useful
    /// for the supervisor composition root to decide whether to spawn the
    /// background fill task at all.
    #[cfg(target_os = "linux")]
    pub async fn pool_size(&self) -> usize {
        self.pool.lock().await.size()
    }

    /// Off-Linux stub — Firecracker is Linux-only, so the pool is always empty.
    #[cfg(not(target_os = "linux"))]
    pub async fn pool_size(&self) -> usize {
        0
    }

    /// Number of warm-pool slots currently in `Available` state (callable from
    /// tests to observe that a fill cycle has run).
    #[cfg(target_os = "linux")]
    pub async fn pool_available(&self) -> usize {
        self.pool.lock().await.available()
    }

    /// Off-Linux stub.
    #[cfg(not(target_os = "linux"))]
    pub async fn pool_available(&self) -> usize {
        0
    }

    /// Drive one `fill()` cycle on the warm pool using the validated
    /// firecracker binary / kernel / rootfs paths from [`FirecrackerConfig`].
    ///
    /// Best-effort: per-slot failures are logged and leave the slot `Empty`
    /// (see [`pool::FirecrackerPool::fill`]). Intended to be called once at
    /// supervisor startup from a detached `tokio::spawn` so that subsequent
    /// `create()` calls can take the fast snapshot-restore path.
    #[cfg(target_os = "linux")]
    pub async fn fill_pool(&self) {
        let binary = self.config.binary_path.to_string_lossy().into_owned();
        let kernel = self.config.kernel_image_path.to_string_lossy().into_owned();
        let rootfs = self.config.rootfs_image_path.to_string_lossy().into_owned();
        let mut pool = self.pool.lock().await;
        pool.fill(&binary, &kernel, &rootfs).await;
    }

    /// Off-Linux stub — Firecracker is Linux-only, so there is no pool to fill.
    #[cfg(not(target_os = "linux"))]
    pub async fn fill_pool(&self) {
        tracing::debug!("FirecrackerCellBackend::fill_pool no-op: target_os != linux");
    }

    /// Number of VMs the backend currently tracks (for tests and operators).
    #[cfg(target_os = "linux")]
    pub async fn tracked_vm_count(&self) -> usize {
        self.running_vms.lock().await.len()
    }

    /// Number of VMs the backend currently tracks (for tests and operators).
    /// Always zero on non-Linux hosts — the backend cannot run there.
    #[cfg(not(target_os = "linux"))]
    pub async fn tracked_vm_count(&self) -> usize {
        0
    }

    /// Wait until `cellos-init` inside the VM reports the cell command's exit
    /// code over vsock, then return it.
    ///
    /// Returns `Err` if the cell is not tracked or the vsock channel closes
    /// before the exit code arrives.  The lock is released before awaiting, so
    /// other operations on the backend can proceed concurrently.
    #[cfg(target_os = "linux")]
    pub async fn wait_for_command_exit(&self, cell_id: &str) -> Result<i32, CellosError> {
        let mut exit_rx = {
            let vms = self.running_vms.lock().await;
            let record = vms.get(cell_id).ok_or_else(|| {
                CellosError::Host(format!(
                    "wait_for_command_exit: no VM tracked for cell {cell_id}"
                ))
            })?;
            record.exit_rx.clone()
        };

        let wait_loop = async {
            // Wait until Some(code) is set by the vsock listener background task.
            loop {
                if let Some(code) = *exit_rx.borrow() {
                    return Ok::<i32, CellosError>(code);
                }
                exit_rx.changed().await.map_err(|_| {
                    CellosError::Host(format!(
                        "vsock exit channel for cell {cell_id} closed without exit code"
                    ))
                })?;
            }
        };

        if self.config.allow_no_vsock {
            // Bounded wait. Surfaces "guest kernel has no vsock" in seconds
            // instead of an indefinite hang. Caller maps the timeout error to
            // a `forced` terminal state in the lifecycle event.
            match tokio::time::timeout(self.config.no_vsock_timeout, wait_loop).await {
                Ok(result) => result,
                Err(_) => Err(CellosError::Host(format!(
                    "vsock exit-code wait timed out after {}s for cell {} \
                     (CELLOS_FIRECRACKER_ALLOW_NO_VSOCK=1). Most likely cause: \
                     guest kernel has no virtio-vsock support — verify \
                     CONFIG_VIRTIO_VSOCKETS=y (NOT VIRTIO_VSOCK — that's a \
                     non-existent symbol) in scripts/firecracker/kernel.config \
                     and rebuild. Set CELLOS_FIRECRACKER_ALLOW_NO_VSOCK=0 to \
                     wait indefinitely.",
                    self.config.no_vsock_timeout.as_secs(),
                    cell_id,
                ))),
            }
        } else {
            wait_loop.await
        }
    }
}

#[cfg(target_os = "linux")]
#[async_trait]
impl CellBackend for FirecrackerCellBackend {
    /// Boot a Firecracker microVM for the cell.
    ///
    /// The VM is configured with the image paths from [`FirecrackerConfig`].
    /// If `spec.environment.imageDigest` is set it is recorded but not yet
    /// verified by this crate (digest verification is a future L2-06 milestone).
    #[instrument(skip(self, spec), fields(cell_id = %spec.spec.id))]
    async fn create(&self, spec: &ExecutionCellDocument) -> Result<CellHandle, CellosError> {
        if spec.spec.id.is_empty() {
            return Err(CellosError::InvalidSpec("spec.id must be non-empty".into()));
        }

        // Mandatory jailer enforcement. Production deployments must run
        // Firecracker under the jailer (chroot + uid/gid drop + seccomp).
        // Operators may opt out for development by setting
        // CELLOS_FIRECRACKER_ALLOW_NO_JAILER=1 which flips this flag to false.
        if self.config.require_jailer && self.config.jailer_binary_path.is_none() {
            return Err(CellosError::Host(
                "jailer is required for production use (set CELLOS_FIRECRACKER_JAILER_BINARY or CELLOS_FIRECRACKER_ALLOW_NO_JAILER=1 to opt out)"
                    .into(),
            ));
        }

        // Refuse to start a cell that declares egress when the host backend
        // cannot enforce it.  This is a fail-closed guard: silently ignoring
        // egressRules would let a spec believe its network is locked down when
        // in fact the VM has unrestricted host networking (or none, depending
        // on the operator's TAP setup).
        let declared_egress: &[EgressRule] = spec
            .spec
            .authority
            .egress_rules
            .as_deref()
            .unwrap_or_default();
        if !self.config.enable_network && !declared_egress.is_empty() {
            return Err(CellosError::Host(
                "spec declares egress_rules but network enforcement is disabled \
                 (set CELLOS_FIRECRACKER_ENABLE_NETWORK=1)"
                    .into(),
            ));
        }

        // Unique run token per cell.
        let run_token = Uuid::new_v4();

        // Compute the API socket path. With jailer active this is inside the
        // chroot; without the jailer it is a per-run file in `socket_dir`.
        let socket_path = resolve_socket_path(&self.config, &spec.spec.id, &run_token);

        // Log declared environment if present.
        if let Some(env) = &spec.spec.environment {
            tracing::info!(
                image_reference = %env.image_reference,
                image_digest = env.image_digest.as_deref().unwrap_or("(not pinned)"),
                template_id = env.template_id.as_deref().unwrap_or("(none)"),
                "cell environment declared"
            );
        }

        // Build the spawn command. When the jailer is configured, invoke it
        // instead of firecracker directly so the VMM ends up in a chroot under
        // a dedicated uid/gid. Otherwise (development mode) spawn firecracker
        // directly.
        //
        // String fields passed via `args(...)` must outlive the array, so we
        // materialize the path strings first and then borrow them.
        let mut cmd = if let Some(jailer_bin) = &self.config.jailer_binary_path {
            let exec_file_str = self.config.binary_path.to_string_lossy().into_owned();
            let uid_str = self.config.jailer_uid.to_string();
            let gid_str = self.config.jailer_gid.to_string();
            let chroot_str = self.config.chroot_base_dir.to_string_lossy().into_owned();
            let mut c = tokio::process::Command::new(jailer_bin);
            let argv = build_jailer_argv(
                spec.spec.id.as_str(),
                exec_file_str.as_str(),
                uid_str.as_str(),
                gid_str.as_str(),
                chroot_str.as_str(),
                self.config.no_seccomp,
            );
            c.args(&argv);
            c
        } else {
            let socket_str = socket_path.to_string_lossy().into_owned();
            let mut c = tokio::process::Command::new(&self.config.binary_path);
            let argv = build_direct_argv(socket_str.as_str(), self.config.no_seccomp);
            c.args(&argv);
            c
        };
        cmd.kill_on_drop(true);

        let child = cmd.spawn().map_err(|e| {
            let bin = if let Some(j) = &self.config.jailer_binary_path {
                j.display().to_string()
            } else {
                self.config.binary_path.display().to_string()
            };
            let label = if self.config.jailer_binary_path.is_some() {
                "jailer"
            } else {
                "firecracker"
            };
            CellosError::Host(format!("spawn {label} ({bin}): {e}"))
        })?;

        tracing::info!(
            cell_id = %spec.spec.id,
            socket = %socket_path.display(),
            "firecracker process spawned"
        );

        // Wait for the socket to become available.
        let client = FirecrackerApiClient::new(&socket_path);
        wait_for_socket_ready(&socket_path, SOCKET_READY_TIMEOUT).await?;

        // Derive the vsock UDS base path.  Firecracker will connect to
        // `<vsock_uds_path>_<port>` when the guest initiates a connection.
        let vsock_uds_path = self.config.socket_dir.join(format!(
            "cellos-vsock-{}-{}.socket",
            spec.spec.id, run_token
        ));

        // FC-18: generate a fresh per-cell HMAC key BEFORE binding the
        // listener, so the same key value flows into both the listener task
        // (for verification) and the kernel cmdline (for the guest's tag
        // computation). A short-read on /dev/urandom is fatal — refuse to
        // launch a cell with weak/known authentication material.
        let exit_hmac_key = generate_exit_hmac_key()?;

        // Start listening for the exit-code report from cellos-init.
        // We must bind the socket BEFORE booting the VM so no connection is missed.
        let (exit_watch_tx, exit_watch_rx) = tokio::sync::watch::channel::<Option<i32>>(None);
        let exit_socket_path = PathBuf::from(format!("{}_9000", vsock_uds_path.display()));
        let exit_socket_path_bg = exit_socket_path.clone();
        let listener_key = exit_hmac_key;
        let listener_cell_id = spec.spec.id.clone();
        tokio::spawn(async move {
            match listen_for_exit_code(&exit_socket_path_bg, &listener_key, &listener_cell_id).await
            {
                Ok(code) => {
                    let _ = exit_watch_tx.send(Some(code));
                }
                Err(e) => {
                    tracing::warn!(error = %e, "vsock exit-code listener failed");
                }
            }
            let _ = std::fs::remove_file(&exit_socket_path_bg);
        });

        // Optionally create a per-cell scratch ext4 image.
        // When scratch_dir is set, the rootfs will be mounted read-only and this
        // writable drive will be attached as /dev/vdb.
        let scratch_image_path = if let Some(scratch_dir) = &self.config.scratch_dir {
            std::fs::create_dir_all(scratch_dir).map_err(|e| {
                CellosError::Host(format!("create scratch_dir {}: {e}", scratch_dir.display()))
            })?;
            let scratch_path = scratch_dir.join(format!(
                "cellos-scratch-{}-{}.ext4",
                spec.spec.id, run_token
            ));
            let scratch_mib = spec
                .spec
                .run
                .as_ref()
                .and_then(|r| r.limits.as_ref())
                .and_then(|l| l.memory_max_bytes)
                .map(|b| ((b / (1024 * 1024)) as u32).clamp(64, 2048))
                .unwrap_or(512);
            create_scratch_image(&scratch_path, scratch_mib).await?;
            Some(scratch_path)
        } else {
            None
        };

        // Provision per-cell network isolation: TAP device + nftables ruleset.
        // The TAP is created BEFORE the API configuration call so Firecracker
        // can `open(/dev/net/tun)` and bind to a device that already exists and
        // is owned by the uid the VMM will drop to.  Failure here is fatal —
        // we do not boot a cell that declared egress without enforcement.
        let cell_short = cell_id_short(&spec.spec.id);
        let tap_iface = if self.config.enable_network {
            let name = create_tap_device(&cell_short, self.config.jailer_uid).await?;
            if let Err(e) = apply_network_policy(&cell_short, &name, declared_egress).await {
                // TAP was created; nftables were not applied. Clean up the TAP
                // before surfacing the error so we don't orphan the device.
                let _ = delete_tap_device(&name).await;
                return Err(e);
            }
            Some(name)
        } else {
            None
        };

        // L2-06-2 warm-pool fast path. Consult the pool BEFORE the cold-boot
        // `configure_vm` + `InstanceStart` sequence. A populated slot lets us
        // skip ~125ms of kernel decompression + init handshake in favour of a
        // ~10ms `PUT /snapshot/load`. When the pool is disabled
        // (`CELLOS_FIRECRACKER_POOL_SIZE` unset/zero) or has no `Available`
        // slot, `checkout` returns `None` and we drop through to the
        // cold-boot path verbatim.
        //
        // Path discipline: `fill_one_slot` writes the paired snapshot files
        // at `/tmp/cellos-pool-<vm_id>.snap` (state) and
        // `/tmp/cellos-pool-<vm_id>.mem` (memory dump). `checkout` returns
        // only the state path; the mem path is derived by suffix-swap so
        // the contract stays a single `Option<PathBuf>` instead of widening
        // to a 2-tuple. If the on-disk convention diverges from this swap,
        // the restore will fail and the caller (this site) falls back to
        // cold boot — the warm pool is a best-effort latency optimisation,
        // not a correctness gate (mirrors the doc on `FirecrackerPool::fill`).
        let (pool_snapshot, pre_checkout_available): (Option<(PathBuf, PathBuf)>, usize) = {
            let mut pool = self.pool.lock().await;
            // Capture Available slot count *before* checkout — the audit
            // event records the supply observed at decision time, not the
            // post-decrement count.
            let pre_available = pool.available();
            let snap = pool.checkout(&spec.spec.id).await.map(|snap_path| {
                let mem_path = snap_path.with_extension("mem");
                (snap_path, mem_path)
            });
            (snap, pre_available)
        };

        // FC-warm-pool audit: emit a single `pool_checkout` CloudEvent per
        // create() call, recording whether the fast path was taken and the
        // pool's pre-checkout supply. Best-effort: a sink failure is logged
        // at warn level and never converted into a create() failure. When
        // no event sink is wired (the default in tests and minimal
        // compositions) this branch is a no-op.
        if let Some(ref event_sink) = self.event_sink {
            let event = cellos_core::events::cloud_event_v1_firecracker_pool_checkout(
                "cellos-host-firecracker",
                &chrono::Utc::now().to_rfc3339(),
                &spec.spec.id,
                pool_snapshot.is_some(),
                pre_checkout_available,
            );
            if let Err(e) = event_sink.emit(&event).await {
                tracing::warn!(
                    target: "cellos.host.firecracker",
                    cell_id = %spec.spec.id,
                    error = %e,
                    "pool_checkout CloudEvent emit failed (best-effort)"
                );
            }
        }

        // Configure the VM, verify artifact digests, and boot — all three can
        // fail.  If any step fails after TAP+nftables are provisioned, clean
        // them up before returning so resources track live cells only.
        let boot_result: Result<VerifiedDigests, CellosError> = async {
            if let Some((snap_path, mem_path)) = pool_snapshot.as_ref() {
                tracing::info!(
                    cell_id = %spec.spec.id,
                    snapshot = %snap_path.display(),
                    mem = %mem_path.display(),
                    "warm-pool fast path: attempting PUT /snapshot/load"
                );
                // Manifest digest verification still applies — a snapshotted
                // VM is only as trustworthy as the kernel/rootfs/firecracker
                // binaries it was captured from. We verify against the
                // current config's pinned digests so a swapped-out artifact
                // on disk still trips the manifest gate.
                let verified = verify_artifacts(&self.config).await?;
                // `restore_into` issues `PUT /snapshot/load` with
                // `resume_vm: true`, so the VM is running on return — no
                // separate `InstanceStart` call is needed on this path.
                pool::restore_into(&client, snap_path, mem_path).await?;
                return Ok(verified);
            }

            configure_vm(
                &client,
                &self.config,
                spec,
                &vsock_uds_path,
                scratch_image_path.as_deref(),
                tap_iface.as_deref(),
                &exit_hmac_key,
            )
            .await?;

            // Verify artifact digests BEFORE booting the VM. If any digest does
            // not match the manifest, refuse to start the cell — running an
            // unverified kernel/rootfs/firecracker binary in a production
            // deployment is exactly the kind of supply-chain failure mode the
            // manifest exists to prevent.
            //
            // FC-08: capture the verified digests so we can surface them on
            // the [`CellHandle`] for the supervisor to embed on
            // `cell.lifecycle.v1.started`.
            let verified = verify_artifacts(&self.config).await?;

            // Boot the VM.
            let status = client
                .put(
                    "/actions",
                    &InstanceAction {
                        action_type: InstanceActionType::InstanceStart,
                    },
                )
                .await?;

            if !status.is_success() {
                return Err(CellosError::Host(format!(
                    "firecracker InstanceStart returned HTTP {status}"
                )));
            }

            Ok(verified)
        }
        .await;

        let verified_digests = match boot_result {
            Ok(v) => v,
            Err(e) => {
                if let Some(ref tap) = tap_iface {
                    let _ = delete_tap_device(tap).await;
                    let _ = remove_network_policy(&cell_short).await;
                }
                // If the warm-pool fast path failed mid-restore, release the
                // slot back to `Empty` so a background filler can re-populate
                // it from a fresh boot. Leaving the slot in `InUse` would
                // permanently shrink the warm pool every time a restore
                // errors. `checkin` is a no-op when the slot wasn't held by
                // this cell, so the call is safe regardless of whether the
                // fast path was taken.
                if pool_snapshot.is_some() {
                    let _ = self.pool.lock().await.checkin(&spec.spec.id).await;
                }
                return Err(e);
            }
        };

        tracing::info!(cell_id = %spec.spec.id, "firecracker VM booted");

        // Track the live VM.
        let chroot_cell_dir = self.config.jailer_binary_path.as_ref().map(|_| {
            let fc_name = self
                .config
                .binary_path
                .file_name()
                .expect("firecracker binary path must have a filename")
                .to_string_lossy()
                .into_owned();
            self.config
                .chroot_base_dir
                .join(fc_name)
                .join(&spec.spec.id)
        });
        // Capture the nft enforcement signal *before* `tap_iface` moves into
        // `VmRecord`. Surfacing this on the returned [`CellHandle`] lets the
        // supervisor emit a `network_enforcement` CloudEvent on the in-VM exit
        // path with parity to the host-subprocess path: `Some(true)` when a TAP
        // was provisioned (which implies `apply_network_policy` ran), or
        // `Some(false)` when networking was explicitly disabled.
        let nft_rules_applied = Some(tap_iface.is_some());

        let graceful_shutdown_timeout = resolve_graceful_shutdown_timeout(&spec.spec);

        self.running_vms.lock().await.insert(
            spec.spec.id.clone(),
            VmRecord {
                socket_path,
                vsock_uds_path,
                child,
                exit_rx: exit_watch_rx,
                chroot_cell_dir,
                scratch_image_path,
                tap_iface,
                graceful_shutdown_timeout,
            },
        );

        Ok(CellHandle {
            cell_id: spec.spec.id.clone(),
            cgroup_path: None,
            nft_rules_applied,
            // FC-08: surface the verified manifest digests so the supervisor
            // can stamp them onto cell.lifecycle.v1.started. When manifest
            // verification was skipped (allow_no_manifest opt-out), all three
            // are `None` and the started event omits the fields entirely.
            kernel_digest_sha256: verified_digests.kernel,
            rootfs_digest_sha256: verified_digests.rootfs,
            firecracker_digest_sha256: verified_digests.firecracker,
        })
    }

    /// Wait for `cellos-init` inside the VM to report the cell command's exit
    /// code over vsock, then return it.
    ///
    /// This overrides the default `None` so the supervisor skips its host-side
    /// `run_cell_command` path and waits for the in-VM result instead.
    async fn wait_for_in_vm_exit(&self, cell_id: &str) -> Option<Result<i32, CellosError>> {
        Some(self.wait_for_command_exit(cell_id).await)
    }

    /// Gracefully shut down the Firecracker VM, then SIGKILL if it does not
    /// exit within the cell's graceful-shutdown window.
    ///
    /// The window is the per-spec `run.limits.gracefulShutdownSeconds` (FC-21),
    /// captured into [`VmRecord::graceful_shutdown_timeout`] at `create()`,
    /// or [`GRACEFUL_SHUTDOWN_TIMEOUT`] when the spec omits the field.
    #[instrument(skip(self, handle), fields(cell_id = %handle.cell_id))]
    async fn destroy(&self, handle: &CellHandle) -> Result<TeardownReport, CellosError> {
        let mut vms = self.running_vms.lock().await;
        let Some(mut record) = vms.remove(&handle.cell_id) else {
            tracing::warn!(cell_id = %handle.cell_id, "destroy called on unknown cell");
            return Ok(TeardownReport {
                cell_id: handle.cell_id.clone(),
                destroyed: false,
                peers_tracked_after: vms.len(),
            });
        };

        // Try graceful shutdown via the API.
        let client = FirecrackerApiClient::new(&record.socket_path);
        let graceful = client
            .put(
                "/actions",
                &InstanceAction {
                    action_type: InstanceActionType::SendCtrlAltDel,
                },
            )
            .await;

        if let Err(e) = graceful {
            tracing::debug!(error = %e, "graceful shutdown request failed — will SIGKILL");
        }

        // Wait for the process to exit, or SIGKILL after timeout. FC-21: prefer
        // the per-spec window captured at create() — falls back to the const.
        let exited = tokio::time::timeout(record.graceful_shutdown_timeout, record.child.wait())
            .await
            .ok();

        if exited.is_none() {
            tracing::warn!(cell_id = %handle.cell_id, "VM did not exit gracefully — sending SIGKILL");
            let _ = record.child.kill().await;
            let _ = record.child.wait().await;
        }

        // Clean up socket files (best effort).
        if record.socket_path.exists() {
            let _ = std::fs::remove_file(&record.socket_path);
        }
        // Best-effort cleanup of jailer chroot tree. Only present when the
        // jailer was active for this cell.
        if let Some(chroot_dir) = record.chroot_cell_dir {
            let _ = std::fs::remove_dir_all(&chroot_dir);
        }
        // Also clean up the vsock UDS base file and any per-port listener sockets.
        let _ = std::fs::remove_file(&record.vsock_uds_path);
        let vsock_exit_socket = PathBuf::from(format!(
            "{}_{VSOCK_EXIT_PORT}",
            record.vsock_uds_path.display()
        ));
        let _ = std::fs::remove_file(&vsock_exit_socket);

        // Remove per-cell scratch image if one was created.
        if let Some(scratch) = record.scratch_image_path {
            let _ = std::fs::remove_file(&scratch);
        }

        // Tear down per-cell network isolation.  Both calls are idempotent and
        // best-effort: a cell whose TAP/nftables setup half-failed during
        // `create()` may have only one of these resources, and we still want
        // `destroy()` to run to completion so the VmRecord stays out of the
        // map.
        if let Some(tap) = record.tap_iface.as_deref() {
            if let Err(e) = delete_tap_device(tap).await {
                tracing::warn!(error = %e, tap = %tap, "delete TAP device failed");
            }
        }
        let cell_short = cell_id_short(&handle.cell_id);
        if let Err(e) = remove_network_policy(&cell_short).await {
            tracing::warn!(error = %e, cell_short = %cell_short, "remove nftables policy failed");
        }

        tracing::info!(cell_id = %handle.cell_id, "firecracker VM destroyed");

        let peers_after = vms.len();
        Ok(TeardownReport {
            cell_id: handle.cell_id.clone(),
            destroyed: true,
            peers_tracked_after: peers_after,
        })
    }
}

// ── Non-Linux stub `CellBackend` impl ────────────────────────────────────────
//
// On Windows/macOS the supervisor's composition root still depends on the
// `FirecrackerCellBackend` type so that `cargo check --workspace` compiles.
// At runtime the backend is only ever selected when
// `CELLOS_CELL_BACKEND=firecracker`, and on a non-Linux host that
// configuration is itself an operator error — every `CellBackend` method
// here returns a clear `Host` error instead of attempting Linux-specific
// I/O. The `from_env()` constructor still succeeds so a developer running
// `cargo check` doesn't need any Firecracker artefacts on disk.

#[cfg(not(target_os = "linux"))]
#[async_trait]
impl CellBackend for FirecrackerCellBackend {
    async fn create(&self, _spec: &ExecutionCellDocument) -> Result<CellHandle, CellosError> {
        Err(CellosError::Host(
            "FirecrackerCellBackend is only supported on Linux \
             (Firecracker requires Linux/KVM); compiled as a stub on this host"
                .into(),
        ))
    }

    async fn destroy(&self, _handle: &CellHandle) -> Result<TeardownReport, CellosError> {
        Err(CellosError::Host(
            "FirecrackerCellBackend is only supported on Linux \
             (Firecracker requires Linux/KVM); compiled as a stub on this host"
                .into(),
        ))
    }
}

// ── Helpers (Linux-only) ──────────────────────────────────────────────────────
//
// Everything below this banner depends on Linux-specific runtime surface:
// `tokio::net::{UnixStream, UnixListener}` (the Firecracker management API
// transport), the `ip`/`nft` host commands, and TAP devices. Gated as one
// block so the cross-platform compile only sees the `FirecrackerConfig`
// data type plus the no-op stub `CellBackend` impl below.

/// Wait for the Firecracker Unix socket to exist and accept connections.
#[cfg(target_os = "linux")]
async fn wait_for_socket_ready(
    socket_path: &Path,
    connect_timeout: Duration,
) -> Result<(), CellosError> {
    let deadline = tokio::time::Instant::now() + connect_timeout;
    loop {
        if socket_path.exists() && tokio::net::UnixStream::connect(socket_path).await.is_ok() {
            return Ok(());
        }
        if tokio::time::Instant::now() >= deadline {
            return Err(CellosError::Host(format!(
                "timed out waiting for Firecracker socket at {} ({}s)",
                socket_path.display(),
                connect_timeout.as_secs()
            )));
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

/// Compute the API socket path for a cell.
///
/// When the jailer is active it creates the socket inside the chroot at
/// `<chroot_base>/<firecracker_filename>/<cell_id>/root/run/firecracker.socket`.
/// Without the jailer the simple per-run path in `socket_dir` is used.
#[cfg(target_os = "linux")]
fn resolve_socket_path(config: &FirecrackerConfig, cell_id: &str, run_token: &Uuid) -> PathBuf {
    if config.jailer_binary_path.is_some() {
        let fc_name = config
            .binary_path
            .file_name()
            .expect("firecracker binary path must have a filename")
            .to_string_lossy()
            .into_owned();
        config
            .chroot_base_dir
            .join(fc_name)
            .join(cell_id)
            .join("root/run/firecracker.socket")
    } else {
        config
            .socket_dir
            .join(format!("cellos-fc-{cell_id}-{run_token}.socket"))
    }
}

/// Create a sparse ext4 scratch image at `path` of `size_mib` MiB.
///
/// Uses `dd` (sparse) to allocate a file then `mkfs.ext4` to format it.
/// Both are standard on any Linux host that can run Firecracker.
#[cfg(target_os = "linux")]
async fn create_scratch_image(path: &Path, size_mib: u32) -> Result<(), CellosError> {
    // Sparse file: write no bytes, just set the file size via seek.
    let dd = tokio::process::Command::new("dd")
        .args([
            "if=/dev/zero",
            &format!("of={}", path.display()),
            "bs=1M",
            "count=0",
            &format!("seek={size_mib}"),
        ])
        .output()
        .await
        .map_err(|e| CellosError::Host(format!("dd for scratch image: {e}")))?;
    if !dd.status.success() {
        return Err(CellosError::Host(format!(
            "dd failed creating scratch image at {}: exit {:?}",
            path.display(),
            dd.status.code()
        )));
    }
    // Format as ext4.
    let mkfs = tokio::process::Command::new("mkfs.ext4")
        .args(["-F", &path.to_string_lossy()])
        .output()
        .await
        .map_err(|e| CellosError::Host(format!("mkfs.ext4 for scratch image: {e}")))?;
    if !mkfs.status.success() {
        return Err(CellosError::Host(format!(
            "mkfs.ext4 failed on {}: exit {:?}",
            path.display(),
            mkfs.status.code()
        )));
    }
    Ok(())
}

/// Send configuration PUT requests to the Firecracker API.
#[cfg(target_os = "linux")]
async fn configure_vm(
    client: &FirecrackerApiClient,
    config: &FirecrackerConfig,
    spec: &ExecutionCellDocument,
    vsock_uds_path: &Path,
    scratch_image_path: Option<&Path>,
    tap_iface: Option<&str>,
    exit_hmac_key: &[u8],
) -> Result<(), CellosError> {
    // L2-06-4: validate the jailer security configuration BEFORE issuing any
    // boot-side API calls. If the operator has somehow landed here with uid=0,
    // gid=0, or chroot=/, refuse to configure the VM — better to fail loudly
    // than to silently boot a cell without the privilege boundary we claim to
    // enforce.
    validate_jailer_security_config(config)?;

    // L2-06-3: machine memory limit derives from `spec.run.limits.memoryMax`,
    // falling back to the static default. Extracted into `derive_mem_size_mib`
    // so the precedence is unit-tested.
    let mem_mib = derive_mem_size_mib(&spec.spec, DEFAULT_MEM_SIZE_MIB);

    let machine_status = client
        .put(
            "/machine-config",
            &MachineConfig {
                vcpu_count: derive_vcpu_count(&spec.spec),
                mem_size_mib: mem_mib,
                track_dirty_pages: false,
            },
        )
        .await?;

    if !machine_status.is_success() {
        return Err(CellosError::Host(format!(
            "firecracker PUT /machine-config returned HTTP {machine_status}"
        )));
    }

    // Boot source — kernel + boot args encoding the cell command.
    //
    // The rootfs boots into cellos-init, which parses `cellos.argv=<base64>`
    // from the kernel cmdline, executes the cell command in-VM, and forwards
    // the exit code to the host over vsock. Remaining L2-06 work is packaging
    // and production hardening rather than host-side subprocess fallback.
    let boot_args = build_boot_args(spec, Some(exit_hmac_key));
    let boot_status = client
        .put(
            "/boot-source",
            &BootSource {
                kernel_image_path: config.kernel_image_path.to_string_lossy().into_owned(),
                boot_args: Some(boot_args),
            },
        )
        .await?;

    if !boot_status.is_success() {
        return Err(CellosError::Host(format!(
            "firecracker PUT /boot-source returned HTTP {boot_status}"
        )));
    }

    // L2-06-1: if the spec pins a content-addressable image digest, hash the
    // configured rootfs file and refuse to attach it unless the on-disk bytes
    // match. The host-wide manifest (verify_artifacts) is the supply-chain
    // boundary (was THIS rootfs binary built by the trusted pipeline?); this
    // is the per-cell content-addressing boundary (is THIS rootfs the exact
    // image the spec asked for?). Both must hold for a production cell to
    // boot.
    //
    // sha256_file streams in 64 KiB chunks; even multi-hundred-MiB rootfs
    // images do not balloon RSS. Hash on the blocking pool so the runtime
    // stays responsive during large-image verifications.
    if let Some(env) = spec.spec.environment.as_ref() {
        if let Some(expected) = env.image_digest.as_ref() {
            let rootfs_owned = config.rootfs_image_path.clone();
            let expected_owned = expected.clone();
            tokio::task::spawn_blocking(move || {
                verify_rootfs_digest(&rootfs_owned, &expected_owned)
            })
            .await
            .map_err(|e| {
                CellosError::Host(format!(
                    "rootfs digest verification task panicked or was cancelled: {e}"
                ))
            })??;
            tracing::info!(
                rootfs = %config.rootfs_image_path.display(),
                expected_digest = %expected,
                "rootfs content-addressing verified (L2-06-1)"
            );
        }
    }

    // Root drive.
    //
    // When a scratch image is configured, the rootfs is mounted read-only so
    // multiple concurrent cells can safely share the same base image; per-cell
    // writes go to the dedicated scratch drive.
    let drive_status = client
        .put(
            "/drives/rootfs",
            &Drive {
                drive_id: "rootfs".into(),
                path_on_host: config.rootfs_image_path.to_string_lossy().into_owned(),
                is_root_device: true,
                is_read_only: scratch_image_path.is_some(),
            },
        )
        .await?;

    if !drive_status.is_success() {
        return Err(CellosError::Host(format!(
            "firecracker PUT /drives/rootfs returned HTTP {drive_status}"
        )));
    }

    // Optional writable scratch drive (attached as /dev/vdb in the guest).
    if let Some(scratch) = scratch_image_path {
        let scratch_status = client
            .put(
                "/drives/scratch",
                &Drive {
                    drive_id: "scratch".into(),
                    path_on_host: scratch.to_string_lossy().into_owned(),
                    is_root_device: false,
                    is_read_only: false,
                },
            )
            .await?;
        if !scratch_status.is_success() {
            return Err(CellosError::Host(format!(
                "firecracker PUT /drives/scratch returned HTTP {scratch_status}"
            )));
        }
    }

    // Virtio-net device, when a host TAP has been provisioned for this cell.
    // The TAP must already exist on the host and be owned by the uid the VMM
    // will run under (Firecracker calls `open(/dev/net/tun)` and binds via
    // `TUNSETIFF` — it does not create the device itself).
    if let Some(tap) = tap_iface {
        let net_status = client
            .put(
                "/network-interfaces/eth0",
                &NetworkInterface {
                    iface_id: "eth0".into(),
                    guest_mac: GUEST_NIC_MAC.into(),
                    host_dev_name: tap.to_owned(),
                },
            )
            .await?;
        if !net_status.is_success() {
            return Err(CellosError::Host(format!(
                "firecracker PUT /network-interfaces/eth0 returned HTTP {net_status}"
            )));
        }
    }

    // Vsock device — enables cellos-init to send the exit code back to the host.
    // Guest-initiated connections to port P arrive at `<vsock_uds_path>_P` on
    // the host.  The host listener for port VSOCK_EXIT_PORT is already bound
    // before this call.
    let vsock_status = client
        .put(
            "/vsock",
            &VsockDevice {
                guest_cid: VSOCK_GUEST_CID,
                uds_path: vsock_uds_path.to_string_lossy().into_owned(),
            },
        )
        .await?;

    if !vsock_status.is_success() {
        return Err(CellosError::Host(format!(
            "firecracker PUT /vsock returned HTTP {vsock_status}"
        )));
    }

    Ok(())
}

/// Build kernel command-line boot args for the cell.
///
/// Standard Linux console args + `cellos.*` parameters consumed by
/// `cellos-init` running as PID 1 inside the guest:
///
/// - `cellos.cell_id` — cell identifier for log attribution
/// - `cellos.vsock_port` — port on the host to send the exit code to
/// - `cellos.argv` — base64-encoded JSON array of strings (the cell command)
/// - `cellos.exit_hmac_key` — FC-18: base64-encoded per-cell HMAC-SHA256 key,
///   plumbed only when `exit_hmac_key.is_some()`. The guest uses this key to
///   authenticate its exit-code report; the host verifies the tag in
///   `listen_for_exit_code`.
///
/// If `spec.run.argv` is empty the `cellos.argv` parameter is omitted; the
/// guest boots normally but `cellos-init` will log an error and exit.
#[cfg(target_os = "linux")]
pub(crate) fn build_boot_args(
    spec: &ExecutionCellDocument,
    exit_hmac_key: Option<&[u8]>,
) -> String {
    // root=/dev/vda mounts the virtio-blk root drive (the rootfs.ext4 image) as
    // the kernel's root filesystem. Without it the kernel panics with
    // "Unable to mount root fs on unknown-block(0,0)" after serial init.
    // ipv6.disable=1: prevent the guest kernel from bringing up an IPv6 stack.
    // The nftables ruleset uses `table ip` (IPv4-only); without this flag a
    // guest workload that pulls in an IPv6 default route bypasses the allowlist
    // entirely. Defense-in-depth: the ip6 table also enforces policy drop (see
    // build_nftables_ruleset), but disabling at boot is the primary control.
    let base = "console=ttyS0 reboot=k panic=1 pci=off ipv6.disable=1 root=/dev/vda rw";
    let cell_id = &spec.spec.id;
    let mut args = format!("{base} cellos.cell_id={cell_id} cellos.vsock_port={VSOCK_EXIT_PORT}");

    // Encode spec.run.argv as base64(json) for cellos-init.
    if let Some(argv) = spec
        .spec
        .run
        .as_ref()
        .map(|r| &r.argv)
        .filter(|a| !a.is_empty())
    {
        if let Ok(json) = serde_json::to_string(argv) {
            let b64 = BASE64_STANDARD.encode(json.as_bytes());
            args.push_str(&format!(" cellos.argv={b64}"));
        }
    }

    // FC-18: plumb the per-cell HMAC key when one was generated. Tokens on
    // the kernel cmdline are space-separated, so base64 (no padding, no
    // whitespace) is the natural encoding — same trick `cellos.argv` uses.
    if let Some(key) = exit_hmac_key {
        let b64 = BASE64_STANDARD.encode(key);
        args.push_str(&format!(" cellos.exit_hmac_key={b64}"));
    }

    args
}

/// Argv passed to the **jailer** binary when running Firecracker under it.
///
/// Extracted as a pure function so the wire-level args can be regression-tested.
/// Two silent reverts in commit d14f134 turned `--level` back into `--log-level`
/// and dropped `root=/dev/vda rw` from boot args; both were caught only at
/// runtime by Firecracker rejecting `--log-level` with exit 153 and the kernel
/// panicking at root mount. Tests asserting these literal strings are present
/// (and the wrong strings are absent) close that regression hole.
#[cfg(target_os = "linux")]
pub(crate) fn build_jailer_argv<'a>(
    spec_id: &'a str,
    exec_file: &'a str,
    uid: &'a str,
    gid: &'a str,
    chroot: &'a str,
    no_seccomp: bool,
) -> Vec<&'a str> {
    let mut argv = vec![
        "--id",
        spec_id,
        "--exec-file",
        exec_file,
        "--uid",
        uid,
        "--gid",
        gid,
        "--chroot-base-dir",
        chroot,
        "--",
        "--api-sock",
        "/run/firecracker.socket",
        "--level",
        "Error",
    ];
    if no_seccomp {
        argv.push("--no-seccomp");
    }
    argv
}

/// Argv passed to the Firecracker binary directly when the jailer is bypassed
/// (development mode — `CELLOS_FIRECRACKER_ALLOW_NO_JAILER=1`).
#[cfg(target_os = "linux")]
pub(crate) fn build_direct_argv(socket_path: &str, no_seccomp: bool) -> Vec<&str> {
    let mut argv = vec!["--api-sock", socket_path, "--level", "Error"];
    if no_seccomp {
        argv.push("--no-seccomp");
    }
    argv
}

// ── Path helpers ──────────────────────────────────────────────────────────────

fn required_absolute_path<F>(
    lookup: &F,
    key: &str,
    description: &str,
) -> Result<PathBuf, CellosError>
where
    F: Fn(&str) -> Option<String>,
{
    let value = lookup(key)
        .ok_or_else(|| missing_env_error(key, description))?
        .trim()
        .to_owned();
    if value.is_empty() {
        return Err(missing_env_error(key, description));
    }
    parse_absolute_path(key, description, &value)
}

fn optional_absolute_path<F>(
    lookup: &F,
    key: &str,
    description: &str,
) -> Result<Option<PathBuf>, CellosError>
where
    F: Fn(&str) -> Option<String>,
{
    let Some(value) = lookup(key) else {
        return Ok(None);
    };
    let value = value.trim();
    if value.is_empty() {
        return Ok(None);
    }
    Ok(Some(parse_absolute_path(key, description, value)?))
}

fn parse_absolute_path(key: &str, description: &str, raw: &str) -> Result<PathBuf, CellosError> {
    let path = Path::new(raw);
    if !path.is_absolute() {
        return Err(CellosError::Host(format!(
            "{key} must be an absolute path to the {description} when CELLOS_CELL_BACKEND=firecracker"
        )));
    }
    Ok(path.to_path_buf())
}

fn missing_env_error(key: &str, description: &str) -> CellosError {
    CellosError::Host(format!(
        "{key} must be set to an absolute path to the {description} when CELLOS_CELL_BACKEND=firecracker"
    ))
}

// ── Artifact manifest verification ────────────────────────────────────────────

/// One parsed entry from the artifact manifest file.
///
/// Lines look like:
///
/// ```text
/// sha256:<hex>  kernel       /opt/cellos/firecracker/vmlinux
/// sha256:<hex>  rootfs       /opt/cellos/firecracker/rootfs.ext4
/// sha256:<hex>  firecracker  /opt/cellos/firecracker/firecracker
/// ```
#[cfg(target_os = "linux")]
#[derive(Debug, Clone, PartialEq, Eq)]
struct ManifestEntry {
    sha256_hex: String,
    role: String,
    path: PathBuf,
}

/// Parse the manifest text. Comment lines (`#`) and blank lines are ignored.
/// Returns an error on any malformed entry — the manifest is a security
/// boundary, so silently skipping garbage lines is unsafe.
#[cfg(target_os = "linux")]
fn parse_manifest(text: &str) -> Result<Vec<ManifestEntry>, CellosError> {
    let mut out = Vec::new();
    for (lineno, raw) in text.lines().enumerate() {
        let line = raw.trim();
        if line.is_empty() || line.starts_with('#') {
            continue;
        }
        // Expect 3 whitespace-delimited fields: sha256:<hex>  <role>  <path>
        let mut parts = line.split_whitespace();
        let digest = parts.next();
        let role = parts.next();
        let path = parts.next();
        let extra = parts.next();
        let (digest, role, path) = match (digest, role, path) {
            (Some(d), Some(r), Some(p)) => (d, r, p),
            _ => {
                return Err(CellosError::Host(format!(
                    "manifest line {}: expected `sha256:<hex>  <role>  <path>`, got: {raw:?}",
                    lineno + 1
                )));
            }
        };
        if extra.is_some() {
            return Err(CellosError::Host(format!(
                "manifest line {}: unexpected trailing field after path",
                lineno + 1
            )));
        }
        let Some(hex) = digest.strip_prefix("sha256:") else {
            return Err(CellosError::Host(format!(
                "manifest line {}: digest field must start with `sha256:`, got: {digest:?}",
                lineno + 1
            )));
        };
        if hex.len() != 64 || !hex.chars().all(|c| c.is_ascii_hexdigit()) {
            return Err(CellosError::Host(format!(
                "manifest line {}: sha256 digest must be 64 hex chars, got: {hex:?}",
                lineno + 1
            )));
        }
        out.push(ManifestEntry {
            sha256_hex: hex.to_ascii_lowercase(),
            role: role.to_string(),
            path: PathBuf::from(path),
        });
    }
    Ok(out)
}

/// Stream-hash a file with SHA256, returning the lowercase hex digest.
///
/// Reads the file in 64 KiB chunks so even multi-hundred-MiB rootfs images do
/// not balloon resident memory.
#[cfg(target_os = "linux")]
fn sha256_file(path: &Path) -> Result<String, CellosError> {
    use sha2::{Digest, Sha256};
    use std::io::Read;

    let mut file = std::fs::File::open(path).map_err(|e| {
        CellosError::Host(format!(
            "open artifact for hashing at {}: {e}",
            path.display()
        ))
    })?;
    let mut hasher = Sha256::new();
    let mut buf = [0u8; 64 * 1024];
    loop {
        let n = file.read(&mut buf).map_err(|e| {
            CellosError::Host(format!(
                "read artifact at {} for hashing: {e}",
                path.display()
            ))
        })?;
        if n == 0 {
            break;
        }
        hasher.update(&buf[..n]);
    }
    let digest = hasher.finalize();
    let mut hex = String::with_capacity(64);
    for byte in digest {
        hex.push_str(&format!("{byte:02x}"));
    }
    Ok(hex)
}

/// L2-06-1 — Verify the SHA256 digest of a rootfs file against an expected
/// value pulled from `spec.environment.imageDigest`. This is content
/// addressing at the *per-cell* layer: even when the host-wide manifest
/// (verify_artifacts) confirms the rootfs binary matches the deployment
/// stamp, a spec can demand a *specific* image digest and the backend must
/// refuse to boot otherwise. Without this gate a poisoned spec referencing
/// the wrong image silently boots the host's default rootfs.
///
/// `expected_sha256` accepts both the bare 64-char hex form and the
/// `sha256:<hex>` prefixed form (cellos-core's `EnvironmentSpec.imageDigest`
/// validates the prefixed form). The comparison is case-insensitive.
///
/// On mismatch returns a forensic-grade error naming the path and both
/// digests so operators can diff them against the spec.
#[cfg(target_os = "linux")]
fn verify_rootfs_digest(path: &Path, expected_sha256: &str) -> Result<String, CellosError> {
    // Normalise: strip optional `sha256:` prefix, lowercase, validate shape.
    let expected_hex = expected_sha256
        .trim()
        .strip_prefix("sha256:")
        .unwrap_or(expected_sha256.trim())
        .to_ascii_lowercase();
    if expected_hex.len() != 64 || !expected_hex.chars().all(|c| c.is_ascii_hexdigit()) {
        return Err(CellosError::Host(format!(
            "verify_rootfs_digest: expected_sha256 must be 64 hex chars (with optional `sha256:` prefix); got {expected_sha256:?}"
        )));
    }
    let actual_hex = sha256_file(path)?;
    if actual_hex != expected_hex {
        return Err(CellosError::Host(format!(
            "rootfs digest mismatch at {}: spec.environment.imageDigest declared sha256:{expected_hex}, on-disk image hashes to sha256:{actual_hex} \
             — refusing to boot a cell against an unverified rootfs (L2-06-1)",
            path.display()
        )));
    }
    Ok(actual_hex)
}

/// L2-06-3 — Derive `mem_size_mib` for the Firecracker `PUT /machine-config`
/// call from the spec's `run.limits.memoryMax`, falling back to the
/// env-derived default when the spec does not declare a memory limit.
///
/// Extracted as a pure function so the precedence (spec → env default) is
/// covered by unit tests without standing up a full backend. The 64-MiB
/// minimum mirrors Firecracker's practical floor — smaller VMs panic during
/// init because the kernel cannot allocate a usable mem_init region.
///
/// Returns `mem_size_mib` clamped to `[64, u32::MAX]`. Callers that need
/// further policy clamping (e.g. fleet-wide max) apply it before passing the
/// spec in.
#[cfg(target_os = "linux")]
fn derive_mem_size_mib(spec: &cellos_core::ExecutionCellSpec, env_default: u32) -> u32 {
    spec.run
        .as_ref()
        .and_then(|r| r.limits.as_ref())
        .and_then(|l| l.memory_max_bytes)
        .map(|bytes| ((bytes / (1024 * 1024)) as u32).max(64))
        .unwrap_or(env_default)
}

/// L2-06-4 — Validate that the live jailer configuration enforces the
/// privilege boundary we depend on for production isolation: non-root uid,
/// non-root gid, and a chroot base that is NOT the filesystem root.
///
/// The constructor for [`FirecrackerConfig`] already rejects uid=0/gid=0
/// (the FC-41 guard), but those checks run at *config-load* time and an
/// in-process mutation or a hand-built `FirecrackerConfig` (tests, future
/// dynamic reconfig) could land at `configure_vm` with insecure values. This
/// function is the second line of defence: called right before
/// `InstanceStart`, it refuses to boot the VM if any of the three invariants
/// fails.
///
/// Errors carry the literal token `L2-06-4` so audit log searches surface
/// them distinctly from the config-load-time FC-41 errors.
///
/// Returns `Ok(())` when the jailer is intentionally disabled (no
/// `jailer_binary_path`) — that path is the operator's explicit dev opt-out,
/// already gated by the loud `CELLOS_FIRECRACKER_ALLOW_NO_JAILER=1` warning
/// in `FirecrackerConfig::from_env`.
#[cfg(target_os = "linux")]
fn validate_jailer_security_config(config: &FirecrackerConfig) -> Result<(), CellosError> {
    // No jailer configured → operator explicitly opted out at config load.
    // Don't double-error here; the config-load WARN is the audit signal.
    if config.jailer_binary_path.is_none() {
        return Ok(());
    }
    if config.jailer_uid == 0 {
        return Err(CellosError::Host(
            "validate_jailer_security_config: jailer_uid=0 — running the jailer as root \
             defeats the privilege boundary that isolates the VMM from the host. \
             [L2-06-4]"
                .into(),
        ));
    }
    if config.jailer_gid == 0 {
        return Err(CellosError::Host(
            "validate_jailer_security_config: jailer_gid=0 — running the jailer in the root \
             group defeats the privilege boundary that isolates the VMM from the host. \
             [L2-06-4]"
                .into(),
        ));
    }
    // chroot_base must be an absolute path AND must not be the filesystem
    // root. `/` as chroot is equivalent to "no chroot at all" — anything the
    // jailer mounts or writes appears in the host's real namespace.
    let chroot = &config.chroot_base_dir;
    if !chroot.is_absolute() {
        return Err(CellosError::Host(format!(
            "validate_jailer_security_config: chroot_base_dir must be an absolute path; got {} [L2-06-4]",
            chroot.display()
        )));
    }
    if chroot == Path::new("/") {
        return Err(CellosError::Host(
            "validate_jailer_security_config: chroot_base_dir=`/` — chroot to filesystem root \
             is functionally no chroot at all. Configure CELLOS_FIRECRACKER_CHROOT_BASE to a \
             dedicated directory like /var/lib/cellos/firecracker. [L2-06-4]"
                .into(),
        ));
    }
    Ok(())
}

/// FC-08 — verified SHA256 hex digests of the boot artifacts that
/// [`verify_artifacts`] hashed and matched against the manifest.
///
/// Surfaced on the [`CellHandle`] so the supervisor can include them on
/// `cell.lifecycle.v1.started`. Lowercase hex (no `sha256:` prefix), exactly
/// 64 chars per field. `firecracker` is `None` when the manifest did not
/// declare a firecracker-role entry (it is optional).
#[cfg(target_os = "linux")]
#[derive(Debug, Clone, Default)]
struct VerifiedDigests {
    kernel: Option<String>,
    rootfs: Option<String>,
    firecracker: Option<String>,
}

/// Verify that the kernel, rootfs (and optionally firecracker) artifacts
/// referenced by the config match the SHA256 digests in the manifest.
///
/// - Returns `Ok(VerifiedDigests::default())` (all `None`) when no manifest is
///   configured AND `allow_no_manifest` is `true` (with a `tracing::warn!` so
///   operators notice they have skipped the verification step).
/// - Returns `Err(CellosError::Host(...))` when no manifest is configured and
///   `allow_no_manifest` is `false` (the default — production posture).
/// - Returns `Err(CellosError::Host(...))` on any parse error, missing role,
///   missing file, read error, or digest mismatch.
/// - Returns `Ok(VerifiedDigests { kernel: Some(_), rootfs: Some(_), firecracker: ... })`
///   on success, with the on-disk hex digests for the supervisor to surface
///   on `cell.lifecycle.v1.started` (FC-08).
#[cfg(target_os = "linux")]
async fn verify_artifacts(config: &FirecrackerConfig) -> Result<VerifiedDigests, CellosError> {
    let Some(manifest_path) = &config.manifest_path else {
        if config.allow_no_manifest {
            tracing::warn!(
                "MANIFEST VERIFICATION DISABLED — pre-boot artifact digest verification is being skipped \
                 because both CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST=1 and \
                 CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY=1 are set. \
                 This is unsafe for production and should only be used for local development."
            );
            return Ok(VerifiedDigests::default());
        }
        return Err(CellosError::Host(
            "firecracker init: CELLOS_FIRECRACKER_MANIFEST is not set \
             — pre-boot artifact digest verification is mandatory by default. \
             Set CELLOS_FIRECRACKER_MANIFEST to a v1 manifest path, or set BOTH \
             CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST=1 AND \
             CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY=1 to opt out \
             (development only — the second flag is a deliberate speed-bump)."
                .into(),
        ));
    };

    // Manifest files are small (a handful of lines); reading synchronously
    // here is fine and avoids pulling in tokio's `fs` feature.
    let text = std::fs::read_to_string(manifest_path).map_err(|e| {
        CellosError::Host(format!(
            "read artifact manifest at {}: {e}",
            manifest_path.display()
        ))
    })?;
    let entries = parse_manifest(&text)?;

    // Build the verification plan. kernel and rootfs are mandatory; firecracker
    // is optional (verified only when an entry is present).
    let kernel_entry = entries.iter().find(|e| e.role == "kernel").ok_or_else(|| {
        CellosError::Host(format!(
            "manifest at {} is missing a `kernel` role entry",
            manifest_path.display()
        ))
    })?;
    let rootfs_entry = entries.iter().find(|e| e.role == "rootfs").ok_or_else(|| {
        CellosError::Host(format!(
            "manifest at {} is missing a `rootfs` role entry",
            manifest_path.display()
        ))
    })?;
    let firecracker_entry = entries.iter().find(|e| e.role == "firecracker");

    // Hash and compare. We intentionally hash the configured paths (not the
    // manifest paths) so a tampered manifest cannot redirect verification to
    // a known-good copy of the file. If the manifest path differs from the
    // configured path, we log it but still verify the configured one.
    let plan: Vec<(&str, &Path, &str, &Path)> = {
        let mut v: Vec<(&str, &Path, &str, &Path)> = vec![
            (
                "kernel",
                config.kernel_image_path.as_path(),
                kernel_entry.sha256_hex.as_str(),
                kernel_entry.path.as_path(),
            ),
            (
                "rootfs",
                config.rootfs_image_path.as_path(),
                rootfs_entry.sha256_hex.as_str(),
                rootfs_entry.path.as_path(),
            ),
        ];
        if let Some(fc) = firecracker_entry {
            v.push((
                "firecracker",
                config.binary_path.as_path(),
                fc.sha256_hex.as_str(),
                fc.path.as_path(),
            ));
        }
        v
    };

    let mut verified = VerifiedDigests::default();
    for (role, configured_path, expected_hex, manifest_decl_path) in plan {
        if configured_path != manifest_decl_path {
            tracing::warn!(
                role,
                configured = %configured_path.display(),
                manifest = %manifest_decl_path.display(),
                "configured artifact path differs from manifest declaration; verifying configured path"
            );
        }
        // sha256_file is sync I/O; offload to the blocking pool so we do not
        // stall the runtime on multi-hundred-MiB rootfs hashes.
        let owned_path = configured_path.to_path_buf();
        let actual_hex = tokio::task::spawn_blocking(move || sha256_file(&owned_path))
            .await
            .map_err(|e| {
                CellosError::Host(format!(
                    "sha256 hashing task for role {role} panicked or was cancelled: {e}"
                ))
            })??;
        if actual_hex != expected_hex {
            // FC-51 + FC-09: emit manifest-failed CloudEvent onto pending buffer
            // before returning the existing host error (sink-unaware call site).
            push_manifest_failed_pending(
                role,
                expected_hex,
                actual_hex.as_str(),
                manifest_path.to_string_lossy().as_ref(),
            );
            return Err(CellosError::Host(format!(
                "artifact digest mismatch for role `{role}` at {}: expected sha256:{expected_hex}, got sha256:{actual_hex}",
                configured_path.display()
            )));
        }
        tracing::info!(
            role,
            path = %configured_path.display(),
            sha256 = %actual_hex,
            "artifact digest verified"
        );
        // FC-08: capture the verified digest so the supervisor can surface it
        // on cell.lifecycle.v1.started. We record only after the on-disk hash
        // matched the manifest, so any value reaching CellHandle is a verified
        // digest by construction.
        match role {
            "kernel" => verified.kernel = Some(actual_hex),
            "rootfs" => verified.rootfs = Some(actual_hex),
            "firecracker" => verified.firecracker = Some(actual_hex),
            _ => {
                // Unknown roles never appear in `plan`; this branch is defensive.
            }
        }
    }

    Ok(verified)
}

// FC-51 + FC-09 — manifest-failed emission seam. `verify_artifacts` is sink-
// unaware (it runs inside `create()`'s boot block); events accumulate on a
// process-wide buffer drained by the supervisor after backend `create()`
// returns Err. Long-term design wires `dyn EventSink` into `CellBackend`.
static MANIFEST_FAILED_PENDING: std::sync::OnceLock<
    std::sync::Mutex<Vec<cellos_core::CloudEventV1>>,
> = std::sync::OnceLock::new();

/// Emit a manifest-failed CloudEvent onto the pending buffer. Public for
/// supervisor drain + FC-51 emission tests; the digest-mismatch branch in
/// `verify_artifacts` calls this with the live verification context.
pub fn push_manifest_failed_pending_for_test(
    role: &str,
    expected_sha256: &str,
    actual_sha256: &str,
    manifest_path: &str,
) {
    push_manifest_failed_pending(role, expected_sha256, actual_sha256, manifest_path);
}

fn push_manifest_failed_pending(role: &str, expected: &str, actual: &str, manifest_path: &str) {
    let data = match cellos_core::manifest_failed_data_v1(role, expected, actual, manifest_path) {
        Ok(d) => d,
        Err(e) => {
            tracing::warn!(error = %e, role, "manifest_failed_data_v1 failed");
            return;
        }
    };
    let ev = cellos_core::CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: "cellos-host-firecracker".into(),
        ty: cellos_core::LIFECYCLE_MANIFEST_FAILED_TYPE.into(),
        datacontenttype: Some("application/json".into()),
        data: Some(data),
        time: None,
        traceparent: None,
    };
    let buf = MANIFEST_FAILED_PENDING.get_or_init(|| std::sync::Mutex::new(Vec::new()));
    if let Ok(mut g) = buf.lock() {
        g.push(ev);
    }
}

/// Drain pending manifest-failed events (consume-on-drain).
pub fn drain_pending_manifest_failed_events() -> Vec<cellos_core::CloudEventV1> {
    let buf = MANIFEST_FAILED_PENDING.get_or_init(|| std::sync::Mutex::new(Vec::new()));
    let mut g = buf.lock().unwrap_or_else(|p| p.into_inner());
    std::mem::take(&mut *g)
}

/// Accept one connection on a Unix socket, read a 4-byte little-endian i32
/// exit code sent by `cellos-init` over vsock (Firecracker proxies the guest
/// vsock connection to this socket), then write a 1-byte ACK back.
///
/// The ACK is part of the exit-code wire protocol — `cellos-init` blocks on
/// reading it before calling `reboot()`. Without the ACK, the guest could
/// power off the VM (and Firecracker could tear down the vsock device)
/// before this read returns, leaving the supervisor to time out and SIGKILL
/// even though the workload exited cleanly. ACK write failures are logged
/// but not propagated: by the time we get here the 4-byte exit code is
/// already committed, and a half-broken ACK path is strictly less bad than
/// failing the whole run.
#[cfg(target_os = "linux")]
/// Bind a Unix-stream listener at `socket_path` and authenticate one
/// exit-code report from the guest's `cellos-init`.
///
/// FC-18: the wire format is `[code:4 LE i32][hmac:32]`. The supervisor
/// recomputes `HMAC-SHA256(hmac_key, code_bytes ‖ cell_id_bytes)` and
/// compares against the received tag. On mismatch (forged or replayed
/// frame) the supervisor logs a structured `vsock_exit_auth_rejected`
/// warning and returns `Err` — the caller treats that as "no exit code
/// received", which causes the supervisor to fall through to the
/// SIGKILL-on-timeout path. Any other code path (`Ok`) is the legitimate
/// cell's exit.
///
/// `hmac_key` is the per-cell key generated in `create()` and plumbed to
/// the guest via `cellos.exit_hmac_key=<base64>`. `cell_id` is bound into
/// the MAC input so a tag minted for one cell cannot be replayed against
/// another cell's listener.
#[cfg(target_os = "linux")]
async fn listen_for_exit_code(
    socket_path: &Path,
    hmac_key: &[u8],
    cell_id: &str,
) -> Result<i32, CellosError> {
    use tokio::io::AsyncWriteExt;

    let listener = UnixListener::bind(socket_path).map_err(|e| {
        CellosError::Host(format!(
            "bind vsock exit listener at {}: {e}",
            socket_path.display()
        ))
    })?;

    let (mut stream, _) = listener.accept().await.map_err(|e| {
        CellosError::Host(format!(
            "accept vsock exit connection at {}: {e}",
            socket_path.display()
        ))
    })?;

    let mut frame = [0u8; EXIT_AUTHED_FRAME_LEN];
    stream.read_exact(&mut frame).await.map_err(|e| {
        CellosError::Host(format!(
            "read vsock exit frame from {}: {e}",
            socket_path.display()
        ))
    })?;

    let mut code_bytes = [0u8; 4];
    code_bytes.copy_from_slice(&frame[..4]);
    let received_tag = &frame[4..];

    if !verify_exit_hmac(hmac_key, &code_bytes, cell_id, received_tag) {
        // Structured rejection: a forged or replayed exit frame. Do NOT ACK
        // the sender — silence forces them to retry/timeout, which is the
        // intended behaviour against an attacker. Returning Err propagates
        // "no legitimate exit observed" up to the supervisor.
        tracing::warn!(
            cell_id = %cell_id,
            socket = %socket_path.display(),
            event = "vsock_exit_auth_rejected",
            "FC-18: rejecting unauthenticated vsock exit frame (HMAC mismatch)"
        );
        return Err(CellosError::Host(format!(
            "vsock_exit_auth_rejected: HMAC mismatch on exit frame for cell {cell_id}"
        )));
    }

    // The ACK byte value is unused by the guest — its arrival is the signal.
    // 0x00 is chosen so any future protocol revision can repurpose non-zero
    // values without ambiguity.
    if let Err(e) = stream.write_all(&[0u8]).await {
        tracing::debug!(
            error = %e,
            socket = %socket_path.display(),
            "vsock ACK write failed (exit code already captured)"
        );
    }

    Ok(i32::from_le_bytes(code_bytes))
}

// ── Network isolation (TAP + nftables) ───────────────────────────────────────

/// Sanitize and truncate a cell-id for use in interface and table names.
///
/// The kernel limits interface names to 15 bytes (`IFNAMSIZ - 1`); combined
/// with the `cfc-` prefix this leaves 11 bytes of slug.  We further constrain
/// to 8 lowercase hex chars derived from SHA-256 of the full cell id — this
/// avoids the zero-padding collision where `"a"` and `"a0000000"` would map
/// to the same slug under a naive right-pad scheme.
#[cfg(target_os = "linux")]
fn cell_id_short(cell_id: &str) -> String {
    use sha2::{Digest, Sha256};
    let digest = Sha256::digest(cell_id.as_bytes());
    format!(
        "{:08x}",
        u32::from_be_bytes([digest[0], digest[1], digest[2], digest[3]])
    )
}

/// Compose the host TAP interface name from a sanitized cell-id slug.
#[cfg(target_os = "linux")]
fn tap_name_for(cell_short: &str) -> String {
    format!("{TAP_NAME_PREFIX}{cell_short}")
}

/// Compose the per-cell nftables table name.
#[cfg(target_os = "linux")]
fn nft_table_name(cell_short: &str) -> String {
    format!("cellos-{cell_short}")
}

/// Create a host TAP interface owned by `uid`, bring it up, and return its
/// name.  Linux-only — returns an error on every other OS.
///
/// The interface is created via `ip tuntap add dev <name> mode tap user <uid>`
/// followed by `ip link set dev <name> up`.  Both commands run via
/// `tokio::process::Command` with each argument passed individually — there is
/// no shell so no injection surface even if a future caller passes hostile
/// input.
#[cfg(target_os = "linux")]
async fn create_tap_device(cell_short: &str, uid: u32) -> Result<String, CellosError> {
    #[cfg(not(target_os = "linux"))]
    {
        let _ = (cell_short, uid);
        Err(CellosError::Host(
            "TAP device creation is only supported on Linux \
             (set CELLOS_FIRECRACKER_ENABLE_NETWORK=0 on this host)"
                .into(),
        ))
    }
    #[cfg(target_os = "linux")]
    {
        let name = tap_name_for(cell_short);
        // Defensive: should never trigger because cell_id_short caps at 8.
        if name.len() > 15 {
            return Err(CellosError::Host(format!(
                "computed TAP name {name:?} exceeds IFNAMSIZ (15)"
            )));
        }

        let uid_str = uid.to_string();
        let add = tokio::process::Command::new("ip")
            .arg("tuntap")
            .arg("add")
            .arg("dev")
            .arg(&name)
            .arg("mode")
            .arg("tap")
            .arg("user")
            .arg(&uid_str)
            .output()
            .await
            .map_err(|e| CellosError::Host(format!("spawn `ip tuntap add` for {name}: {e}")))?;
        if !add.status.success() {
            return Err(CellosError::Host(format!(
                "`ip tuntap add dev {name}` failed: exit {:?} stderr={}",
                add.status.code(),
                String::from_utf8_lossy(&add.stderr).trim()
            )));
        }

        let up = tokio::process::Command::new("ip")
            .arg("link")
            .arg("set")
            .arg("dev")
            .arg(&name)
            .arg("up")
            .output()
            .await
            .map_err(|e| CellosError::Host(format!("spawn `ip link set up` for {name}: {e}")))?;
        if !up.status.success() {
            // Best-effort rollback so we don't leak a half-configured TAP.
            let _ = delete_tap_device(&name).await;
            return Err(CellosError::Host(format!(
                "`ip link set dev {name} up` failed: exit {:?} stderr={}",
                up.status.code(),
                String::from_utf8_lossy(&up.stderr).trim()
            )));
        }

        Ok(name)
    }
}

/// Remove a host TAP interface created by [`create_tap_device`].  Idempotent —
/// returns `Ok(())` even when the device does not exist, so this is safe to
/// call from `destroy()` regardless of how far `create()` got.
#[cfg(target_os = "linux")]
async fn delete_tap_device(name: &str) -> Result<(), CellosError> {
    #[cfg(not(target_os = "linux"))]
    {
        let _ = name;
        Ok(())
    }
    #[cfg(target_os = "linux")]
    {
        let out = tokio::process::Command::new("ip")
            .arg("link")
            .arg("delete")
            .arg(name)
            .output()
            .await
            .map_err(|e| CellosError::Host(format!("spawn `ip link delete` for {name}: {e}")))?;
        if out.status.success() {
            return Ok(());
        }
        // Idempotent: missing device is not a failure.  `ip` writes
        // "Cannot find device" or "does not exist" to stderr in that case.
        let stderr = String::from_utf8_lossy(&out.stderr);
        if stderr.contains("Cannot find device") || stderr.contains("does not exist") {
            return Ok(());
        }
        Err(CellosError::Host(format!(
            "`ip link delete {name}` failed: exit {:?} stderr={}",
            out.status.code(),
            stderr.trim()
        )))
    }
}

// FC-32 integration-test exposure: doc-hidden, Linux-only, additive only.
#[cfg(target_os = "linux")]
#[doc(hidden)]
pub mod __fc32 {
    pub fn cell_id_short(id: &str) -> String {
        super::cell_id_short(id)
    }
    pub fn tap_name_for(s: &str) -> String {
        super::tap_name_for(s)
    }
    pub async fn create_tap_device(s: &str, uid: u32) -> Result<String, super::CellosError> {
        super::create_tap_device(s, uid).await
    }
    pub async fn delete_tap_device(name: &str) -> Result<(), super::CellosError> {
        super::delete_tap_device(name).await
    }
}

// FC-18 integration-test exposure: doc-hidden, additive. Mirrors `__fc32`.
// Platform-agnostic because the verification primitive is.
#[doc(hidden)]
pub mod __fc18 {
    pub const EXIT_HMAC_KEY_LEN: usize = super::EXIT_HMAC_KEY_LEN;
    pub const EXIT_HMAC_TAG_LEN: usize = super::EXIT_HMAC_TAG_LEN;
    pub fn verify_exit_hmac(
        key: &[u8],
        exit_code_bytes: &[u8; 4],
        cell_id: &str,
        received_tag: &[u8],
    ) -> bool {
        super::verify_exit_hmac(key, exit_code_bytes, cell_id, received_tag)
    }
}

/// Build the nftables ruleset text for a cell.
///
/// The ruleset creates a fresh `ip` family table named `cellos-<cell_short>`
/// with two chains:
///
/// - `egress` (hooked at `forward`, priority `filter`): default DROP for
///   packets originating on the cell's TAP interface, with explicit ACCEPT
///   rules for each declared destination.
/// - `output` (hooked at `output`, priority `filter`): no-op accept — keeps
///   host-originated traffic unaffected.
///
/// The function is intentionally synchronous and free of process I/O so it
/// can be exercised by unit tests on any platform.  Hostnames in
/// `egress_rules` are emitted as inline comments only when they fail to parse
/// as IP literals; nftables itself does **not** resolve hostnames, so a
/// non-IP `host` becomes a dropped rule plus a `# unresolved` comment.
/// Resolution to IPs is the caller's responsibility (see
/// [`apply_network_policy`]).
#[cfg(target_os = "linux")]
fn build_nftables_ruleset(
    cell_short: &str,
    tap_iface: &str,
    egress_rules: &[EgressRule],
) -> String {
    use std::fmt::Write as _;

    let table = nft_table_name(cell_short);
    let mut s = String::new();
    let _ = writeln!(s, "table ip {table} {{");
    let _ = writeln!(s, "    chain egress {{");
    let _ = writeln!(
        s,
        "        type filter hook forward priority filter; policy drop;"
    );
    let _ = writeln!(s, "        ct state established,related accept");

    for rule in egress_rules {
        // Normalize protocol: tcp by default; "udp" → udp; anything else → tcp
        // (including "https" and "dns-acknowledged" which are application-level
        // and ride on tcp/udp respectively at the IP layer).
        let l4 = match rule.protocol.as_deref().map(|p| p.to_ascii_lowercase()) {
            Some(ref p) if p == "udp" => "udp",
            Some(ref p) if p == "dns-acknowledged" => "udp",
            _ => "tcp",
        };

        match rule.host.parse::<std::net::IpAddr>() {
            Ok(std::net::IpAddr::V4(ip)) => {
                let _ = writeln!(
                    s,
                    "        iifname \"{tap_iface}\" ip daddr {ip} {l4} dport {port} accept",
                    port = rule.port
                );
            }
            Ok(std::net::IpAddr::V6(_)) => {
                // table ip is IPv4-only; IPv6 addresses cannot be expressed as
                // ip daddr rules.  Emit a comment so operators can see the rule
                // was declared but silently skipped rather than crashing nft.
                let _ = writeln!(
                    s,
                    "        # skipped IPv6 {host:?} port {port} {l4} — table ip is IPv4-only",
                    host = rule.host,
                    port = rule.port
                );
            }
            Err(_) => {
                // Hostname not pre-resolved: emit a comment so operators see
                // that the rule was declared but cannot be applied without DNS.
                let _ = writeln!(
                    s,
                    "        # unresolved host {host:?} port {port} {l4} — no accept rule",
                    host = rule.host,
                    port = rule.port
                );
            }
        }
    }

    let _ = writeln!(s, "        iifname \"{tap_iface}\" drop");
    let _ = writeln!(s, "    }}");
    let _ = writeln!(s, "}}");

    // IPv6 sibling table: policy drop by default, with per-rule accept entries
    // for declared IPv6 destinations. Defense-in-depth alongside ipv6.disable=1
    // in the kernel boot args — if the boot flag is ever removed, this table
    // ensures IPv6 traffic is still allowlist-filtered.
    let _ = writeln!(s, "table ip6 {table} {{");
    let _ = writeln!(s, "    chain egress {{");
    let _ = writeln!(
        s,
        "        type filter hook forward priority filter; policy drop;"
    );
    let _ = writeln!(s, "        ct state established,related accept");

    for rule in egress_rules {
        let l4 = match rule.protocol.as_deref().map(|p| p.to_ascii_lowercase()) {
            Some(ref p) if p == "udp" => "udp",
            Some(ref p) if p == "dns-acknowledged" => "udp",
            _ => "tcp",
        };
        if let Ok(std::net::IpAddr::V6(ip)) = rule.host.parse::<std::net::IpAddr>() {
            let _ = writeln!(
                s,
                "        iifname \"{tap_iface}\" ip6 daddr {ip} {l4} dport {port} accept",
                port = rule.port
            );
        }
    }

    let _ = writeln!(s, "        iifname \"{tap_iface}\" drop");
    let _ = writeln!(s, "    }}");
    let _ = writeln!(s, "}}");
    s
}

/// Resolve hostnames in `egress_rules` to a flat list of `EgressRule` entries
/// whose `host` field is an IP literal.  Each input rule may expand to zero
/// or more output rules (zero on resolution failure, multiple when DNS
/// returns several addresses).
#[cfg(target_os = "linux")]
async fn resolve_egress_targets(egress_rules: &[EgressRule]) -> Vec<EgressRule> {
    let mut resolved = Vec::with_capacity(egress_rules.len());
    for rule in egress_rules {
        if rule.host.parse::<std::net::IpAddr>().is_ok() {
            resolved.push(rule.clone());
            continue;
        }
        match tokio::net::lookup_host((rule.host.as_str(), rule.port)).await {
            Ok(addrs) => {
                let mut any = false;
                for sa in addrs {
                    any = true;
                    resolved.push(EgressRule {
                        host: sa.ip().to_string(),
                        port: rule.port,
                        protocol: rule.protocol.clone(),
                        dns_egress_justification: rule.dns_egress_justification.clone(),
                    });
                }
                if !any {
                    tracing::warn!(host = %rule.host, "DNS returned no addresses; egress rule skipped");
                }
            }
            Err(e) => {
                tracing::warn!(error = %e, host = %rule.host, "DNS resolution failed; egress rule skipped");
            }
        }
    }
    resolved
}

/// Apply the per-cell nftables ruleset by piping the text from
/// [`build_nftables_ruleset`] into `nft -f -`.  Linux-only.
///
/// Hostnames in `egress_rules` are resolved to IPs via the host's resolver
/// before the ruleset is generated; rules whose host fails to resolve are
/// dropped with a warning (the default-DROP policy still applies).
#[cfg(target_os = "linux")]
async fn apply_network_policy(
    cell_short: &str,
    tap_iface: &str,
    egress_rules: &[EgressRule],
) -> Result<(), CellosError> {
    #[cfg(not(target_os = "linux"))]
    {
        let _ = (cell_short, tap_iface, egress_rules);
        Err(CellosError::Host(
            "nftables policy enforcement is only supported on Linux".into(),
        ))
    }
    #[cfg(target_os = "linux")]
    {
        // Make sure no stale table from a previous run with the same cell-id
        // short slug is still present — `nft -f -` with `add table` would
        // append rather than replace.
        let _ = remove_network_policy(cell_short).await;

        let resolved = resolve_egress_targets(egress_rules).await;
        let ruleset = build_nftables_ruleset(cell_short, tap_iface, &resolved);

        let mut child = tokio::process::Command::new("nft")
            .arg("-f")
            .arg("-")
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            .spawn()
            .map_err(|e| CellosError::Host(format!("spawn nft: {e}")))?;

        if let Some(mut stdin) = child.stdin.take() {
            stdin.write_all(ruleset.as_bytes()).await.map_err(|e| {
                CellosError::Host(format!("write nftables ruleset to nft stdin: {e}"))
            })?;
            stdin
                .shutdown()
                .await
                .map_err(|e| CellosError::Host(format!("close nft stdin: {e}")))?;
        }

        let output = child
            .wait_with_output()
            .await
            .map_err(|e| CellosError::Host(format!("wait for nft: {e}")))?;
        if !output.status.success() {
            return Err(CellosError::Host(format!(
                "nft -f - rejected ruleset for {cell_short}: exit {:?} stderr={}",
                output.status.code(),
                String::from_utf8_lossy(&output.stderr).trim()
            )));
        }
        Ok(())
    }
}

/// Drop the per-cell nftables table.  Idempotent — succeeds even when the
/// table does not exist, so it is safe to call from `destroy()` after a
/// half-failed `create()`.
#[cfg(target_os = "linux")]
async fn remove_network_policy(cell_short: &str) -> Result<(), CellosError> {
    #[cfg(not(target_os = "linux"))]
    {
        let _ = cell_short;
        Ok(())
    }
    #[cfg(target_os = "linux")]
    {
        let table = nft_table_name(cell_short);
        let out = tokio::process::Command::new("nft")
            .arg("delete")
            .arg("table")
            .arg("ip")
            .arg(&table)
            .output()
            .await
            .map_err(|e| CellosError::Host(format!("spawn `nft delete table {table}`: {e}")))?;
        if out.status.success() {
            return Ok(());
        }
        let stderr = String::from_utf8_lossy(&out.stderr);
        // nftables emits "No such file or directory" / "does not exist" when
        // the table is absent — that is the expected idempotent path.
        if stderr.contains("No such file or directory")
            || stderr.contains("does not exist")
            || stderr.contains("Could not process rule")
        {
            return Ok(());
        }
        Err(CellosError::Host(format!(
            "`nft delete table ip {table}` failed: exit {:?} stderr={}",
            out.status.code(),
            stderr.trim()
        )))
    }
}

// ── Tests ─────────────────────────────────────────────────────────────────────
//
// The test module is Linux-only because nearly every helper it exercises
// (`listen_for_exit_code`, `build_jailer_argv`, `build_nftables_ruleset`,
// `verify_artifacts`, …) is itself Linux-only — Firecracker is a Linux/KVM
// VMM. Cross-platform `cargo check --workspace` only needs the public API
// surface to compile, not the test bodies.

#[cfg(all(test, target_os = "linux"))]
mod tests {
    use super::*;

    #[test]
    fn config_parses_required_paths() {
        let config = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/firecracker/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/firecracker/vmlinux.bin".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/firecracker/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_JAILER_BINARY" => Some("/opt/firecracker/jailer".into()),
            "CELLOS_FIRECRACKER_CHROOT_BASE" => Some("/var/lib/cellos/firecracker".into()),
            // Opt out of mandatory manifest enforcement; this test exercises
            // path parsing, not the manifest guard.
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();

        assert_eq!(
            config.binary_path,
            PathBuf::from("/opt/firecracker/firecracker")
        );
        assert_eq!(
            config.kernel_image_path,
            PathBuf::from("/opt/firecracker/vmlinux.bin")
        );
        assert_eq!(
            config.rootfs_image_path,
            PathBuf::from("/opt/firecracker/rootfs.ext4")
        );
        assert_eq!(
            config.jailer_binary_path,
            Some(PathBuf::from("/opt/firecracker/jailer"))
        );
        assert_eq!(
            config.chroot_base_dir,
            PathBuf::from("/var/lib/cellos/firecracker")
        );
    }

    #[test]
    fn config_socket_dir_defaults_to_tmp() {
        let config = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/firecracker/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/firecracker/vmlinux.bin".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/firecracker/rootfs.ext4".into()),
            // Opt out of mandatory manifest enforcement; this test exercises
            // socket-dir defaulting, not the manifest guard.
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();
        assert_eq!(config.socket_dir, PathBuf::from("/tmp"));
    }

    #[test]
    fn config_requires_absolute_paths() {
        let err = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/firecracker/vmlinux.bin".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/firecracker/rootfs.ext4".into()),
            _ => None,
        })
        .unwrap_err();

        assert!(err
            .to_string()
            .contains("CELLOS_FIRECRACKER_BINARY must be an absolute path"));
    }

    #[test]
    fn config_requires_binary_path() {
        let err = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/firecracker/vmlinux.bin".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/firecracker/rootfs.ext4".into()),
            _ => None,
        })
        .unwrap_err();

        assert!(err
            .to_string()
            .contains("CELLOS_FIRECRACKER_BINARY must be set"));
    }

    #[test]
    fn build_boot_args_includes_cell_id_and_vsock_port() {
        let doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "my-cell-001",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 }
            }
        }))
        .unwrap();
        let args = build_boot_args(&doc, None);
        assert!(args.contains("console=ttyS0"));
        assert!(args.contains("cellos.cell_id=my-cell-001"));
        assert!(
            args.contains("cellos.vsock_port=9000"),
            "vsock port must be encoded"
        );
    }

    #[test]
    fn build_boot_args_encodes_argv_when_run_present() {
        use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
        use base64::Engine;

        let doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "argv-cell",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 },
                "run": { "argv": ["echo", "hello world"] }
            }
        }))
        .unwrap();
        let args = build_boot_args(&doc, None);

        // Extract cellos.argv=<b64> token and decode it.
        let b64 = args
            .split_ascii_whitespace()
            .find(|t| t.starts_with("cellos.argv="))
            .expect("cellos.argv not found in boot args")
            .strip_prefix("cellos.argv=")
            .unwrap();

        let json = BASE64_STANDARD.decode(b64).expect("base64 decode");
        let decoded: Vec<String> = serde_json::from_slice(&json).expect("json decode");
        assert_eq!(decoded, vec!["echo", "hello world"]);
    }

    /// Regression: commit d14f134 dropped the `root=/dev/vda rw` boot arg.
    /// The kernel then panicked with "Unable to mount root fs on
    /// unknown-block(0,0)" after serial init. The fix was re-applied in
    /// 95a1941. This test pins the requirement so the same revert cannot
    /// happen silently again.
    #[test]
    fn build_boot_args_includes_root_dev_vda_rw() {
        let doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "root-arg-cell",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 }
            }
        }))
        .unwrap();
        let args = build_boot_args(&doc, None);
        assert!(
            args.contains("root=/dev/vda"),
            "boot args MUST set root=/dev/vda — without it kernel panics at root mount. args={args:?}"
        );
        assert!(
            args.contains(" rw"),
            "rootfs must be mounted read-write (cellos-init writes to /proc, /sys mounts). args={args:?}"
        );
        assert!(
            args.contains("console=ttyS0"),
            "console=ttyS0 required for supervisor to read kernel stdout. args={args:?}"
        );
    }

    /// Regression: commit d14f134 reverted the Firecracker arg name from
    /// `--level` to the (never-valid) `--log-level`. Firecracker exited at
    /// startup with `ParseArguments(UnexpectedArgument("log-level"))` and the
    /// supervisor timed out waiting for the API socket. Re-applied in 7f190d7.
    /// This test pins both the correct arg name AND the absence of the wrong
    /// one — a future agent's whole-file copy from a stale base would clobber
    /// the fix without this assertion.
    #[test]
    fn firecracker_argv_uses_level_not_log_level() {
        // Direct (no-jailer) path
        let direct = build_direct_argv("/tmp/fc.sock", false);
        assert!(
            direct.contains(&"--level"),
            "direct argv must contain --level: {direct:?}"
        );
        assert!(
            !direct.contains(&"--log-level"),
            "direct argv must NOT contain --log-level (Firecracker rejects it): {direct:?}"
        );

        // Jailer path
        let jailer = build_jailer_argv(
            "cell-1",
            "/usr/bin/firecracker",
            "1000",
            "1000",
            "/tmp",
            false,
        );
        assert!(
            jailer.contains(&"--level"),
            "jailer argv must contain --level: {jailer:?}"
        );
        assert!(
            !jailer.contains(&"--log-level"),
            "jailer argv must NOT contain --log-level (Firecracker rejects it): {jailer:?}"
        );
    }

    /// Pin the structural shape of the jailer argv (positional ordering, the
    /// `--` separator, the in-jail socket path). Catches accidental reorders
    /// or omissions that would still parse but produce wrong-looking VMs.
    #[test]
    fn build_jailer_argv_has_required_positionals_and_separator() {
        let argv = build_jailer_argv(
            "my-cell",
            "/usr/bin/firecracker",
            "1000",
            "1000",
            "/srv/fc",
            false,
        );
        // ID + exec-file + uid + gid + chroot + -- + api-sock + level
        assert_eq!(argv[0], "--id");
        assert_eq!(argv[1], "my-cell");
        assert_eq!(argv[2], "--exec-file");
        assert_eq!(argv[3], "/usr/bin/firecracker");
        // The `--` separator must precede the firecracker-side args.
        let dash_dash = argv.iter().position(|a| *a == "--").expect("missing --");
        // After --: api-sock + value + level + value
        assert_eq!(argv[dash_dash + 1], "--api-sock");
        assert_eq!(
            argv[dash_dash + 2],
            "/run/firecracker.socket",
            "in-jail socket path must be /run/firecracker.socket"
        );
    }

    #[test]
    fn build_boot_args_omits_argv_when_run_absent() {
        let doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "no-run-cell",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 }
            }
        }))
        .unwrap();
        let args = build_boot_args(&doc, None);
        assert!(
            !args.contains("cellos.argv="),
            "cellos.argv must be absent when spec.run is missing"
        );
    }

    /// FC-18: when an HMAC key is provided, `build_boot_args` must emit a
    /// `cellos.exit_hmac_key=<base64>` token whose decoded bytes equal the
    /// provided key. Pin both the prefix and the round-trip so a future
    /// rename or accidental hex-encoding regresses loudly.
    #[test]
    fn build_boot_args_includes_exit_hmac_key_when_provided() {
        let doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "fc18-cell",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 }
            }
        }))
        .unwrap();
        let key = [0x9Au8; 32];
        let args = build_boot_args(&doc, Some(&key));
        let token = args
            .split_ascii_whitespace()
            .find(|t| t.starts_with("cellos.exit_hmac_key="))
            .expect("FC-18 hmac key token missing");
        let b64 = token.strip_prefix("cellos.exit_hmac_key=").unwrap();
        let decoded = BASE64_STANDARD.decode(b64).expect("base64 decode");
        assert_eq!(decoded, key.to_vec());
    }

    /// FC-18: when no key is provided, the token MUST be absent — the host
    /// listener will then fail-shut on the missing-tag wire shape.
    #[test]
    fn build_boot_args_omits_exit_hmac_key_when_absent() {
        let doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "no-fc18-cell",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 }
            }
        }))
        .unwrap();
        let args = build_boot_args(&doc, None);
        assert!(
            !args.contains("cellos.exit_hmac_key="),
            "cellos.exit_hmac_key must be absent when no key is provided"
        );
    }

    /// Pure helper: a tag minted under the wrong key MUST NOT verify.
    /// This is the FC-18 acceptance property — extracted as a unit test
    /// here so it locks the contract independent of any vsock plumbing.
    #[test]
    fn verify_exit_hmac_rejects_wrong_key() {
        let real_key = [0x01u8; 32];
        let attacker_key = [0x02u8; 32];
        let cell_id = "cell-x";
        let code: i32 = 0;
        let attacker_tag = fc18_compute_tag(&attacker_key, code, cell_id);
        assert!(!verify_exit_hmac(
            &real_key,
            &code.to_le_bytes(),
            cell_id,
            &attacker_tag
        ));
    }

    /// Pure helper: a tag minted for a different cell-id MUST NOT verify
    /// even with the right key — this is the cross-cell replay defence.
    #[test]
    fn verify_exit_hmac_rejects_wrong_cell_id() {
        let key = [0x01u8; 32];
        let code: i32 = 0;
        let other_tag = fc18_compute_tag(&key, code, "other-cell");
        assert!(!verify_exit_hmac(
            &key,
            &code.to_le_bytes(),
            "this-cell",
            &other_tag
        ));
    }

    /// Pure helper: legitimate tag verifies cleanly.
    #[test]
    fn verify_exit_hmac_accepts_legitimate_tag() {
        let key = [0x77u8; 32];
        let cell_id = "the-real-cell";
        let code: i32 = 137;
        let tag = fc18_compute_tag(&key, code, cell_id);
        assert!(verify_exit_hmac(&key, &code.to_le_bytes(), cell_id, &tag));
    }

    /// Pure helper: a tag of the wrong length MUST NOT verify.
    /// Defends against truncation or padding attacks.
    #[test]
    fn verify_exit_hmac_rejects_wrong_length() {
        let key = [0x01u8; 32];
        let code: i32 = 0;
        let bogus = [0u8; 16];
        assert!(!verify_exit_hmac(
            &key,
            &code.to_le_bytes(),
            "any-cell",
            &bogus
        ));
    }

    /// Helper for the listener tests below: precomputes the FC-18 tag a
    /// legitimate guest would produce. Mirrors `cellos-init::compute_exit_hmac`
    /// — kept inline so the test crate doesn't need to depend on
    /// `cellos-init`'s binary surface.
    fn fc18_compute_tag(key: &[u8], code: i32, cell_id: &str) -> [u8; 32] {
        use hmac::{digest::KeyInit, Hmac, Mac};
        use sha2::Sha256;
        type HmacSha256 = Hmac<Sha256>;
        let mut mac = HmacSha256::new_from_slice(key).expect("any key length");
        mac.update(&code.to_le_bytes());
        mac.update(cell_id.as_bytes());
        let tag = mac.finalize().into_bytes();
        let mut out = [0u8; 32];
        out.copy_from_slice(&tag);
        out
    }

    /// Verifies the vsock exit-code listener round-trip: write the FC-18
    /// authenticated frame (4-byte LE code + 32-byte HMAC tag), read the
    /// 1-byte ACK back, verify the listener reads the correct i32. The ACK
    /// read mirrors what cellos-init does in production — the listener
    /// writes the byte unconditionally after a successful HMAC verify.
    #[tokio::test]
    async fn listen_for_exit_code_round_trip() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        use tokio::net::UnixStream;

        let dir = tempfile::tempdir().expect("tmpdir");
        let socket_path = dir.path().join("test_exit.socket");
        let key = [0xAAu8; 32];
        let cell_id = "test-cell-rt";

        // Spawn the listener first.
        let path_clone = socket_path.clone();
        let key_clone = key;
        let cell_id_clone = cell_id.to_string();
        let handle = tokio::spawn(async move {
            listen_for_exit_code(&path_clone, &key_clone, &cell_id_clone).await
        });

        // Give the listener a moment to bind before connecting.
        tokio::time::sleep(Duration::from_millis(10)).await;

        let mut stream = UnixStream::connect(&socket_path)
            .await
            .expect("connect to listener");
        let code = 42i32;
        let tag = fc18_compute_tag(&key, code, cell_id);
        stream
            .write_all(&code.to_le_bytes())
            .await
            .expect("write exit code");
        stream.write_all(&tag).await.expect("write hmac tag");

        let mut ack = [0u8; 1];
        stream.read_exact(&mut ack).await.expect("read ACK");
        assert_eq!(ack[0], 0x00, "host writes 0x00 ACK after verifying frame");

        let received = handle.await.expect("join").expect("listen_for_exit_code");
        assert_eq!(received, 42);
    }

    #[tokio::test]
    async fn listen_for_exit_code_negative_exit_code() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        use tokio::net::UnixStream;

        let dir = tempfile::tempdir().expect("tmpdir");
        let socket_path = dir.path().join("test_exit_neg.socket");
        let key = [0x55u8; 32];
        let cell_id = "test-cell-neg";

        let path_clone = socket_path.clone();
        let key_clone = key;
        let cell_id_clone = cell_id.to_string();
        let handle = tokio::spawn(async move {
            listen_for_exit_code(&path_clone, &key_clone, &cell_id_clone).await
        });

        tokio::time::sleep(Duration::from_millis(10)).await;

        let mut stream = UnixStream::connect(&socket_path).await.expect("connect");
        let code = -1i32;
        let tag = fc18_compute_tag(&key, code, cell_id);
        stream.write_all(&code.to_le_bytes()).await.expect("write");
        stream.write_all(&tag).await.expect("write hmac tag");

        let mut ack = [0u8; 1];
        stream.read_exact(&mut ack).await.expect("read ACK");
        assert_eq!(ack[0], 0x00);

        let received = handle.await.expect("join").expect("listen");
        assert_eq!(received, -1);
    }

    /// Asserts the ACK protocol contract specifically: after the
    /// authenticated frame is read AND verified, the listener writes
    /// exactly one byte and no more, and that byte is the agreed-upon 0x00
    /// sentinel. Reading 2 bytes must hit EOF on the second byte — the
    /// listener closes the stream after the ACK.
    #[tokio::test]
    async fn listen_for_exit_code_writes_exactly_one_ack_byte() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        use tokio::net::UnixStream;

        let dir = tempfile::tempdir().expect("tmpdir");
        let socket_path = dir.path().join("test_exit_ack.socket");
        let key = [0x33u8; 32];
        let cell_id = "test-cell-ack";

        let path_clone = socket_path.clone();
        let key_clone = key;
        let cell_id_clone = cell_id.to_string();
        let handle = tokio::spawn(async move {
            listen_for_exit_code(&path_clone, &key_clone, &cell_id_clone).await
        });

        tokio::time::sleep(Duration::from_millis(10)).await;

        let mut stream = UnixStream::connect(&socket_path).await.expect("connect");
        let code = 7i32;
        let tag = fc18_compute_tag(&key, code, cell_id);
        stream
            .write_all(&code.to_le_bytes())
            .await
            .expect("write exit code");
        stream.write_all(&tag).await.expect("write hmac tag");

        // Drain everything the host sends. The contract is "exactly one byte
        // then EOF", so read_to_end should yield a single 0x00.
        let mut sink = Vec::new();
        stream.read_to_end(&mut sink).await.expect("drain ack");
        assert_eq!(
            sink,
            vec![0x00],
            "listener must write exactly one 0x00 ACK byte then close"
        );

        let code_received = handle.await.expect("join").expect("listen");
        assert_eq!(code_received, 7);
    }

    /// FC-18: the listener MUST reject a frame whose HMAC does not verify.
    /// Behavioural contract: no ACK is written, and the spawned task
    /// surfaces a `vsock_exit_auth_rejected` error rather than a clean
    /// exit code. This is the property that prevents a different process
    /// (CID 3, another cell) from forging the legitimate cell's exit.
    #[tokio::test]
    async fn listen_for_exit_code_rejects_forged_hmac() {
        use tokio::io::AsyncWriteExt;
        use tokio::net::UnixStream;

        let dir = tempfile::tempdir().expect("tmpdir");
        let socket_path = dir.path().join("test_exit_forged.socket");
        let real_key = [0x01u8; 32];
        let attacker_key = [0x02u8; 32];
        let cell_id = "test-cell-forged";

        let path_clone = socket_path.clone();
        let key_clone = real_key;
        let cell_id_clone = cell_id.to_string();
        let handle = tokio::spawn(async move {
            listen_for_exit_code(&path_clone, &key_clone, &cell_id_clone).await
        });

        tokio::time::sleep(Duration::from_millis(10)).await;

        let mut stream = UnixStream::connect(&socket_path).await.expect("connect");
        // Attacker sends a plausible exit code with a tag minted under the
        // wrong key — this is exactly the FC-18 acceptance scenario.
        let attacker_tag = fc18_compute_tag(&attacker_key, 0, cell_id);
        stream
            .write_all(&0i32.to_le_bytes())
            .await
            .expect("write code");
        stream.write_all(&attacker_tag).await.expect("write tag");

        let result = handle.await.expect("join");
        match result {
            Err(CellosError::Host(msg)) => {
                assert!(
                    msg.contains("vsock_exit_auth_rejected"),
                    "expected FC-18 rejection marker in error, got: {msg}"
                );
            }
            other => panic!("expected Err(Host(vsock_exit_auth_rejected)), got {other:?}"),
        }
    }

    #[test]
    fn config_parses_jailer_uid_and_gid() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_JAILER_UID" => Some("10100".into()),
            "CELLOS_FIRECRACKER_JAILER_GID" => Some("10200".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();
        assert_eq!(cfg.jailer_uid, 10100);
        assert_eq!(cfg.jailer_gid, 10200);
    }

    #[test]
    fn config_jailer_uid_gid_default_to_10002() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            // Tests that don't probe manifest behaviour opt out of the
            // mandatory-by-default manifest guard so they exercise their
            // own subject without tripping the unrelated check.
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();
        assert_eq!(cfg.jailer_uid, 10002);
        assert_eq!(cfg.jailer_gid, 10002);
    }

    #[test]
    fn resolve_socket_path_without_jailer_uses_socket_dir() {
        use uuid::Uuid;
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            // Tests that don't probe manifest behaviour opt out of the
            // mandatory-by-default manifest guard so they exercise their
            // own subject without tripping the unrelated check.
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();
        let token = Uuid::nil();
        let path = resolve_socket_path(&cfg, "test-cell", &token);
        assert!(
            path.starts_with(&cfg.socket_dir),
            "expected socket in socket_dir, got {path:?}"
        );
        assert!(path.to_string_lossy().contains("test-cell"));
    }

    #[test]
    fn resolve_socket_path_with_jailer_uses_chroot_base() {
        use uuid::Uuid;
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_JAILER_BINARY" => Some("/opt/fc/jailer".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();
        let token = Uuid::nil();
        let path = resolve_socket_path(&cfg, "test-cell", &token);
        // Expected: <chroot_base>/firecracker/test-cell/root/run/firecracker.socket
        let expected = cfg
            .chroot_base_dir
            .join("firecracker")
            .join("test-cell")
            .join("root/run/firecracker.socket");
        assert_eq!(path, expected, "got: {path:?}");
    }

    /// Verifies that `create()` now attempts to spawn the VMM binary and fails
    /// with a spawn error (not "not implemented") when the binary does not exist.
    /// This proves the L2-06 lifecycle path is wired, not just a stub.
    #[tokio::test]
    async fn create_fails_with_spawn_error_when_binary_missing() {
        let backend = FirecrackerCellBackend::new(FirecrackerConfig {
            binary_path: PathBuf::from("/nonexistent/firecracker"),
            kernel_image_path: PathBuf::from("/opt/firecracker/vmlinux.bin"),
            rootfs_image_path: PathBuf::from("/opt/firecracker/rootfs.ext4"),
            jailer_binary_path: None,
            chroot_base_dir: PathBuf::from("/var/lib/cellos/firecracker"),
            socket_dir: PathBuf::from("/tmp"),
            jailer_uid: 10002,
            jailer_gid: 10002,
            scratch_dir: None,
            manifest_path: None,
            // This pre-existing test validates spawn-error wiring without the
            // jailer; opt out of jailer enforcement so we still hit the spawn path.
            require_jailer: false,
            // Manifest verification opted out for the same reason — keep the
            // test exercising the spawn path, not the new pre-boot guard.
            allow_no_manifest: true,
            // Network enforcement off so this test exercises the spawn path
            // without requiring `ip` / `nft` / CAP_NET_ADMIN in CI.
            enable_network: false,
            allow_no_vsock: false,
            no_vsock_timeout: std::time::Duration::from_secs(5),
            no_seccomp: false,
        });
        let doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "spawn-err-test",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 }
            }
        }))
        .unwrap();

        let err = backend.create(&doc).await.unwrap_err();
        let msg = err.to_string();
        // Must NOT contain the old scaffold message.
        assert!(
            !msg.contains("not implemented"),
            "expected spawn error, got old scaffold message: {msg}"
        );
        // Must mention "spawn" or "nonexistent" (OS spawn failure).
        assert!(
            msg.contains("spawn") || msg.contains("nonexistent") || msg.contains("No such file"),
            "expected spawn error message, got: {msg}"
        );
    }

    // ── derive_vcpu_count tests ─────────────────────────────────────────────

    /// Build an `ExecutionCellSpec` whose `run.limits.cpu_max` matches the args.
    /// All other fields are populated from a minimal valid JSON document so
    /// this stays robust to future required fields on `ExecutionCellSpec`.
    fn make_spec_with_cpu(
        quota_micros: u64,
        period_micros: Option<u64>,
    ) -> cellos_core::ExecutionCellSpec {
        let mut doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "vcpu-test",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 },
                "run": { "argv": [] }
            }
        }))
        .unwrap();
        let run = doc.spec.run.as_mut().expect("run present");
        run.limits = Some(cellos_core::RunLimits {
            memory_max_bytes: None,
            cpu_max: Some(cellos_core::RunCpuMax {
                quota_micros,
                period_micros,
            }),
            graceful_shutdown_seconds: None,
        });
        doc.spec
    }

    #[test]
    fn derive_vcpu_count_no_limits_returns_default() {
        let doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "no-limits",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 }
            }
        }))
        .unwrap();
        assert_eq!(derive_vcpu_count(&doc.spec), DEFAULT_VCPU_COUNT);
    }

    #[test]
    fn derive_vcpu_count_one_full_core() {
        // quota=100_000, period=100_000 → 1 vCPU
        let spec = make_spec_with_cpu(100_000, Some(100_000));
        assert_eq!(derive_vcpu_count(&spec), 1);
    }

    #[test]
    fn derive_vcpu_count_fractional_rounds_up() {
        // quota=50_000, period=100_000 → 0.5 core → 1 vCPU (round up)
        let spec = make_spec_with_cpu(50_000, Some(100_000));
        assert_eq!(derive_vcpu_count(&spec), 1);
    }

    #[test]
    fn derive_vcpu_count_two_cores() {
        // quota=200_000, period=100_000 → 2 vCPU
        let spec = make_spec_with_cpu(200_000, Some(100_000));
        assert_eq!(derive_vcpu_count(&spec), 2);
    }

    #[test]
    fn derive_vcpu_count_clamped_at_32() {
        // quota=10_000_000, period=100_000 → 100 vCPU → clamped to 32
        let spec = make_spec_with_cpu(10_000_000, Some(100_000));
        assert_eq!(derive_vcpu_count(&spec), 32);
    }

    #[test]
    fn derive_vcpu_count_default_period() {
        // period=None defaults to 100_000; quota=150_000 → 2 vCPU (rounds up)
        let spec = make_spec_with_cpu(150_000, None);
        assert_eq!(derive_vcpu_count(&spec), 2);
    }

    // ── scratch_dir config tests ───────────────────────────────────────────

    #[test]
    fn config_parses_scratch_dir() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_SCRATCH_DIR" => Some("/var/lib/cellos/scratch".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();
        assert_eq!(
            cfg.scratch_dir,
            Some(PathBuf::from("/var/lib/cellos/scratch"))
        );
    }

    #[test]
    fn config_scratch_dir_absent_when_not_set() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            // Tests that don't probe manifest behaviour opt out of the
            // mandatory-by-default manifest guard so they exercise their
            // own subject without tripping the unrelated check.
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();
        assert_eq!(cfg.scratch_dir, None);
    }

    // ── manifest + verify_artifacts tests ──────────────────────────────────

    /// Build a minimal `FirecrackerConfig` for verify_artifacts tests. All
    /// path fields point to the supplied tempdir so we can scribble files
    /// without hitting the host filesystem.
    fn cfg_for_verify(
        kernel: Option<PathBuf>,
        rootfs: Option<PathBuf>,
        manifest: Option<PathBuf>,
    ) -> FirecrackerConfig {
        FirecrackerConfig {
            binary_path: PathBuf::from("/opt/fc/firecracker"),
            kernel_image_path: kernel.unwrap_or_else(|| PathBuf::from("/opt/fc/vmlinux")),
            rootfs_image_path: rootfs.unwrap_or_else(|| PathBuf::from("/opt/fc/rootfs.ext4")),
            jailer_binary_path: None,
            chroot_base_dir: PathBuf::from("/var/lib/cellos/firecracker"),
            socket_dir: PathBuf::from("/tmp"),
            jailer_uid: 10002,
            jailer_gid: 10002,
            scratch_dir: None,
            manifest_path: manifest,
            require_jailer: false,
            // Default to dev opt-out so that existing tests that pass
            // `manifest=None` continue to verify the "skip" path. Tests that
            // exercise the mandatory-by-default behaviour flip this to false
            // explicitly.
            allow_no_manifest: true,
            enable_network: false,
            allow_no_vsock: false,
            no_vsock_timeout: std::time::Duration::from_secs(5),
            no_seccomp: false,
        }
    }

    #[tokio::test]
    async fn verify_artifacts_skips_when_no_manifest_and_opt_out() {
        // No manifest configured AND allow_no_manifest=true → returns Ok(())
        // and does not touch any file.
        let cfg = cfg_for_verify(None, None, None);
        verify_artifacts(&cfg).await.expect(
            "verify_artifacts should succeed when manifest_path is None and allow_no_manifest=true",
        );
    }

    #[tokio::test]
    async fn verify_artifacts_errors_when_no_manifest_and_no_opt_out() {
        // Mandatory-by-default posture: manifest unset and allow_no_manifest
        // unset → hard error from verify_artifacts. The error message must
        // also name the second escape-hatch flag because the dev opt-out
        // is two-flag (FC-05 / SEAM-23 hardening).
        let mut cfg = cfg_for_verify(None, None, None);
        cfg.allow_no_manifest = false;
        let err = verify_artifacts(&cfg)
            .await
            .expect_err("verify_artifacts must reject missing manifest by default");
        let msg = err.to_string();
        assert!(
            msg.contains("CELLOS_FIRECRACKER_MANIFEST is not set"),
            "expected manifest-missing error, got: {msg}"
        );
        assert!(
            msg.contains("CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST"),
            "error must mention the dev opt-out hint, got: {msg}"
        );
        assert!(
            msg.contains("CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY"),
            "error must name the paired escape-hatch flag (FC-05 hardening), got: {msg}"
        );
    }

    #[tokio::test]
    async fn verify_artifacts_fails_on_wrong_hash() {
        use std::io::Write;
        let dir = tempfile::tempdir().expect("tmpdir");
        let kernel_path = dir.path().join("vmlinux");
        let rootfs_path = dir.path().join("rootfs.ext4");
        std::fs::File::create(&kernel_path)
            .expect("kernel")
            .write_all(b"hello kernel")
            .expect("write kernel");
        std::fs::File::create(&rootfs_path)
            .expect("rootfs")
            .write_all(b"hello rootfs")
            .expect("write rootfs");

        // Manifest with a deliberately wrong hash for kernel (all zeroes).
        let manifest_path = dir.path().join("manifest.txt");
        let manifest = format!(
            "# CellOS Firecracker artifact manifest v1\n\
             sha256:{wrong_kernel}  kernel  {kernel}\n\
             sha256:{wrong_rootfs}  rootfs  {rootfs}\n",
            wrong_kernel = "0".repeat(64),
            wrong_rootfs = "0".repeat(64),
            kernel = kernel_path.display(),
            rootfs = rootfs_path.display(),
        );
        std::fs::write(&manifest_path, manifest).expect("write manifest");

        let cfg = cfg_for_verify(
            Some(kernel_path.clone()),
            Some(rootfs_path),
            Some(manifest_path),
        );

        let err = verify_artifacts(&cfg)
            .await
            .expect_err("expected digest mismatch error");
        let msg = err.to_string();
        assert!(
            msg.contains("digest mismatch"),
            "expected digest mismatch error, got: {msg}"
        );
        assert!(
            msg.contains("kernel"),
            "expected the failing role to be reported, got: {msg}"
        );
    }

    #[tokio::test]
    async fn verify_artifacts_succeeds_on_correct_hash() {
        use std::io::Write;
        let dir = tempfile::tempdir().expect("tmpdir");
        let kernel_path = dir.path().join("vmlinux");
        let rootfs_path = dir.path().join("rootfs.ext4");

        let kernel_bytes: &[u8] = b"hello kernel";
        let rootfs_bytes: &[u8] = b"hello rootfs";
        std::fs::File::create(&kernel_path)
            .expect("kernel")
            .write_all(kernel_bytes)
            .expect("write kernel");
        std::fs::File::create(&rootfs_path)
            .expect("rootfs")
            .write_all(rootfs_bytes)
            .expect("write rootfs");

        // Compute the real digests via the same helper used in the code path.
        let kernel_hex = sha256_file(&kernel_path).expect("hash kernel");
        let rootfs_hex = sha256_file(&rootfs_path).expect("hash rootfs");

        let manifest_path = dir.path().join("manifest.txt");
        let manifest = format!(
            "# good manifest\n\
             sha256:{kernel_hex}  kernel  {kernel}\n\
             sha256:{rootfs_hex}  rootfs  {rootfs}\n",
            kernel = kernel_path.display(),
            rootfs = rootfs_path.display(),
        );
        std::fs::write(&manifest_path, manifest).expect("write manifest");

        let cfg = cfg_for_verify(Some(kernel_path), Some(rootfs_path), Some(manifest_path));

        verify_artifacts(&cfg)
            .await
            .expect("verify_artifacts should succeed when digests match");
    }

    /// E1 / FC-01 acceptance gate: when the on-disk SHA256 differs from the
    /// stamped manifest, `verify_artifacts` MUST fail closed with a forensic
    /// error message that names the role and echoes the manifest-declared
    /// digest. This is the hook every E2E test relies on — without it, a
    /// silently-mismatched kernel could boot.
    ///
    /// We can't reach `create()`'s `InstanceStart` path on non-Linux without a
    /// real KVM + firecracker binary, so this test pins the digest-mismatch
    /// error contract on the same `verify_artifacts` function `create()`
    /// invokes (lib.rs `boot_result` block, BEFORE `InstanceStart`). On
    /// Linux CI the firecracker-e2e job exercises the full end-to-end path.
    #[tokio::test]
    async fn verify_artifacts_fc01_digest_mismatch_names_role_and_expected_digest() {
        use std::io::Write;
        let dir = tempfile::tempdir().expect("tmpdir");
        let kernel_path = dir.path().join("vmlinux");
        let rootfs_path = dir.path().join("rootfs.ext4");
        std::fs::File::create(&kernel_path)
            .expect("kernel")
            .write_all(b"on-disk kernel bytes")
            .expect("write kernel");
        std::fs::File::create(&rootfs_path)
            .expect("rootfs")
            .write_all(b"on-disk rootfs bytes")
            .expect("write rootfs");

        // Stamp the manifest with a deliberately wrong (but well-formed)
        // kernel digest. The rootfs digest is correct, so the failure must
        // attribute to the kernel role specifically.
        let real_rootfs_hex = sha256_file(&rootfs_path).expect("hash rootfs");
        let wrong_kernel_hex = "f".repeat(64);

        let manifest_path = dir.path().join("manifest.txt");
        let manifest = format!(
            "# FC-01 fail-closed fixture\n\
             sha256:{wrong_kernel_hex}  kernel  {kernel}\n\
             sha256:{real_rootfs_hex}  rootfs  {rootfs}\n",
            kernel = kernel_path.display(),
            rootfs = rootfs_path.display(),
        );
        std::fs::write(&manifest_path, manifest).expect("write manifest");

        let cfg = cfg_for_verify(
            Some(kernel_path.clone()),
            Some(rootfs_path),
            Some(manifest_path),
        );
        let err = verify_artifacts(&cfg)
            .await
            .expect_err("digest mismatch must fail closed");
        let msg = err.to_string();

        // Forensic contract: error names the role and echoes the
        // manifest-declared (expected) digest. Operators reading supervisor
        // logs must be able to compare both digests without re-hashing.
        assert!(
            msg.contains("kernel"),
            "error must name the failing role; got: {msg}"
        );
        assert!(
            msg.contains(&wrong_kernel_hex),
            "error must echo the manifest-declared (expected) digest; got: {msg}"
        );
        assert!(
            msg.contains("digest mismatch"),
            "error must use the canonical `digest mismatch` phrase \
             (runbook + log-grep contract); got: {msg}"
        );
    }

    /// E1 / FC-01 + FC-51 wiring gate: on a real digest mismatch,
    /// `verify_artifacts` MUST push a `cell.lifecycle.v1.manifest_failed`
    /// CloudEvent onto the process-wide pending buffer that the supervisor
    /// drains via `drain_pending_manifest_failed_events`. The
    /// `fc51_manifest_failed_emission.rs` integration test covers the
    /// test-seam (`push_manifest_failed_pending_for_test`) but NOT the live
    /// emission path inside `verify_artifacts`. Without this test the wiring
    /// is theoretical — a refactor that drops the
    /// `push_manifest_failed_pending(role, …)` call inside the mismatch
    /// branch would still pass the error-string assertions above while
    /// silently regressing the CloudEvent contract supervisor.rs depends
    /// on for FC-09 (visible-failure-on-mismatch).
    #[tokio::test]
    async fn verify_artifacts_fc01_digest_mismatch_emits_manifest_failed_event() {
        use std::io::Write;
        // Drain any events left behind by a previously-running test in the
        // same process. The pending buffer is process-wide (OnceLock<Mutex<…>>)
        // and cargo runs the module tests on a shared runtime, so we cannot
        // assume the buffer is empty at entry.
        let _pre = drain_pending_manifest_failed_events();

        let dir = tempfile::tempdir().expect("tmpdir");
        let kernel_path = dir.path().join("vmlinux");
        let rootfs_path = dir.path().join("rootfs.ext4");
        std::fs::File::create(&kernel_path)
            .expect("kernel")
            .write_all(b"on-disk kernel bytes")
            .expect("write kernel");
        std::fs::File::create(&rootfs_path)
            .expect("rootfs")
            .write_all(b"on-disk rootfs bytes")
            .expect("write rootfs");

        let real_rootfs_hex = sha256_file(&rootfs_path).expect("hash rootfs");
        let wrong_kernel_hex = "e".repeat(64);

        let manifest_path = dir.path().join("manifest.txt");
        let manifest = format!(
            "# FC-51 emission fixture\n\
             sha256:{wrong_kernel_hex}  kernel  {kernel}\n\
             sha256:{real_rootfs_hex}  rootfs  {rootfs}\n",
            kernel = kernel_path.display(),
            rootfs = rootfs_path.display(),
        );
        std::fs::write(&manifest_path, &manifest).expect("write manifest");

        let cfg = cfg_for_verify(
            Some(kernel_path),
            Some(rootfs_path),
            Some(manifest_path.clone()),
        );

        // Must fail closed BEFORE any boot action could be taken.
        let _err = verify_artifacts(&cfg)
            .await
            .expect_err("digest mismatch must fail closed");

        // …AND the live emission path must have pushed the event the
        // supervisor will drain for FC-09 (`cell.lifecycle.v1.manifest_failed`).
        let drained = drain_pending_manifest_failed_events();
        assert!(
            !drained.is_empty(),
            "verify_artifacts must emit a manifest_failed CloudEvent on \
             digest mismatch (FC-51 wiring); pending buffer was empty"
        );
        let ev = drained
            .iter()
            .find(|e| e.ty == cellos_core::LIFECYCLE_MANIFEST_FAILED_TYPE)
            .expect("at least one event must use LIFECYCLE_MANIFEST_FAILED_TYPE");
        assert_eq!(ev.source, "cellos-host-firecracker");
        let data_str = ev
            .data
            .as_ref()
            .and_then(|d| serde_json::to_string(d).ok())
            .unwrap_or_default();
        assert!(
            data_str.contains("kernel"),
            "event data must name the failing role; got: {data_str}"
        );
        assert!(
            data_str.contains(&wrong_kernel_hex),
            "event data must echo the manifest-declared (expected) digest; got: {data_str}"
        );
    }

    /// E1 / FC-01: when no kernel role is declared in the manifest, the
    /// verification step fails closed with an explicit error. A manifest that
    /// silently skips the kernel role is just as dangerous as one with the
    /// wrong digest — both leave the host booting an unverified kernel image.
    #[tokio::test]
    async fn verify_artifacts_fc01_missing_kernel_role_fails_closed() {
        use std::io::Write;
        let dir = tempfile::tempdir().expect("tmpdir");
        let rootfs_path = dir.path().join("rootfs.ext4");
        std::fs::File::create(&rootfs_path)
            .expect("rootfs")
            .write_all(b"rootfs bytes")
            .expect("write rootfs");
        let real_rootfs_hex = sha256_file(&rootfs_path).expect("hash rootfs");

        let manifest_path = dir.path().join("manifest.txt");
        let manifest = format!(
            "# missing kernel role\n\
             sha256:{real_rootfs_hex}  rootfs  {rootfs}\n",
            rootfs = rootfs_path.display(),
        );
        std::fs::write(&manifest_path, manifest).expect("write manifest");

        let cfg = cfg_for_verify(None, Some(rootfs_path), Some(manifest_path));
        let err = verify_artifacts(&cfg)
            .await
            .expect_err("manifest without `kernel` role must fail closed");
        assert!(
            err.to_string().contains("kernel"),
            "error must name the missing role; got: {err}"
        );
    }

    #[test]
    fn parse_manifest_rejects_short_digest() {
        let text = "sha256:deadbeef  kernel  /opt/fc/vmlinux\n";
        let err = parse_manifest(text).expect_err("expected parse error");
        assert!(err.to_string().contains("64 hex chars"), "got: {err}");
    }

    #[test]
    fn parse_manifest_rejects_missing_prefix() {
        let text = "deadbeef  kernel  /opt/fc/vmlinux\n";
        let err = parse_manifest(text).expect_err("expected parse error");
        assert!(err.to_string().contains("sha256:"), "got: {err}");
    }

    #[test]
    fn parse_manifest_skips_comments_and_blanks() {
        let hex = "a".repeat(64);
        let text = format!(
            "# header\n\
             \n\
             sha256:{hex}  kernel  /opt/fc/vmlinux\n\
             \n\
             # trailing comment\n"
        );
        let entries = parse_manifest(&text).expect("parse");
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].role, "kernel");
        assert_eq!(entries[0].sha256_hex, hex);
        assert_eq!(entries[0].path, PathBuf::from("/opt/fc/vmlinux"));
    }

    // ── jailer enforcement tests ───────────────────────────────────────────

    #[tokio::test]
    async fn create_fails_when_jailer_required_but_not_configured() {
        let backend = FirecrackerCellBackend::new(FirecrackerConfig {
            binary_path: PathBuf::from("/nonexistent/firecracker"),
            kernel_image_path: PathBuf::from("/opt/firecracker/vmlinux.bin"),
            rootfs_image_path: PathBuf::from("/opt/firecracker/rootfs.ext4"),
            jailer_binary_path: None,
            chroot_base_dir: PathBuf::from("/var/lib/cellos/firecracker"),
            socket_dir: PathBuf::from("/tmp"),
            jailer_uid: 10002,
            jailer_gid: 10002,
            scratch_dir: None,
            manifest_path: None,
            require_jailer: true,
            // Tests for the jailer guard must short-circuit BEFORE the manifest
            // guard runs, so opt out of manifest enforcement here.
            allow_no_manifest: true,
            enable_network: false,
            allow_no_vsock: false,
            no_vsock_timeout: std::time::Duration::from_secs(5),
            no_seccomp: false,
        });
        let doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "jailer-required-test",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 }
            }
        }))
        .unwrap();

        let err = backend.create(&doc).await.unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("jailer is required"),
            "expected jailer-required error, got: {msg}"
        );
        // We must fail BEFORE attempting to spawn — confirm no spawn-error keywords leak.
        assert!(
            !msg.contains("spawn"),
            "create() should reject before spawning, got: {msg}"
        );
    }

    #[tokio::test]
    async fn create_allows_no_jailer_when_opt_out() {
        // require_jailer=false → create() proceeds past the jailer guard and
        // hits the spawn path, which then fails because /nonexistent does not
        // exist. The point is to prove we got past the new guard.
        let backend = FirecrackerCellBackend::new(FirecrackerConfig {
            binary_path: PathBuf::from("/nonexistent/firecracker"),
            kernel_image_path: PathBuf::from("/opt/firecracker/vmlinux.bin"),
            rootfs_image_path: PathBuf::from("/opt/firecracker/rootfs.ext4"),
            jailer_binary_path: None,
            chroot_base_dir: PathBuf::from("/var/lib/cellos/firecracker"),
            socket_dir: PathBuf::from("/tmp"),
            jailer_uid: 10002,
            jailer_gid: 10002,
            scratch_dir: None,
            manifest_path: None,
            require_jailer: false,
            // Test exercises the jailer opt-out path; opt out of manifest
            // enforcement too so we still hit the spawn path.
            allow_no_manifest: true,
            enable_network: false,
            allow_no_vsock: false,
            no_vsock_timeout: std::time::Duration::from_secs(5),
            no_seccomp: false,
        });
        let doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "jailer-optout-test",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 }
            }
        }))
        .unwrap();

        let err = backend.create(&doc).await.unwrap_err();
        let msg = err.to_string();
        assert!(
            !msg.contains("jailer is required"),
            "should have bypassed jailer guard with require_jailer=false, got: {msg}"
        );
    }

    /// Doc contract: when `enable_network=false`, the Firecracker backend must
    /// surface `nft_rules_applied = Some(false)` on the returned [`CellHandle`]
    /// so the supervisor emits a `network_enforcement` CloudEvent (with
    /// `applied=false`) on the in-VM exit path. Without this signal the event
    /// is silently dropped — the observability gap fixed by this commit.
    ///
    /// We can't reach the `Ok(CellHandle ..)` branch from a unit test on
    /// macOS (it requires KVM + a real firecracker binary), so this test
    /// pins the construction expression directly. If the production code
    /// stops using `Some(tap_iface.is_some())` this test must be updated in
    /// lockstep — that is the point.
    #[test]
    fn cell_handle_nft_signal_when_network_disabled() {
        // Mirrors the production expression in `create()` for the
        // `enable_network=false` path, where `tap_iface` is `None`.
        let tap_iface: Option<String> = None;
        let handle = CellHandle {
            cell_id: "doc-contract".to_string(),
            cgroup_path: None,
            nft_rules_applied: Some(tap_iface.is_some()),
            kernel_digest_sha256: None,
            rootfs_digest_sha256: None,
            firecracker_digest_sha256: None,
        };
        assert_eq!(
            handle.nft_rules_applied,
            Some(false),
            "enable_network=false must surface Some(false) for parity with the \
             host-subprocess path so network_enforcement is still observable"
        );
    }

    /// Doc contract: when `enable_network=true` and TAP+nftables were
    /// provisioned, `nft_rules_applied` must be `Some(true)`.
    #[test]
    fn cell_handle_nft_signal_when_network_enabled_and_tap_provisioned() {
        let tap_iface: Option<String> = Some("tap-doc-contract".to_string());
        let handle = CellHandle {
            cell_id: "doc-contract".to_string(),
            cgroup_path: None,
            nft_rules_applied: Some(tap_iface.is_some()),
            kernel_digest_sha256: None,
            rootfs_digest_sha256: None,
            firecracker_digest_sha256: None,
        };
        assert_eq!(handle.nft_rules_applied, Some(true));
    }

    #[test]
    fn config_require_jailer_defaults_to_true() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            // Tests that don't probe manifest behaviour opt out of the
            // mandatory-by-default manifest guard so they exercise their
            // own subject without tripping the unrelated check.
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();
        assert!(cfg.require_jailer, "require_jailer must default to true");
    }

    #[test]
    fn config_require_jailer_flips_off_when_allow_no_jailer() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_JAILER" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();
        assert!(
            !cfg.require_jailer,
            "ALLOW_NO_JAILER=1 must flip require_jailer to false"
        );
    }

    #[test]
    fn config_manifest_path_parsed_when_set() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_MANIFEST" => Some("/etc/cellos/manifest.txt".into()),
            _ => None,
        })
        .unwrap();
        assert_eq!(
            cfg.manifest_path,
            Some(PathBuf::from("/etc/cellos/manifest.txt"))
        );
    }

    #[test]
    fn config_manifest_path_absent_when_not_set() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            // Tests that don't probe manifest behaviour opt out of the
            // mandatory-by-default manifest guard so they exercise their
            // own subject without tripping the unrelated check.
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();
        assert_eq!(cfg.manifest_path, None);
    }

    // ── manifest mandatory-by-default guard (SEAM-23) ───────────────────────
    //
    // These four tests cover every (manifest, allow_no_manifest) combination
    // the operator can produce via env vars:
    //
    //   manifest set,   opt-out unset → ok (production posture)
    //   manifest unset, opt-out unset → hard error (mandatory by default)
    //   manifest unset, opt-out=1     → ok with WARN (development)
    //   manifest set,   opt-out=1     → hard error (inconsistent config)

    // ── ALLOW_NO_VSOCK opt-out (catches "kernel built without vsock"
    //    silently hanging the supervisor instead of failing fast) ────────────

    #[test]
    fn config_allow_no_vsock_default_off() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_MANIFEST" => Some("/etc/cellos/manifest.txt".into()),
            _ => None,
        })
        .expect("base config must build");
        assert!(
            !cfg.allow_no_vsock,
            "allow_no_vsock must default to false (production posture: wait \
             for authenticated in-VM exit code, no timeout)"
        );
        // Default timeout is honoured even when opt-out is off; only consulted
        // when allow_no_vsock=true.
        assert_eq!(cfg.no_vsock_timeout, std::time::Duration::from_secs(5));
    }

    #[test]
    fn config_allow_no_vsock_set_with_default_timeout() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_MANIFEST" => Some("/etc/cellos/manifest.txt".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_VSOCK" => Some("1".into()),
            _ => None,
        })
        .expect("opt-out must build");
        assert!(cfg.allow_no_vsock);
        assert_eq!(cfg.no_vsock_timeout, std::time::Duration::from_secs(5));
    }

    #[test]
    fn config_allow_no_vsock_with_custom_timeout() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_MANIFEST" => Some("/etc/cellos/manifest.txt".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_VSOCK" => Some("1".into()),
            "CELLOS_FIRECRACKER_NO_VSOCK_TIMEOUT_SECS" => Some("30".into()),
            _ => None,
        })
        .expect("custom timeout must build");
        assert_eq!(cfg.no_vsock_timeout, std::time::Duration::from_secs(30));
    }

    #[test]
    fn config_no_vsock_timeout_falls_back_on_garbage() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_MANIFEST" => Some("/etc/cellos/manifest.txt".into()),
            "CELLOS_FIRECRACKER_NO_VSOCK_TIMEOUT_SECS" => Some("not-a-number".into()),
            _ => None,
        })
        .expect("garbage timeout falls back, doesn't error");
        assert_eq!(cfg.no_vsock_timeout, std::time::Duration::from_secs(5));
    }

    #[test]
    fn config_manifest_set_opt_out_unset_is_ok() {
        // Production posture: manifest pinned, no opt-out → from_lookup succeeds.
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_MANIFEST" => Some("/etc/cellos/manifest.txt".into()),
            _ => None,
        })
        .expect("manifest set + opt-out unset must succeed");
        assert_eq!(
            cfg.manifest_path,
            Some(PathBuf::from("/etc/cellos/manifest.txt"))
        );
        assert!(
            !cfg.allow_no_manifest,
            "allow_no_manifest must default to false"
        );
    }

    #[test]
    fn config_manifest_unset_opt_out_unset_is_error() {
        // Mandatory-by-default: missing manifest with no opt-out is a hard
        // error from from_lookup. The error message must mention both env
        // var names so the operator knows how to fix it. The error must
        // also name the second escape-hatch flag because the dev opt-out
        // is two-flag (FC-05 / SEAM-23 hardening).
        let err = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            _ => None,
        })
        .expect_err("missing manifest must be rejected by default");
        let msg = err.to_string();
        assert!(
            msg.contains("CELLOS_FIRECRACKER_MANIFEST is not set"),
            "expected manifest-missing error, got: {msg}"
        );
        assert!(
            msg.contains("CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST"),
            "error must mention dev opt-out hint, got: {msg}"
        );
        assert!(
            msg.contains("CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY"),
            "error must mention the paired second escape-hatch flag, got: {msg}"
        );
        // Mirror the jailer guard: error is prefixed with `firecracker init:`
        // so it's easy to distinguish from runtime errors in supervisor logs.
        assert!(
            msg.contains("firecracker init"),
            "error must include `firecracker init` prefix, got: {msg}"
        );
    }

    #[test]
    fn config_manifest_unset_opt_out_set_is_ok() {
        // Development opt-out: missing manifest is allowed when the operator
        // has explicitly opted out via CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST=1.
        // (The WARN log is emitted as a side effect; we don't assert on it
        // here, but the behaviour mirrors CELLOS_FIRECRACKER_ALLOW_NO_JAILER.)
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .expect("manifest unset + opt-out set must succeed (dev mode)");
        assert_eq!(cfg.manifest_path, None);
        assert!(
            cfg.allow_no_manifest,
            "ALLOW_NO_MANIFEST=1 must flip allow_no_manifest to true"
        );
    }

    #[test]
    fn config_manifest_set_opt_out_set_is_inconsistent_error() {
        // Inconsistent config: both manifest and opt-out set. Almost certainly
        // an operator mistake (e.g. dev env file shipped to production), so
        // we fail closed with a clear error rather than silently picking one.
        let err = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_MANIFEST" => Some("/etc/cellos/manifest.txt".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .expect_err("conflicting manifest + opt-out config must be rejected");
        let msg = err.to_string();
        assert!(
            msg.contains("mutually exclusive"),
            "error must explain conflict, got: {msg}"
        );
        assert!(
            msg.contains("CELLOS_FIRECRACKER_MANIFEST")
                && msg.contains("CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST"),
            "error must name both env vars, got: {msg}"
        );
        assert!(
            msg.contains("firecracker init"),
            "error must include `firecracker init` prefix, got: {msg}"
        );
    }

    // ── two-flag escape-hatch handshake (FC-05 hardening / E2 brief) ───────
    //
    // The `_REALLY` flag exists because a single env var can leak from a dev
    // `.env` to a production rollout. Requiring both flags forces the
    // operator to make the trade-off on the same line, on purpose.

    #[test]
    fn config_first_flag_alone_without_second_is_error() {
        // Operator sets `CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST=1` but forgot
        // (or is testing a leaked env var) — verification stays on, and
        // from_lookup MUST surface an explicit, named error.
        let err = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            // Note: NOT setting `_REALLY`. This is the escape-hatch leak.
            _ => None,
        })
        .expect_err("first flag alone must NOT be accepted");
        let msg = err.to_string();
        assert!(
            msg.contains("CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY"),
            "error must name the paired flag, got: {msg}"
        );
        assert!(
            msg.contains("firecracker init"),
            "error must include `firecracker init` prefix, got: {msg}"
        );
    }

    #[test]
    fn config_second_flag_alone_without_first_is_error() {
        // Symmetric guard: setting `_REALLY=1` without the primary flag is
        // also rejected. This catches a different operator mistake (operator
        // remembered the second flag from a runbook but the first flag was
        // unset by an automation tool, or someone pre-set `_REALLY` thinking
        // it would help).
        let err = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            // Note: NOT setting the primary flag.
            _ => None,
        })
        .expect_err("second flag alone must NOT be accepted");
        let msg = err.to_string();
        assert!(
            msg.contains("two-flag"),
            "error must explain the two-flag handshake, got: {msg}"
        );
        assert!(
            msg.contains("CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST"),
            "error must name the primary flag, got: {msg}"
        );
        assert!(
            msg.contains("firecracker init"),
            "error must include `firecracker init` prefix, got: {msg}"
        );
    }

    #[test]
    fn config_both_flags_set_flips_allow_no_manifest_to_true() {
        // The intended dev-mode posture: BOTH flags, no manifest path. This
        // is the only way to disable digest verification.
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .expect("both flags set + no manifest must succeed (dev mode)");
        assert!(
            cfg.allow_no_manifest,
            "both flags set must flip allow_no_manifest to true"
        );
    }

    #[test]
    fn config_neither_flag_with_manifest_is_production_posture() {
        // Default production posture: manifest pinned, neither escape-hatch
        // flag set. allow_no_manifest must remain false.
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_MANIFEST" => Some("/etc/cellos/manifest.txt".into()),
            _ => None,
        })
        .expect("manifest set + neither opt-out flag must succeed");
        assert!(
            !cfg.allow_no_manifest,
            "production posture must keep allow_no_manifest=false"
        );
    }

    // ── network enforcement (TAP + nftables) ───────────────────────────────

    /// `IFNAMSIZ` on Linux is 16 (15 usable bytes + NUL).  The TAP name is
    /// `cfc-` (4) + 8-char slug = 12 bytes — well under the limit.  This test
    /// guards the invariant against future prefix changes.
    #[test]
    fn tap_name_stays_within_ifnamsiz() {
        let short = cell_id_short("0123456789abcdef-extra-tail-noise");
        assert_eq!(short.len(), 8, "slug must be exactly 8 chars");
        let name = tap_name_for(&short);
        assert!(
            name.len() <= 15,
            "TAP name {name:?} exceeds IFNAMSIZ (15): len={}",
            name.len()
        );
        assert!(name.starts_with("cfc-"));
    }

    /// `cell_id_short` must produce an 8-char hex slug and must NOT collide
    /// when two cell ids share the same alphanumeric prefix (the old zero-pad
    /// scheme made `"a"` and `"a0000000"` identical — this guards against that
    /// regression).
    #[test]
    fn cell_id_short_no_collision_on_short_input() {
        let s = cell_id_short("ab");
        assert_eq!(s, "fb8e20fc");
        assert_eq!(s.len(), 8);
        // Demonstrate no collision: "ab" and "ab000000" must differ.
        assert_ne!(s, cell_id_short("ab000000"));
    }

    /// `cell_id_short` must produce a stable 8-char hex slug even when the
    /// cell id contains non-alphanumeric characters (dashes, slashes, etc.).
    #[test]
    fn cell_id_short_stable_for_non_alphanumeric_input() {
        let s = cell_id_short("MY-Cell/01!@#");
        assert_eq!(s, "485deb36");
        assert_eq!(s.len(), 8);
        assert!(s.chars().all(|c| c.is_ascii_hexdigit()));
    }

    /// Refuse-to-start guard: a spec that declares any egress rule MUST be
    /// rejected when the backend was configured with `enable_network: false`.
    /// This prevents a fail-open mode where the spec believes its network is
    /// locked down but the host backend silently ignores the rules.
    #[tokio::test]
    async fn create_fails_when_network_disabled_but_egress_declared() {
        let backend = FirecrackerCellBackend::new(FirecrackerConfig {
            binary_path: PathBuf::from("/nonexistent/firecracker"),
            kernel_image_path: PathBuf::from("/opt/firecracker/vmlinux.bin"),
            rootfs_image_path: PathBuf::from("/opt/firecracker/rootfs.ext4"),
            jailer_binary_path: None,
            chroot_base_dir: PathBuf::from("/var/lib/cellos/firecracker"),
            socket_dir: PathBuf::from("/tmp"),
            jailer_uid: 10002,
            jailer_gid: 10002,
            scratch_dir: None,
            manifest_path: None,
            require_jailer: false,
            // Test asserts the egress guard fires before any other check; opt
            // out of manifest enforcement so we don't trip that one first.
            allow_no_manifest: true,
            enable_network: false,
            allow_no_vsock: false,
            no_vsock_timeout: std::time::Duration::from_secs(5),
            no_seccomp: false,
        });
        let doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "egress-disabled-test",
                "authority": {
                    "secretRefs": [],
                    "egressRules": [
                        { "host": "api.example.com", "port": 443, "protocol": "https" }
                    ]
                },
                "lifetime": { "ttlSeconds": 60 }
            }
        }))
        .unwrap();

        let err = backend.create(&doc).await.expect_err("create must fail");
        let msg = err.to_string();
        assert!(
            msg.contains("egress_rules"),
            "error must mention egress_rules; got: {msg}"
        );
        // And critically: we must have failed BEFORE any spawn attempt, so
        // the message must NOT mention the missing firecracker binary.
        assert!(
            !msg.contains("/nonexistent/firecracker"),
            "guard must short-circuit before spawn; got: {msg}"
        );
    }

    /// The nftables ruleset formatter is a pure function — exercise it on a
    /// known set of rules and assert the structurally important strings are
    /// present.  This test deliberately avoids invoking `nft` so it runs in
    /// CI on macOS as well as Linux.
    #[test]
    fn nftables_ruleset_format() {
        let rules = vec![
            EgressRule {
                host: "10.0.0.1".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            },
            EgressRule {
                host: "192.168.5.5".into(),
                port: 53,
                protocol: Some("dns-acknowledged".into()),
                dns_egress_justification: Some("operator-approved DNS".into()),
            },
            EgressRule {
                host: "203.0.113.7".into(),
                port: 22,
                protocol: Some("tcp".into()),
                dns_egress_justification: None,
            },
        ];
        let ruleset = build_nftables_ruleset("abcd1234", "cfc-abcd1234", &rules);

        // Table is per-cell.
        assert!(
            ruleset.contains("table ip cellos-abcd1234"),
            "missing per-cell table; got:\n{ruleset}"
        );
        // Default-DROP forward chain hooked at filter priority.
        assert!(
            ruleset.contains("type filter hook forward priority filter; policy drop;"),
            "missing default-drop policy; got:\n{ruleset}"
        );
        // Established/related must always be allowed for return traffic.
        assert!(
            ruleset.contains("ct state established,related accept"),
            "missing conntrack accept; got:\n{ruleset}"
        );
        // Each declared destination produces an explicit accept rule on the
        // cell's TAP interface, with the correct L4 protocol mapping.
        assert!(
            ruleset.contains("iifname \"cfc-abcd1234\" ip daddr 10.0.0.1 tcp dport 443 accept"),
            "missing https accept; got:\n{ruleset}"
        );
        assert!(
            ruleset.contains("iifname \"cfc-abcd1234\" ip daddr 192.168.5.5 udp dport 53 accept"),
            "missing dns-acknowledged accept (must map to udp); got:\n{ruleset}"
        );
        assert!(
            ruleset.contains("iifname \"cfc-abcd1234\" ip daddr 203.0.113.7 tcp dport 22 accept"),
            "missing tcp accept; got:\n{ruleset}"
        );
        // Trailing explicit drop on the TAP — defence-in-depth alongside the
        // chain-level policy drop in case future edits change the chain
        // policy without updating the rules.
        assert!(
            ruleset.contains("iifname \"cfc-abcd1234\" drop"),
            "missing per-iface drop; got:\n{ruleset}"
        );
    }

    /// Hostnames that aren't IP literals must be emitted as comments — not as
    /// `accept` rules — because `nft` does not perform name resolution.  The
    /// caller (`apply_network_policy`) is responsible for pre-resolving;
    /// the formatter must be safe when given a non-IP host as a defensive
    /// fallback.
    #[test]
    fn nftables_ruleset_emits_comment_for_unresolved_hostname() {
        let rules = vec![EgressRule {
            host: "api.example.com".into(),
            port: 443,
            protocol: Some("https".into()),
            dns_egress_justification: None,
        }];
        let ruleset = build_nftables_ruleset("xyz12345", "cfc-xyz12345", &rules);
        assert!(
            ruleset.contains("# unresolved host \"api.example.com\""),
            "missing unresolved-host comment; got:\n{ruleset}"
        );
        assert!(
            !ruleset.contains("daddr api.example.com"),
            "must not emit hostname as IP literal; got:\n{ruleset}"
        );
    }

    /// `enable_network` defaults to the platform default when no env var is
    /// set: true on Linux, false elsewhere.
    #[test]
    fn config_enable_network_default_matches_platform() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            // Tests that don't probe manifest behaviour opt out of the
            // mandatory-by-default manifest guard so they exercise their
            // own subject without tripping the unrelated check.
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();
        assert_eq!(cfg.enable_network, NETWORK_DEFAULT_ENABLED);
    }

    /// `CELLOS_FIRECRACKER_ENABLE_NETWORK=0` forces network off regardless of
    /// platform default.
    #[test]
    fn config_enable_network_env_off_overrides_default() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_ENABLE_NETWORK" => Some("0".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();
        assert!(!cfg.enable_network);
    }

    /// `CELLOS_FIRECRACKER_ENABLE_NETWORK=1` forces network on regardless of
    /// platform default.
    #[test]
    fn config_enable_network_env_on_overrides_default() {
        let cfg = FirecrackerConfig::from_lookup(|key| match key {
            "CELLOS_FIRECRACKER_BINARY" => Some("/opt/fc/firecracker".into()),
            "CELLOS_FIRECRACKER_KERNEL_IMAGE" => Some("/opt/fc/vmlinux".into()),
            "CELLOS_FIRECRACKER_ROOTFS_IMAGE" => Some("/opt/fc/rootfs.ext4".into()),
            "CELLOS_FIRECRACKER_ENABLE_NETWORK" => Some("true".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST" => Some("1".into()),
            "CELLOS_FIRECRACKER_ALLOW_NO_MANIFEST_REALLY" => Some("1".into()),
            _ => None,
        })
        .unwrap();
        assert!(cfg.enable_network);
    }

    // ── L2-06-1 — verify_rootfs_digest tests ───────────────────────────────

    /// A rootfs file whose on-disk SHA256 matches the declared digest must
    /// be accepted. Tests both the bare-hex and `sha256:`-prefixed forms.
    #[test]
    fn verify_rootfs_digest_accepts_matching_hash() {
        use std::io::Write;
        let dir = tempfile::tempdir().expect("tmpdir");
        let path = dir.path().join("rootfs.ext4");
        let bytes: &[u8] = b"deterministic rootfs bytes for L2-06-1 test";
        std::fs::File::create(&path)
            .expect("create rootfs")
            .write_all(bytes)
            .expect("write rootfs");
        let expected = sha256_file(&path).expect("hash rootfs");

        // Bare hex form (no prefix) → ok.
        verify_rootfs_digest(&path, &expected).expect("bare-hex digest must verify");
        // Prefixed `sha256:` form → ok (matches the EnvironmentSpec contract).
        let prefixed = format!("sha256:{expected}");
        verify_rootfs_digest(&path, &prefixed).expect("sha256:-prefixed digest must verify");
        // Mixed-case must round-trip after lowercasing.
        let upper = expected.to_ascii_uppercase();
        verify_rootfs_digest(&path, &upper)
            .expect("uppercase digest must verify (case-insensitive)");
    }

    /// A rootfs file whose on-disk SHA256 does NOT match the declared digest
    /// must be rejected with a forensic error that names the path and both
    /// digests so an operator can diff them against the spec.
    #[test]
    fn verify_rootfs_digest_rejects_mismatched_hash() {
        use std::io::Write;
        let dir = tempfile::tempdir().expect("tmpdir");
        let path = dir.path().join("rootfs.ext4");
        std::fs::File::create(&path)
            .expect("create rootfs")
            .write_all(b"the on-disk bytes")
            .expect("write rootfs");
        // A well-formed but deliberately wrong digest.
        let wrong = "f".repeat(64);

        let err =
            verify_rootfs_digest(&path, &wrong).expect_err("digest mismatch must fail closed");
        let msg = err.to_string();
        assert!(
            msg.contains("rootfs digest mismatch"),
            "error must use canonical phrasing for log-grep contract; got: {msg}"
        );
        assert!(
            msg.contains("L2-06-1"),
            "error must carry the L2-06-1 audit tag; got: {msg}"
        );
        assert!(
            msg.contains(&wrong),
            "error must echo the declared (expected) digest; got: {msg}"
        );
    }

    /// A malformed expected digest (wrong length, non-hex chars) is rejected
    /// without touching the rootfs file. This is the input-validation guard
    /// at the API boundary — without it a typo'd spec produces a confusing
    /// "mismatch" error on a bogus hex string.
    #[test]
    fn verify_rootfs_digest_rejects_malformed_expected() {
        let dir = tempfile::tempdir().expect("tmpdir");
        let path = dir.path().join("rootfs.ext4");
        std::fs::write(&path, b"any bytes").expect("write");

        // Too short.
        let err = verify_rootfs_digest(&path, "deadbeef").expect_err("short hex must reject");
        assert!(err.to_string().contains("64 hex chars"), "got: {err}");

        // Right length, wrong charset.
        let bad_chars = "z".repeat(64);
        let err = verify_rootfs_digest(&path, &bad_chars).expect_err("non-hex chars must reject");
        assert!(err.to_string().contains("64 hex chars"), "got: {err}");
    }

    // ── L2-06-3 — derive_mem_size_mib tests ────────────────────────────────

    /// `spec.run.limits.memoryMax=512 MiB` → `mem_size_mib = 512`.
    /// This is the production happy-path: the supervisor admits a spec, the
    /// admitted memory cap flows directly into the Firecracker machine
    /// configuration, and the VM boots with the right RAM allocation.
    #[test]
    fn derive_mem_size_mib_uses_spec_when_present() {
        let mut doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "mem-test",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 },
                "run": { "argv": [] }
            }
        }))
        .unwrap();
        let run = doc.spec.run.as_mut().expect("run present");
        run.limits = Some(cellos_core::RunLimits {
            memory_max_bytes: Some(512 * 1024 * 1024),
            cpu_max: None,
            graceful_shutdown_seconds: None,
        });
        assert_eq!(derive_mem_size_mib(&doc.spec, DEFAULT_MEM_SIZE_MIB), 512);
    }

    /// No `run.limits` → fall back to the env-derived default. The default
    /// argument is `DEFAULT_MEM_SIZE_MIB` (128 MiB) in production; pass a
    /// different sentinel here to prove the fallback is honoured rather than
    /// hardcoded.
    #[test]
    fn derive_mem_size_mib_falls_back_to_env_default() {
        let doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "mem-default",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 }
            }
        }))
        .unwrap();
        // Sentinel default (256) chosen so we'd notice if 128 leaked.
        assert_eq!(derive_mem_size_mib(&doc.spec, 256), 256);
    }

    /// Sub-MiB allocations clamp UP to 64 MiB (the practical Firecracker
    /// floor below which the guest kernel panics during init).
    #[test]
    fn derive_mem_size_mib_clamps_to_64_minimum() {
        let mut doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "tiny",
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 },
                "run": { "argv": [] }
            }
        }))
        .unwrap();
        let run = doc.spec.run.as_mut().expect("run present");
        run.limits = Some(cellos_core::RunLimits {
            memory_max_bytes: Some(1024), // 1 KiB → would be 0 MiB
            cpu_max: None,
            graceful_shutdown_seconds: None,
        });
        assert_eq!(derive_mem_size_mib(&doc.spec, DEFAULT_MEM_SIZE_MIB), 64);
    }

    // ── L2-06-4 — validate_jailer_security_config tests ────────────────────

    /// A well-formed config (non-root uid/gid, non-`/` chroot, jailer set)
    /// passes validation. This is the production posture.
    #[test]
    fn validate_jailer_security_config_accepts_safe_defaults() {
        let cfg = FirecrackerConfig {
            binary_path: PathBuf::from("/opt/fc/firecracker"),
            kernel_image_path: PathBuf::from("/opt/fc/vmlinux"),
            rootfs_image_path: PathBuf::from("/opt/fc/rootfs.ext4"),
            jailer_binary_path: Some(PathBuf::from("/opt/fc/jailer")),
            chroot_base_dir: PathBuf::from("/var/lib/cellos/firecracker"),
            socket_dir: PathBuf::from("/tmp"),
            jailer_uid: 10002,
            jailer_gid: 10002,
            scratch_dir: None,
            manifest_path: None,
            require_jailer: true,
            allow_no_manifest: true,
            enable_network: false,
            allow_no_vsock: false,
            no_vsock_timeout: std::time::Duration::from_secs(5),
            no_seccomp: false,
        };
        validate_jailer_security_config(&cfg).expect("safe defaults must validate");
    }

    /// uid=0 must be rejected with an `L2-06-4` audit tag.
    #[test]
    fn validate_jailer_security_config_rejects_root_uid() {
        let cfg = FirecrackerConfig {
            binary_path: PathBuf::from("/opt/fc/firecracker"),
            kernel_image_path: PathBuf::from("/opt/fc/vmlinux"),
            rootfs_image_path: PathBuf::from("/opt/fc/rootfs.ext4"),
            jailer_binary_path: Some(PathBuf::from("/opt/fc/jailer")),
            chroot_base_dir: PathBuf::from("/var/lib/cellos/firecracker"),
            socket_dir: PathBuf::from("/tmp"),
            jailer_uid: 0,
            jailer_gid: 10002,
            scratch_dir: None,
            manifest_path: None,
            require_jailer: true,
            allow_no_manifest: true,
            enable_network: false,
            allow_no_vsock: false,
            no_vsock_timeout: std::time::Duration::from_secs(5),
            no_seccomp: false,
        };
        let err = validate_jailer_security_config(&cfg).expect_err("uid=0 must be rejected");
        let msg = err.to_string();
        assert!(msg.contains("jailer_uid=0"), "got: {msg}");
        assert!(msg.contains("L2-06-4"), "audit tag missing: {msg}");
    }

    /// gid=0 must be rejected with an `L2-06-4` audit tag.
    #[test]
    fn validate_jailer_security_config_rejects_root_gid() {
        let cfg = FirecrackerConfig {
            binary_path: PathBuf::from("/opt/fc/firecracker"),
            kernel_image_path: PathBuf::from("/opt/fc/vmlinux"),
            rootfs_image_path: PathBuf::from("/opt/fc/rootfs.ext4"),
            jailer_binary_path: Some(PathBuf::from("/opt/fc/jailer")),
            chroot_base_dir: PathBuf::from("/var/lib/cellos/firecracker"),
            socket_dir: PathBuf::from("/tmp"),
            jailer_uid: 10002,
            jailer_gid: 0,
            scratch_dir: None,
            manifest_path: None,
            require_jailer: true,
            allow_no_manifest: true,
            enable_network: false,
            allow_no_vsock: false,
            no_vsock_timeout: std::time::Duration::from_secs(5),
            no_seccomp: false,
        };
        let err = validate_jailer_security_config(&cfg).expect_err("gid=0 must be rejected");
        let msg = err.to_string();
        assert!(msg.contains("jailer_gid=0"), "got: {msg}");
        assert!(msg.contains("L2-06-4"), "audit tag missing: {msg}");
    }

    /// chroot_base_dir=`/` must be rejected — that's equivalent to no chroot.
    #[test]
    fn validate_jailer_security_config_rejects_root_chroot() {
        let cfg = FirecrackerConfig {
            binary_path: PathBuf::from("/opt/fc/firecracker"),
            kernel_image_path: PathBuf::from("/opt/fc/vmlinux"),
            rootfs_image_path: PathBuf::from("/opt/fc/rootfs.ext4"),
            jailer_binary_path: Some(PathBuf::from("/opt/fc/jailer")),
            chroot_base_dir: PathBuf::from("/"),
            socket_dir: PathBuf::from("/tmp"),
            jailer_uid: 10002,
            jailer_gid: 10002,
            scratch_dir: None,
            manifest_path: None,
            require_jailer: true,
            allow_no_manifest: true,
            enable_network: false,
            allow_no_vsock: false,
            no_vsock_timeout: std::time::Duration::from_secs(5),
            no_seccomp: false,
        };
        let err = validate_jailer_security_config(&cfg).expect_err("chroot=/ must be rejected");
        let msg = err.to_string();
        assert!(msg.contains("chroot_base_dir=`/`"), "got: {msg}");
        assert!(msg.contains("L2-06-4"), "audit tag missing: {msg}");
    }

    /// No jailer configured (dev opt-out via `CELLOS_FIRECRACKER_ALLOW_NO_JAILER=1`)
    /// → validation short-circuits with `Ok(())`. The config-load WARN is the
    /// audit signal for that posture; double-erroring here would prevent
    /// developers from running the backend at all.
    #[test]
    fn validate_jailer_security_config_passes_when_jailer_disabled() {
        let cfg = FirecrackerConfig {
            binary_path: PathBuf::from("/opt/fc/firecracker"),
            kernel_image_path: PathBuf::from("/opt/fc/vmlinux"),
            rootfs_image_path: PathBuf::from("/opt/fc/rootfs.ext4"),
            jailer_binary_path: None, // dev mode
            chroot_base_dir: PathBuf::from("/var/lib/cellos/firecracker"),
            socket_dir: PathBuf::from("/tmp"),
            jailer_uid: 10002,
            jailer_gid: 10002,
            scratch_dir: None,
            manifest_path: None,
            require_jailer: false,
            allow_no_manifest: true,
            enable_network: false,
            allow_no_vsock: false,
            no_vsock_timeout: std::time::Duration::from_secs(5),
            no_seccomp: false,
        };
        validate_jailer_security_config(&cfg).expect("jailer disabled must short-circuit Ok");
    }

    /// `build_jailer_argv` includes `--uid`, `--gid`, `--chroot-base-dir` with
    /// the values handed to it. This is the wire-level pin for L2-06-4: the
    /// validation function above guards the *config*, this test guards the
    /// *argv* that the live jailer process actually sees. If a future refactor
    /// drops one of these flags, the jailer no longer enforces the privilege
    /// boundary even with a correct config — both tests must pass.
    #[test]
    fn build_jailer_argv_includes_uid_gid_chroot_flags() {
        let argv = build_jailer_argv(
            "cell-l2-06-4",
            "/opt/fc/firecracker",
            "10002",
            "10003",
            "/var/lib/cellos/firecracker",
            false,
        );
        // --uid must appear with its value immediately after.
        let uid_pos = argv
            .iter()
            .position(|a| *a == "--uid")
            .expect("--uid must be present");
        assert_eq!(argv[uid_pos + 1], "10002");
        // --gid must appear with its value immediately after.
        let gid_pos = argv
            .iter()
            .position(|a| *a == "--gid")
            .expect("--gid must be present");
        assert_eq!(argv[gid_pos + 1], "10003");
        // --chroot-base-dir must appear with its value immediately after.
        let chroot_pos = argv
            .iter()
            .position(|a| *a == "--chroot-base-dir")
            .expect("--chroot-base-dir must be present");
        assert_eq!(argv[chroot_pos + 1], "/var/lib/cellos/firecracker");
    }
}