supermachine 0.7.69

//! 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, HashMap};
use std::path::{Path, PathBuf};
use std::time::Duration;

use super::dockerfile::{
    BaseImage, CopyFlags, Dockerfile, 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);

/// 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>,
}

/// The result of a build: the bootable image snapshot + its run-config.
pub struct BuildOutcome {
    pub image: Image,
    pub config: ImageConfig,
}

/// 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)
        .build()?;
    let vm = pool.acquire()?;
    await_agent_ready(&vm);

    let mut state = BuildState::default();
    let mut config = ImageConfig::default();
    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 })
}

/// 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()),
        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());
    Image::from_oci(&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,
}

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 image = Image::from_snapshot(
            stages[final_idx]
                .as_ref()
                .expect("final stage built")
                .image
                .snapshot_path(),
        )?;
        let config = configs[final_idx].take().expect("final stage config");
        persist_run_config(image.snapshot_path(), &config)?;
        Ok(BuildOutcome { image, config })
    }

    /// 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) = self.build_stage(stage, &base, context_dir, prior)?;
        let built = StageBuilt {
            name: stage.name.clone(),
            key,
            image: Image::from_snapshot(image.snapshot_path())?,
        };
        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),
                None => Image::from_oci(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) = self.build_stage(stage, base, context_dir, &[])?;
        persist_run_config(image.snapshot_path(), &config)?;
        Ok(BuildOutcome { image, config })
    }

    /// 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), 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) {
                resume = i + 1;
                resume_image = Some(Image::from_snapshot(&dir)?);
            } else {
                break;
            }
        }

        // 3. Replay state + config for the cached prefix (no VM).
        let mut state = BuildState::default();
        let mut config = ImageConfig::default();
        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);

        // 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));
        }

        // 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)
            .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 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) {
                final_image = Some(Image::from_snapshot(&dir)?);
            } else 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");
                final_image = Some(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)?
                });
            } 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)?;
                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);
            }
            prev_snapshot_dir = dir;
        }

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

/// 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 snapshot dir (matches what `vm.snapshot()` writes).
fn is_snapshot_dir(dir: &Path) -> bool {
    dir.join("restore.snap").is_file() && dir.join("metadata.json").is_file()
}

/// 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,
    }
}

/// Patch a built snapshot's `metadata.json` with the Dockerfile's run-config
/// (effective CMD, WORKDIR, USER, ENV), 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()
    {
        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()));
            }
        }
    }

    let out = serde_json::to_string_pretty(&meta)
        .map_err(|e| Error::vm_msg(format!("serialize metadata.json: {e}")))?;
    std::fs::write(&meta_path, out).map_err(Error::Io)?;
    Ok(())
}

/// 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)
        .build()?;
    let vm = pool.acquire()?;
    await_agent_ready(&vm);
    let argv: Vec<&str> = vec![
        "tar",
        "-cf",
        "-",
        "--exclude=./proc",
        "--exclude=./sys",
        "--exclude=./dev",
        "--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)
}

/// 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> {
    use std::io::Write as _;

    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 tar_bytes = boot_and_tar_rootfs(image)?;
    let diff_id = sha256_hex(&tar_bytes);
    let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
    gz.write_all(&tar_bytes).map_err(Error::Io)?;
    let layer_gz = gz.finish().map_err(Error::Io)?;
    let layer_digest = sha256_hex(&layer_gz);
    let layer_size = layer_gz.len() as u64;
    write_blob(&layer_digest, &layer_gz)?;

    // 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(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: tar_bytes.len() as u64,
    })
}

/// 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::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)
            .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 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() {
        // Empty/absent cache dir is not an error — nothing to persist.
        return Ok(());
    }
    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('/');
        ar.append_path_with_name(host, entry).map_err(Error::Io)?;
    }
    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(())
}

/// 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 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"));
    }
}