supermachine 0.7.103

//! The snapshot-as-layer executor: drive a parsed Dockerfile's instructions
//! through `restore(base) → exec(RUN) → snapshot()` inside a microVM, so the
//! resulting snapshot is directly the bootable build artifact (the fused
//! build+run thesis — see docs/design/in-vm-builder-2026-06-05.md).
//!
//! Proven viable end-to-end by `integration_builder_snapshot_layer`: a
//! snapshot taken after a RUN captures the overlayfs (tmpfs-in-RAM) changes.
//!
//! Supported: FROM (registry via Image::from_oci, a prior stage, or `scratch`
//! → a minimal base), RUN (shell + exec + heredoc form, with accumulated
//! ENV/WORKDIR, and `--mount=type=cache/secret/bind/tmpfs`), ENV, WORKDIR
//! (relative chains), USER, COPY/ADD (host-context, dir/file/trailing-slash
//! semantics, --chown/--chmod), `ADD <url>` (self-contained host fetch),
//! COPY --from=<stage>, multi-stage builds, and run-config persistence
//! (ENTRYPOINT/CMD/WORKDIR/USER/ENV → image metadata) — all with a
//! content-addressed per-layer snapshot cache (`Builder`). [`commit_squashfs`]
//! flattens a built image to a read-only squashfs for runtime RAM-density.

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::io::Read as _;
use std::path::{Path, PathBuf};
use std::time::Duration;

use super::dockerfile::{
    BaseImage, CopyFlags, Dockerfile, Healthcheck, Instruction, MountKind, RunMount, ShellOrExec,
    Stage,
};
use super::subst::resolve_stage_instructions;
use crate::api::{Error, Image};

/// Apply Dockerfile build-time `$VAR` substitution to every stage, returning a
/// resolved `Dockerfile` whose instruction text has all `ARG`/`ENV` references
/// expanded (`--build-arg` overriding `ARG` defaults). Each stage's scope is
/// seeded with the global (pre-`FROM`) args. See [`super::subst`].
fn resolve_dockerfile(df: &Dockerfile, build_args: &BTreeMap<String, String>) -> Dockerfile {
    Dockerfile {
        global_args: df.global_args.clone(),
        stages: df
            .stages
            .iter()
            .map(|s| Stage {
                base: s.base.clone(),
                name: s.name.clone(),
                platform: s.platform.clone(),
                instructions: resolve_stage_instructions(
                    &s.instructions,
                    &df.global_args,
                    build_args,
                ),
            })
            .collect(),
    }
}

/// Default per-RUN timeout. Builds can be slow (package installs, compiles).
const RUN_TIMEOUT: Duration = Duration::from_secs(1800);

/// Default RAM ceiling for build-step VMs, in MiB.
///
/// A build VM's writable rootfs is an overlay whose **upper is a tmpfs in
/// guest RAM** (init-oci / KVM init both mount it with default options →
/// ~50% of guest RAM). So the per-`RUN` write budget — every file a step
/// adds, plus apt/pip's transient download+unpack scratch — is bounded by
/// guest RAM, *not* host disk. Inheriting a runtime image's modest RAM (e.g.
/// 256–2048 MiB → ~128 MiB–1 GiB of tmpfs) made a realistic `RUN apt-get
/// install …` (multi-GB) die on `ENOSPC` mid-step even with hundreds of GiB
/// free on the host.
///
/// Build VMs therefore boot with a *generous* ceiling. This is nearly free:
/// the override is a pure runtime CoW page-fault ceiling (see
/// [`crate::api::PoolBuilder::with_memory_mib`]) — host pages commit only
/// when the guest actually writes them — and snapshots are **sparse** (only
/// non-zero pages hit disk, see `vmm::snapshot::write_sparse_ram`), so a step
/// that writes 2 GiB yields a ~2 GiB layer regardless of an 8 GiB ceiling.
/// Raise/lower via `SUPERMACHINE_BUILD_MEMORY_MIB`.
const BUILD_MEMORY_DEFAULT_MIB: u32 = 8192;

/// Resolve a build-step VM's RAM ceiling from the (optional)
/// `SUPERMACHINE_BUILD_MEMORY_MIB` value and the base image's own baked
/// ceiling. The env value wins when present and sane (≥ 256 MiB); otherwise
/// [`BUILD_MEMORY_DEFAULT_MIB`]. Either way we never go *below* the base
/// image's baked memory — a deliberately large base is honored, never shrunk.
/// Pure function (env parsing factored out) so the policy is unit-testable.
fn resolve_build_memory(env: Option<&str>, base_mem: u32) -> u32 {
    let want = env
        .and_then(|s| s.trim().parse::<u32>().ok())
        .filter(|&m| m >= 256)
        .unwrap_or(BUILD_MEMORY_DEFAULT_MIB);
    want.max(base_mem)
}

/// The RAM ceiling to boot a build-step VM with, given the base image it boots
/// from. Reads `SUPERMACHINE_BUILD_MEMORY_MIB`; see [`resolve_build_memory`].
fn build_memory_mib(base_mem: u32) -> u32 {
    resolve_build_memory(
        std::env::var("SUPERMACHINE_BUILD_MEMORY_MIB")
            .ok()
            .as_deref(),
        base_mem,
    )
}

/// Resolve a cold `FROM <reference>` registry base **for building**.
///
/// A build VM boots by *restoring* its base snapshot, and a restored guest's
/// RAM is fixed at whatever the snapshot was baked with — you can't grow it at
/// restore time (see `vmm::runner`: restore takes `s.memory.len()`). So to give
/// the build VM enough RAM for a heavy `RUN`'s tmpfs upper, the base itself must
/// be **baked** at the generous build ceiling, not merely booted with a runtime
/// memory override. We therefore bake the `FROM` image at
/// [`build_memory_mib`] — under a *build-specific snapshot name* so it never
/// thrashes the small-RAM **runtime** bake of the same image (which lives under
/// the plain image-derived name). The bake is sparse + CoW, so the larger RAM
/// size costs only what a step actually writes.
/// The FROM base image's inherited runtime config, read from its baked snapshot
/// metadata. Dockerfile instructions modify this config rather than starting
/// from an empty one (for example, `FROM postgres` followed only by `ENV` must
/// retain postgres' ENTRYPOINT and CMD).
///
/// The bake records the resolved image `Env` in the snapshot's `metadata.json`
/// under `native_bake_inputs.env.env` (a `{ "K": "V", ... }` object); a
/// dedicated top-level `image_env` is preferred when present (forward-compat).
/// Reading from the snapshot is robust to caching — no registry re-inspect.
fn base_image_config(base: &Image) -> ImageConfig {
    let snap = base.snapshot_path();
    let meta_path = if snap.is_dir() {
        snap.join("metadata.json")
    } else {
        match snap.parent() {
            Some(p) => p.join("metadata.json"),
            None => return ImageConfig::default(),
        }
    };
    let Ok(text) = std::fs::read_to_string(&meta_path) else {
        return ImageConfig::default();
    };
    let Ok(meta) = serde_json::from_str::<serde_json::Value>(&text) else {
        return ImageConfig::default();
    };
    let env_obj = meta
        .get("image_env")
        .and_then(|v| v.as_object())
        .or_else(|| {
            meta.pointer("/native_bake_inputs/env/env")
                .and_then(|v| v.as_object())
        });
    let mut config = ImageConfig::default();
    if let Some(obj) = env_obj {
        config.env = obj
            .iter()
            .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
            .collect();
        config.env.sort_by(|a, b| a.0.cmp(&b.0));
    }
    let string_array = |key: &str| {
        meta.get(key).and_then(|v| v.as_array()).map(|values| {
            values
                .iter()
                .filter_map(|v| v.as_str().map(str::to_owned))
                .collect::<Vec<_>>()
        })
    };
    config.cmd = string_array("cmd").map(ShellOrExec::Exec);
    config.entrypoint = string_array("entrypoint").map(ShellOrExec::Exec);
    config.workdir = meta
        .get("working_dir")
        .and_then(|v| v.as_str())
        .map(str::to_owned);
    config.user = meta.get("user").and_then(|v| v.as_str()).map(str::to_owned);
    config.shell = string_array("shell");
    config
}

fn seed_from_base(base: &Image) -> (BuildState, ImageConfig) {
    let config = base_image_config(base);
    let state = BuildState {
        env: config.env.iter().cloned().collect(),
        cwd: config.workdir.clone(),
        user: config.user.clone(),
        shell: config.shell.clone(),
    };
    (state, config)
}

fn from_oci_for_build(reference: &str) -> Result<Image, Error> {
    let mem = build_memory_mib(0);
    Image::builder(reference)
        .with_memory_mib(mem)
        .with_name(format!(
            "{}__smbuild{mem}m",
            crate::bake::snapshot_name_for_image(reference)
        ))
        .build()
}

/// How many connection-SETUP EOFs to absorb while waiting for a freshly-booted
/// build VM's in-guest exec agent to begin `listen()`ing on the exec vsock
/// port. The agent comes up asynchronously *after* the VM boots, so the FIRST
/// exec on a new VM can win the race and get an immediate EOF
/// (`ExecEof{frames:0}` — `is_setup_failure`). That's harmless (the command
/// never started) but must be retried, not surfaced as a build failure — it was
/// an intermittent `RUN step failed … ExecEof` flake on loaded hosts.
/// `output_resilient` backs off linearly (50 ms·attempt) and returns the
/// instant the agent answers, so this is a patience cap, not a fixed wait.
const AGENT_READY_RETRIES: u8 = 40;

/// Block until the build VM's in-guest exec agent is ready to serve (or give up
/// after [`AGENT_READY_RETRIES`]), by issuing a trivial no-op exec that absorbs
/// the post-boot setup-EOF race. Once this returns, every later exec on `vm` is
/// race-free. Best-effort: if the agent never comes up, the first *real* exec
/// surfaces the actual error, so we don't hard-fail here. Call once right after
/// acquiring/booting each build VM.
fn await_agent_ready(vm: &crate::api::Vm) {
    let _ = vm
        .exec_builder()
        .timeout(Duration::from_secs(30))
        .argv(["true"].iter().copied())
        .output_resilient(AGENT_READY_RETRIES);
}

/// Mutable build state applied to subsequent RUN steps (Docker semantics:
/// ENV/WORKDIR/USER accumulate and affect later instructions).
#[derive(Default)]
struct BuildState {
    env: BTreeMap<String, String>,
    cwd: Option<String>,
    user: Option<String>,
    /// The `SHELL` interpreter for shell-form `RUN` (and shell-form CMD/
    /// ENTRYPOINT). `None` = Docker's default `["/bin/sh", "-c"]`. A `SHELL`
    /// instruction sets it for the rest of the stage (Docker semantics).
    shell: Option<Vec<String>>,
}

/// Metadata an instruction declared (for the image run-config). Collected
/// now; wiring into the snapshot's metadata is a follow-up.
#[derive(Debug, Default, Clone)]
pub struct ImageConfig {
    pub env: Vec<(String, String)>,
    pub workdir: Option<String>,
    pub user: Option<String>,
    pub entrypoint: Option<ShellOrExec>,
    pub cmd: Option<ShellOrExec>,
    pub exposed_ports: Vec<String>,
    pub labels: Vec<(String, String)>,
    /// `SHELL` interpreter for shell-form CMD/ENTRYPOINT resolution + recorded
    /// in the OCI export config (Docker's `Shell` field). `None` = default sh.
    pub shell: Option<Vec<String>>,
    /// `VOLUME` mount points (anonymous-volume declarations) — recorded in the
    /// OCI export config's `Volumes` for round-trip fidelity.
    pub volumes: Vec<String>,
    /// `STOPSIGNAL` — recorded in the OCI export config's `StopSignal`.
    pub stop_signal: Option<String>,
    /// `HEALTHCHECK` — recorded in the OCI export config's `Healthcheck` and in
    /// the snapshot metadata. The builder does not run the probe; it round-trips
    /// the directive into the built image.
    pub healthcheck: Option<Healthcheck>,
    /// `ONBUILD` triggers (verbatim instruction text) — recorded as the image's
    /// `OnBuild` list (they run when a downstream build uses this image as base).
    pub onbuild: Vec<String>,
}

/// The result of a build: the bootable image snapshot + its run-config.
pub struct BuildOutcome {
    pub image: Image,
    pub config: ImageConfig,
    /// The final stage's layer-chain cache key — a content address of
    /// the whole build (base identity + every resolved instruction +
    /// COPY/ADD source digests fold into it transitively). Two builds
    /// with the same key produce byte-identical images, so embedders
    /// can use it to short-circuit downstream work (e.g. the napi
    /// dockerfile→export→warm-bake pipeline returns its cached warm
    /// snapshot without re-exporting when this key is unchanged).
    /// `None` on the keyless [`build_linear`] test entry point.
    pub layer_key: Option<String>,
    /// Snapshot-layer keys produced by this Dockerfile build. These are useful
    /// for pruning current-build layer snapshots after the final image has been
    /// exported into a standalone artifact.
    pub layer_keys: Vec<String>,
}

/// Build the final (last) stage of `df` by executing its instructions in a
/// VM booted from `base`, snapshotting the result to `dest`.
///
/// `base` stands in for the resolved `FROM` (v1 takes it explicitly rather
/// than pulling/baking the base image — base resolution is a follow-up).
pub fn build_linear(
    df: &Dockerfile,
    base: &Image,
    context_dir: &Path,
    dest: impl AsRef<Path>,
) -> Result<BuildOutcome, Error> {
    let stage = df
        .stages
        .last()
        .ok_or_else(|| Error::vm_msg("Dockerfile has no stages"))?;
    // Resolve $VAR / ${VAR} (ARG defaults + ENV scope); no --build-arg overrides
    // on this simple linear entry point (use `Builder` for those).
    let resolved_instrs =
        resolve_stage_instructions(&stage.instructions, &df.global_args, &BTreeMap::new());

    let pool = base
        .pool()
        .min(1)
        .max(1)
        .restore_on_release(false)
        // Sentry: serve from a per-build private reflink clone + suppress the base
        // workload (additive / backend-neutral; KVM/HVF ignore it).
        .for_build(true)
        // Generous RAM so a heavy RUN's tmpfs-upper writes don't ENOSPC; CoW so
        // it costs only what's actually written (see BUILD_MEMORY_DEFAULT_MIB).
        .with_memory_mib(build_memory_mib(base.memory_mib()))
        .build()?;
    let vm = pool.acquire()?;
    await_agent_ready(&vm);

    let (mut state, mut config) = seed_from_base(base);
    for instr in &resolved_instrs {
        apply_instr(
            Some(&vm),
            instr,
            &mut state,
            &mut config,
            context_dir,
            &[],
            None,
        )?;
    }

    let image = vm.snapshot(dest.as_ref())?;
    Ok(BuildOutcome {
        image,
        config,
        layer_key: None,
        layer_keys: Vec::new(),
    })
}

/// Apply one instruction. When `vm` is `Some`, the filesystem effects (RUN,
/// COPY/ADD, WORKDIR's mkdir) execute in that VM; when `None`, only the
/// build state + image config are updated — used to replay a cached layer
/// prefix whose FS effects are already baked into the cached snapshot.
fn apply_instr(
    vm: Option<&crate::api::Vm>,
    instr: &Instruction,
    state: &mut BuildState,
    config: &mut ImageConfig,
    context_dir: &Path,
    prior: &[Option<StageBuilt>],
    mount_cache: Option<&Path>,
) -> Result<(), Error> {
    match instr {
        Instruction::Run { run, mounts } => {
            if let Some(vm) = vm {
                run_with_mounts(vm, run, mounts, state, context_dir, prior, mount_cache)?;
            }
        }
        Instruction::Env(pairs) => {
            for (k, v) in pairs {
                state.env.insert(k.clone(), v.clone());
                config.env.push((k.clone(), v.clone()));
            }
        }
        Instruction::Workdir(dir) => {
            // WORKDIR may be relative to the previous WORKDIR (Docker chains).
            let resolved = if dir.starts_with('/') {
                dir.clone()
            } else {
                let base = state.cwd.as_deref().unwrap_or("/");
                format!("{}/{}", base.trim_end_matches('/'), dir)
            };
            if let Some(vm) = vm {
                // Docker's WORKDIR creates the directory if absent.
                run_step(
                    vm,
                    &ShellOrExec::Exec(vec!["mkdir".into(), "-p".into(), resolved.clone()]),
                    state,
                )?;
            }
            state.cwd = Some(resolved.clone());
            config.workdir = Some(resolved);
        }
        Instruction::User(u) => {
            state.user = Some(u.clone());
            config.user = Some(u.clone());
        }
        Instruction::Entrypoint(e) => config.entrypoint = Some(e.clone()),
        Instruction::Cmd(c) => config.cmd = Some(c.clone()),
        Instruction::Expose(ports) => config.exposed_ports.extend(ports.iter().cloned()),
        Instruction::Label(pairs) => config.labels.extend(pairs.iter().cloned()),
        Instruction::Shell(argv) => {
            // Docker semantics: SHELL sets the interpreter for shell-form RUN
            // (executed in the build VM) AND shell-form CMD/ENTRYPOINT for the
            // rest of the stage. Both the build state (RUN) and the image config
            // (CMD/ENTRYPOINT resolution + OCI export) track it.
            state.shell = Some(argv.clone());
            config.shell = Some(argv.clone());
        }
        Instruction::Volume(mounts) => config.volumes.extend(mounts.iter().cloned()),
        Instruction::StopSignal(sig) => config.stop_signal = Some(sig.clone()),
        // Metadata-only directives: round-tripped into the image run-config, not
        // executed at build time (HEALTHCHECK replaces any prior; ONBUILD appends).
        Instruction::Healthcheck(hc) => config.healthcheck = Some(hc.clone()),
        Instruction::Onbuild(trigger) => config.onbuild.push(trigger.clone()),
        Instruction::Arg { .. } => {}
        Instruction::Copy {
            sources,
            dest,
            flags,
        } => {
            if let Some(vm) = vm {
                stage_or_from(vm, sources, dest, flags, context_dir, state, prior)?;
            }
        }
        Instruction::Add {
            sources,
            dest,
            flags,
        } => {
            if let Some(vm) = vm {
                // ADD (unlike COPY) accepts URLs: fetch them on the host and
                // write them into the guest; the rest behave like COPY.
                let (urls, locals): (Vec<String>, Vec<String>) =
                    sources.iter().cloned().partition(|s| is_url(s));
                if !urls.is_empty() {
                    add_urls(vm, &urls, dest, flags, state.cwd.as_deref())?;
                }
                if !locals.is_empty() {
                    stage_or_from(vm, &locals, dest, flags, context_dir, state, prior)?;
                }
            }
        }
    }
    Ok(())
}

/// Shared COPY/ADD staging: `--from=<stage>` extracts from a prior stage, else
/// stage from the host build context.
fn stage_or_from(
    vm: &crate::api::Vm,
    sources: &[String],
    dest: &str,
    flags: &CopyFlags,
    context_dir: &Path,
    state: &BuildState,
    prior: &[Option<StageBuilt>],
) -> Result<(), Error> {
    match &flags.from {
        Some(name) => {
            let src = resolve_stage(prior, name)?;
            let source = Image::from_snapshot(src.image.snapshot_path())?;
            copy_from_stage(vm, &source, sources, dest, flags, state.cwd.as_deref())
        }
        None => stage_copy(vm, sources, dest, flags, context_dir, state.cwd.as_deref()),
    }
}

/// `ADD <url>`: fetch each URL on the host (self-contained rustls client) and
/// write it into the guest. Docker semantics: a trailing-slash `dest` is a
/// directory (filename inferred from the URL); otherwise `dest` is the file.
fn add_urls(
    vm: &crate::api::Vm,
    urls: &[String],
    dest: &str,
    flags: &CopyFlags,
    workdir: Option<&str>,
) -> Result<(), Error> {
    let wd = workdir.unwrap_or("/");
    let abs_dest = if dest.starts_with('/') {
        dest.to_string()
    } else {
        format!("{}/{}", wd.trim_end_matches('/'), dest)
    };
    let dest_is_dir = dest.ends_with('/') || urls.len() > 1;

    for url in urls {
        let bytes = super::fetch::http_get(url)?;
        let target = if dest_is_dir {
            let name = url_filename(url).ok_or_else(|| {
                Error::vm_msg(format!("ADD url: cannot infer filename from {url}"))
            })?;
            format!("{}/{}", abs_dest.trim_end_matches('/'), name)
        } else {
            abs_dest.clone()
        };
        // Ensure the parent dir exists, then write the bytes in.
        if let Some((parent, _)) = target.rsplit_once('/') {
            if !parent.is_empty() {
                let _ = vm
                    .exec_builder()
                    .argv(["mkdir", "-p", parent].iter().copied())
                    .output();
            }
        }
        vm.write_file(&target, &bytes).map_err(Error::Io)?;
        if let Some(chmod) = &flags.chmod {
            let _ = vm
                .exec_builder()
                .argv(["chmod", chmod.as_str(), target.as_str()].iter().copied())
                .output();
        }
        if let Some(chown) = &flags.chown {
            let _ = vm
                .exec_builder()
                .argv(["chown", chown.as_str(), target.as_str()].iter().copied())
                .output();
        }
    }
    Ok(())
}

/// Whether a COPY/ADD source is an http(s) URL (vs a build-context path).
fn is_url(s: &str) -> bool {
    s.starts_with("http://") || s.starts_with("https://")
}

/// The filename a URL `ADD`s to a directory dest: the last non-empty path
/// segment, with any query/fragment stripped.
fn url_filename(url: &str) -> Option<String> {
    let after_scheme = url.split_once("://").map(|(_, r)| r).unwrap_or(url);
    let path = after_scheme.split(['?', '#']).next().unwrap_or("");
    path.rsplit('/')
        .find(|s| !s.is_empty())
        .map(|s| s.to_owned())
}

/// Resolve `FROM scratch` to a bootable base. A microVM always needs a
/// kernel + a minimal userland to boot (and the builder's COPY/RUN paths use
/// in-guest `tar`/`sh`), so "scratch" maps to a tiny base image rather than a
/// truly empty rootfs — the right call for a microVM substrate (Firecracker &
/// co. do the same: there's always an init). The extra busybox applets are
/// harmless; the image's run-config is still whatever the Dockerfile declares.
/// Override the base with `SUPERMACHINE_SCRATCH_BASE` (e.g. a distroless ref).
fn scratch_base() -> Result<Image, Error> {
    let reference = std::env::var("SUPERMACHINE_SCRATCH_BASE")
        .unwrap_or_else(|_| "busybox:stable-musl".to_owned());
    // Baked at the build RAM ceiling (so heavy RUNs fit), like any FROM base.
    from_oci_for_build(&reference)
}

/// The set of stage indices a stage depends on: its `FROM <stage>` base plus
/// every `COPY/ADD --from=<stage|idx>`. References to registry images (not a
/// known stage name, not a stage index) are not dependencies.
fn stage_dependencies(stage: &Stage, n: usize, names: &HashMap<&str, usize>) -> Vec<usize> {
    let mut deps = Vec::new();
    let push = |deps: &mut Vec<usize>, r: &str| {
        if let Some(&i) = names.get(r) {
            deps.push(i);
        } else if let Ok(i) = r.parse::<usize>() {
            if i < n {
                deps.push(i);
            }
        }
    };
    match &stage.base {
        // `FROM <name>` is a stage dep only if `<name>` is a known stage.
        BaseImage::Image(reference) if names.contains_key(reference.as_str()) => {
            push(&mut deps, reference)
        }
        BaseImage::Stage(name) => push(&mut deps, name),
        _ => {}
    }
    for instr in &stage.instructions {
        let from = match instr {
            Instruction::Copy { flags, .. } | Instruction::Add { flags, .. } => {
                flags.from.as_deref()
            }
            _ => None,
        };
        if let Some(r) = from {
            push(&mut deps, r);
        }
    }
    deps.sort_unstable();
    deps.dedup();
    deps
}

/// Resolve a `COPY --from=<stage>` reference by stage name or numeric index.
/// `prior` is indexed by stage index (`None` = not built / not a dependency).
fn resolve_stage<'a>(
    prior: &'a [Option<StageBuilt>],
    reference: &str,
) -> Result<&'a StageBuilt, Error> {
    prior
        .iter()
        .flatten()
        .rev()
        .find(|s| s.name.as_deref() == Some(reference))
        .or_else(|| {
            reference
                .parse::<usize>()
                .ok()
                .and_then(|i| prior.get(i))
                .and_then(|o| o.as_ref())
        })
        .ok_or_else(|| Error::vm_msg(format!("COPY --from unknown stage `{reference}`")))
}

/// A caching in-VM Dockerfile builder. Each instruction produces a
/// content-addressed layer snapshot under `cache_dir/layers/<key>`; a
/// rebuild restores the longest unchanged prefix from cache and re-executes
/// only what changed. A fully-cached rebuild boots **no VM at all** — it
/// just returns the final cached layer.
pub struct Builder {
    cache_dir: PathBuf,
    /// `--build-arg` overrides for `ARG` substitution (name → value). Empty by
    /// default, so `ARG` defaults apply.
    build_args: BTreeMap<String, String>,
}

/// A built stage: its name (if `FROM ... AS <name>`), its final-layer cache
/// key (the identity that `COPY --from` folds into its own key, so a change
/// in the source stage busts the consumer), and a handle to its image.
struct StageBuilt {
    name: Option<String>,
    key: String,
    image: Image,
    layer_keys: Vec<String>,
}

#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
struct CompactLayerManifest {
    version: u32,
    kind: String,
    parent_snapshot: String,
    rootfs: String,
    diff_blob: Option<String>,
    diff_id: Option<String>,
    layer_digest: Option<String>,
    layer_size: Option<u64>,
    created_by_version: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct FsEntryMeta {
    kind: FsEntryKind,
    mode: u32,
    uid: u64,
    gid: u64,
    mtime: i64,
    size: u64,
    digest: Option<String>,
    link_target: Option<PathBuf>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
enum FsEntryKind {
    Dir,
    File,
    Symlink,
}

impl Builder {
    pub fn new(cache_dir: impl Into<PathBuf>) -> Self {
        Self {
            cache_dir: cache_dir.into(),
            build_args: BTreeMap::new(),
        }
    }

    /// Set `--build-arg` overrides for `ARG` substitution. Replaces any
    /// previously-set build args.
    pub fn build_args(mut self, args: BTreeMap<String, String>) -> Self {
        self.build_args = args;
        self
    }

    fn layer_dir(&self, key: &str) -> PathBuf {
        self.cache_dir.join("layers").join(key)
    }

    /// Build every stage of `df` (resolving each `FROM` — a registry ref via
    /// [`Image::from_oci`], or a prior stage), with layer caching and
    /// `COPY --from=<stage>`. **Independent stages build concurrently** in
    /// their own VMs (a stage runs once all stages it references via `FROM` or
    /// `--from` are built), bounded by `SUPERMACHINE_BUILD_PARALLELISM`
    /// (default 4). Returns the final stage's image. The full "Dockerfile in →
    /// bootable image out" entry point.
    pub fn build_dockerfile(
        &self,
        df: &Dockerfile,
        context_dir: &Path,
    ) -> Result<BuildOutcome, Error> {
        if df.stages.is_empty() {
            return Err(Error::vm_msg("Dockerfile has no stages"));
        }
        // Resolve $VAR / ${VAR} (ARG + ENV scope, --build-arg overrides) ONCE up
        // front; cache keys and execution both then run on substituted text.
        let resolved = resolve_dockerfile(df, &self.build_args);
        let df = &resolved;
        let n = df.stages.len();

        // Stage name → index, and the dependency edges (a stage references
        // earlier stages via `FROM <stage>` or `COPY/ADD --from=<stage|idx>`).
        let names: HashMap<&str, usize> = df
            .stages
            .iter()
            .enumerate()
            .filter_map(|(i, s)| s.name.as_deref().map(|nm| (nm, i)))
            .collect();
        let deps: Vec<Vec<usize>> = df
            .stages
            .iter()
            .map(|s| stage_dependencies(s, n, &names))
            .collect();

        let mut stages: Vec<Option<StageBuilt>> = (0..n).map(|_| None).collect();
        let mut configs: Vec<Option<ImageConfig>> = (0..n).map(|_| None).collect();
        let mut done = vec![false; n];
        let mut remaining = n;

        let parallelism = std::env::var("SUPERMACHINE_BUILD_PARALLELISM")
            .ok()
            .and_then(|s| s.parse::<usize>().ok())
            .unwrap_or(4)
            .max(1);

        // Schedule in waves: build all stages whose deps are satisfied, in
        // parallel (capped), then advance.
        while remaining > 0 {
            let ready: Vec<usize> = (0..n)
                .filter(|&i| !done[i] && deps[i].iter().all(|&d| done[d]))
                .collect();
            if ready.is_empty() {
                return Err(Error::vm_msg("Dockerfile has a stage dependency cycle"));
            }
            for chunk in ready.chunks(parallelism) {
                let wave: Vec<(usize, Result<(StageBuilt, ImageConfig), Error>)> =
                    std::thread::scope(|scope| {
                        let handles: Vec<(usize, _)> = chunk
                            .iter()
                            .map(|&i| {
                                let stages_ref = &stages;
                                let names_ref = &names;
                                (
                                    i,
                                    scope.spawn(move || {
                                        self.build_one_stage(
                                            &df.stages[i],
                                            context_dir,
                                            names_ref,
                                            stages_ref,
                                        )
                                    }),
                                )
                            })
                            .collect();
                        handles
                            .into_iter()
                            .map(|(i, h)| {
                                let r = h.join().unwrap_or_else(|_| {
                                    Err(Error::vm_msg(format!("stage {i} build panicked")))
                                });
                                (i, r)
                            })
                            .collect()
                    });
                for (i, r) in wave {
                    let (sb, cfg) = r?;
                    stages[i] = Some(sb);
                    configs[i] = Some(cfg);
                    done[i] = true;
                    remaining -= 1;
                }
            }
        }

        // The final stage (last in the file) is the output; persist its config.
        let final_idx = n - 1;
        let final_stage = stages[final_idx].as_ref().expect("final stage built");
        let layer_key = Some(final_stage.key.clone());
        let layer_keys: Vec<String> = stages
            .iter()
            .filter_map(|stage| stage.as_ref())
            .flat_map(|stage| stage.layer_keys.iter().cloned())
            .collect();
        let config = configs[final_idx].take().expect("final stage config");
        // The outcome image must LAUNCH the Dockerfile's run-config, not the
        // FROM base's: the final RUN-layer snapshot freezes init-oci AFTER it
        // already parsed the BASE image's staged /.supermachine-cmd, so every
        // restore of that snapshot forks the base's CMD (task #126; repro:
        // integration_workload_contract::workload_sees_image_env_and_workdir).
        // Finalize with a fresh bake of the exported rootfs — the bake
        // restages cmd/workdir/user/env into the rootfs before boot, exactly
        // like the npm dockerfile-source flow's first-run export + re-bake.
        // Content-addressed by the final layer key, so a cached rebuild
        // still boots no VM at all.
        //
        // FAST PATH: a Dockerfile that adds NO run-config (pure RUN/COPY —
        // common for build/intermediate stages) leaves the base's staged
        // /.supermachine-cmd correct, so the final layer snapshot already
        // launches the right command. Skip the re-bake (a full boot + tar +
        // re-bake) and return the layer image directly — exactly the
        // pre-#126 behavior, which was correct for these. Mirrors
        // `persist_run_config`'s own "nothing to persist" guard so the two
        // never disagree about whether the build touched the run-config.
        let touches_run_config = effective_cmd(&config).is_some()
            || config.workdir.is_some()
            || config.user.is_some()
            || !config.env.is_empty()
            || config.stop_signal.is_some()
            || config.healthcheck.is_some()
            || !config.onbuild.is_empty();
        if !touches_run_config {
            let image = Image::from_snapshot(final_stage.image.snapshot_path())?;
            return Ok(BuildOutcome {
                image,
                config,
                layer_key,
                layer_keys,
            });
        }
        let image = self.finalize_baked_image(final_stage, &config, &layer_keys)?;
        persist_run_config(image.snapshot_path(), &config)?;
        Ok(BuildOutcome {
            image,
            config,
            layer_key,
            layer_keys,
        })
    }

    /// Produce the build's OUTCOME image: a fresh warm bake of the final
    /// stage's rootfs with the resolved run-config staged in. Cached under
    /// `<cache>/baked/<final-layer-key>/image` — a loadable cache hit
    /// returns without booting anything (and a version-skewed or otherwise
    /// unloadable cached bake is treated as a miss and rebuilt, mirroring
    /// the layer-cache semantics above).
    fn finalize_baked_image(
        &self,
        final_stage: &StageBuilt,
        config: &ImageConfig,
        layer_keys: &[String],
    ) -> Result<Image, Error> {
        let baked_dir = self.cache_dir.join("baked").join(&final_stage.key);
        if let Some(parent) = baked_dir.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        // Serialize concurrent finalizes of the SAME final-layer key. The baked
        // snapshot + its intermediate `oci-layout/` are written IN-PLACE under
        // `baked_dir`; a peer build's reuse-check (`from_snapshot`) and the bake's
        // own `oci-layout` read are otherwise lock-free, so two producers racing the
        // same `baked_dir` tear it ("bake: parse OCI manifest … EOF", torn snapshot).
        // Whoever wins builds; the rest take the cache hit below. Mirrors the per-layer
        // CacheLock in build_stage; held for the whole finalize via RAII.
        let _baked_lock = CacheLock::acquire(&baked_dir);
        let baked_name = format!(
            "dfbuild_{}",
            &final_stage.key[..final_stage.key.len().min(24)]
        );
        let snap_dir = baked_dir.join(&baked_name);
        match Image::from_snapshot(&snap_dir) {
            Ok(img) => return Ok(img),
            Err(_) if snap_dir.exists() => {
                // Treat an unloadable baked image as a cache miss, but do not remove it
                // here. Readers/verifier cells are lock-free and may be serving from
                // this path right now; the later snapshot publish path replaces it
                // atomically when the rebuild succeeds.
            }
            Err(_) => {}
        }

        #[cfg(target_arch = "aarch64")]
        const OCI_ARCH: &str = "arm64";
        #[cfg(target_arch = "x86_64")]
        const OCI_ARCH: &str = "amd64";

        // Export boots the final layer snapshot once and streams the rootfs
        // out; the OCI config carries the run-config the bake then stages.
        let layout_dir = baked_dir.join("oci-layout");
        let _ = std::fs::remove_dir_all(&layout_dir);
        export_oci(&final_stage.image, config, OCI_ARCH, &layout_dir)?;
        if uses_full_rootfs_dockerfile_layers(&final_stage.image)
            && dockerfile_layer_prune_enabled()
        {
            let pruned = prune_dockerfile_layer_snapshots(&self.cache_dir, layer_keys)?;
            if pruned > 0 {
                eprintln!(
                    "supermachine(builder): pruned {pruned} Dockerfile layer snapshot(s) \
                     after OCI export to reduce peak bake disk usage"
                );
            }
        }

        // NOTE the snapshot NAME is content-addressed (`baked_name` above):
        // the warm-pool cache keys by name (`<name>__kvmwarm__…`), so a
        // constant name here would make every dockerfile build share ONE
        // global warm snapshot — pool acquires would restore a FOREIGN
        // image's state (first observed as RUN outputs "missing" from
        // freshly built images on the box).
        let image = Image::builder("dockerfile-built")
            .with_oci_layout(&layout_dir)
            .with_snapshots_dir(&baked_dir)
            .with_name(&baked_name)
            .build()?;
        // The bake extracted everything it needs into the snapshot dir +
        // layer cache; the layout (a full gzipped rootfs copy) is dead
        // weight from here.
        let _ = std::fs::remove_dir_all(&layout_dir);
        Ok(image)
    }

    /// Resolve one stage's base and build it, producing its [`StageBuilt`]
    /// handle + run-config. `prior` is indexed by stage index (deps are built).
    fn build_one_stage(
        &self,
        stage: &Stage,
        context_dir: &Path,
        names: &HashMap<&str, usize>,
        prior: &[Option<StageBuilt>],
    ) -> Result<(StageBuilt, ImageConfig), Error> {
        let base = self.resolve_base(stage, names, prior)?;
        let (image, config, key, layer_keys) =
            self.build_stage(stage, &base, context_dir, prior)?;
        let built = StageBuilt {
            name: stage.name.clone(),
            key,
            image: Image::from_snapshot(image.snapshot_path())?,
            layer_keys,
        };
        Ok((built, config))
    }

    /// Resolve a stage's `FROM`: a prior stage (by name), a registry image, or
    /// scratch.
    fn resolve_base(
        &self,
        stage: &Stage,
        names: &HashMap<&str, usize>,
        prior: &[Option<StageBuilt>],
    ) -> Result<Image, Error> {
        let from_stage = |di: usize| -> Result<Image, Error> {
            let dep = prior[di]
                .as_ref()
                .ok_or_else(|| Error::vm_msg("FROM references a stage that isn't built"))?;
            Image::from_snapshot(dep.image.snapshot_path())
        };
        match &stage.base {
            BaseImage::Image(reference) => match names.get(reference.as_str()) {
                Some(&di) => from_stage(di),
                // A registry FROM base, baked at the build RAM ceiling (so heavy
                // RUN steps' tmpfs-upper writes fit) — see `from_oci_for_build`.
                None => from_oci_for_build(reference),
            },
            BaseImage::Stage(name) => {
                let di = *names
                    .get(name.as_str())
                    .ok_or_else(|| Error::vm_msg(format!("FROM unknown stage `{name}`")))?;
                from_stage(di)
            }
            BaseImage::Scratch => scratch_base(),
        }
    }

    /// Build the final stage of `df` on an explicit `base` (no FROM
    /// resolution, no prior stages). The single-stage convenience.
    pub fn build(
        &self,
        df: &Dockerfile,
        base: &Image,
        context_dir: &Path,
    ) -> Result<BuildOutcome, Error> {
        let stage = df
            .stages
            .last()
            .ok_or_else(|| Error::vm_msg("Dockerfile has no stages"))?;
        let (image, config, key, layer_keys) = self.build_stage(stage, base, context_dir, &[])?;
        persist_run_config(image.snapshot_path(), &config)?;
        Ok(BuildOutcome {
            image,
            config,
            layer_key: Some(key),
            layer_keys,
        })
    }

    /// Build one `stage` on `base` with per-layer caching, resolving
    /// `COPY --from` against already-built `prior` stages. Returns the
    /// image, its run-config, and the final-layer cache key.
    fn build_stage(
        &self,
        stage: &super::dockerfile::Stage,
        base: &Image,
        context_dir: &Path,
        prior: &[Option<StageBuilt>],
    ) -> Result<(Image, ImageConfig, String, Vec<String>), Error> {
        let instrs = &stage.instructions;

        // 1. Chained content-addressed cache keys (parent ⧺ instruction;
        //    COPY/ADD fold a content hash of the sources, or — for
        //    `--from=<stage>` — the source stage's identity key).
        let base_key = sha256_hex(base.snapshot_path().to_string_lossy().as_bytes());
        let mut keys = Vec::with_capacity(instrs.len());
        let mut prev = base_key.clone();
        for instr in instrs {
            let key = sha256_hex(
                format!("{prev}\n{}", instr_repr(instr, context_dir, prior)?).as_bytes(),
            );
            keys.push(key.clone());
            prev = key;
        }

        // 2. Longest cached prefix (hits always form a prefix).
        let mut resume = 0usize;
        let mut resume_image: Option<Image> = None;
        for (i, key) in keys.iter().enumerate() {
            let dir = self.layer_dir(key);
            if !is_snapshot_dir(&dir) {
                break;
            }
            if !has_required_layer_runtime_metadata(&dir) {
                eprintln!(
                    "[builder] cached layer {} missing runtime/network metadata; rebuilding from instruction {i}",
                    dir.display()
                );
                break;
            }
            if compact_layer_manifest_path(&dir).is_file() {
                // Compact full-rootfs layers are self-describing. Do NOT
                // materialize each prefix while scanning, or a cache hit would
                // recreate the full Dockerfile layer chain and lose the storage
                // win. Materialize only the final selected resume layer below.
                resume = i + 1;
                resume_image = None;
                continue;
            }
            match Image::from_snapshot(&dir) {
                Ok(img) => {
                    resume = i + 1;
                    resume_image = Some(img);
                }
                Err(e) => {
                    // Unloadable cached layer (most commonly: baked under a
                    // previous binary version, which from_snapshot now hard-
                    // refuses) = cache MISS from here on, not a build failure.
                    //
                    // Do NOT `remove_dir_all(&dir)` here. This longest-prefix scan
                    // is LOCK-FREE (no per-key CacheLock), and `is_snapshot_dir` +
                    // `from_snapshot` are two separate stats — a transient load
                    // failure is frequently a peer producer mid-commit
                    // (`write_sentry_snapshot_dir`'s remove→rename) or a peer
                    // resume-cloning this exact dir, NOT a genuinely stale layer.
                    // A lock-free remove here races that peer: it deletes a
                    // just-published tree (or one being cloned) out from under it,
                    // which is the concurrent torn-tree flake
                    // (sentry_concurrent_layer_cache_no_torn_tree: "metadata.json
                    // EOF", "cannot read ld.so", "No such file or directory").
                    // Treat it as a clean cache miss and rebuild: build_stage's
                    // step 6 re-publishes this key under its per-key CacheLock,
                    // atomically overwriting any genuinely stale dir.
                    eprintln!(
                        "[builder] cached layer {} unloadable ({e}); rebuilding from instruction {i}",
                        dir.display()
                    );
                    break;
                }
            }
        }

        // 3. Replay state + config for the cached prefix (no VM).
        // Docker stages inherit the base image's complete runtime config. The
        // child instructions below override individual fields as they replay.
        let (mut state, mut config) = seed_from_base(base);
        for instr in &instrs[..resume] {
            apply_instr(
                None,
                instr,
                &mut state,
                &mut config,
                context_dir,
                prior,
                None,
            )?;
        }

        let final_key = keys.last().cloned().unwrap_or(base_key);
        if resume > 0 && resume_image.is_none() {
            let dir = self.layer_dir(&keys[resume - 1]);
            materialize_compact_layer_if_needed(&dir)?;
            resume_image = Some(Image::from_snapshot(&dir)?);
        }

        // 4. Fully cached → return the final cached layer; no VM boot.
        if resume == instrs.len() {
            let image = match resume_image {
                Some(img) => img,
                None => Image::from_snapshot(base.snapshot_path())?,
            };
            return Ok((image, config, final_key, keys));
        }

        // 5. Boot from the resume point: the longest cached layer if any, else
        //    the `base` itself. Use `base` directly (not a re-wrap via
        //    from_snapshot) so a fresh cold-boot base — e.g. `FROM alpine`
        //    resolved to an `Image::from_oci` that hasn't been baked to a
        //    snapshot yet (no restore.snap on disk, the common KVM case) — still
        //    boots: `base.pool()` lazily bakes/warms it. (from_snapshot required
        //    an existing restore.snap and so failed for registry bases.)
        let boot_from: &Image = resume_image.as_ref().unwrap_or(base);
        let pool = boot_from
            .pool()
            .min(1)
            .max(1)
            .restore_on_release(false)
            // Sentry: each stage gets its OWN pool → its OWN per-build private
            // reflink clone of the (resume / base) rootfs, so concurrent stages
            // can't collide and a failed RUN can't contaminate the warm base; the
            // base workload is also suppressed. Backend-neutral (KVM/HVF ignore it).
            .for_build(true)
            // Generous RAM so a heavy RUN's tmpfs-upper writes don't ENOSPC; CoW so
            // it costs only what's actually written (see BUILD_MEMORY_DEFAULT_MIB).
            .with_memory_mib(build_memory_mib(boot_from.memory_mib()))
            .build()?;
        let vm = pool.acquire()?;
        await_agent_ready(&vm);

        // 6. Execute the uncached suffix. Only FS-mutating instructions
        //    (RUN/COPY/ADD/WORKDIR) take a real snapshot (a full ~290 ms RAM
        //    save); metadata-only instructions (ENV/CMD/ENTRYPOINT/LABEL/…)
        //    don't change the filesystem, so their layer is a cheap clonefile
        //    of the previous snapshot — same FS, instant, and the run-config is
        //    replayed on resume. This keeps a cache key per instruction without
        //    paying a RAM save for steps that didn't touch the disk.
        let run_cache = self.cache_dir.join("runmounts");
        let mut prev_snapshot_dir: PathBuf = if resume > 0 {
            self.layer_dir(&keys[resume - 1])
        } else {
            base.snapshot_path()
                .parent()
                .ok_or_else(|| Error::vm_msg("base snapshot has no parent directory"))?
                .to_path_buf()
        };
        // Track the nearest preceding FULL KVM snapshot. When a layer's immediate
        // predecessor is a KVM DIFF (not a valid `save_diff` base), we diff
        // against this full instead of paying a full save — so restore stays
        // depth ≤2 (one full + one overlay) while almost every layer is a cheap
        // diff. Seeded from the resume/base snapshot if it's already a full.
        let mut last_full_snap: Option<PathBuf> = {
            let p = prev_snapshot_dir.join("restore.snap");
            is_full_kvm_snapshot(&p).then_some(p)
        };
        let prune_incremental_full_rootfs_layers =
            uses_full_rootfs_dockerfile_layers(boot_from) && dockerfile_layer_prune_enabled();
        let mut final_image: Option<Image> = None;
        for i in resume..instrs.len() {
            apply_instr(
                Some(&vm),
                &instrs[i],
                &mut state,
                &mut config,
                context_dir,
                prior,
                Some(&run_cache),
            )?;
            let dir = self.layer_dir(&keys[i]);
            if let Some(parent) = dir.parent() {
                std::fs::create_dir_all(parent).map_err(Error::Io)?;
            }
            // Serialize concurrent producers of this exact layer key — two
            // independent stages built in parallel from the same base, or two
            // build processes sharing the cache_dir. Whoever wins writes the
            // snapshot; the others reuse it (identical instructions → identical
            // result). Uncontended flock is ~µs, so this is free for the
            // common single-builder case.
            let _lock = CacheLock::acquire(&dir);
            if is_snapshot_dir(&dir) && has_required_layer_runtime_metadata(&dir) {
                final_image = Some(Image::from_snapshot(&dir)?);
            } else {
                if is_snapshot_dir(&dir) && !has_required_layer_runtime_metadata(&dir) {
                    let _ = std::fs::remove_dir_all(&dir);
                    std::fs::create_dir_all(&dir).map_err(Error::Io)?;
                }
                if mutates_fs(&instrs[i]) {
                    // Differential snapshot: the worker clonefiles the base + writes
                    // only changed pages instead of a full ~290ms RAM save.
                    //   1. Predecessor is a valid diff base (a full, or any HVF
                    //      snapshot) → diff against it (smallest delta; HVF keeps its
                    //      existing per-layer chaining).
                    //   2. Predecessor is a KVM diff (SMSNAP*D — diff-of-diff isn't
                    //      supported) → diff against the nearest FULL instead, so we
                    //      still avoid a full save and restore stays depth ≤2.
                    //   3. No usable base at all (leading FS layer on a cold-boot
                    //      base) → full save.
                    let prev_snap = prev_snapshot_dir.join("restore.snap");
                    let image = if prev_snap.is_file() && base_is_diffable(&prev_snap) {
                        vm.snapshot_diff(&dir, &prev_snap)?
                    } else if let Some(full) = last_full_snap.as_ref().filter(|p| p.is_file()) {
                        vm.snapshot_diff(&dir, full)?
                    } else {
                        vm.snapshot(&dir)?
                    };
                    if prune_incremental_full_rootfs_layers {
                        commit_compact_layer(&dir, &prev_snapshot_dir, &self.cache_dir)?;
                    }
                    final_image = Some(image);
                } else if prev_snapshot_dir.join("restore.snap").is_file() {
                    // Metadata-only layer (ENV/CMD/…): the FS is identical to the
                    // previous layer, so clone its snapshot dir (instant CoW) — but
                    // only when that predecessor IS a real snapshot dir.
                    clone_snapshot_dir(&prev_snapshot_dir, &dir)?;
                    if prune_incremental_full_rootfs_layers {
                        commit_noop_compact_layer(&dir, &prev_snapshot_dir)?;
                        materialize_compact_layer_if_needed(&dir)?;
                    }
                    final_image = Some(Image::from_snapshot(&dir)?);
                } else {
                    // No predecessor snapshot to clone — this happens for a LEADING
                    // metadata-only instruction on a cold-boot base (e.g. KVM
                    // `FROM alpine` → a squashfs image with no restore.snap). Capture
                    // a real snapshot of the booted VM (its FS == the base) so the
                    // layer is a VALID, cacheable snapshot dir; cloning the bare
                    // cold-boot dir would yield a dir with no restore.snap that never
                    // cache-hits on rebuild.
                    final_image = Some(vm.snapshot(&dir)?);
                }
            }
            // Remember this layer as the nearest full if it is one (a valid
            // future diff base); diff/clone-of-diff layers leave it unchanged.
            let snap = dir.join("restore.snap");
            if is_full_kvm_snapshot(&snap) {
                last_full_snap = Some(snap);
            }
            if prune_incremental_full_rootfs_layers && mutates_fs(&instrs[i]) && i > 0 {
                for prev_dir in
                    prune_earlier_full_rootfs_layer_snapshots(&self.cache_dir, &keys, i, &dir)?
                {
                    eprintln!(
                        "supermachine(builder): pruned earlier Dockerfile layer snapshot {} \
                         during build to reduce peak disk usage",
                        prev_dir.display()
                    );
                }
            }
            prev_snapshot_dir = dir;
        }

        Ok((
            final_image.expect("non-empty suffix"),
            config,
            final_key,
            keys,
        ))
    }
}

/// Some no-KVM backends currently store each filesystem-mutating Dockerfile
/// layer as a full rootfs directory instead of a compact page/file diff. Once
/// mutating layer `i` is published, the live build no longer needs earlier
/// same-stage layers: the running build VM has its own private rootfs and the
/// final export reads only the final layer. Keep the stage base (`i == 0`) so
/// prior-stage final images remain available for COPY --from.
///
/// Metadata-only layer dirs are not self-contained for these backends today:
/// they clone only snapshot metadata files, whose rootfs pointer can point at
/// the previous layer's rootfs. Therefore callers must invoke this only after a
/// mutating instruction has produced a fresh self-contained rootfs.
fn prune_earlier_full_rootfs_layer_snapshots(
    cache_dir: &Path,
    layer_keys: &[String],
    i: usize,
    current_dir: &Path,
) -> Result<Vec<PathBuf>, Error> {
    if i == 0 {
        return Ok(Vec::new());
    }
    let mut pruned = Vec::new();
    let mut seen = BTreeSet::new();
    for key in layer_keys[..i].iter().filter(|key| seen.insert(*key)) {
        let prev_dir = cache_dir.join("layers").join(key);
        if prev_dir == current_dir || !prev_dir.exists() {
            continue;
        }
        match prune_layer_materialization_or_dir(&prev_dir) {
            Ok(()) => pruned.push(prev_dir),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => {
                return Err(Error::vm_msg(format!(
                    "prune earlier Dockerfile layer snapshot {}: {e}",
                    prev_dir.display()
                )));
            }
        }
    }
    Ok(pruned)
}

fn compact_layer_manifest_path(dir: &Path) -> PathBuf {
    dir.join("supermachine-layer.json")
}

fn read_compact_layer_manifest(dir: &Path) -> Result<Option<CompactLayerManifest>, Error> {
    let path = compact_layer_manifest_path(dir);
    if !path.is_file() {
        return Ok(None);
    }
    let text = std::fs::read_to_string(&path).map_err(Error::Io)?;
    let manifest = serde_json::from_str::<CompactLayerManifest>(&text)
        .map_err(|e| Error::vm_msg(format!("parse {}: {e}", path.display())))?;
    Ok(Some(manifest))
}

fn snapshot_rootfs_dir(dir: &Path) -> Result<Option<PathBuf>, Error> {
    let meta_path = dir.join("metadata.json");
    if !meta_path.is_file() {
        return Ok(None);
    }
    let text = std::fs::read_to_string(&meta_path).map_err(Error::Io)?;
    let meta = serde_json::from_str::<serde_json::Value>(&text)
        .map_err(|e| Error::vm_msg(format!("parse {}: {e}", meta_path.display())))?;
    let Some(backend) = meta.get("backend").and_then(|v| v.as_str()) else {
        return Ok(None);
    };
    let key = match backend {
        "sentry" => "sentry_rootfs",
        "runsc" => "runsc_rootfs",
        "kboxlike" => "kboxlike_rootfs",
        _ => return Ok(None),
    };
    let Some(raw) = meta.get(key).and_then(|v| v.as_str()) else {
        return Ok(None);
    };
    let path = PathBuf::from(raw);
    Ok(Some(if path.is_absolute() {
        path
    } else {
        dir.join(path)
    }))
}

fn snapshot_rootfs_metadata_key(dir: &Path) -> Result<Option<&'static str>, Error> {
    let meta_path = dir.join("metadata.json");
    if !meta_path.is_file() {
        return Ok(None);
    }
    let text = std::fs::read_to_string(&meta_path).map_err(Error::Io)?;
    let meta = serde_json::from_str::<serde_json::Value>(&text)
        .map_err(|e| Error::vm_msg(format!("parse {}: {e}", meta_path.display())))?;
    Ok(match meta.get("backend").and_then(|v| v.as_str()) {
        Some("sentry") => Some("sentry_rootfs"),
        Some("runsc") => Some("runsc_rootfs"),
        Some("kboxlike") => Some("kboxlike_rootfs"),
        _ => None,
    })
}

fn set_snapshot_rootfs_metadata(dir: &Path, rootfs: &str) -> Result<(), Error> {
    let Some(key) = snapshot_rootfs_metadata_key(dir)? else {
        return Ok(());
    };
    let meta_path = dir.join("metadata.json");
    let text = std::fs::read_to_string(&meta_path).map_err(Error::Io)?;
    let mut meta = serde_json::from_str::<serde_json::Value>(&text)
        .map_err(|e| Error::vm_msg(format!("parse {}: {e}", meta_path.display())))?;
    let obj = meta
        .as_object_mut()
        .ok_or_else(|| Error::vm_msg("metadata.json is not an object"))?;
    obj.insert(key.to_owned(), serde_json::json!(rootfs));
    let tmp = dir.join(format!(".metadata.json.tmp.{}", std::process::id()));
    std::fs::write(
        &tmp,
        serde_json::to_vec_pretty(&meta)
            .map_err(|e| Error::vm_msg(format!("serialize metadata.json: {e}")))?,
    )
    .map_err(Error::Io)?;
    std::fs::rename(&tmp, &meta_path).map_err(Error::Io)?;
    Ok(())
}

fn commit_compact_layer(
    layer_dir: &Path,
    parent_dir: &Path,
    cache_dir: &Path,
) -> Result<(), Error> {
    let Some(rootfs) = snapshot_rootfs_dir(layer_dir)? else {
        return Ok(());
    };
    let Some(parent_rootfs) = snapshot_rootfs_dir(parent_dir)? else {
        return Ok(());
    };
    if !rootfs.is_dir() || !parent_rootfs.is_dir() {
        return Ok(());
    }

    let blobs = cache_dir.join("blobs/sha256");
    let diff = write_rootfs_diff_layer(&parent_rootfs, &rootfs, &blobs)?;
    let manifest = CompactLayerManifest {
        version: 1,
        kind: "supermachine-dockerfile-rootfs-diff".to_owned(),
        parent_snapshot: parent_dir.to_string_lossy().into_owned(),
        rootfs: rootfs
            .strip_prefix(layer_dir)
            .unwrap_or(rootfs.as_path())
            .to_string_lossy()
            .into_owned(),
        diff_blob: Some(format!("blobs/sha256/{}", diff.layer_digest)),
        diff_id: Some(diff.diff_id),
        layer_digest: Some(format!("sha256:{}", diff.layer_digest)),
        layer_size: Some(diff.layer_size),
        created_by_version: env!("CARGO_PKG_VERSION").to_owned(),
    };
    let tmp = layer_dir.join(format!(
        ".supermachine-layer.json.tmp.{}",
        std::process::id()
    ));
    let text = serde_json::to_vec_pretty(&manifest)
        .map_err(|e| Error::vm_msg(format!("serialize compact layer manifest: {e}")))?;
    std::fs::write(&tmp, text).map_err(Error::Io)?;
    std::fs::rename(&tmp, compact_layer_manifest_path(layer_dir)).map_err(Error::Io)?;
    Ok(())
}

fn commit_noop_compact_layer(layer_dir: &Path, parent_dir: &Path) -> Result<(), Error> {
    set_snapshot_rootfs_metadata(layer_dir, "rootfs")?;
    let manifest = CompactLayerManifest {
        version: 1,
        kind: "supermachine-dockerfile-rootfs-diff".to_owned(),
        parent_snapshot: parent_dir.to_string_lossy().into_owned(),
        rootfs: "rootfs".to_owned(),
        diff_blob: None,
        diff_id: None,
        layer_digest: None,
        layer_size: Some(0),
        created_by_version: env!("CARGO_PKG_VERSION").to_owned(),
    };
    let tmp = layer_dir.join(format!(
        ".supermachine-layer.json.tmp.{}",
        std::process::id()
    ));
    let text = serde_json::to_vec_pretty(&manifest)
        .map_err(|e| Error::vm_msg(format!("serialize compact layer manifest: {e}")))?;
    std::fs::write(&tmp, text).map_err(Error::Io)?;
    std::fs::rename(&tmp, compact_layer_manifest_path(layer_dir)).map_err(Error::Io)?;
    Ok(())
}

fn materialize_compact_layer_if_needed(layer_dir: &Path) -> Result<(), Error> {
    let Some(manifest) = read_compact_layer_manifest(layer_dir)? else {
        return Ok(());
    };
    let rootfs = layer_dir.join(&manifest.rootfs);
    if rootfs.is_dir() {
        return Ok(());
    }

    let parent = PathBuf::from(&manifest.parent_snapshot);
    materialize_compact_layer_if_needed(&parent)?;
    let parent_rootfs = snapshot_rootfs_dir(&parent)?.ok_or_else(|| {
        Error::vm_msg(format!(
            "compact layer {} parent has no host rootfs",
            layer_dir.display()
        ))
    })?;
    if !parent_rootfs.is_dir() {
        return Err(Error::vm_msg(format!(
            "compact layer {} parent rootfs missing: {}",
            layer_dir.display(),
            parent_rootfs.display()
        )));
    }

    let staging = layer_dir.join(format!(
        ".rootfs.materialize.{}.{}",
        std::process::id(),
        unique_layer_suffix()
    ));
    let _ = std::fs::remove_dir_all(&staging);
    copy_tree_preserving(&parent_rootfs, &staging).map_err(Error::Io)?;
    if let Some(diff_blob) = &manifest.diff_blob {
        let diff_path = if Path::new(diff_blob).is_absolute() {
            PathBuf::from(diff_blob)
        } else {
            layer_dir
                .parent()
                .and_then(|p| p.parent())
                .unwrap_or(layer_dir)
                .join(diff_blob)
        };
        apply_oci_diff_layer(&diff_path, &staging)?;
    }
    if let Some(parent) = rootfs.parent() {
        std::fs::create_dir_all(parent).map_err(Error::Io)?;
    }
    match std::fs::rename(&staging, &rootfs) {
        Ok(()) => {
            if compact_layer_manifest_path(&parent).is_file() {
                let _ = prune_layer_materialization_or_dir(&parent);
            }
            Ok(())
        }
        Err(e) => {
            let _ = std::fs::remove_dir_all(&staging);
            Err(Error::Io(e))
        }
    }
}

fn prune_layer_materialization_or_dir(layer_dir: &Path) -> std::io::Result<()> {
    if let Ok(Some(manifest)) = read_compact_layer_manifest(layer_dir) {
        let rootfs = layer_dir.join(manifest.rootfs);
        match std::fs::remove_dir_all(rootfs) {
            Ok(()) => Ok(()),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(e) => Err(e),
        }
    } else {
        std::fs::remove_dir_all(layer_dir)
    }
}

fn unique_layer_suffix() -> u128 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0)
}

/// Whether `image` uses a Dockerfile-layer storage format where each mutating
/// layer is physically a full rootfs snapshot today. These backends need
/// aggressive pruning until the builder grows a real OCI/overlay diff content
/// store.
fn uses_full_rootfs_dockerfile_layers(image: &Image) -> bool {
    image.is_runsc() || image.is_sentry() || image.is_kboxlike()
}

/// Whether full-rootfs Dockerfile builds should discard current-build layer
/// snapshots after exporting the final image into a standalone OCI layout.
///
/// Defaults ON for no-KVM full-rootfs backends because keeping every
/// Dockerfile layer as a complete rootfs can push browser-sized bakes over
/// small cloud-sandbox quotas before the final warm snapshot is made. Set
/// `SUPERMACHINE_PRUNE_DOCKERFILE_LAYERS=0` to preserve partial rebuild cache
/// behavior on large disks. The legacy runsc-specific knob is still honored.
pub fn dockerfile_layer_prune_enabled() -> bool {
    let value = std::env::var("SUPERMACHINE_PRUNE_DOCKERFILE_LAYERS")
        .ok()
        .or_else(|| std::env::var("SUPERMACHINE_RUNSC_PRUNE_DOCKERFILE_LAYERS").ok());
    match value {
        Some(v) => !matches!(
            v.trim().to_ascii_lowercase().as_str(),
            "0" | "false" | "no" | "off" | "never"
        ),
        None => true,
    }
}

/// Backward-compatible alias for embedders that still use the old runsc-only
/// name. The policy is now backend-neutral.
pub fn runsc_prune_dockerfile_layers_enabled() -> bool {
    dockerfile_layer_prune_enabled()
}

/// Release the current Dockerfile build's materialized layer rootfs trees from
/// `cache_dir/layers`. Compact diff-backed layers keep their metadata + diff
/// blob, so future builds can still cache-hit and materialize them on demand.
/// Legacy non-compact layers are removed whole.
pub fn prune_dockerfile_layer_snapshots(
    cache_dir: &Path,
    layer_keys: &[String],
) -> Result<usize, Error> {
    let mut pruned = 0usize;
    let mut seen = BTreeSet::new();
    for key in layer_keys {
        if !seen.insert(key) {
            continue;
        }
        let dir = cache_dir.join("layers").join(key);
        if !dir.exists() {
            continue;
        }
        prune_layer_materialization_or_dir(&dir).map_err(|e| {
            Error::vm_msg(format!(
                "prune Dockerfile layer materialization {}: {e}",
                dir.display()
            ))
        })?;
        pruned += 1;
    }
    Ok(pruned)
}

/// Whether `snap` can serve as a base for a differential snapshot. A KVM
/// differential snapshot (`SMSNAP4D`/`SMSNAP5D` magic) CANNOT — KVM's `save_diff`
/// requires a full base (no diff-of-diff), so a layer following a diff layer must
/// take a full snapshot. Any other magic (a KVM full `SMSNAP04`/`SMSNAP05`, or an
/// HVF snapshot whose format differs entirely) is treated as diffable, preserving
/// HVF's existing per-layer diff chaining. Unreadable → assume diffable (the
/// caller already checked the file exists; let the diff path surface a real error).
fn base_is_diffable(snap: &Path) -> bool {
    use std::io::Read as _;
    let mut magic = [0u8; 8];
    match std::fs::File::open(snap).and_then(|mut f| f.read_exact(&mut magic)) {
        // A KVM differential snapshot's magic is `SMSNAP` + version + `D`
        // (SMSNAP4D/5D/6D/…). Detect the trailing `D` generically so a new diff
        // version never silently regresses to diff-of-diff again. KVM full magics
        // (SMSNAP04/05/06, trailing digit) and HVF snapshots (magic
        // `SMSNAP\x08\x00`, trailing NUL) stay diffable — preserving HVF chaining.
        Ok(()) => !(magic.starts_with(b"SMSNAP") && magic[7] == b'D'),
        Err(_) => true,
    }
}

/// Whether `snap` is a KVM FULL snapshot (`SMSNAP04`/`05`/`06`/… — `SMSNAP` +
/// trailing version digit), i.e. a valid base for `save_diff`. Used to track the
/// nearest preceding full so a layer after a KVM diff can diff against it instead
/// of paying a full save. HVF snapshots (trailing NUL) and KVM diffs (trailing
/// `D`) return false.
fn is_full_kvm_snapshot(snap: &Path) -> bool {
    use std::io::Read as _;
    let mut magic = [0u8; 8];
    std::fs::File::open(snap)
        .and_then(|mut f| f.read_exact(&mut magic))
        .is_ok()
        && magic.starts_with(b"SMSNAP")
        && magic[7].is_ascii_digit()
}

/// A valid, COMPLETE layer snapshot dir — the longest-cached-prefix predicate the
/// per-layer cache (`build_stage` step 2) scans with, so it must recognize every
/// backend's snapshot AND never accept a half-written one.
///
/// Backend-neutral, two arms:
///   - **KVM/HVF** — a RAM-dump snapshot: `restore.snap` + `metadata.json` (the
///     pair `vm.snapshot()` writes). Unchanged.
///   - **sentry** — a no-virt layer has no `restore.snap` (the warm state is the
///     filesystem, not guest pages): `write_sentry_snapshot_dir` emits
///     `metadata.json` (`"backend":"sentry"`) + a sibling `rootfs/` dir. Accept
///     that shape, gated on the `backend` field being `"sentry"` SPECIFICALLY
///     (parsed from `metadata.json`) — NOT on bare `rootfs/` presence — so a
///     half-written KVM dir (a `rootfs/` staged before `restore.snap` lands) is
///     never mistaken for a complete sentry layer.
///
/// Without the sentry arm the cache never hits on a no-virt host (every rebuild
/// re-runs from scratch), and `boot_from` never resumes from a sentry layer.
fn is_snapshot_dir(dir: &Path) -> bool {
    let has_meta = dir.join("metadata.json").is_file();
    if has_meta && compact_layer_manifest_path(dir).is_file() {
        return true;
    }
    // KVM/HVF: the RAM-dump pair.
    if has_meta && dir.join("restore.snap").is_file() {
        return true;
    }
    // Sentry: a self-contained rootfs copy + sentry metadata (no restore.snap).
    // Require BOTH a sibling `rootfs/` dir AND `metadata.json`'s backend=="sentry"
    // so the predicate can't false-positive on a half-written KVM layer dir.
    has_meta
        && dir.join("rootfs").is_dir()
        && std::fs::read_to_string(dir.join("metadata.json"))
            .ok()
            .and_then(|t| serde_json::from_str::<serde_json::Value>(&t).ok())
            .and_then(|v| {
                v.get("backend")
                    .and_then(|b| b.as_str())
                    .map(|s| s == "sentry")
            })
            .unwrap_or(false)
}

#[cfg(all(target_os = "macos", target_arch = "aarch64"))]
fn has_required_layer_runtime_metadata(dir: &Path) -> bool {
    let Ok(text) = std::fs::read_to_string(dir.join("metadata.json")) else {
        return false;
    };
    let Ok(meta) = serde_json::from_str::<serde_json::Value>(&text) else {
        return false;
    };
    match meta
        .get("backend")
        .and_then(|v| v.as_str())
        .unwrap_or("hvf")
    {
        "hvf" => {
            meta.get("baked_by_version")
                .and_then(|v| v.as_str())
                .is_some()
                && meta.get("runtime_sha16").and_then(|v| v.as_str()).is_some()
                && meta.get("tsi_token").and_then(|v| v.as_str()).is_some()
        }
        _ => true,
    }
}

#[cfg(not(all(target_os = "macos", target_arch = "aarch64")))]
fn has_required_layer_runtime_metadata(_dir: &Path) -> bool {
    true
}

/// Whether an instruction mutates the guest filesystem — i.e. warrants its own
/// snapshot. Metadata-only instructions (ENV/USER/LABEL/EXPOSE/ARG/CMD/
/// ENTRYPOINT/STOPSIGNAL/SHELL/VOLUME) leave the FS unchanged, so their layer
/// can reuse the previous snapshot.
fn mutates_fs(instr: &Instruction) -> bool {
    matches!(
        instr,
        Instruction::Run { .. }
            | Instruction::Copy { .. }
            | Instruction::Add { .. }
            | Instruction::Workdir(_)
    )
}

/// Materialize `dst` as a copy of snapshot dir `src`, preferring APFS
/// `clonefile` (instant copy-on-write — `restore.snap` shares blocks until
/// written) and falling back to a byte copy. Used for metadata-only layers,
/// whose filesystem is identical to the previous layer's.
fn clone_snapshot_dir(src: &Path, dst: &Path) -> Result<(), Error> {
    std::fs::create_dir_all(dst).map_err(Error::Io)?;
    for entry in std::fs::read_dir(src).map_err(Error::Io)? {
        let entry = entry.map_err(Error::Io)?;
        if !entry.file_type().map_err(Error::Io)?.is_file() {
            continue;
        }
        clone_or_copy_file(&entry.path(), &dst.join(entry.file_name()))?;
    }
    Ok(())
}

fn clone_or_copy_file(src: &Path, dst: &Path) -> Result<(), Error> {
    let _ = std::fs::remove_file(dst);
    #[cfg(target_os = "macos")]
    {
        use std::os::unix::ffi::OsStrExt;
        if let (Ok(s), Ok(d)) = (
            std::ffi::CString::new(src.as_os_str().as_bytes()),
            std::ffi::CString::new(dst.as_os_str().as_bytes()),
        ) {
            // clonefile(2): instant APFS CoW clone. Falls through to a byte
            // copy on failure (e.g. cross-device).
            if unsafe { libc::clonefile(s.as_ptr(), d.as_ptr(), 0) } == 0 {
                return Ok(());
            }
        }
    }
    std::fs::copy(src, dst).map_err(Error::Io)?;
    Ok(())
}

/// Resolve Dockerfile ENTRYPOINT + CMD into the effective argv (Docker
/// rules): shell-form ENTRYPOINT ignores CMD; exec-form ENTRYPOINT takes
/// CMD as default args (exec form only); else CMD alone. None → keep the
/// base image's command.
fn shellexec_to_argv(s: &ShellOrExec) -> Vec<String> {
    shellexec_to_argv_with(s, None)
}

/// Resolve a [`ShellOrExec`] to argv, wrapping shell-form in the given `SHELL`
/// interpreter (Docker default `["/bin/sh", "-c"]` when `shell` is `None`).
fn shellexec_to_argv_with(s: &ShellOrExec, shell: Option<&[String]>) -> Vec<String> {
    match s {
        ShellOrExec::Exec(a) => a.clone(),
        ShellOrExec::Shell(c) => {
            let mut argv: Vec<String> = shell
                .map(<[String]>::to_vec)
                .unwrap_or_else(|| vec!["/bin/sh".into(), "-c".into()]);
            argv.push(c.clone());
            argv
        }
    }
}

fn effective_cmd(config: &ImageConfig) -> Option<Vec<String>> {
    let shell = config.shell.as_deref();
    let to_argv = |s: &ShellOrExec| shellexec_to_argv_with(s, shell);
    match (&config.entrypoint, &config.cmd) {
        (Some(ep @ ShellOrExec::Shell(_)), _) => Some(to_argv(ep)),
        (Some(ShellOrExec::Exec(ep)), cmd) => {
            let mut argv = ep.clone();
            if let Some(ShellOrExec::Exec(c)) = cmd {
                argv.extend(c.clone());
            }
            Some(argv)
        }
        (None, Some(cmd)) => Some(to_argv(cmd)),
        (None, None) => None,
    }
}

/// Serialize a [`Healthcheck`] into Docker's image-config `Healthcheck` shape:
/// `{"Test": [...]}` where the test is `["NONE"]` (disabled), `["CMD-SHELL", s]`
/// (shell form), or `["CMD", arg...]` (exec form), plus any tuning fields. The
/// durations round-trip as the raw directive strings (e.g. `"30s"`) — the
/// builder records the directive, it doesn't run the probe.
fn healthcheck_json(hc: &Healthcheck) -> serde_json::Value {
    match hc {
        Healthcheck::None => serde_json::json!({ "Test": ["NONE"] }),
        Healthcheck::Cmd {
            test,
            interval,
            timeout,
            start_period,
            start_interval,
            retries,
        } => {
            let test_arr: Vec<String> = match test {
                ShellOrExec::Shell(s) => vec!["CMD-SHELL".to_string(), s.clone()],
                ShellOrExec::Exec(a) => {
                    let mut v = vec!["CMD".to_string()];
                    v.extend(a.iter().cloned());
                    v
                }
            };
            let mut obj = serde_json::Map::new();
            obj.insert("Test".into(), serde_json::json!(test_arr));
            let mut put = |k: &str, val: &Option<String>| {
                if let Some(v) = val {
                    obj.insert(k.into(), serde_json::json!(v));
                }
            };
            put("Interval", interval);
            put("Timeout", timeout);
            put("StartPeriod", start_period);
            put("StartInterval", start_interval);
            put("Retries", retries);
            serde_json::Value::Object(obj)
        }
    }
}

/// Patch a built snapshot's `metadata.json` with the Dockerfile's run-config
/// (effective CMD, WORKDIR, USER, ENV, STOPSIGNAL, HEALTHCHECK, ONBUILD), so the
/// built image runs the right default command like a Docker image.
/// `snapshot_file` is the `restore.snap` path ([`Image::snapshot_path`]);
/// metadata.json is its sibling.
fn persist_run_config(snapshot_file: &Path, config: &ImageConfig) -> Result<(), Error> {
    // Nothing to persist → don't touch the file (avoids mutating a base
    // image when a stage has no run-config instructions).
    if effective_cmd(config).is_none()
        && config.workdir.is_none()
        && config.user.is_none()
        && config.env.is_empty()
        && config.stop_signal.is_none()
        && config.healthcheck.is_none()
        && config.onbuild.is_empty()
    {
        return Ok(());
    }
    // `snapshot_path()` may be the snapshot dir or the `restore.snap` file
    // depending on how the Image was constructed — find metadata.json either
    // way (it's a sibling of restore.snap / inside the snapshot dir).
    let meta_path = if snapshot_file.is_dir() {
        snapshot_file.join("metadata.json")
    } else {
        snapshot_file
            .parent()
            .ok_or_else(|| Error::vm_msg("snapshot path has no parent directory"))?
            .join("metadata.json")
    };
    let text = std::fs::read_to_string(&meta_path).map_err(Error::Io)?;
    let mut meta: serde_json::Value = serde_json::from_str(&text)
        .map_err(|e| Error::vm_msg(format!("parse metadata.json: {e}")))?;
    let obj = meta
        .as_object_mut()
        .ok_or_else(|| Error::vm_msg("metadata.json is not an object"))?;

    if let Some(cmd) = effective_cmd(config) {
        obj.insert(
            "cmd".into(),
            serde_json::Value::Array(cmd.into_iter().map(serde_json::Value::String).collect()),
        );
    }
    if let Some(wd) = &config.workdir {
        obj.insert("working_dir".into(), serde_json::Value::String(wd.clone()));
    }
    if let Some(u) = &config.user {
        obj.insert("user".into(), serde_json::Value::String(u.clone()));
    }
    if !config.env.is_empty() {
        let env = obj
            .entry("env")
            .or_insert_with(|| serde_json::Value::Object(Default::default()));
        if let Some(map) = env.as_object_mut() {
            for (k, v) in &config.env {
                map.insert(k.clone(), serde_json::Value::String(v.clone()));
            }
        }
    }
    if let Some(sig) = &config.stop_signal {
        obj.insert("stop_signal".into(), serde_json::Value::String(sig.clone()));
    }
    if let Some(hc) = &config.healthcheck {
        obj.insert("healthcheck".into(), healthcheck_json(hc));
    }
    if !config.onbuild.is_empty() {
        obj.insert(
            "onbuild".into(),
            serde_json::Value::Array(
                config
                    .onbuild
                    .iter()
                    .cloned()
                    .map(serde_json::Value::String)
                    .collect(),
            ),
        );
    }

    let out = serde_json::to_string_pretty(&meta)
        .map_err(|e| Error::vm_msg(format!("serialize metadata.json: {e}")))?;
    // Atomic publish. Many builds re-persist the SAME content-addressed
    // baked/layer `metadata.json` concurrently (build_dockerfile calls this on
    // the finalized image AFTER the per-key CacheLock is released, so the writes
    // are unsynchronized). A plain truncate+write (`std::fs::write`) is observable
    // 0-bytes by a peer mid-write: another `persist_run_config`'s own read above,
    // a finalize `from_snapshot` reuse-check, or a verify cell's `sentry_sandbox`
    // all then see "parse metadata.json EOF". The from_snapshot EOF in particular
    // trips `finalize_baked_image`'s rebuild branch, which `remove_dir_all`s the
    // baked snapshot dir out from under live verify cells ("cannot read guest exe"
    // / "libm.so.6 missing" — the torn-tree symptom). Write a UNIQUE sibling tmp
    // (pid alone is insufficient — concurrent builds are threads in one process)
    // then rename(2): the live path is always a complete file.
    let parent = meta_path
        .parent()
        .ok_or_else(|| Error::vm_msg("metadata.json has no parent directory"))?;
    let seq = META_TMP_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    let tmp = parent.join(format!(".metadata.json.tmp.{}.{seq}", std::process::id()));
    std::fs::write(&tmp, out).map_err(Error::Io)?;
    if let Err(e) = std::fs::rename(&tmp, &meta_path) {
        let _ = std::fs::remove_file(&tmp);
        return Err(Error::Io(e));
    }
    Ok(())
}

/// Monotonic sequence for unique `metadata.json` staging-tmp names. Concurrent
/// builds run as threads in a single process, so `pid` alone collides — fold in
/// a process-wide counter so each atomic write stages to its own tmp.
static META_TMP_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// The result of committing a built image's filesystem to a read-only
/// squashfs (see [`commit_squashfs`]).
#[derive(Debug, Clone)]
pub struct CommitOutcome {
    /// Path to the produced read-only `rootfs.squashfs`.
    pub squashfs: PathBuf,
    /// Path to the `commit.json` manifest.
    pub manifest: PathBuf,
    /// Size of the squashfs in bytes.
    pub bytes: u64,
    /// hex(SHA-256) of the squashfs (content address).
    pub sha256: String,
}

/// Flatten a built image's full root filesystem into a content-addressed,
/// read-only squashfs under `dest` (+ a `commit.json` manifest). This is the
/// **runtime RAM-density** artifact: a snapshot-layer keeps the build's files
/// in the guest's tmpfs overlay (private RAM per VM); committing them to a
/// read-only squashfs lets every booted VM share those pages copy-on-write,
/// so density scales with VM count instead of each VM paying private RAM.
///
/// Mechanism (boot → tar the rootfs out → `mksquashfs`, reusing the bake's
/// zstd args): validated on HVF. **Gated on the x86 KVM box:** wiring the
/// committed squashfs back in as a bake lower (boot from it with an empty
/// tmpfs upper) + measuring multi-VM RAM density at scale — that's where this
/// optimization pays off and is meaningfully measured. The squashfs is
/// written in-process (backhand) — no `mksquashfs` subprocess.
pub fn commit_squashfs(image: &Image, dest: &Path) -> Result<CommitOutcome, Error> {
    use std::io::Read as _;

    std::fs::create_dir_all(dest).map_err(Error::Io)?;
    let stage = dest.join("rootfs");
    let _ = std::fs::remove_dir_all(&stage);
    std::fs::create_dir_all(&stage).map_err(Error::Io)?;

    // 1. Boot the image, stream its full rootfs tar out, unpack it to the stage.
    let tar_bytes = boot_and_tar_rootfs(image)?;
    tar::Archive::new(&tar_bytes[..])
        .unpack(&stage)
        .map_err(Error::Io)?;

    // 2. Build the squashfs of the staged tree in-process (zstd level 3, the
    //    same compression the bake uses; all-root so the committed rootfs is
    //    root-owned regardless of the host uid). No `mksquashfs` subprocess.
    let squashfs = dest.join("rootfs.squashfs");
    let _ = std::fs::remove_file(&squashfs);
    crate::bake::squashfs::write_squashfs(
        &stage,
        &squashfs,
        &crate::bake::squashfs::Ownership::AllRoot,
    )
    .map_err(|e| Error::vm_msg(format!("commit: squashfs: {e}")))?;
    let _ = std::fs::remove_dir_all(&stage);

    // 3. Content-address + manifest.
    let size = std::fs::metadata(&squashfs).map_err(Error::Io)?.len();
    let sha = {
        use ring::digest::{Context, SHA256};
        let mut f = std::fs::File::open(&squashfs).map_err(Error::Io)?;
        let mut ctx = Context::new(&SHA256);
        let mut buf = [0u8; 64 * 1024];
        loop {
            let n = f.read(&mut buf).map_err(Error::Io)?;
            if n == 0 {
                break;
            }
            ctx.update(&buf[..n]);
        }
        let d = ctx.finish();
        use std::fmt::Write as _;
        let mut s = String::with_capacity(64);
        for b in d.as_ref() {
            let _ = write!(s, "{b:02x}");
        }
        s
    };
    let manifest = dest.join("commit.json");
    let manifest_json = serde_json::json!({
        "kind": "supermachine-commit-squashfs",
        "source_snapshot": image.snapshot_path().to_string_lossy(),
        "squashfs": "rootfs.squashfs",
        "bytes": size,
        "sha256": sha,
        "committed_by_version": env!("CARGO_PKG_VERSION"),
    });
    std::fs::write(
        &manifest,
        serde_json::to_string_pretty(&manifest_json).unwrap_or_default(),
    )
    .map_err(Error::Io)?;

    Ok(CommitOutcome {
        squashfs,
        manifest,
        bytes: size,
        sha256: sha,
    })
}

/// Commit a built image to a read-only squashfs AND wrap it as a KVM-bootable
/// [`Image`] in one step — closing the in-VM builder's RAM-density loop
/// (component 3 of `docs/design/in-vm-builder-2026-06-05.md`).
///
/// A built (snapshot-layer) image bakes every RUN's outputs into the guest's
/// tmpfs overlay upper — captured in the snapshot as a FULL guest-RAM dump
/// (`restore.snap`, the whole `memory_mib`), so the artifact is hundreds of MiB
/// regardless of how small the actual files are, and each restore CoW-maps that
/// whole RAM image. This flattens the rootfs to a compressed read-only squashfs
/// ([`commit_squashfs`]) and re-bakes a cold-boot KVM image that mounts the
/// squashfs read-only under an EMPTY tmpfs upper. The payoff:
///   - the artifact shrinks to ~the filesystem size (measured ~7.5× smaller for
///     a 64 MiB-blob build: 512 MiB snapshot → 68 MiB squashfs), and
///   - the committed squashfs is read-only, so the host page cache holds ONE
///     copy of its blocks shared across every VM booted from it (each VM pages
///     in only what it reads), instead of each VM CoW-mapping a private
///     hundreds-of-MiB RAM snapshot. (Maximal zero-copy sharing of *decompressed*
///     file pages into guests is the virtio-fs + DAX path; this virtio-blk route
///     already removes the inflated per-image snapshot RAM.)
///
/// `dest` ends up self-contained (`rootfs.squashfs` + `commit.json` + `kernel`
/// + `agent.cpio` + `metadata.json`) and copy-free (the commit writes the
/// squashfs straight into `dest`, where the bake reuses it in place). Returns
/// the commit manifest and the bootable image.
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub fn commit_kvm_bootable(image: &Image, dest: &Path) -> Result<(CommitOutcome, Image), Error> {
    let commit = commit_squashfs(image, dest)?;
    let bootable = Image::bake_kvm_from_squashfs_auto(&commit.squashfs, dest)?;
    Ok((commit, bootable))
}

/// Boot `image` and stream its full root filesystem out as an uncompressed tar
/// (minus the virtual filesystems). Streamed over the exec STDOUT channel
/// (chunked, unbounded) — NOT `read_file`, whose single CONTROL frame caps at
/// 16 MiB and can't move a real rootfs. Shared by [`commit_squashfs`] and
/// [`export_oci`].
fn boot_and_tar_rootfs(image: &Image) -> Result<Vec<u8>, Error> {
    let pool = image
        .pool()
        .min(1)
        .max(1)
        .restore_on_release(false)
        // Sentry: don't auto-launch the image's baked workload while we tar its
        // rootfs out (we only want a read of the FS, not the CMD running). The
        // per-build private clone is harmless here. Backend-neutral.
        .for_build(true)
        .build()?;
    let vm = pool.acquire()?;
    await_agent_ready(&vm);
    // The base image's layer may not ship /tmp at all (node:20-slim and the
    // playwright base have no /tmp entry), and the build VM's RUN-time /tmp is an
    // uncommitted tmpfs mount — so without this a clean export has NO /tmp dir and
    // the warm-bake guest boots without one (the warmup then dies on "cannot
    // create /tmp/...: Directory nonexistent"). Pre-create the FHS-mandatory dirs
    // in the build VM's writable overlay (best-effort; mkdir -p no-ops if present;
    // 1777 for the world-writable temp dirs) so the tar below captures them as
    // empty dirs. Excluding only their CONTENTS (./tmp/*, ./run/*) keeps the dirs.
    let _ = vm
        .exec_builder()
        .timeout(RUN_TIMEOUT)
        .argv(
            [
                "sh",
                "-c",
                "mkdir -p /tmp /run /var/tmp; chmod 1777 /tmp /var/tmp 2>/dev/null || true",
            ]
            .iter()
            .copied(),
        )
        .output();
    let argv: Vec<&str> = vec![
        "tar",
        "-cf",
        "-",
        "--exclude=./proc",
        "--exclude=./sys",
        "--exclude=./dev",
        // Exclude the CONTENTS of /tmp and /run (ephemeral runtime state), but
        // KEEP the empty dirs themselves: they are FHS-mandatory and the
        // warm-bake guest needs them (npm writes /tmp/*, postgres uses
        // /run/postgresql, and squashfs.rs symlinks /var/run -> /run).
        // Excluding `./tmp`/`./run` outright dropped the dirs, so a clean
        // dockerfile-source warm-bake booted with NO /tmp → the warmup died on
        // "cannot create /tmp/offline-npmrc: Directory nonexistent". /proc, /sys,
        // /dev stay fully excluded — those are re-provided by the guest at boot.
        "--exclude=./tmp/*",
        "--exclude=./run/*",
        "-C",
        "/",
        ".",
    ];
    let out = vm
        .exec_builder()
        .timeout(RUN_TIMEOUT)
        .argv(argv.iter().copied())
        .output()
        .map_err(Error::Io)?;
    if !out.success() {
        return Err(Error::vm_msg(format!(
            "tar rootfs failed: {}",
            String::from_utf8_lossy(&out.stderr).trim()
        )));
    }
    Ok(out.stdout)
}

#[cfg(any(
    all(target_os = "macos", target_arch = "aarch64"),
    all(target_os = "linux", target_arch = "x86_64")
))]
struct OciLayerStreamOutcome {
    diff_id: String,
    layer_digest: String,
    layer_size: u64,
    tar_bytes: u64,
}

#[cfg(any(
    all(target_os = "macos", target_arch = "aarch64"),
    all(target_os = "linux", target_arch = "x86_64")
))]
struct HashingWriter<W> {
    inner: W,
    ctx: ring::digest::Context,
    bytes: u64,
}

#[cfg(any(
    all(target_os = "macos", target_arch = "aarch64"),
    all(target_os = "linux", target_arch = "x86_64")
))]
impl<W> HashingWriter<W> {
    fn new(inner: W) -> Self {
        Self {
            inner,
            ctx: ring::digest::Context::new(&ring::digest::SHA256),
            bytes: 0,
        }
    }

    fn finish(self) -> (W, String, u64) {
        (self.inner, digest_to_hex(self.ctx.finish()), self.bytes)
    }
}

#[cfg(any(
    all(target_os = "macos", target_arch = "aarch64"),
    all(target_os = "linux", target_arch = "x86_64")
))]
impl<W: std::io::Write> std::io::Write for HashingWriter<W> {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        let n = self.inner.write(buf)?;
        self.ctx.update(&buf[..n]);
        self.bytes += n as u64;
        Ok(n)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        self.inner.flush()
    }
}

#[cfg(any(
    all(target_os = "macos", target_arch = "aarch64"),
    all(target_os = "linux", target_arch = "x86_64")
))]
fn digest_to_hex(d: ring::digest::Digest) -> String {
    use std::fmt::Write as _;

    let mut s = String::with_capacity(64);
    for b in d.as_ref() {
        let _ = write!(s, "{b:02x}");
    }
    s
}

struct OciDiffWriter<W: std::io::Write> {
    gz: flate2::write::GzEncoder<HashingWriter<W>>,
    diff_ctx: ring::digest::Context,
    tar_bytes: u64,
}

impl<W: std::io::Write> OciDiffWriter<W> {
    fn new(inner: W) -> Self {
        Self {
            gz: flate2::write::GzEncoder::new(
                HashingWriter::new(inner),
                flate2::Compression::default(),
            ),
            diff_ctx: ring::digest::Context::new(&ring::digest::SHA256),
            tar_bytes: 0,
        }
    }

    fn finish(self) -> std::io::Result<(W, String, String, u64, u64)> {
        let hashing = self.gz.finish()?;
        let (inner, layer_digest, layer_size) = hashing.finish();
        Ok((
            inner,
            digest_to_hex(self.diff_ctx.finish()),
            layer_digest,
            layer_size,
            self.tar_bytes,
        ))
    }
}

impl<W: std::io::Write> std::io::Write for OciDiffWriter<W> {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.diff_ctx.update(buf);
        self.tar_bytes += buf.len() as u64;
        self.gz.write(buf)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        self.gz.flush()
    }
}

fn write_rootfs_diff_layer(
    parent_rootfs: &Path,
    current_rootfs: &Path,
    blobs: &Path,
) -> Result<OciLayerStreamOutcome, Error> {
    std::fs::create_dir_all(blobs).map_err(Error::Io)?;
    let tmp = blobs.join(format!(
        ".diff-layer.tmp.{}.{}",
        std::process::id(),
        unique_layer_suffix()
    ));
    let file = std::fs::File::create(&tmp).map_err(Error::Io)?;
    let diff_writer = OciDiffWriter::new(file);
    let mut tar = tar::Builder::new(diff_writer);

    let parent = collect_rootfs_entries(parent_rootfs)?;
    let current = collect_rootfs_entries(current_rootfs)?;
    for (rel, old) in &parent {
        match current.get(rel) {
            None => append_whiteout(&mut tar, rel)?,
            Some(new) if old.kind != new.kind => append_whiteout(&mut tar, rel)?,
            _ => {}
        }
    }
    for (rel, new) in &current {
        if rel.as_os_str().is_empty() {
            continue;
        }
        let changed = parent.get(rel) != Some(new);
        let type_replaced = parent
            .get(rel)
            .map(|old| old.kind != new.kind)
            .unwrap_or(false);
        if changed || type_replaced {
            append_fs_entry_to_tar(&mut tar, current_rootfs, rel, new)?;
        }
    }

    let diff_writer = tar.into_inner().map_err(Error::Io)?;
    let (file, diff_id, layer_digest, layer_size, tar_bytes) =
        diff_writer.finish().map_err(Error::Io)?;
    drop(file);
    let final_path = blobs.join(&layer_digest);
    if final_path.is_file() {
        let _ = std::fs::remove_file(&tmp);
    } else {
        std::fs::rename(&tmp, &final_path).map_err(Error::Io)?;
    }
    Ok(OciLayerStreamOutcome {
        diff_id,
        layer_digest,
        layer_size,
        tar_bytes,
    })
}

fn collect_rootfs_entries(root: &Path) -> Result<BTreeMap<PathBuf, FsEntryMeta>, Error> {
    let mut out = BTreeMap::new();
    let mut stack = vec![PathBuf::new()];
    while let Some(rel) = stack.pop() {
        let path = root.join(&rel);
        let md = std::fs::symlink_metadata(&path).map_err(Error::Io)?;
        let meta = fs_entry_meta(&path, &md)?;
        out.insert(rel.clone(), meta);
        if md.file_type().is_dir() {
            for entry in std::fs::read_dir(&path).map_err(Error::Io)? {
                let entry = entry.map_err(Error::Io)?;
                stack.push(rel.join(entry.file_name()));
            }
        }
    }
    Ok(out)
}

fn fs_entry_meta(path: &Path, md: &std::fs::Metadata) -> Result<FsEntryMeta, Error> {
    use std::os::unix::fs::{MetadataExt, PermissionsExt};

    let ft = md.file_type();
    let (kind, digest, link_target) = if ft.is_dir() {
        (FsEntryKind::Dir, None, None)
    } else if ft.is_file() {
        (FsEntryKind::File, Some(file_sha256(path)?), None)
    } else if ft.is_symlink() {
        (
            FsEntryKind::Symlink,
            None,
            Some(std::fs::read_link(path).map_err(Error::Io)?),
        )
    } else {
        // The no-KVM snapshot writers skip sockets/FIFOs/devices; ignore any
        // transient one that appears in a build rootfs diff scan.
        (FsEntryKind::File, Some(String::new()), None)
    };
    Ok(FsEntryMeta {
        kind,
        mode: md.permissions().mode(),
        uid: md.uid() as u64,
        gid: md.gid() as u64,
        mtime: md.mtime(),
        size: md.size(),
        digest,
        link_target,
    })
}

fn file_sha256(path: &Path) -> Result<String, Error> {
    let mut f = std::fs::File::open(path).map_err(Error::Io)?;
    let mut ctx = ring::digest::Context::new(&ring::digest::SHA256);
    let mut buf = [0u8; 128 * 1024];
    loop {
        let n = f.read(&mut buf).map_err(Error::Io)?;
        if n == 0 {
            break;
        }
        ctx.update(&buf[..n]);
    }
    Ok(digest_to_hex(ctx.finish()))
}

fn append_whiteout<W: std::io::Write>(tar: &mut tar::Builder<W>, rel: &Path) -> Result<(), Error> {
    let Some(name) = rel.file_name() else {
        return Ok(());
    };
    let marker_name = format!(".wh.{}", name.to_string_lossy());
    let marker = rel
        .parent()
        .filter(|p| !p.as_os_str().is_empty())
        .map(|p| p.join(&marker_name))
        .unwrap_or_else(|| PathBuf::from(marker_name));
    let mut header = tar::Header::new_gnu();
    header.set_entry_type(tar::EntryType::Regular);
    header.set_size(0);
    header.set_mode(0);
    tar.append_data(&mut header, marker, std::io::empty())
        .map_err(Error::Io)?;
    Ok(())
}

fn append_fs_entry_to_tar<W: std::io::Write>(
    tar: &mut tar::Builder<W>,
    root: &Path,
    rel: &Path,
    meta: &FsEntryMeta,
) -> Result<(), Error> {
    let full = root.join(rel);
    let mut header = tar::Header::new_gnu();
    header.set_mode(meta.mode);
    header.set_uid(meta.uid);
    header.set_gid(meta.gid);
    header.set_mtime(meta.mtime.max(0) as u64);
    match meta.kind {
        FsEntryKind::Dir => {
            header.set_entry_type(tar::EntryType::Directory);
            header.set_size(0);
            tar.append_data(&mut header, rel, std::io::empty())
                .map_err(Error::Io)?;
        }
        FsEntryKind::File => {
            header.set_entry_type(tar::EntryType::Regular);
            header.set_size(meta.size);
            let mut f = std::fs::File::open(&full).map_err(Error::Io)?;
            tar.append_data(&mut header, rel, &mut f)
                .map_err(Error::Io)?;
        }
        FsEntryKind::Symlink => {
            header.set_entry_type(tar::EntryType::Symlink);
            header.set_size(0);
            if let Some(target) = &meta.link_target {
                tar.append_link(&mut header, rel, target)
                    .map_err(Error::Io)?;
            } else {
                tar.append_data(&mut header, rel, std::io::empty())
                    .map_err(Error::Io)?;
            }
        }
    }
    Ok(())
}

fn apply_oci_diff_layer(diff_blob: &Path, rootfs: &Path) -> Result<(), Error> {
    let file = std::fs::File::open(diff_blob).map_err(Error::Io)?;
    let gz = flate2::read::GzDecoder::new(file);
    let mut archive = tar::Archive::new(gz);
    for entry in archive.entries().map_err(Error::Io)? {
        let mut entry = entry.map_err(Error::Io)?;
        let path = entry.path().map_err(Error::Io)?.to_path_buf();
        if is_oci_whiteout_path(&path) {
            apply_whiteout(rootfs, &path)?;
            continue;
        }
        entry.unpack_in(rootfs).map_err(Error::Io)?;
    }
    Ok(())
}

fn is_oci_whiteout_path(path: &Path) -> bool {
    path.file_name()
        .and_then(|n| n.to_str())
        .map(|n| n.starts_with(".wh."))
        .unwrap_or(false)
}

fn apply_whiteout(rootfs: &Path, marker: &Path) -> Result<(), Error> {
    let Some(name) = marker.file_name().and_then(|n| n.to_str()) else {
        return Ok(());
    };
    if name == ".wh..wh..opq" {
        let dir = marker
            .parent()
            .map(|p| rootfs.join(p))
            .unwrap_or_else(|| rootfs.to_path_buf());
        if let Ok(read) = std::fs::read_dir(&dir) {
            for entry in read.flatten() {
                let path = entry.path();
                let file_name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
                if file_name.starts_with(".wh.") {
                    continue;
                }
                let md = std::fs::symlink_metadata(&path).map_err(Error::Io)?;
                if md.file_type().is_dir() {
                    std::fs::remove_dir_all(path).map_err(Error::Io)?;
                } else {
                    std::fs::remove_file(path).map_err(Error::Io)?;
                }
            }
        }
        return Ok(());
    }
    let target_name = name.trim_start_matches(".wh.");
    let target = marker
        .parent()
        .map(|p| rootfs.join(p).join(target_name))
        .unwrap_or_else(|| rootfs.join(target_name));
    match std::fs::symlink_metadata(&target) {
        Ok(md) if md.file_type().is_dir() => std::fs::remove_dir_all(target).map_err(Error::Io)?,
        Ok(_) => std::fs::remove_file(target).map_err(Error::Io)?,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
        Err(e) => return Err(Error::Io(e)),
    }
    Ok(())
}

fn copy_tree_preserving(src: &Path, dst: &Path) -> std::io::Result<()> {
    use std::collections::HashMap;
    use std::ffi::CString;
    use std::os::unix::ffi::OsStrExt;
    use std::os::unix::fs::MetadataExt;

    fn set_owner(path: &Path, uid: u32, gid: u32, follow: bool) -> std::io::Result<()> {
        let c = CString::new(path.as_os_str().as_bytes()).map_err(|_| {
            std::io::Error::new(std::io::ErrorKind::InvalidInput, "path contains NUL")
        })?;
        let r = unsafe {
            if follow {
                libc::chown(c.as_ptr(), uid as libc::uid_t, gid as libc::gid_t)
            } else {
                libc::lchown(c.as_ptr(), uid as libc::uid_t, gid as libc::gid_t)
            }
        };
        if r == 0 {
            Ok(())
        } else {
            Err(std::io::Error::last_os_error())
        }
    }

    std::fs::create_dir_all(dst)?;
    let root_md = std::fs::symlink_metadata(src)?;
    std::fs::set_permissions(dst, root_md.permissions())?;
    let _ = set_owner(dst, root_md.uid(), root_md.gid(), true);
    let mut hardlinks: HashMap<(u64, u64), PathBuf> = HashMap::new();
    let mut stack = vec![(src.to_path_buf(), dst.to_path_buf())];
    while let Some((s, d)) = stack.pop() {
        for entry in std::fs::read_dir(&s)? {
            let entry = entry?;
            let sp = entry.path();
            let dp = d.join(entry.file_name());
            let md = std::fs::symlink_metadata(&sp)?;
            let ft = md.file_type();
            if ft.is_dir() {
                std::fs::create_dir_all(&dp)?;
                std::fs::set_permissions(&dp, md.permissions())?;
                let _ = set_owner(&dp, md.uid(), md.gid(), true);
                stack.push((sp, dp));
            } else if ft.is_symlink() {
                let target = std::fs::read_link(&sp)?;
                let _ = std::fs::remove_file(&dp);
                std::os::unix::fs::symlink(target, &dp)?;
                let _ = set_owner(&dp, md.uid(), md.gid(), false);
            } else if ft.is_file() {
                if md.nlink() > 1 {
                    let key = (md.dev(), md.ino());
                    if let Some(first) = hardlinks.get(&key) {
                        let _ = std::fs::remove_file(&dp);
                        std::fs::hard_link(first, &dp)?;
                        let _ = set_owner(&dp, md.uid(), md.gid(), true);
                        continue;
                    }
                    hardlinks.insert(key, dp.clone());
                }
                if crate::bake::clonefile_one(&sp, &dp).is_err() {
                    let _ = std::fs::remove_file(&dp);
                    std::fs::copy(&sp, &dp)?;
                }
                std::fs::set_permissions(&dp, md.permissions())?;
                let _ = set_owner(&dp, md.uid(), md.gid(), true);
            }
        }
    }
    Ok(())
}

/// Boot `image`, tar its rootfs, and write the OCI gzip layer blob while the
/// tar stream is still flowing. This keeps Dockerfile warm-bake export bounded
/// by small IO buffers instead of holding both the full tar and the gzip layer
/// in the embedding Node process.
#[cfg(any(
    all(target_os = "macos", target_arch = "aarch64"),
    all(target_os = "linux", target_arch = "x86_64")
))]
fn stream_rootfs_to_oci_layer(image: &Image, blobs: &Path) -> Result<OciLayerStreamOutcome, Error> {
    use std::io::{Read as _, Write as _};

    let pool = image
        .pool()
        .min(1)
        .max(1)
        .restore_on_release(false)
        .for_build(true)
        .build()?;
    let vm = pool.acquire()?;
    await_agent_ready(&vm);

    let _ = vm
        .exec_builder()
        .timeout(RUN_TIMEOUT)
        .argv(
            [
                "sh",
                "-c",
                "mkdir -p /tmp /run /var/tmp; chmod 1777 /tmp /var/tmp 2>/dev/null || true",
            ]
            .iter()
            .copied(),
        )
        .output();

    let argv: Vec<&str> = vec![
        "tar",
        "-cf",
        "-",
        "--exclude=./proc",
        "--exclude=./sys",
        "--exclude=./dev",
        "--exclude=./tmp/*",
        "-C",
        "/",
        ".",
    ];

    let mut child = vm
        .exec_builder()
        .argv(argv.iter().copied())
        .spawn()
        .map_err(Error::Io)?;
    drop(child.stdin());

    let mut stdout = child
        .stdout()
        .ok_or_else(|| Error::vm_msg("tar rootfs: missing stdout stream"))?;
    let stderr_handle = child.stderr().map(|mut stderr| {
        std::thread::spawn(move || {
            let mut buf = Vec::new();
            let _ = stderr.read_to_end(&mut buf);
            buf
        })
    });

    std::fs::create_dir_all(blobs).map_err(Error::Io)?;
    let tmp = blobs.join(format!(
        ".layer.tmp.{}.{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0)
    ));

    let file = std::fs::File::create(&tmp).map_err(Error::Io)?;
    let hashing_file = HashingWriter::new(file);
    let mut gz = flate2::write::GzEncoder::new(hashing_file, flate2::Compression::default());
    let mut diff_ctx = ring::digest::Context::new(&ring::digest::SHA256);
    let mut tar_bytes = 0u64;
    let mut buf = [0u8; 128 * 1024];

    loop {
        let n = stdout.read(&mut buf).map_err(Error::Io)?;
        if n == 0 {
            break;
        }
        diff_ctx.update(&buf[..n]);
        tar_bytes += n as u64;
        gz.write_all(&buf[..n]).map_err(Error::Io)?;
    }

    let hashing_file = gz.finish().map_err(Error::Io)?;
    let (file, layer_digest, layer_size) = hashing_file.finish();
    drop(file);

    let status = child.wait().map_err(Error::Io)?;
    let stderr = stderr_handle
        .and_then(|h| h.join().ok())
        .unwrap_or_default();
    if !status.success() {
        let _ = std::fs::remove_file(&tmp);
        return Err(Error::vm_msg(format!(
            "tar rootfs failed: {}",
            String::from_utf8_lossy(&stderr).trim()
        )));
    }

    let layer_path = blobs.join(&layer_digest);
    std::fs::rename(&tmp, layer_path).map_err(Error::Io)?;

    Ok(OciLayerStreamOutcome {
        diff_id: digest_to_hex(diff_ctx.finish()),
        layer_digest,
        layer_size,
        tar_bytes,
    })
}

/// The result of exporting a built image as an OCI image layout (see
/// [`export_oci`]).
#[derive(Debug, Clone)]
pub struct OciExportOutcome {
    /// The OCI image-layout directory (`oci-layout` + `index.json` + `blobs/`).
    pub layout_dir: PathBuf,
    /// `sha256:…` digest of the image manifest (what a registry addresses).
    pub manifest_digest: String,
    /// Uncompressed size of the single rootfs layer tar, in bytes.
    pub layer_bytes: u64,
}

/// Export a built image as a standard **OCI image layout** under `dest` — the
/// interoperable artifact (`skopeo copy oci:<dest> docker://…`, or any
/// OCI-aware tool), as opposed to the fused snapshot/squashfs run artifacts.
///
/// Flattens the rootfs to a SINGLE gzipped layer (boot → stream tar → gzip),
/// writes an OCI image config carrying the built run-config (Env/Cmd/Entrypoint/
/// WorkingDir/User from `config`) + the layer's `diff_id`, an OCI image
/// manifest, and the `oci-layout` + `index.json` envelope. All blobs are
/// content-addressed (`blobs/sha256/<hex>`). `arch` is the image's CPU arch
/// (e.g. `"amd64"`); the export targets `os=linux`.
///
/// One squashed layer (not per-build-layer) keeps export self-contained and is
/// what most consumers want; per-layer OCI export can layer on top later.
// Portable: boots via the pool + streams `tar` over exec, then gzip + sha256 +
// JSON + file writes — no backend-specific code. Available on both HVF and KVM
// (HVF needs it for the source:"dockerfile" + warmup chain: export → re-bake
// via oci-layout with the warmup/mounts).
#[cfg(any(
    all(target_os = "macos", target_arch = "aarch64"),
    all(target_os = "linux", target_arch = "x86_64")
))]
pub fn export_oci(
    image: &Image,
    config: &ImageConfig,
    arch: &str,
    dest: &Path,
) -> Result<OciExportOutcome, Error> {
    let blobs = dest.join("blobs/sha256");
    std::fs::create_dir_all(&blobs).map_err(Error::Io)?;
    let write_blob = |digest_hex: &str, bytes: &[u8]| -> Result<(), Error> {
        std::fs::write(blobs.join(digest_hex), bytes).map_err(Error::Io)
    };

    // 1. rootfs → one tar layer; diff_id = sha256(tar), then gzip it; the layer
    //    blob digest = sha256(gzip). (OCI: diff_id is over the *uncompressed*
    //    tar, the descriptor digest is over the compressed blob.)
    let layer = stream_rootfs_to_oci_layer(image, &blobs)?;
    let diff_id = layer.diff_id;
    let layer_digest = layer.layer_digest;
    let layer_size = layer.layer_size;

    // 2. OCI image config (carries the run-config + the layer diff_id).
    let env: Vec<String> = config.env.iter().map(|(k, v)| format!("{k}={v}")).collect();
    let mut cfg = serde_json::Map::new();
    if !env.is_empty() {
        cfg.insert("Env".into(), serde_json::json!(env));
    }
    let shell = config.shell.as_deref();
    if let Some(entry) = config
        .entrypoint
        .as_ref()
        .map(|e| shellexec_to_argv_with(e, shell))
    {
        cfg.insert("Entrypoint".into(), serde_json::json!(entry));
    }
    if let Some(cmd) = config
        .cmd
        .as_ref()
        .map(|c| shellexec_to_argv_with(c, shell))
    {
        cfg.insert("Cmd".into(), serde_json::json!(cmd));
    }
    if let Some(wd) = &config.workdir {
        cfg.insert("WorkingDir".into(), serde_json::json!(wd));
    }
    if let Some(user) = &config.user {
        cfg.insert("User".into(), serde_json::json!(user));
    }
    if !config.exposed_ports.is_empty() {
        let ports: serde_json::Map<String, serde_json::Value> = config
            .exposed_ports
            .iter()
            .map(|p| (format!("{p}/tcp"), serde_json::json!({})))
            .collect();
        cfg.insert("ExposedPorts".into(), serde_json::Value::Object(ports));
    }
    if !config.labels.is_empty() {
        let labels: serde_json::Map<String, serde_json::Value> = config
            .labels
            .iter()
            .map(|(k, v)| (k.clone(), serde_json::json!(v)))
            .collect();
        cfg.insert("Labels".into(), serde_json::Value::Object(labels));
    }
    if !config.volumes.is_empty() {
        // OCI `Volumes` is a set: map each mount point to an empty object.
        let vols: serde_json::Map<String, serde_json::Value> = config
            .volumes
            .iter()
            .map(|v| (v.clone(), serde_json::json!({})))
            .collect();
        cfg.insert("Volumes".into(), serde_json::Value::Object(vols));
    }
    if let Some(sig) = &config.stop_signal {
        cfg.insert("StopSignal".into(), serde_json::json!(sig));
    }
    if let Some(hc) = &config.healthcheck {
        cfg.insert("Healthcheck".into(), healthcheck_json(hc));
    }
    if !config.onbuild.is_empty() {
        // Docker image-config `OnBuild`: the trigger instructions (verbatim).
        cfg.insert("OnBuild".into(), serde_json::json!(config.onbuild));
    }
    if let Some(sh) = &config.shell {
        // Docker image-config extension; round-trips SHELL for downstream builds.
        cfg.insert("Shell".into(), serde_json::json!(sh));
    }
    let image_config = serde_json::json!({
        "architecture": arch,
        "os": "linux",
        "config": serde_json::Value::Object(cfg),
        "rootfs": { "type": "layers", "diff_ids": [format!("sha256:{diff_id}")] },
        "history": [{ "created_by": "supermachine in-VM builder (squashed export)" }],
    });
    let config_bytes =
        serde_json::to_vec(&image_config).map_err(|e| Error::vm_msg(format!("oci config: {e}")))?;
    let config_digest = sha256_hex(&config_bytes);
    let config_size = config_bytes.len() as u64;
    write_blob(&config_digest, &config_bytes)?;

    // 3. OCI image manifest → blob.
    let manifest = serde_json::json!({
        "schemaVersion": 2,
        "mediaType": "application/vnd.oci.image.manifest.v1+json",
        "config": {
            "mediaType": "application/vnd.oci.image.config.v1+json",
            "digest": format!("sha256:{config_digest}"),
            "size": config_size,
        },
        "layers": [{
            "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
            "digest": format!("sha256:{layer_digest}"),
            "size": layer_size,
        }],
    });
    let manifest_bytes =
        serde_json::to_vec(&manifest).map_err(|e| Error::vm_msg(format!("oci manifest: {e}")))?;
    let manifest_digest = sha256_hex(&manifest_bytes);
    write_blob(&manifest_digest, &manifest_bytes)?;

    // 4. oci-layout marker + index.json envelope.
    std::fs::write(dest.join("oci-layout"), r#"{"imageLayoutVersion":"1.0.0"}"#)
        .map_err(Error::Io)?;
    let index = serde_json::json!({
        "schemaVersion": 2,
        "mediaType": "application/vnd.oci.image.index.v1+json",
        "manifests": [{
            "mediaType": "application/vnd.oci.image.manifest.v1+json",
            "digest": format!("sha256:{manifest_digest}"),
            "size": manifest_bytes.len(),
            "platform": { "architecture": arch, "os": "linux" },
        }],
    });
    std::fs::write(
        dest.join("index.json"),
        serde_json::to_vec_pretty(&index).map_err(|e| Error::vm_msg(format!("oci index: {e}")))?,
    )
    .map_err(Error::Io)?;

    Ok(OciExportOutcome {
        layout_dir: dest.to_path_buf(),
        manifest_digest: format!("sha256:{manifest_digest}"),
        layer_bytes: layer.tar_bytes,
    })
}

/// hex(SHA-256(bytes)) via `ring` (no subprocess).
fn sha256_hex(bytes: &[u8]) -> String {
    use std::fmt::Write as _;
    let d = ring::digest::digest(&ring::digest::SHA256, bytes);
    let mut s = String::with_capacity(64);
    for b in d.as_ref() {
        let _ = write!(s, "{b:02x}");
    }
    s
}

/// Deterministic per-instruction representation for the cache key. For
/// COPY/ADD it folds in a content hash of the source bytes, so editing a
/// copied file busts the cache.
fn instr_repr(
    instr: &Instruction,
    context_dir: &Path,
    prior: &[Option<StageBuilt>],
) -> Result<String, Error> {
    Ok(match instr {
        Instruction::Run { run, mounts } => format!("RUN {mounts:?} {}", describe(run)),
        Instruction::Env(p) => format!("ENV {p:?}"),
        Instruction::Workdir(d) => format!("WORKDIR {d}"),
        Instruction::User(u) => format!("USER {u}"),
        Instruction::Entrypoint(e) => format!("ENTRYPOINT {}", describe(e)),
        Instruction::Cmd(c) => format!("CMD {}", describe(c)),
        Instruction::Expose(p) => format!("EXPOSE {p:?}"),
        Instruction::Label(p) => format!("LABEL {p:?}"),
        Instruction::Arg { name, default } => format!("ARG {name}={default:?}"),
        Instruction::Volume(v) => format!("VOLUME {v:?}"),
        Instruction::StopSignal(s) => format!("STOPSIGNAL {s}"),
        Instruction::Shell(s) => format!("SHELL {s:?}"),
        Instruction::Healthcheck(hc) => format!("HEALTHCHECK {hc:?}"),
        Instruction::Onbuild(t) => format!("ONBUILD {t}"),
        Instruction::Copy {
            sources,
            dest,
            flags,
        } => format!(
            "COPY {flags:?} {sources:?} {dest} {}",
            copy_cache_content(sources, context_dir, flags, prior)?
        ),
        Instruction::Add {
            sources,
            dest,
            flags,
        } => format!(
            "ADD {flags:?} {sources:?} {dest} {}",
            copy_cache_content(sources, context_dir, flags, prior)?
        ),
    })
}

/// The COPY/ADD content component of a cache key: the source stage's
/// identity key for `--from=<stage>`, else a hash of the host source bytes.
fn copy_cache_content(
    sources: &[String],
    context_dir: &Path,
    flags: &CopyFlags,
    prior: &[Option<StageBuilt>],
) -> Result<String, Error> {
    match &flags.from {
        Some(name) => Ok(format!("from={}", resolve_stage(prior, name)?.key)),
        None => copy_content_hash(sources, context_dir),
    }
}

/// `COPY --from=<stage>`: extract `sources` from the source stage's image to
/// a host temp dir, then stage them into the current VM at `dest` (reusing
/// the host-context COPY path). One tar out of the source VM, one in.
fn copy_from_stage(
    current_vm: &crate::api::Vm,
    source: &Image,
    sources: &[String],
    dest: &str,
    flags: &CopyFlags,
    workdir: Option<&str>,
) -> Result<(), Error> {
    let tmp = std::env::temp_dir().join(format!(
        "sm-copyfrom-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0)
    ));
    std::fs::create_dir_all(&tmp).map_err(Error::Io)?;

    let result = (|| {
        // 1. Tar the sources inside the source image, read the archive out.
        let pool = source
            .pool()
            .min(1)
            .max(1)
            .restore_on_release(false)
            // Sentry: suppress the source image's baked workload while we tar
            // `COPY --from` sources out of it; per-build private clone is harmless.
            // Backend-neutral (KVM/HVF ignore it).
            .for_build(true)
            .build()?;
        let svm = pool.acquire()?;
        await_agent_ready(&svm);
        let mut argv: Vec<String> = vec![
            "tar".into(),
            "-cf".into(),
            "/tmp/.sm-from.tar".into(),
            "-C".into(),
            "/".into(),
        ];
        for s in sources {
            argv.push(s.trim_start_matches('/').to_string());
        }
        let out = svm
            .exec_builder()
            .timeout(RUN_TIMEOUT)
            .argv(argv.iter().map(String::as_str))
            .output()
            .map_err(Error::Io)?;
        if !out.success() {
            return Err(Error::vm_msg(format!(
                "COPY --from tar failed in source stage: {}",
                String::from_utf8_lossy(&out.stderr).trim()
            )));
        }
        let bytes = svm
            .read_file_with_max_bytes("/tmp/.sm-from.tar", 4 * 1024 * 1024 * 1024)
            .map_err(Error::Io)?;
        tar::Archive::new(&bytes[..])
            .unpack(&tmp)
            .map_err(Error::Io)?;

        // 2. Stage from the host temp into the current VM. The extracted
        //    paths are the source paths (relative to /), so treat them as
        //    context-relative sources and reuse the COPY path.
        let rel: Vec<String> = sources
            .iter()
            .map(|s| s.trim_start_matches('/').to_string())
            .collect();
        let clean = CopyFlags {
            from: None,
            chown: flags.chown.clone(),
            chmod: flags.chmod.clone(),
        };
        stage_copy(current_vm, &rel, dest, &clean, &tmp, workdir)
    })();

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

/// Confine a COPY/ADD source to the build context. A hostile Dockerfile must
/// not read host files via `COPY ../../etc/passwd` or a symlink that escapes
/// the context — the context is uploaded by an untrusted user in a
/// build-as-a-service setting. Rejects absolute paths and any `..` component
/// up front (matching Docker, which forbids `COPY ../x`), then — defense in
/// depth against symlinks — requires the canonical path to stay within the
/// canonical context dir.
fn confine_to_context(context_dir: &Path, src: &str) -> Result<PathBuf, Error> {
    let rel = Path::new(src);
    if rel.is_absolute()
        || rel
            .components()
            .any(|c| matches!(c, std::path::Component::ParentDir))
    {
        return Err(Error::vm_msg(format!(
            "COPY/ADD source escapes the build context: {src}"
        )));
    }
    let host = context_dir.join(rel);
    if let (Ok(canon_ctx), Ok(canon_host)) = (context_dir.canonicalize(), host.canonicalize()) {
        if !canon_host.starts_with(&canon_ctx) {
            return Err(Error::vm_msg(format!(
                "COPY/ADD source resolves outside the build context (symlink?): {src}"
            )));
        }
    }
    Ok(host)
}

/// Hash the (context-relative path, file bytes) of all COPY/ADD sources, so
/// the cache key changes iff a copied file changes.
fn copy_content_hash(sources: &[String], context_dir: &Path) -> Result<String, Error> {
    let mut entries: Vec<(String, String)> = Vec::new();
    for src in sources {
        // URL sources (ADD <url>) aren't context files; skip — the URL string
        // is folded into the key via instr_repr.
        if is_url(src) {
            continue;
        }
        let host = confine_to_context(context_dir, src)?;
        let mut files = Vec::new();
        if host.is_dir() {
            walk_files(&host, &mut files).map_err(Error::Io)?;
        } else if host.is_file() {
            files.push(host.clone());
        }
        for f in files {
            let rel = f
                .strip_prefix(context_dir)
                .unwrap_or(&f)
                .to_string_lossy()
                .into_owned();
            let bytes = std::fs::read(&f).map_err(Error::Io)?;
            entries.push((rel, sha256_hex(&bytes)));
        }
    }
    entries.sort();
    let mut buf = String::new();
    for (rel, hash) in &entries {
        buf.push_str(rel);
        buf.push('\0');
        buf.push_str(hash);
        buf.push('\0');
    }
    Ok(sha256_hex(buf.as_bytes()))
}

/// Execute one RUN step with the accumulated env/cwd. `user` is recorded but
/// not yet applied (uid switching is a follow-up).
fn run_step(vm: &crate::api::Vm, cmd: &ShellOrExec, state: &BuildState) -> Result<(), Error> {
    let argv: Vec<String> = match cmd {
        // Shell form: run through the configured SHELL (Docker default
        // `["/bin/sh", "-c"]`), appending the command string as the final arg.
        ShellOrExec::Shell(s) => {
            let mut argv = state
                .shell
                .clone()
                .unwrap_or_else(|| vec!["/bin/sh".into(), "-c".into()]);
            argv.push(s.clone());
            argv
        }
        ShellOrExec::Exec(a) => a.clone(),
    };

    let mut b = vm.exec_builder().timeout(RUN_TIMEOUT);
    for (k, v) in &state.env {
        b = b.env(k, v);
    }
    if let Some(cwd) = &state.cwd {
        b = b.cwd(cwd);
    }
    let out = b
        .argv(argv.iter().map(String::as_str))
        .output()
        .map_err(Error::Io)?;

    if !out.success() {
        let stderr = String::from_utf8_lossy(&out.stderr);
        let shown = stderr.trim();
        let tail = &shown[shown.len().saturating_sub(2000)..];
        return Err(Error::vm_msg(format!(
            "RUN step failed: `{}`\n--- stderr (tail) ---\n{tail}",
            describe(cmd),
        )));
    }
    Ok(())
}

/// Execute a `RUN` carrying `--mount` directives: set each mount up in the
/// guest, run the command, then tear them down — persisting `type=cache`
/// contents back to the host and removing cache/secret/bind/tmpfs contents so
/// they never land in the layer snapshot. Teardown runs even on failure (so a
/// secret can't leak into the image).
fn run_with_mounts(
    vm: &crate::api::Vm,
    cmd: &ShellOrExec,
    mounts: &[RunMount],
    state: &BuildState,
    context_dir: &Path,
    prior: &[Option<StageBuilt>],
    mount_cache: Option<&Path>,
) -> Result<(), Error> {
    if mounts.is_empty() {
        return run_step(vm, cmd, state);
    }
    let mut teardowns: Vec<Teardown> = Vec::new();
    for m in mounts {
        setup_mount(
            vm,
            m,
            context_dir,
            prior,
            mount_cache,
            state.user.as_deref(),
            &mut teardowns,
        )?;
    }
    let result = run_step(vm, cmd, state);
    let mut teardown_err = None;
    for t in teardowns.into_iter().rev() {
        if let Err(e) = t.run(vm) {
            teardown_err.get_or_insert(e);
        }
    }
    match result {
        Err(e) => Err(e), // the RUN failure is the most informative
        Ok(()) => teardown_err.map(Err).unwrap_or(Ok(())),
    }
}

/// What to do with a mount after the RUN completes. `stash` (if any) is the
/// path the target's pre-existing image content was moved aside to during
/// setup — it's moved back so a mount never destroys base content.
enum Teardown {
    /// Persist the guest dir's contents to the host cache, then remove it and
    /// restore any stashed original. `_lock` serializes concurrent builds that
    /// share this cache id (released when the teardown is consumed).
    PersistCache {
        guest_target: String,
        host_cache: PathBuf,
        stash: Option<String>,
        _lock: Option<CacheLock>,
    },
    /// Remove a guest path (ephemeral cache, secret, bind, tmpfs) so it's not
    /// captured in the layer snapshot, then restore any stashed original.
    Remove {
        guest_path: String,
        stash: Option<String>,
    },
}

impl Teardown {
    fn run(self, vm: &crate::api::Vm) -> Result<(), Error> {
        match self {
            Teardown::PersistCache {
                guest_target,
                host_cache,
                stash,
                _lock,
            } => {
                persist_cache_from_guest(vm, &guest_target, &host_cache)?;
                remove_guest_path(vm, &guest_target);
                restore_stash(vm, stash.as_deref(), &guest_target);
                // _lock drops here, after the cache has been persisted.
                Ok(())
            }
            Teardown::Remove { guest_path, stash } => {
                remove_guest_path(vm, &guest_path);
                restore_stash(vm, stash.as_deref(), &guest_path);
                Ok(())
            }
        }
    }
}

/// Set up one mount before the RUN, recording how to tear it down. Any target
/// that already exists in the image is stashed aside first (and restored at
/// teardown) so the mount is transparent to the image — matching Docker, where
/// a mount shadows the target during the RUN and the original survives.
fn setup_mount(
    vm: &crate::api::Vm,
    m: &RunMount,
    context_dir: &Path,
    prior: &[Option<StageBuilt>],
    mount_cache: Option<&Path>,
    user: Option<&str>,
    teardowns: &mut Vec<Teardown>,
) -> Result<(), Error> {
    match &m.kind {
        MountKind::Cache => {
            let target = m
                .target
                .clone()
                .ok_or_else(|| Error::vm_msg("RUN --mount=type=cache requires target"))?;
            let stash = stash_target(vm, &target)?;
            ensure_guest_dir(vm, &target)?;
            match mount_cache {
                Some(root) => {
                    let id =
                        m.id.clone()
                            .unwrap_or_else(|| sha256_hex(target.as_bytes()));
                    let host_cache = root.join(sanitize_id(&id));
                    // Serialize concurrent builds sharing this cache id (held
                    // until the cache is persisted at teardown) so a parallel
                    // build can't read a half-written cache or interleave its
                    // own writes. Best-effort: proceed unlocked if it fails.
                    let _lock = CacheLock::acquire(&host_cache);
                    if host_cache.is_dir() {
                        restore_cache_into_guest(vm, &host_cache, &target)?;
                    }
                    teardowns.push(Teardown::PersistCache {
                        guest_target: target,
                        host_cache,
                        stash,
                        _lock,
                    });
                }
                // No persistence root (build_linear): ephemeral, just clean up.
                None => teardowns.push(Teardown::Remove {
                    guest_path: target,
                    stash,
                }),
            }
        }
        MountKind::Secret => {
            let target = m
                .target
                .clone()
                .unwrap_or_else(|| format!("/run/secrets/{}", m.id.as_deref().unwrap_or("secret")));
            let stash = stash_target(vm, &target)?;
            match secret_source(m) {
                Some(src) if src.is_file() => {
                    let bytes = std::fs::read(&src).map_err(Error::Io)?;
                    write_guest_file(vm, &target, &bytes)?;
                    // Owned by the build's USER (so a non-root RUN can read it),
                    // mode 0400 — matching BuildKit's per-user secret mount.
                    if let Some(u) = user {
                        let _ = vm
                            .exec_builder()
                            .argv(["chown", u, target.as_str()].iter().copied())
                            .output();
                    }
                    let _ = vm
                        .exec_builder()
                        .argv(["chmod", "0400", target.as_str()].iter().copied())
                        .output();
                }
                _ if m.required => {
                    // Restore the stash before bailing so the failed build's VM
                    // is left consistent (it's discarded anyway, but be tidy).
                    restore_stash(vm, stash.as_deref(), &target);
                    return Err(Error::vm_msg(format!(
                        "RUN --mount=type=secret id={:?} is required but no source was provided \
                         (set SUPERMACHINE_BUILD_SECRET_<ID> or the mount's src=)",
                        m.id
                    )));
                }
                // Optional + absent: present an empty file (Docker behavior).
                _ => write_guest_file(vm, &target, b"")?,
            }
            teardowns.push(Teardown::Remove {
                guest_path: target,
                stash,
            });
        }
        MountKind::Bind => {
            let target = m
                .target
                .clone()
                .ok_or_else(|| Error::vm_msg("RUN --mount=type=bind requires target"))?;
            let stash = stash_target(vm, &target)?;
            match &m.from {
                Some(stage_ref) => {
                    let src = resolve_stage(prior, stage_ref)?;
                    let source_img = Image::from_snapshot(src.image.snapshot_path())?;
                    let sub = m.source.clone().unwrap_or_else(|| "/".to_owned());
                    copy_from_stage(
                        vm,
                        &source_img,
                        &[sub],
                        &target,
                        &CopyFlags::default(),
                        None,
                    )?;
                }
                None => {
                    let sub = m.source.clone().unwrap_or_else(|| ".".to_owned());
                    ensure_guest_dir(vm, &target)?;
                    stage_copy(
                        vm,
                        &[sub],
                        &format!("{}/", target.trim_end_matches('/')),
                        &CopyFlags::default(),
                        context_dir,
                        None,
                    )?;
                }
            }
            teardowns.push(Teardown::Remove {
                guest_path: target,
                stash,
            });
        }
        MountKind::Tmpfs => {
            let target = m
                .target
                .clone()
                .ok_or_else(|| Error::vm_msg("RUN --mount=type=tmpfs requires target"))?;
            let stash = stash_target(vm, &target)?;
            ensure_guest_dir(vm, &target)?;
            teardowns.push(Teardown::Remove {
                guest_path: target,
                stash,
            });
        }
        // SSH agent forwarding / unknown kinds: unsupported, best-effort no-op.
        MountKind::Ssh | MountKind::Other(_) => {}
    }
    Ok(())
}

/// An exclusive cross-process lock on a `type=cache` id, held for the whole
/// restore→RUN→persist cycle so concurrent builds sharing the id (a shared
/// `cache_dir`) can't corrupt each other's cache. Advisory `flock(LOCK_EX)` on
/// a sibling `<host_cache>.lock`; released on Drop or process death.
pub(super) struct CacheLock {
    fd: std::os::unix::io::RawFd,
}

impl CacheLock {
    /// Acquire the lock (blocking). `None` on any failure — the caller then
    /// proceeds unlocked (no worse than having no lock at all).
    fn acquire(host_cache: &Path) -> Option<Self> {
        use std::os::unix::io::IntoRawFd;
        let mut lock_os = host_cache.as_os_str().to_owned();
        lock_os.push(".lock");
        let lock_path = PathBuf::from(lock_os);
        if let Some(parent) = lock_path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        let file = std::fs::OpenOptions::new()
            .create(true)
            .write(true)
            .truncate(false)
            .open(&lock_path)
            .ok()?;
        let fd = file.into_raw_fd();
        let rc = unsafe { libc::flock(fd, libc::LOCK_EX) };
        if rc != 0 {
            unsafe { libc::close(fd) };
            return None;
        }
        Some(CacheLock { fd })
    }
}

impl Drop for CacheLock {
    fn drop(&mut self) {
        unsafe {
            libc::flock(self.fd, libc::LOCK_UN);
            libc::close(self.fd);
        }
    }
}

/// If `target` already exists in the image, move it aside to a unique stash
/// path (returned) so a mount can't destroy pre-existing content on teardown.
fn stash_target(vm: &crate::api::Vm, target: &str) -> Result<Option<String>, Error> {
    let t = target.trim_end_matches('/');
    if t.is_empty() || t == "/" {
        return Ok(None);
    }
    let exists = vm
        .exec_builder()
        .argv(["test", "-e", target].iter().copied())
        .output()
        .map(|o| o.success())
        .unwrap_or(false);
    if !exists {
        return Ok(None);
    }
    let stash = format!("{t}.sm-mount-stash.{}", unique_suffix());
    let out = vm
        .exec_builder()
        .argv(["mv", target, stash.as_str()].iter().copied())
        .output()
        .map_err(Error::Io)?;
    if !out.success() {
        return Err(Error::vm_msg(format!(
            "RUN --mount: could not stash pre-existing {target}: {}",
            String::from_utf8_lossy(&out.stderr).trim()
        )));
    }
    Ok(Some(stash))
}

/// Move a stashed original back to `target` (best-effort), after the mount's
/// own content has been removed.
fn restore_stash(vm: &crate::api::Vm, stash: Option<&str>, target: &str) {
    if let Some(s) = stash {
        let _ = vm
            .exec_builder()
            .argv(["mv", s, target].iter().copied())
            .output();
    }
}

/// A process-unique suffix (pid + nanos) for transient guest paths.
fn unique_suffix() -> String {
    format!(
        "{}.{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0)
    )
}

/// Resolve a secret's host source: an explicit `src=` on the mount, else the
/// `SUPERMACHINE_BUILD_SECRET_<ID>` env var (uppercased, `-`→`_`).
fn secret_source(m: &RunMount) -> Option<PathBuf> {
    if let Some(src) = &m.source {
        return Some(PathBuf::from(src));
    }
    let id = m.id.as_deref()?;
    let var = format!(
        "SUPERMACHINE_BUILD_SECRET_{}",
        id.to_ascii_uppercase().replace('-', "_")
    );
    std::env::var(var).ok().map(PathBuf::from)
}

/// A safe single path component for a cache id (no traversal / separators).
fn sanitize_id(id: &str) -> String {
    let cleaned: String = id
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
                c
            } else {
                '_'
            }
        })
        .collect();
    if cleaned.is_empty() {
        sha256_hex(id.as_bytes())
    } else {
        cleaned
    }
}

fn ensure_guest_dir(vm: &crate::api::Vm, dir: &str) -> Result<(), Error> {
    let out = vm
        .exec_builder()
        .argv(["mkdir", "-p", dir].iter().copied())
        .output()
        .map_err(Error::Io)?;
    if !out.success() {
        return Err(Error::vm_msg(format!(
            "RUN --mount: mkdir {dir} failed: {}",
            String::from_utf8_lossy(&out.stderr).trim()
        )));
    }
    Ok(())
}

fn write_guest_file(vm: &crate::api::Vm, path: &str, bytes: &[u8]) -> Result<(), Error> {
    if let Some((parent, _)) = path.rsplit_once('/') {
        if !parent.is_empty() {
            ensure_guest_dir(vm, parent)?;
        }
    }
    vm.write_file(path, bytes).map_err(Error::Io)
}

/// Best-effort `rm -rf` of a guest path, with a guard so we never wipe `/`.
fn remove_guest_path(vm: &crate::api::Vm, path: &str) {
    let p = path.trim_end_matches('/');
    if p.is_empty() || p == "/" {
        return;
    }
    let _ = vm
        .exec_builder()
        .argv(["rm", "-rf", path].iter().copied())
        .output();
}

/// Restore a host cache directory's contents into the guest at `target`
/// (host-built tar → one write → one extract).
fn restore_cache_into_guest(
    vm: &crate::api::Vm,
    host_cache: &Path,
    target: &str,
) -> Result<(), Error> {
    let mut ar = tar::Builder::new(Vec::new());
    ar.append_dir_all(".", host_cache).map_err(Error::Io)?;
    let tarball = ar.into_inner().map_err(Error::Io)?;
    if tarball.is_empty() {
        return Ok(());
    }
    let staged = "/tmp/.sm-cache-in.tar";
    vm.write_file(staged, &tarball).map_err(Error::Io)?;
    let out = vm
        .exec_builder()
        .timeout(RUN_TIMEOUT)
        .argv(
            ["tar", "-xf", staged, "--no-same-owner", "-C", target]
                .iter()
                .copied(),
        )
        .output()
        .map_err(Error::Io)?;
    let _ = vm
        .exec_builder()
        .argv(["rm", "-f", staged].iter().copied())
        .output();
    if !out.success() {
        return Err(Error::vm_msg(format!(
            "RUN --mount=cache: restore extract failed: {}",
            String::from_utf8_lossy(&out.stderr).trim()
        )));
    }
    Ok(())
}

/// Persist the guest `target` directory's contents back to the host cache
/// (one in-guest tar → read out → host extract), so the next build reuses it.
fn persist_cache_from_guest(
    vm: &crate::api::Vm,
    target: &str,
    host_cache: &Path,
) -> Result<(), Error> {
    let target_exists = vm
        .exec_builder()
        .argv(["test", "-d", target].iter().copied())
        .output()
        .map(|o| o.success())
        .unwrap_or(false);
    if !target_exists {
        return Ok(());
    }
    ensure_guest_dir(vm, "/tmp")?;
    let staged = "/tmp/.sm-cache-out.tar";
    let out = vm
        .exec_builder()
        .timeout(RUN_TIMEOUT)
        .argv(["tar", "-cf", staged, "-C", target, "."].iter().copied())
        .output()
        .map_err(Error::Io)?;
    if !out.success() {
        return Err(Error::vm_msg(format!(
            "RUN --mount=cache: persist tar failed: {}",
            String::from_utf8_lossy(&out.stderr).trim()
        )));
    }
    let bytes = vm
        .read_file_with_max_bytes(staged, 4 * 1024 * 1024 * 1024)
        .map_err(Error::Io)?;
    let _ = vm
        .exec_builder()
        .argv(["rm", "-f", staged].iter().copied())
        .output();
    std::fs::create_dir_all(host_cache).map_err(Error::Io)?;
    tar::Archive::new(&bytes[..])
        .unpack(host_cache)
        .map_err(Error::Io)?;
    Ok(())
}

fn describe(cmd: &ShellOrExec) -> String {
    match cmd {
        ShellOrExec::Shell(s) => s.clone(),
        ShellOrExec::Exec(a) => a.join(" "),
    }
}

/// Stage `COPY`/`ADD` sources from the host build context into the guest,
/// honoring Docker's file/dir/trailing-slash dest semantics. Builds an
/// in-memory tar named by the final guest paths, writes it into the VM in
/// one shot, and extracts it at `/` — one write + one exec regardless of
/// file count (a `node_modules` COPY is not N vsock round-trips).
fn stage_copy(
    vm: &crate::api::Vm,
    sources: &[String],
    dest: &str,
    flags: &CopyFlags,
    context_dir: &Path,
    workdir: Option<&str>,
) -> Result<(), Error> {
    if flags.from.is_some() {
        return Err(Error::vm_msg(
            "COPY --from (multi-stage) is not yet supported by the executor",
        ));
    }
    let wd = workdir.unwrap_or("/");
    let abs_dest = if dest.starts_with('/') {
        dest.to_string()
    } else {
        format!("{}/{}", wd.trim_end_matches('/'), dest)
    };
    let single_dir = sources.len() == 1 && context_dir.join(&sources[0]).is_dir();
    let dest_is_dir = dest.ends_with('/') || sources.len() > 1 || single_dir;

    // Resolve sources → (host file, final absolute guest path) pairs.
    let mut pairs: Vec<(PathBuf, String)> = Vec::new();
    for src in sources {
        let host_src = confine_to_context(context_dir, src)?;
        if host_src.is_dir() {
            // Directory CONTENTS copy into dest (Docker semantics).
            let mut files = Vec::new();
            walk_files(&host_src, &mut files).map_err(Error::Io)?;
            for f in files {
                let rel = f.strip_prefix(&host_src).unwrap_or(&f);
                let guest = format!(
                    "{}/{}",
                    abs_dest.trim_end_matches('/'),
                    rel.to_string_lossy()
                );
                pairs.push((f, guest));
            }
        } else if host_src.is_file() {
            let guest = if dest_is_dir {
                let name = host_src
                    .file_name()
                    .map(|n| n.to_string_lossy().into_owned())
                    .unwrap_or_default();
                format!("{}/{}", abs_dest.trim_end_matches('/'), name)
            } else {
                abs_dest.clone()
            };
            pairs.push((host_src, guest));
        } else {
            return Err(Error::vm_msg(format!(
                "COPY/ADD source not found in build context: {src}"
            )));
        }
    }
    if pairs.is_empty() {
        return Ok(());
    }

    // In-memory tar named by final guest paths (relative to `/`).
    let mut ar = tar::Builder::new(Vec::new());
    for (host, guest) in &pairs {
        let entry = guest.trim_start_matches('/');
        append_host_file_to_tar(&mut ar, host, Path::new(entry))?;
    }
    let tarball = ar.into_inner().map_err(Error::Io)?;

    // One write + one extract. `--no-same-owner` lands files as root:root
    // (Docker's COPY default); `--chown` overrides below.
    let staged = "/tmp/.sm-copy.tar";
    vm.write_file(staged, &tarball).map_err(Error::Io)?;
    let extract = vm
        .exec_builder()
        .timeout(RUN_TIMEOUT)
        .argv(
            ["tar", "-xf", staged, "--no-same-owner", "-C", "/"]
                .iter()
                .copied(),
        )
        .output()
        .map_err(Error::Io)?;
    let _ = vm
        .exec_builder()
        .argv(["rm", "-f", staged].iter().copied())
        .output();
    if !extract.success() {
        return Err(Error::vm_msg(format!(
            "COPY/ADD extract failed: {}",
            String::from_utf8_lossy(&extract.stderr).trim()
        )));
    }

    if let Some(chown) = &flags.chown {
        let argv = ["chown", "-R", chown.as_str(), abs_dest.as_str()];
        let _ = vm.exec_builder().argv(argv.iter().copied()).output();
    }
    if let Some(chmod) = &flags.chmod {
        let argv = ["chmod", "-R", chmod.as_str(), abs_dest.as_str()];
        let _ = vm.exec_builder().argv(argv.iter().copied()).output();
    }
    Ok(())
}

fn append_host_file_to_tar<W: std::io::Write>(
    tar: &mut tar::Builder<W>,
    host: &Path,
    entry: &Path,
) -> Result<(), Error> {
    use std::os::unix::fs::{MetadataExt, PermissionsExt};

    let md = std::fs::metadata(host).map_err(Error::Io)?;
    let mut header = tar::Header::new_gnu();
    header.set_entry_type(tar::EntryType::Regular);
    header.set_size(md.len());
    header.set_mode(md.permissions().mode());
    header.set_uid(md.uid() as u64);
    header.set_gid(md.gid() as u64);
    header.set_mtime(md.mtime().max(0) as u64);
    let mut file = std::fs::File::open(host).map_err(Error::Io)?;
    tar.append_data(&mut header, entry, &mut file)
        .map_err(Error::Io)?;
    Ok(())
}

/// Recursively collect regular files under `dir`. (Symlinks skipped in v1.)
fn walk_files(dir: &Path, out: &mut Vec<PathBuf>) -> std::io::Result<()> {
    for entry in std::fs::read_dir(dir)? {
        let entry = entry?;
        let ft = entry.file_type()?;
        let path = entry.path();
        if ft.is_dir() {
            walk_files(&path, out)?;
        } else if ft.is_file() {
            out.push(path);
        }
    }
    Ok(())
}

#[cfg(test)]
mod shell_volume_stopsignal_tests {
    use super::*;
    use crate::builder::dockerfile::Instruction;
    use std::path::Path;

    fn apply(instr: &Instruction, state: &mut BuildState, config: &mut ImageConfig) {
        // vm=None: metadata-only instructions update state/config without a VM.
        apply_instr(None, instr, state, config, Path::new("."), &[], None).unwrap();
    }

    #[test]
    fn shell_form_wraps_in_configured_shell() {
        // Default SHELL is `/bin/sh -c`.
        assert_eq!(
            shellexec_to_argv_with(&ShellOrExec::Shell("echo hi".into()), None),
            vec!["/bin/sh", "-c", "echo hi"]
        );
        // A configured SHELL (e.g. bash) wraps the command string.
        let bash = [
            "/bin/bash".to_string(),
            "-eo".into(),
            "pipefail".into(),
            "-c".into(),
        ];
        assert_eq!(
            shellexec_to_argv_with(&ShellOrExec::Shell("set -x; make".into()), Some(&bash)),
            vec!["/bin/bash", "-eo", "pipefail", "-c", "set -x; make"]
        );
        // Exec form is passed through verbatim regardless of SHELL.
        assert_eq!(
            shellexec_to_argv_with(
                &ShellOrExec::Exec(vec!["/usr/bin/true".into()]),
                Some(&bash)
            ),
            vec!["/usr/bin/true"]
        );
    }

    #[test]
    fn shell_instruction_sets_state_and_config() {
        let mut state = BuildState::default();
        let mut config = ImageConfig::default();
        let sh = vec!["/bin/bash".to_string(), "-c".into()];
        apply(&Instruction::Shell(sh.clone()), &mut state, &mut config);
        assert_eq!(state.shell.as_deref(), Some(sh.as_slice()));
        assert_eq!(config.shell.as_deref(), Some(sh.as_slice()));
    }

    #[test]
    fn effective_cmd_honors_configured_shell() {
        let mut config = ImageConfig {
            cmd: Some(ShellOrExec::Shell("nginx -g 'daemon off;'".into())),
            ..Default::default()
        };
        // Default shell.
        assert_eq!(
            effective_cmd(&config).unwrap(),
            vec!["/bin/sh", "-c", "nginx -g 'daemon off;'"]
        );
        // After a SHELL instruction, shell-form CMD uses it.
        config.shell = Some(vec!["/bin/bash".into(), "-c".into()]);
        assert_eq!(
            effective_cmd(&config).unwrap(),
            vec!["/bin/bash", "-c", "nginx -g 'daemon off;'"]
        );
    }

    #[test]
    fn base_image_config_reads_snapshot_metadata() {
        // base_image_config reads the FROM base's recorded image config from its
        // snapshot metadata.json, preferring a top-level `image_env`, else
        // `native_bake_inputs.env.env`, returning sorted (K,V) pairs.
        let dir = std::env::temp_dir().join(format!(
            "sm-baseenv-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos())
                .unwrap_or(0)
        ));
        std::fs::create_dir_all(&dir).unwrap();
        // Case A: native_bake_inputs.env.env (what the bake actually writes).
        std::fs::write(
            dir.join("metadata.json"),
            serde_json::json!({
                "native_bake_inputs": { "env": { "env": {
                    "PLAYWRIGHT_BROWSERS_PATH": "/ms-playwright",
                    "LANG": "C.UTF-8",
                    "PATH": "/usr/local/bin:/usr/bin"
                }, "secrets": {} } }
            })
            .to_string(),
        )
        .unwrap();
        // restore.snap present so snapshot_path()/is_dir resolution works.
        std::fs::write(dir.join("restore.snap"), b"x").unwrap();
        let img = Image::from_snapshot(&dir).expect("load snapshot");
        let env = base_image_config(&img).env;
        assert_eq!(
            env,
            vec![
                ("LANG".to_string(), "C.UTF-8".to_string()),
                ("PATH".to_string(), "/usr/local/bin:/usr/bin".to_string()),
                (
                    "PLAYWRIGHT_BROWSERS_PATH".to_string(),
                    "/ms-playwright".to_string()
                ),
            ],
            "sorted base env from native_bake_inputs.env.env"
        );

        // Case B: a top-level `image_env` wins over native_bake_inputs.
        std::fs::write(
            dir.join("metadata.json"),
            serde_json::json!({
                "image_env": { "FOO": "bar" },
                "native_bake_inputs": { "env": { "env": { "FOO": "ignored" } } }
            })
            .to_string(),
        )
        .unwrap();
        let img = Image::from_snapshot(&dir).expect("reload");
        assert_eq!(
            base_image_config(&img).env,
            vec![("FOO".to_string(), "bar".to_string())]
        );

        // Case C: no env recorded (e.g. scratch) → empty.
        std::fs::write(
            dir.join("metadata.json"),
            serde_json::json!({ "cmd": ["sh"] }).to_string(),
        )
        .unwrap();
        let img = Image::from_snapshot(&dir).expect("reload2");
        let config = base_image_config(&img);
        assert!(config.env.is_empty());
        assert_eq!(config.cmd, Some(ShellOrExec::Exec(vec!["sh".into()])));

        std::fs::write(
            dir.join("metadata.json"),
            serde_json::json!({
                "cmd": ["postgres"],
                "entrypoint": ["docker-entrypoint.sh"],
                "working_dir": "/var/lib/postgresql",
                "user": "postgres",
                "shell": ["/bin/ash", "-c"]
            })
            .to_string(),
        )
        .unwrap();
        let img = Image::from_snapshot(&dir).expect("reload3");
        let config = base_image_config(&img);
        assert_eq!(config.cmd, Some(ShellOrExec::Exec(vec!["postgres".into()])));
        assert_eq!(
            config.entrypoint,
            Some(ShellOrExec::Exec(vec!["docker-entrypoint.sh".into()]))
        );
        assert_eq!(config.workdir.as_deref(), Some("/var/lib/postgresql"));
        assert_eq!(config.user.as_deref(), Some("postgres"));
        assert_eq!(
            config.shell.as_deref(),
            Some(["/bin/ash".to_string(), "-c".to_string()].as_slice())
        );

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

    #[test]
    fn build_memory_policy() {
        // No env → the generous default, but never below the base's own RAM.
        assert_eq!(resolve_build_memory(None, 2048), BUILD_MEMORY_DEFAULT_MIB);
        // A base larger than the default is honored (never shrunk).
        assert_eq!(resolve_build_memory(None, 16384), 16384);
        // A sane env override wins over the default…
        assert_eq!(resolve_build_memory(Some("4096"), 2048), 4096);
        // …but still can't shrink a larger base.
        assert_eq!(resolve_build_memory(Some("4096"), 6000), 6000);
        // Whitespace tolerated.
        assert_eq!(resolve_build_memory(Some("  3000 "), 1024), 3000);
        // Nonsense / too-small / non-numeric env → fall back to the default.
        assert_eq!(
            resolve_build_memory(Some("0"), 2048),
            BUILD_MEMORY_DEFAULT_MIB
        );
        assert_eq!(
            resolve_build_memory(Some("128"), 2048),
            BUILD_MEMORY_DEFAULT_MIB
        ); // < 256 floor
        assert_eq!(
            resolve_build_memory(Some("xyz"), 2048),
            BUILD_MEMORY_DEFAULT_MIB
        );
        assert_eq!(
            resolve_build_memory(Some(""), 512),
            BUILD_MEMORY_DEFAULT_MIB
        );
    }

    #[test]
    fn incremental_full_rootfs_prune_removes_earlier_layers_after_mutating_layer() {
        let dir = std::env::temp_dir().join(format!(
            "sm-full-rootfs-prune-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos())
                .unwrap_or(0)
        ));
        let layers = dir.join("layers");
        let keys = vec!["a".to_string(), "b".to_string(), "c".to_string()];
        for key in &keys {
            let layer = layers.join(key);
            std::fs::create_dir_all(&layer).unwrap();
            std::fs::write(layer.join("metadata.json"), "{}").unwrap();
        }

        let current = layers.join("b");
        let pruned = prune_earlier_full_rootfs_layer_snapshots(&dir, &keys, 1, &current)
            .expect("prune previous");
        assert_eq!(pruned, vec![layers.join("a")]);
        assert!(!layers.join("a").exists(), "previous layer was pruned");
        assert!(layers.join("b").exists(), "current layer is kept");
        assert!(layers.join("c").exists(), "future layer is untouched");

        std::fs::create_dir_all(layers.join("a")).unwrap();
        let current = layers.join("c");
        let pruned = prune_earlier_full_rootfs_layer_snapshots(&dir, &keys, 2, &current)
            .expect("prune all earlier");
        assert_eq!(pruned, vec![layers.join("a"), layers.join("b")]);
        assert!(!layers.join("a").exists(), "earlier layer was pruned");
        assert!(!layers.join("b").exists(), "previous layer was pruned");
        assert!(layers.join("c").exists(), "current layer is kept");

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

    #[test]
    fn compact_rootfs_diff_round_trips_files_symlinks_and_deletions() {
        let dir = unique_test_dir("sm-rootfs-diff-roundtrip");
        let parent = dir.join("parent");
        let current = dir.join("current");
        let replay = dir.join("replay");
        let blobs = dir.join("blobs/sha256");
        std::fs::create_dir_all(parent.join("dir")).unwrap();
        std::fs::write(parent.join("dir/changed.txt"), b"old").unwrap();
        std::fs::write(parent.join("dir/deleted.txt"), b"delete me").unwrap();
        std::fs::write(parent.join("same.txt"), b"same").unwrap();

        std::fs::create_dir_all(current.join("dir")).unwrap();
        std::fs::write(current.join("dir/changed.txt"), b"new").unwrap();
        std::fs::write(current.join("dir/added.txt"), b"add me").unwrap();
        std::fs::write(current.join("same.txt"), b"same").unwrap();
        #[cfg(unix)]
        std::os::unix::fs::symlink("../same.txt", current.join("dir/link")).unwrap();

        let layer = write_rootfs_diff_layer(&parent, &current, &blobs).expect("write diff");
        assert!(blobs.join(&layer.layer_digest).is_file());

        copy_tree_preserving(&parent, &replay).unwrap();
        apply_oci_diff_layer(&blobs.join(&layer.layer_digest), &replay).expect("apply diff");
        assert_eq!(
            collect_rootfs_entries(&replay).unwrap(),
            collect_rootfs_entries(&current).unwrap()
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn compact_rootfs_diff_round_trips_long_paths() {
        let dir = unique_test_dir("sm-rootfs-diff-long-path");
        let parent = dir.join("parent");
        let current = dir.join("current");
        let replay = dir.join("replay");
        let blobs = dir.join("blobs/sha256");
        let long_dir = PathBuf::from(
            "node_modules/playwright-core/.local-browsers/ms-playwright/chromium-1217/chrome-linux64/PrivacySandboxAttestationsPreloaded",
        );
        assert!(
            long_dir.as_os_str().len() > 100,
            "test must exceed the raw tar header path field"
        );

        std::fs::create_dir_all(&parent).unwrap();
        std::fs::create_dir_all(current.join(&long_dir)).unwrap();
        std::fs::write(current.join(&long_dir).join("manifest.json"), b"{}").unwrap();
        std::fs::write(
            current
                .join(&long_dir)
                .join("privacy-sandbox-attestations.dat"),
            b"dat",
        )
        .unwrap();

        let layer = write_rootfs_diff_layer(&parent, &current, &blobs).expect("write diff");
        copy_tree_preserving(&parent, &replay).unwrap();
        apply_oci_diff_layer(&blobs.join(&layer.layer_digest), &replay).expect("apply diff");
        assert_eq!(
            collect_rootfs_entries(&replay).unwrap(),
            collect_rootfs_entries(&current).unwrap()
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn compact_rootfs_diff_round_trips_node_base_npm_paths() {
        let dir = unique_test_dir("sm-rootfs-diff-node-npm-path");
        let parent = dir.join("parent");
        let current = dir.join("current");
        let replay = dir.join("replay");
        let blobs = dir.join("blobs/sha256");
        let npm_dir = PathBuf::from(
            "usr/local/lib/node_modules/npm/node_modules/exponential-backoff/dist/delay/skip-first",
        );

        std::fs::create_dir_all(&parent).unwrap();
        std::fs::create_dir_all(current.join(&npm_dir)).unwrap();
        std::fs::write(
            current.join(&npm_dir).join("skip-first.js"),
            b"module.exports = true;\n",
        )
        .unwrap();

        let layer = write_rootfs_diff_layer(&parent, &current, &blobs).expect("write diff");
        copy_tree_preserving(&parent, &replay).unwrap();
        apply_oci_diff_layer(&blobs.join(&layer.layer_digest), &replay).expect("apply diff");
        assert_eq!(
            collect_rootfs_entries(&replay).unwrap(),
            collect_rootfs_entries(&current).unwrap()
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn stage_copy_tar_round_trips_long_context_paths() {
        let dir = unique_test_dir("sm-stage-copy-long-path");
        let host = dir.join("Preview/index.tsx");
        let entry = PathBuf::from(
            "app/.claude/worktrees/agent-a2c1a7a53930819d4/app/components/funnels/FunnelEditContent/Preview/index.tsx",
        );
        let replay = dir.join("replay");
        assert!(
            entry.as_os_str().len() > 100,
            "test must exceed the raw tar header path field"
        );

        std::fs::create_dir_all(host.parent().unwrap()).unwrap();
        std::fs::write(&host, b"export const Preview = true;\n").unwrap();

        let mut ar = tar::Builder::new(Vec::new());
        append_host_file_to_tar(&mut ar, &host, &entry).expect("append long COPY path");
        let tarball = ar.into_inner().unwrap();
        std::fs::create_dir_all(&replay).unwrap();
        tar::Archive::new(&tarball[..]).unpack(&replay).unwrap();
        assert_eq!(
            std::fs::read(replay.join(&entry)).unwrap(),
            b"export const Preview = true;\n"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn compact_layer_materializes_from_parent_and_prunes_only_rootfs() {
        let dir = unique_test_dir("sm-compact-layer-materialize");
        let cache = dir.join("cache");
        let base = cache.join("base");
        let layer = cache.join("layers/l1");
        let base_rootfs = base.join("rootfs");
        let layer_rootfs = layer.join("rootfs");
        std::fs::create_dir_all(&base_rootfs).unwrap();
        std::fs::write(base_rootfs.join("a.txt"), b"base").unwrap();
        std::fs::write(
            base.join("metadata.json"),
            serde_json::json!({
                "backend": "kboxlike",
                "kboxlike_rootfs": "rootfs",
                "memory_mib": 512,
                "vcpus": 1
            })
            .to_string(),
        )
        .unwrap();

        std::fs::create_dir_all(&layer_rootfs).unwrap();
        std::fs::write(layer_rootfs.join("a.txt"), b"changed").unwrap();
        std::fs::write(layer_rootfs.join("b.txt"), b"added").unwrap();
        std::fs::write(
            layer.join("metadata.json"),
            serde_json::json!({
                "backend": "kboxlike",
                "kboxlike_rootfs": "rootfs",
                "memory_mib": 512,
                "vcpus": 1
            })
            .to_string(),
        )
        .unwrap();
        std::fs::write(layer.join("restore.snap"), b"kboxlike-rootfs\n").unwrap();

        let expected = collect_rootfs_entries(&layer_rootfs).unwrap();
        commit_compact_layer(&layer, &base, &cache).expect("commit compact layer");
        assert!(compact_layer_manifest_path(&layer).is_file());

        prune_layer_materialization_or_dir(&layer).expect("prune materialized rootfs");
        assert!(!layer_rootfs.exists());
        assert!(compact_layer_manifest_path(&layer).is_file());
        assert!(layer.join("metadata.json").is_file());

        materialize_compact_layer_if_needed(&layer).expect("materialize compact layer");
        assert_eq!(collect_rootfs_entries(&layer_rootfs).unwrap(), expected);
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn noop_compact_layer_repoints_metadata_and_materializes_parent_rootfs() {
        let dir = unique_test_dir("sm-noop-compact-layer");
        let base = dir.join("base");
        let layer = dir.join("layers/l1");
        std::fs::create_dir_all(base.join("rootfs")).unwrap();
        std::fs::write(base.join("rootfs/base.txt"), b"base").unwrap();
        std::fs::write(
            base.join("metadata.json"),
            serde_json::json!({
                "backend": "kboxlike",
                "kboxlike_rootfs": "rootfs",
                "memory_mib": 512,
                "vcpus": 1
            })
            .to_string(),
        )
        .unwrap();
        std::fs::create_dir_all(&layer).unwrap();
        std::fs::write(
            layer.join("metadata.json"),
            serde_json::json!({
                "backend": "kboxlike",
                "kboxlike_rootfs": base.join("rootfs").to_string_lossy(),
                "memory_mib": 512,
                "vcpus": 1
            })
            .to_string(),
        )
        .unwrap();
        std::fs::write(layer.join("restore.snap"), b"kboxlike-rootfs\n").unwrap();

        commit_noop_compact_layer(&layer, &base).expect("commit no-op compact layer");
        assert_eq!(
            snapshot_rootfs_dir(&layer).unwrap().unwrap(),
            layer.join("rootfs")
        );
        materialize_compact_layer_if_needed(&layer).expect("materialize no-op compact layer");
        assert_eq!(
            collect_rootfs_entries(&layer.join("rootfs")).unwrap(),
            collect_rootfs_entries(&base.join("rootfs")).unwrap()
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    fn unique_test_dir(prefix: &str) -> PathBuf {
        std::env::temp_dir().join(format!(
            "{prefix}-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos())
                .unwrap_or(0)
        ))
    }

    #[test]
    fn volume_and_stopsignal_collected_into_config() {
        let mut state = BuildState::default();
        let mut config = ImageConfig::default();
        apply(
            &Instruction::Volume(vec!["/data".into(), "/cache".into()]),
            &mut state,
            &mut config,
        );
        apply(
            &Instruction::Volume(vec!["/more".into()]),
            &mut state,
            &mut config,
        );
        apply(
            &Instruction::StopSignal("SIGQUIT".into()),
            &mut state,
            &mut config,
        );
        assert_eq!(config.volumes, vec!["/data", "/cache", "/more"]);
        assert_eq!(config.stop_signal.as_deref(), Some("SIGQUIT"));
    }

    #[test]
    fn healthcheck_and_onbuild_collected_into_config() {
        use crate::builder::dockerfile::Healthcheck;
        let mut state = BuildState::default();
        let mut config = ImageConfig::default();
        let hc = Healthcheck::Cmd {
            test: ShellOrExec::Shell("curl -f http://localhost/ || exit 1".into()),
            interval: Some("30s".into()),
            timeout: Some("3s".into()),
            start_period: None,
            start_interval: None,
            retries: Some("3".into()),
        };
        apply(
            &Instruction::Healthcheck(hc.clone()),
            &mut state,
            &mut config,
        );
        apply(
            &Instruction::Onbuild("COPY . /app/src".into()),
            &mut state,
            &mut config,
        );
        apply(
            &Instruction::Onbuild("RUN make".into()),
            &mut state,
            &mut config,
        );
        // A later HEALTHCHECK replaces an earlier one (Docker semantics).
        apply(
            &Instruction::Healthcheck(Healthcheck::None),
            &mut state,
            &mut config,
        );
        assert_eq!(config.healthcheck, Some(Healthcheck::None));
        assert_eq!(config.onbuild, vec!["COPY . /app/src", "RUN make"]);
    }

    #[test]
    fn persist_run_config_records_stopsignal_healthcheck_onbuild() {
        use crate::builder::dockerfile::Healthcheck;
        let dir = std::env::temp_dir().join(format!(
            "sm-persistcfg-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos())
                .unwrap_or(0)
        ));
        std::fs::create_dir_all(&dir).unwrap();
        // A baseline metadata.json the builder patches in place.
        std::fs::write(dir.join("metadata.json"), "{}").unwrap();
        std::fs::write(dir.join("restore.snap"), b"x").unwrap();

        let config = ImageConfig {
            stop_signal: Some("SIGQUIT".into()),
            healthcheck: Some(Healthcheck::Cmd {
                test: ShellOrExec::Shell("curl -f http://localhost/ || exit 1".into()),
                interval: Some("30s".into()),
                timeout: Some("3s".into()),
                start_period: None,
                start_interval: None,
                retries: Some("3".into()),
            }),
            onbuild: vec!["RUN make".into()],
            ..Default::default()
        };
        // snapshot_file = the metadata.json's sibling restore.snap path.
        persist_run_config(&dir.join("restore.snap"), &config).expect("persist");

        let meta: serde_json::Value =
            serde_json::from_slice(&std::fs::read(dir.join("metadata.json")).unwrap()).unwrap();
        assert_eq!(
            meta.get("stop_signal").and_then(|v| v.as_str()),
            Some("SIGQUIT")
        );
        let hc = meta.get("healthcheck").expect("healthcheck recorded");
        let test = hc
            .get("Test")
            .and_then(|v| v.as_array())
            .expect("Test array");
        assert_eq!(test[0].as_str(), Some("CMD-SHELL"));
        assert!(test[1].as_str().unwrap().contains("curl -f"));
        assert_eq!(hc.get("Interval").and_then(|v| v.as_str()), Some("30s"));
        assert_eq!(hc.get("Retries").and_then(|v| v.as_str()), Some("3"));
        let ob = meta
            .get("onbuild")
            .and_then(|v| v.as_array())
            .expect("onbuild array");
        assert_eq!(ob[0].as_str(), Some("RUN make"));

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

    /// The longest-cached-prefix predicate recognizes BOTH a KVM/HVF RAM-dump
    /// snapshot (`restore.snap` + `metadata.json`) AND a no-virt sentry layer
    /// (`metadata.json` with `backend=="sentry"` + a sibling `rootfs/` dir),
    /// while rejecting incomplete dirs — so the sentry arm is backend-neutral and
    /// cannot shift KVM cache behavior nor false-positive on a half-written layer.
    #[test]
    fn is_snapshot_dir_recognizes_both_backends() {
        let root = std::env::temp_dir().join(format!(
            "sm-issnap-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos())
                .unwrap_or(0)
        ));
        std::fs::create_dir_all(&root).unwrap();

        // Arm 1 — KVM/HVF full dir: restore.snap + metadata.json → true.
        let kvm = root.join("kvm");
        std::fs::create_dir_all(&kvm).unwrap();
        std::fs::write(kvm.join("restore.snap"), b"SMSNAP04....").unwrap();
        std::fs::write(kvm.join("metadata.json"), b"{\"backend\":\"kvm\"}").unwrap();
        assert!(is_snapshot_dir(&kvm), "KVM full dir is a complete snapshot");

        // Arm 2 — bare cold-boot dir: a rootfs/ but NO metadata.json → false (a
        // half-written KVM layer must never be mistaken for a complete one).
        let bare = root.join("bare");
        std::fs::create_dir_all(bare.join("rootfs")).unwrap();
        assert!(
            !is_snapshot_dir(&bare),
            "rootfs/ alone (no metadata) is incomplete"
        );

        // Arm 3 — sentry layer dir: metadata.json backend=sentry + rootfs/ → true.
        let sentry = root.join("sentry");
        std::fs::create_dir_all(sentry.join("rootfs")).unwrap();
        std::fs::write(sentry.join("metadata.json"), b"{\"backend\":\"sentry\"}").unwrap();
        assert!(
            is_snapshot_dir(&sentry),
            "sentry layer (backend=sentry + rootfs/) is complete"
        );

        // Arm 3b — defense: metadata present + rootfs/ but backend != sentry and no
        // restore.snap (a half-written KVM dir whose rootfs/ staged first) → false.
        let halfkvm = root.join("halfkvm");
        std::fs::create_dir_all(halfkvm.join("rootfs")).unwrap();
        std::fs::write(halfkvm.join("metadata.json"), b"{\"backend\":\"kvm\"}").unwrap();
        assert!(
            !is_snapshot_dir(&halfkvm),
            "KVM metadata + rootfs/ but no restore.snap is NOT a complete layer"
        );

        // Arm 4 — empty dir → false.
        let empty = root.join("empty");
        std::fs::create_dir_all(&empty).unwrap();
        assert!(!is_snapshot_dir(&empty), "empty dir is not a snapshot");

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