//! VM runner — the reusable library entry point behind the
//! `supermachine-worker` command-line harness. Owns VM boot/restore,
//! virtio device wiring, vCPU dispatch, snapshot triggers, and
//! pool-worker warm-restore loops.
//!
//! VM-execution module — gated to a platform with a hypervisor backend
//! (macOS/HVF today; broaden to include KVM when the Linux backend lands).
#![cfg(all(target_os = "macos", target_arch = "aarch64"))]

use std::fmt;
use std::os::unix::net::UnixStream;
use std::time::SystemTime;

use crate::vmm::pool::PoolWorker;
use crate::vmm::resources::{ResourceError, VmResources};
use crate::vmm::tls::TlsConfig;

#[derive(Default)]
pub struct RunOptions {
    pub tls: Option<TlsConfig>,
    pub pool_sock: Option<UnixStream>,
    pub pool_worker: Option<PoolWorker>,
    /// Experimental warm-pool shortcut used to prove whether full HVF GIC blob
    /// restore is required for a snapshot class.
    pub experimental_skip_warm_gic_restore: bool,
    /// 0.7.43+: explicit "this is a bake-then-pool worker" flag,
    /// set by `bin/worker.rs::main` based on the CLI inputs it
    /// observed. The runner can't infer this reliably from
    /// resources alone because the kcache-HIT path sets
    /// `resources.restore_from = Some(<cache>)` but still wants
    /// bake-then-pool semantics (BAKE_READY emit, no initial
    /// DONE write); the regular pool-worker spawn ALSO has
    /// `resources.restore_from = Some(...)` but wants DONE.
    /// `bin/worker.rs` distinguishes via the CLI flag
    /// (`--kcache-restore-from` vs `--restore-from`) and passes
    /// the resolved bool here.
    pub bake_then_pool: bool,
    /// 0.7.44+ when true, `runner::run` ends the process via
    /// `libc::_exit(0)` after draining in-flight saves, bypassing
    /// Rust's natural unwind (which would drop `MicroVm` and call
    /// `hv_vcpu_destroy` on the live HVF state while secondary
    /// vCPU + muxer-io + acceptor threads are still running →
    /// SIGSEGV under FORKS=15 load, surfaces as
    /// `supervisor closed mid-protocol` in the bake driver).
    ///
    /// Set by the standalone `supermachine-worker` binary, which
    /// always exits after one VM lifecycle. The embedded napi
    /// path (which keeps the lib process alive across many VMs)
    /// leaves this false → normal unwind so subsequent MicroVms
    /// can be created.
    ///
    /// The OS reaps memory, sockets, mmaps, and HVF state when
    /// the process dies, so skipping Rust drops + C++ static
    /// destructors here is correctness-preserving.
    pub exit_after_run: bool,
    /// 0.7.56+: optional PL011 output sink, applied to the built VM's
    /// `Vmm::serial` before the vCPU runs. Carries the file opened by
    /// `bin/worker.rs` from `--log-sink FILE` (kept as `Option<File>`
    /// rather than a path so the binary owns the CLI error reporting).
    /// `None` (the `Default`) leaves output on stdout — matching the
    /// in-process / embedded path, which never set the old process
    /// global.
    pub log_sink: Option<std::fs::File>,
    /// 0.7.56+: enable the PL011 boot-line marker scanner (heartbeat /
    /// parked / pre-exec / kcache / smpark-gpa) on the built VM. Set by
    /// `bin/worker.rs` for any snapshot-trigger mode; `false` (the
    /// `Default`) on the embedded path, exactly reproducing the old
    /// process-global default that only `bin/worker.rs` ever flipped.
    pub heartbeat_detection: bool,
    /// 0.7.56+: spawn the kcache-HIT resume thread after the VM is
    /// built — it re-pushes the 'R\n' resume token until init-oci acks
    /// (`serial.kcache_resumed`). Set by `bin/worker.rs` from
    /// `SUPERMACHINE_KCACHE_RESUME`. Was a thread spawned in
    /// `bin/worker.rs::main` against process-global serial state before
    /// 0.7.56; moved here so it can capture this VM's `Vmm::serial`.
    pub kcache_resume: bool,
    /// 0.7.56+: spawn the pre-exec-sync resume thread after the VM is
    /// built — re-pushes 'P\n' until init-oci acks
    /// (`serial.pre_exec_sync_resumed`). Set by `bin/worker.rs` from
    /// `SUPERMACHINE_PRE_EXEC_SYNC_RESUME`. Same migration as
    /// `kcache_resume`.
    pub pre_exec_sync_resume: bool,
}

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct RunReport {
    pub warm_restores: u64,
}

#[cfg(all(target_os = "macos", target_arch = "aarch64"))]
struct WarmSnapshotCache {
    path: String,
    file_len: u64,
    modified: Option<SystemTime>,
    file: std::fs::File,
    snap: crate::vmm::snapshot::Snapshot,
    ram_offset: u64,
    memory_bytes: usize,
}

#[derive(Debug)]
pub enum RunError {
    Build(crate::vmm::builder::BuildError),
    Hvf(crate::hypervisor::ActiveError),
    MmapCow {
        path: String,
        source: std::io::Error,
    },
    Pool(crate::vmm::pool::PoolError),
    Resource(ResourceError),
    SnapshotLoad {
        path: String,
        source: crate::vmm::snapshot::FileError,
    },
    ThreadSpawn {
        name: String,
        source: std::io::Error,
    },
    Tls(crate::vmm::tls::StartError),
    UnexpectedProofOfLifeExit {
        reason: crate::hypervisor::VcpuExit,
        ec: u64,
    },
    VsockMux(crate::vmm::vsock_mux::StartError),
    Worker(crate::vmm::worker::WorkerError),
}

impl fmt::Display for RunError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            RunError::Build(e) => write!(f, "{e}"),
            RunError::Hvf(e) => write!(f, "HVF operation failed: {e:?}"),
            RunError::MmapCow { path, source } => {
                write!(f, "mmap CoW snapshot RAM {path}: {source}")
            }
            RunError::Pool(e) => write!(f, "{e}"),
            RunError::Resource(e) => write!(f, "{e}"),
            RunError::SnapshotLoad { path, source } => {
                write!(f, "load snapshot {path}: {source:?}")
            }
            RunError::ThreadSpawn { name, source } => {
                write!(f, "spawn thread {name}: {source}")
            }
            RunError::Tls(e) => write!(f, "{e}"),
            RunError::UnexpectedProofOfLifeExit { reason, ec } => {
                write!(
                    f,
                    "unexpected proof-of-life exit: {reason:?} ESR_EL2 EC={ec:#x}"
                )
            }
            RunError::VsockMux(e) => write!(f, "{e}"),
            RunError::Worker(e) => write!(f, "{e}"),
        }
    }
}

impl std::error::Error for RunError {}

impl From<ResourceError> for RunError {
    fn from(value: ResourceError) -> Self {
        Self::Resource(value)
    }
}

impl From<crate::vmm::builder::BuildError> for RunError {
    fn from(value: crate::vmm::builder::BuildError) -> Self {
        Self::Build(value)
    }
}

impl From<crate::hypervisor::ActiveError> for RunError {
    fn from(value: crate::hypervisor::ActiveError) -> Self {
        Self::Hvf(value)
    }
}

impl From<crate::vmm::pool::PoolError> for RunError {
    fn from(value: crate::vmm::pool::PoolError) -> Self {
        Self::Pool(value)
    }
}

impl From<crate::vmm::worker::WorkerError> for RunError {
    fn from(value: crate::vmm::worker::WorkerError) -> Self {
        Self::Worker(value)
    }
}

impl From<crate::vmm::vsock_mux::StartError> for RunError {
    fn from(value: crate::vmm::vsock_mux::StartError) -> Self {
        Self::VsockMux(value)
    }
}

impl From<crate::vmm::tls::StartError> for RunError {
    fn from(value: crate::vmm::tls::StartError) -> Self {
        Self::Tls(value)
    }
}

#[cfg(all(target_os = "macos", target_arch = "aarch64"))]
pub fn run(resources: &VmResources, options: RunOptions) -> Result<RunReport, RunError> {
    // Bake-then-pool relaxes the "snapshot trigger requires out_path"
    // rule. The runner doesn't actually capture in that mode — it
    // signals BAKE_READY on the supervisor socket and hands the
    // SNAPSHOT_ASYNC / SNAPSHOT RPCs to the host instead.
    //
    // 0.7.43+: the bake-then-pool decision is made by
    // `bin/worker.rs::main` (which sees the CLI flags) and
    // propagated here via `options.bake_then_pool`. The runner
    // cannot infer reliably from resources alone — the
    // kcache-HIT path has `restore_from = Some(<cache>)` AND
    // wants bake-then-pool; the regular pool-worker spawn ALSO
    // has `restore_from = Some(...)` but wants DONE.
    let bake_then_pool = options.bake_then_pool;
    if !bake_then_pool {
        resources.validate_for_run()?;
    } else if resources.memory_bytes() == 0 {
        return Err(RunError::Resource(
            crate::vmm::resources::ResourceError::ZeroMemory,
        ));
    } else if resources.vcpus == 0 {
        return Err(RunError::Resource(
            crate::vmm::resources::ResourceError::ZeroVcpus,
        ));
    } else if resources.kernel_path.is_none() && resources.restore_from.is_none() {
        // 0.7.43+ : bake-then-pool now permits restore_from
        // (kcache-HIT path + general pool spawn_one). Accept
        // either a kernel or a snapshot; reject if both absent.
        return Err(RunError::Resource(
            crate::vmm::resources::ResourceError::MissingKernel,
        ));
    }
    run_kernel(
        resources,
        options.tls,
        options.pool_sock,
        options.pool_worker,
        options.experimental_skip_warm_gic_restore,
        bake_then_pool,
        options.exit_after_run,
        options.log_sink,
        options.heartbeat_detection,
        options.kcache_resume,
        options.pre_exec_sync_resume,
    )
}

#[cfg(all(target_os = "macos", target_arch = "aarch64"))]
pub fn run_proof_of_life() -> Result<(), RunError> {
    use crate::arch::aarch64::layout;
    use crate::vmm::vstate::{MicroVm, TEST_PROGRAM};

    eprintln!("supermachine: HVF init test");
    let vm = MicroVm::new(64 * 1024 * 1024)?;
    eprintln!(
        "  VM created, RAM mapped at GPA 0x{:x}, {} MiB",
        vm.ram_gpa,
        vm.ram_size / (1024 * 1024)
    );

    let entry = vm.ram_gpa + layout::KERNEL_LOAD_OFFSET;
    // SAFETY: TEST_PROGRAM is 8 bytes, well within RAM.
    unsafe {
        vm.write_ram(entry, &TEST_PROGRAM);
    }
    vm.set_boot_cpsr()?;
    vm.set_pc(entry)?;
    eprintln!("  PC set to 0x{entry:x}, CPSR=EL1h (DAIF masked)");

    eprintln!("  running vCPU…");
    let exit = vm.run_once()?;
    // A guest HVC traps as VcpuExit::Exception; ESR_EL2 is the `syndrome`, whose
    // EC field (bits 31:26) is 0x16 for "HVC executed in AArch64".
    let esr = match exit {
        crate::hypervisor::VcpuExit::Exception { syndrome, .. } => syndrome,
        _ => 0,
    };
    let ec = (esr >> 26) & 0x3f;
    eprintln!("  exit: {exit:?}  ESR_EL2=0x{esr:x}  EC={ec:#x}");

    if matches!(exit, crate::hypervisor::VcpuExit::Exception { .. }) && ec == 0x16 {
        eprintln!("  PASS: HVF round-trip working — guest executed HVC #0");
        Ok(())
    } else {
        eprintln!("  UNEXPECTED exit; HVF probably misconfigured");
        Err(RunError::UnexpectedProofOfLifeExit { reason: exit, ec })
    }
}

#[cfg(all(target_os = "macos", target_arch = "aarch64"))]
// 0.7.43+: `bake_then_pool_from_caller` is the explicit signal
// from `bin/worker.rs::main` of whether this worker is operating
// in bake-then-pool mode. See `RunOptions::bake_then_pool` for
// rationale.
fn run_kernel(
    resources: &VmResources,
    tls_cfg: Option<TlsConfig>,
    pool_sock: Option<UnixStream>,
    pool_worker: Option<PoolWorker>,
    option_skip_warm_gic_restore: bool,
    bake_then_pool_from_caller: bool,
    exit_after_run: bool,
    // 0.7.56+ per-VM PL011 config (see RunOptions). Applied to
    // `Vmm::serial` after build, before the vCPU runs.
    log_sink: Option<std::fs::File>,
    heartbeat_detection: bool,
    kcache_resume: bool,
    pre_exec_sync_resume: bool,
) -> Result<RunReport, RunError> {
    use crate::vmm::builder;
    use crate::vmm::pool::{PoolControl, WarmRestoreTimings};
    use crate::vmm::snapshot;
    use crate::vmm::worker::{self, DispatchSnapshot};

    let kernel_path = resources.kernel_path.as_deref();
    let initrd_path = resources.initrd_path.as_deref();
    let cmdline = resources.cmdline.as_str();
    let mem_size = resources.memory_bytes();
    let blk_paths = &resources.block_devices;
    let n_vcpus = resources.vcpus;
    let snapshot_after_ms = resources.snapshot.after_ms;
    let snapshot_at = resources.snapshot.at_heartbeat;
    let snapshot_on_listener = resources.snapshot.on_listener;
    let snapshot_on_pre_exec = resources.snapshot.on_pre_exec;
    let quiesce_ms = resources.snapshot.quiesce_ms;
    let snapshot_out = resources.snapshot.out_path.as_deref();
    let balloon_target_pages = resources.balloon_target_pages;
    let restore_from = resources.restore_from.as_deref();
    let cow_restore = resources.cow_restore;
    let vsock_mux_path = resources.endpoints.vsock_mux.as_deref();
    let http_port_addr = resources.endpoints.http_port.as_deref();
    let vsock_mux_handoff_path = resources.endpoints.vsock_mux_handoff.as_deref();
    let vsock_exec_path = resources.endpoints.vsock_exec.as_deref();
    let vsock_exec_guest_port = resources
        .endpoints
        .vsock_exec_guest_port
        .unwrap_or(crate::vmm::resources::DEFAULT_EXEC_GUEST_PORT);
    let timings = crate::trace::enabled("timings");
    let skip_warm_gic_restore = option_skip_warm_gic_restore
        || std::env::var_os("SUPERMACHINE_SKIP_WARM_GIC_RESTORE").is_some();
    // Cycle-restore RAM remap uses the MAP_FIXED variant
    // (`remap_cow_from_file_fixed`): it re-maps the snapshot at the
    // SAME host VA instead of munmap + fresh-VA mmap. This is
    // correct-by-construction for device safety. Every virtio
    // `GuestMem` caches the RAM base pointer at construction (queue.rs)
    // and is NOT refreshed on cycle-restore; with the non-fixed path,
    // if the kernel handed back a different VA the muxer's queue
    // pointers would dangle (use-after-munmap → intermittent SIGSEGV /
    // silent corruption). MAP_FIXED pins the VA — asserted in the fixed
    // remap — so the cached pointers stay valid across every cycle. It
    // is also equal-or-faster (one mmap, no separate munmap), so this
    // is a pure reliability win with no perf cost. (Was toggled off as
    // a debug knob in 0.7.38; re-enabled in 0.7.53 after the restore
    // audit identified the dangling-pointer hazard.)
    let fixed_warm_ram_remap = true;
    let run_t0 = std::time::Instant::now();

    let mut cow_ram: Option<(*mut u8, usize)> = None;
    let restore = match restore_from {
        Some(p) => Some(if cow_restore {
            eprintln!("supermachine: restoring from {p} (CoW mmap)");
            let t0 = std::time::Instant::now();
            let (snap, ram_offset, memory_bytes) =
                snapshot::load_meta(p).map_err(|source| RunError::SnapshotLoad {
                    path: p.to_string(),
                    source,
                })?;
            if timings {
                eprintln!(
                    "[timing] restore.load_meta={}us total={}us",
                    t0.elapsed().as_micros(),
                    run_t0.elapsed().as_micros()
                );
            }
            let t0 = std::time::Instant::now();
            let (ptr, len) =
                snapshot::mmap_ram_cow_at(p, ram_offset, memory_bytes).map_err(|source| {
                    RunError::MmapCow {
                        path: p.to_string(),
                        source,
                    }
                })?;
            if timings {
                eprintln!(
                    "[timing] restore.mmap_cow={}us total={}us",
                    t0.elapsed().as_micros(),
                    run_t0.elapsed().as_micros()
                );
            }
            cow_ram = Some((ptr, len));
            snap
        } else {
            eprintln!("supermachine: restoring from {p}");
            let t0 = std::time::Instant::now();
            snapshot::load_from_file(p)
                .map_err(|source| RunError::SnapshotLoad {
                    path: p.to_string(),
                    source,
                })
                .inspect(|_| {
                    if timings {
                        eprintln!(
                            "[timing] restore.load_full={}us total={}us",
                            t0.elapsed().as_micros(),
                            run_t0.elapsed().as_micros()
                        );
                    }
                })?
        }),
        None => None,
    };

    if restore.is_none() {
        eprintln!("supermachine: kernel boot");
        eprintln!("  kernel    : {}", kernel_path.unwrap_or(""));
        if let Some(p) = initrd_path {
            eprintln!("  initramfs : {p}");
        }
        eprintln!("  cmdline   : {cmdline}");
        eprintln!("  memory    : {} MiB", mem_size / (1024 * 1024));
        for p in blk_paths {
            eprintln!("  blk       : {p}");
        }
    } else if let Some(s) = restore.as_ref() {
        eprintln!(
            "  memory    : {} MiB (from snapshot)",
            s.memory.len() / (1024 * 1024)
        );
    }

    let restore_memory_len = restore.as_ref().map(|s| s.memory.len());
    let t0 = std::time::Instant::now();
    let mut vmm = builder::build_vmm(resources, cow_ram, restore_memory_len)?;
    if timings {
        eprintln!(
            "[timing] restore.build_vmm={}us total={}us",
            t0.elapsed().as_micros(),
            run_t0.elapsed().as_micros()
        );
    }

    // 0.7.56+: apply per-VM PL011 config carried in RunOptions. These
    // used to be process-global side effects set by `bin/worker.rs`
    // BEFORE the VM existed; now the serial state is per-VM (built
    // above), so we apply them here, after build and before the vCPU
    // runs (no guest output happens until dispatch). The embedded /
    // in-process path leaves these at their `Default` (None / false),
    // exactly reproducing the old globals it never set.
    if let Some(sink) = log_sink {
        vmm.serial.set_log_sink_file(sink);
    }
    vmm.serial.set_heartbeat_detection(heartbeat_detection);

    // 0.7.56+: kcache-HIT / pre-exec-sync resume threads. Spawned here
    // (capturing THIS VM's `serial`) instead of in `bin/worker.rs::main`
    // — they re-push the unblock token to init-oci, which is paused
    // inside `read(0)` in the restored snapshot, and stop once init-oci
    // writes its "resumed" marker. Detached helpers (≤3 s); the Arc
    // keeps `serial` alive for their lifetime. Gated on the RunOptions
    // flags `bin/worker.rs` derives from the resume env vars, so the
    // embedded path (flags false) never spawns them — same as before.
    if kcache_resume {
        let serial = vmm.serial.clone();
        std::thread::Builder::new()
            .name("supermachine-kcache-resume".into())
            .stack_size(64 * 1024)
            .spawn(move || {
                use std::sync::atomic::Ordering as O;
                // Let the runner reach restore + GIC + vCPU launch.
                std::thread::sleep(std::time::Duration::from_millis(50));
                // Retry every 100 ms, up to 30 attempts (3 s); stop as
                // soon as init-oci writes the resumed marker so we don't
                // pile extra bytes into the post-pivot workload's TTY.
                for _ in 0..30 {
                    if serial.kcache_resumed.load(O::SeqCst) {
                        break;
                    }
                    serial.push_rx_byte(b'R');
                    serial.push_rx_byte(b'\n');
                    std::thread::sleep(std::time::Duration::from_millis(100));
                }
            })
            .ok();
    }
    if pre_exec_sync_resume {
        let serial = vmm.serial.clone();
        std::thread::Builder::new()
            .name("supermachine-pre-exec-resume".into())
            .stack_size(64 * 1024)
            .spawn(move || {
                use std::sync::atomic::Ordering as O;
                std::thread::sleep(std::time::Duration::from_millis(50));
                for _ in 0..30 {
                    if serial.pre_exec_sync_resumed.load(O::SeqCst) {
                        break;
                    }
                    serial.push_rx_byte(b'P');
                    serial.push_rx_byte(b'\n');
                    std::thread::sleep(std::time::Duration::from_millis(100));
                }
            })
            .ok();
    }

    // Bind host-facing endpoints BEFORE restore. The lib's
    // `wait_for_socket` (and the router's worker-readiness
    // probe) detects the unix socket existing as the "spawn
    // done" signal — moving the bind earlier means clients
    // stop polling sooner and pay less spawn-roundtrip cost.
    //
    // Safety: each accept handler calls `wait_for_host_port`,
    // which polls the muxer's TSI listener registry until the
    // guest's listener registers (post-restore). So if a
    // client connects before restore completes, the kernel
    // queues the connection, accept fires, the handler blocks
    // in `wait_for_host_port` until the guest is up, then
    // proceeds. No data is misrouted.
    if let Some(c) = tls_cfg {
        crate::vmm::tls::start(c, vmm.vsock.clone())?;
    }
    if let Some(p) = vsock_mux_path {
        crate::vmm::vsock_mux::start(p, vmm.vsock.clone(), None)?;
    }
    if let Some(addr) = http_port_addr {
        crate::vmm::vsock_mux::start_tcp(addr, vmm.vsock.clone(), None)?;
    }
    if let Some(p) = vsock_mux_handoff_path {
        crate::vmm::vsock_mux::start_handoff(p, vmm.vsock.clone(), None)?;
    }
    if let Some(p) = vsock_exec_path {
        crate::vmm::vsock_mux::start_exec(p, vmm.vsock.clone(), vsock_exec_guest_port)?;
    }
    if timings {
        eprintln!(
            "[timing] restore.endpoints_ready={}us",
            run_t0.elapsed().as_micros()
        );
    }

    // Compute per-secondary `PerVcpuState` BEFORE we spawn so we
    // can hand each secondary its target state at thread-create
    // time. Used by both branches below.
    let secondary_states: Vec<Option<crate::vmm::snapshot::PerVcpuState>> = restore
        .as_ref()
        .map(|s| {
            (1..n_vcpus as usize)
                .map(|i| s.per_vcpu.get(i).cloned())
                .collect()
        })
        .unwrap_or_else(|| (1..n_vcpus).map(|_| None).collect());

    // Cold-restore with multi-vCPU: SPAWN SECONDARIES FIRST, BEFORE
    // we call `vmm.restore_snapshot` (which internally invokes
    // `hv_gic_set_state` on the opaque GIC blob). Apple's
    // `hv_gic_set_state` applies state for every PE described in
    // the blob; if a PE's `hv_vcpu_create` hasn't happened yet the
    // state for that slot is silently dropped (HVF doesn't buffer
    // for not-yet-created vCPUs). When that PE is later created
    // it comes up in HVF's default per-PE GIC state — WAKER
    // ProcessorSleep=1, CTLR=0, IGRPMODR0=0 — which we cannot
    // overwrite because HVF doesn't expose those registers via
    // per-register accessors. The kernel's view of that PE looks
    // online (it onlined during the original bake before snapshot),
    // but the redistributor refuses to route SGIs/PPIs to it. IPIs
    // from `wake_up()` calls on other CPUs land on a redistributor
    // in ProcessorSleep state and are lost; any task in WFI on
    // that vCPU stays asleep forever.
    //
    // Symptom: post-restore amd64 processes spawned through
    // rosettad's UNIX-DGRAM protocol intermittently deadlock with
    // both the wrapper and rosettad sleeping in
    // `__skb_wait_for_more_packets`. ~13 % failure rate at 4
    // vCPUs, 0 % at 1 vCPU (no secondaries, no missed PE).
    //
    // Fix: spawn each secondary thread → it calls Vcpu::new()
    // (which is `hv_vcpu_create`) + `register_secondary` → boot
    // thread waits for all N-1 handles to be registered → calls
    // `vmm.restore_snapshot` (the GIC blob now lands on every PE)
    // → sets `coord.gic_restore_done = true` → secondaries spin
    // out of their wait in `maybe_apply_restore`'s sibling barrier
    // (`run_secondary_inner`'s gic-done check) and apply their
    // per-PE register state on top of a coherent redistributor.
    //
    // Cold-boot (no restore) is unaffected: no GIC blob is
    // installed, the kernel will set up each PE's GIC state
    // during its own SMP bring-up via PSCI CPU_ON.
    for idx in 1..n_vcpus {
        let coord_c = vmm.coord.clone();
        let bus_c = vmm.bus.clone();
        // Each secondary needs its own clone of the DaxSession Arcs so
        // it can route stage-2 faults into `handle_stage2_fault`
        // independently — same shape as the boot vCPU's
        // `dispatch_vcpu(&vmm.dax_sessions)` call below.
        let dax_c = vmm.dax_sessions.clone();
        let name = format!("vcpu-{idx}");
        let st = secondary_states
            .get((idx - 1) as usize)
            .cloned()
            .unwrap_or(None);
        std::thread::Builder::new()
            .name(name.clone())
            // 0.7.43+: cap vCPU thread stack at 256 KiB (default
            // is 2 MiB on macOS). The HVF run loop is a tight
            // ioctl/exit/dispatch cycle with no deep recursion; the
            // 8-vCPU upper bound × stack-virtual = 16 MiB → 2 MiB
            // saved across all vCPUs is mostly virtual address
            // space, but on a worker with many ephemeral spawn
            // paths it also drops resident pages on the warm
            // stack-pages by a few KiB per thread. Mirrors the
            // size we already use in `bin/router.rs`.
            .stack_size(256 * 1024)
            .spawn(move || worker::run_secondary(idx, coord_c, bus_c, dax_c, st))
            .map_err(|source| RunError::ThreadSpawn { name, source })?;
    }

    let mut first_restore_us: u128 = 0;
    if let Some(snap) = restore.as_ref() {
        // For multi-vCPU restore: wait for every secondary thread
        // to have created its HVF vCPU and registered with coord
        // before we install the GIC blob. See block-comment above
        // for the full rationale (HVF drops blob state for
        // not-yet-existing PEs).
        if n_vcpus > 1 {
            let target = (n_vcpus - 1) as usize;
            let t_barrier = std::time::Instant::now();
            let deadline = t_barrier + std::time::Duration::from_secs(5);
            loop {
                let registered = vmm.coord.secondary_handles.lock().unwrap().len();
                if registered >= target {
                    break;
                }
                if std::time::Instant::now() > deadline {
                    eprintln!(
                        "supermachine: warning: only {registered}/{target} secondaries \
                         registered after 5s; proceeding with hv_gic_set_state but \
                         missing PEs will come up with HVF default GIC state — IRQ \
                         delivery on those vCPUs is likely to stall."
                    );
                    break;
                }
                std::hint::spin_loop();
            }
            if timings {
                eprintln!(
                    "[timing] restore.secondary_barrier={}us",
                    t_barrier.elapsed().as_micros()
                );
            }
        }
        let t0 = std::time::Instant::now();
        vmm.restore_snapshot(snap)?;
        first_restore_us = t0.elapsed().as_micros();
        // Release secondaries — they're spinning in their
        // `gic_restore_done` wait inside `run_secondary_inner`
        // and need this signal to proceed with their per-PE
        // register state restore on top of the now-coherent
        // redistributor blob.
        vmm.coord
            .gic_restore_done
            .store(true, std::sync::atomic::Ordering::Release);
        if timings {
            eprintln!(
                "[timing] restore.state={}us total={}us",
                first_restore_us,
                run_t0.elapsed().as_micros()
            );
        }
        eprintln!(
            "  restored in {first_restore_us} us  (devices={} listeners={})",
            snap.virtio.devices.len(),
            snap.virtio.vsock_listeners.len()
        );

        // Hydrate per-mount PosixFs inode tables from the sidecar
        // (0.7.6+). Without this, every restore starts with an empty
        // FUSE `nodeid → host_path` table — the guest's pre-snapshot
        // dentry cache then references nodeids the daemon doesn't
        // know about, surfacing as MODULE_NOT_FOUND / EISDIR on any
        // path not walked during the warmup callback.
        if let Some(p) = restore_from {
            if !vmm.posix_fs.is_empty() {
                let t0 = std::time::Instant::now();
                let sidecar_path = snapshot::posix_fs_sidecar_path(p);
                match snapshot::read_posix_fs_sidecar(&sidecar_path) {
                    Ok(Some(blobs)) => {
                        if blobs.len() != vmm.posix_fs.len() {
                            eprintln!(
                                "supermachine: warning: posix-fs sidecar has {} entries but VM has {} mounts; \
                                 hydrating only the matching prefix (warm-restore dentry behaviour will be \
                                 degraded for any mount index past the prefix)",
                                blobs.len(),
                                vmm.posix_fs.len()
                            );
                        }
                        let n = blobs.len().min(vmm.posix_fs.len());
                        let mut ok = 0usize;
                        for i in 0..n {
                            if blobs[i].is_empty() {
                                continue;
                            }
                            match vmm.posix_fs[i].restore_state(&blobs[i]) {
                                Ok(()) => ok += 1,
                                Err(e) => {
                                    eprintln!(
                                        "supermachine: warning: posix-fs sidecar restore failed for \
                                         mount {i}: {e} — warm restore will fall back to lazy LOOKUP \
                                         (paths not walked during warmup may return ENOENT)"
                                    );
                                }
                            }
                        }
                        if timings {
                            eprintln!(
                                "[timing] restore.posix_fs_sidecar={}us mounts_restored={}/{}",
                                t0.elapsed().as_micros(),
                                ok,
                                vmm.posix_fs.len()
                            );
                        }
                    }
                    Ok(None) => {
                        // No sidecar — this is a pre-0.7.6 snapshot
                        // (or a CoW path that didn't write one). The
                        // existing warning emitted when bake version
                        // != runtime version already covers the
                        // user-facing diagnosis; stay quiet here so
                        // 0.7.5 → 0.7.6 upgrades don't spam.
                    }
                    Err(e) => {
                        eprintln!(
                            "supermachine: warning: posix-fs sidecar {sidecar_path} could not be \
                             read ({e:?}) — warm restore will fall back to lazy LOOKUP"
                        );
                    }
                }
            }
            // DAX sidecar — hydrate each DaxSession's slot table
            // with metadata-only entries (host_va = None). The first
            // guest stage-2 fault into each slot triggers
            // `handle_stage2_fault`, which calls backend.dax_map +
            // mapper.map to re-establish the host mmap + HVF stage-2
            // mapping on demand. Without this hook, a cycle-restored
            // VM SIGBUSes on the first guest access to a DAX page
            // that wasn't in the working set pre-snapshot (Prisma's
            // tokio worker on cycle-restored Remix server, field repro).
            if !vmm.dax_sessions.is_empty() {
                let t0 = std::time::Instant::now();
                let sidecar_path = snapshot::dax_sidecar_path(p);
                match snapshot::read_dax_sidecar(&sidecar_path) {
                    Ok(Some(blobs)) => {
                        let n = blobs.len().min(vmm.dax_sessions.len());
                        let mut ok = 0usize;
                        for i in 0..n {
                            if blobs[i].is_empty() {
                                continue;
                            }
                            match vmm.dax_sessions[i].restore_state(&blobs[i]) {
                                Ok(()) => ok += 1,
                                Err(e) => eprintln!(
                                    "supermachine: warning: dax sidecar restore failed for \
                                     mount {i}: {e} — DAX-mapped pages from pre-snapshot will \
                                     SIGBUS on first access"
                                ),
                            }
                        }
                        if timings {
                            eprintln!(
                                "[timing] restore.dax_sidecar={}us sessions_restored={}/{}",
                                t0.elapsed().as_micros(),
                                ok,
                                vmm.dax_sessions.len()
                            );
                        }
                    }
                    Ok(None) => {
                        // No sidecar — pre-fix snapshot. Same behaviour
                        // as before this fix landed: any DAX page not
                        // paged-in pre-snapshot will SIGBUS on first
                        // access. Stay quiet — the user-facing
                        // diagnostic (bake version mismatch) already
                        // covers the upgrade-needed case.
                    }
                    Err(e) => eprintln!(
                        "supermachine: warning: dax sidecar {sidecar_path} could not be \
                         read ({e:?}) — DAX-mapped pages from pre-snapshot will SIGBUS \
                         on first access"
                    ),
                }
            }
        }
    }

    // 0.7.44+ DEFAULT-OFF balloon-at-restore.
    //
    // The previous behavior inflated to 70% at every restore so
    // idle host RSS would drop to ~25% of memory_mib. Two
    // observations made that wrong:
    //
    //   1. RSS savings at restore are illusory. The snapshot is
    //      mmap'd CoW; until the guest touches a page, the page
    //      isn't in the worker's RSS. madvise(MADV_FREE) on
    //      never-touched CoW pages is a no-op for RSS, and the
    //      idle worker baseline already measures ~50 MiB
    //      regardless of balloon target.
    //
    //   2. Linux's virtio-balloon subtracts inflated pages from
    //      MemTotal (via `totalram_pages_sub` in the driver). A
    //      512-MiB guest with the old 70% target reported
    //      MemTotal=~215 MiB — directly causing the
    //      `image.test.ts > respects memoryMib` failure AND the
    //      vcpu_matrix host_proxy_parallel_4MiB guest OOM-kills
    //      (50× 4 MiB urllib reads need ~300 MiB free; the
    //      configured 1024 MiB guest had only ~300 MiB MemTotal
    //      after balloon inflate, leaving zero headroom for the
    //      Python runtime).
    //
    // Net: balloon-at-restore was pure cost, zero benefit. Pool
    // restore-on-release already reclaims touched pages via snapshot
    // CoW remap — that's where the real RSS reclaim happens.
    //
    // Opt back in via SUPERMACHINE_BALLOON_AT_RESTORE=1 for users
    // running long-lived VMs that touch + free large arenas without
    // a snapshot restore between exec cycles. Default: skip.
    if std::env::var("SUPERMACHINE_BALLOON_AT_RESTORE")
        .as_deref()
        .map(|v| v == "1" || v == "true")
        .unwrap_or(false)
    {
        if let Some(pages) = balloon_target_pages {
            vmm.balloon.request_inflate(pages);
            eprintln!("  balloon: requested inflate of {pages} pages (opt-in)");
        }
    }

    // Secondary threads were spawned earlier (before
    // `vmm.restore_snapshot`) so that `hv_gic_set_state` would
    // apply to every PE — see the long block-comment above the
    // spawn loop. Drop the variable here only to silence "unused"
    // — the actual move into each thread happened at spawn time.
    let _ = secondary_states;

    eprintln!("  vCPU launched ({n_vcpus} total), dispatch loop running\n");
    if timings && restore.is_some() {
        eprintln!(
            "[timing] restore.vcpu_launched={}us",
            run_t0.elapsed().as_micros()
        );
    }

    let pool_mode = pool_sock.is_some() || pool_worker.is_some();
    // Bake-then-pool: pool ctl is connected AND the worker was
    // started without --snapshot-out. The dispatch loop signals
    // readiness via `BAKE_READY` and hands control to the host
    // for warmup + final SNAPSHOT.
    //
    // 0.7.43+: propagated from `bin/worker.rs::main`. See
    // `RunOptions::bake_then_pool`.
    let bake_then_pool = bake_then_pool_from_caller;
    let mut bake_ready_signaled = false;
    let transport_idle = pool_mode.then(|| {
        let vsock = vmm.vsock.clone();
        std::sync::Arc::new(move || vsock.is_transport_idle())
            as std::sync::Arc<dyn Fn() -> bool + Send + Sync>
    });
    // Suppress the initial DONE write in bake-then-pool mode even
    // when restore happened. The bake driver isn't expecting it;
    // it's waiting for BAKE_READY (which the dispatch loop emits
    // after init-oci's readiness trigger fires post-resume).
    let initial_done_us = if bake_then_pool {
        None
    } else {
        restore.is_some().then_some(first_restore_us)
    };
    let mut pool = PoolControl::start(
        pool_sock.as_ref(),
        pool_worker,
        initial_done_us,
        restore
            .is_some()
            .then(|| vmm.vsock.muxer().first_host_port())
            .flatten(),
        crate::hypervisor::HypervisorVcpu::exit_token(&vmm.vm.vcpu),
        transport_idle,
        vmm.serial.clone(),
    )?;
    let mut report = RunReport::default();
    let mut warm_snapshot_cache: Option<WarmSnapshotCache> = None;

    // In-flight async-save threads. Drained at QUIT (and on
    // every snapshot RPC entry, to keep the list bounded). Each
    // thread writes a CompactSnapshot to disk via `.partial`+
    // rename. Holding the join handles here lets QUIT block
    // until pending writes hit disk — without this, a `QUIT`
    // immediately after `SNAPSHOT_ASYNC` would race the worker
    // exit against the in-flight save and leave a `.partial`.
    //
    // Each entry also retains an `Arc<CompactSnapshot>` so a
    // follow-up diff-save (Sync with `base_path` matching this
    // entry's path) can reuse the in-memory base for clonefile-
    // based diff capture without re-reading from disk.
    struct InFlightAsyncSave {
        path: String,
        handle: std::thread::JoinHandle<()>,
        snapshot: std::sync::Arc<crate::vmm::snapshot::CompactSnapshot>,
    }
    let mut in_flight_async_saves: Vec<InFlightAsyncSave> = Vec::new();

    // Lazy-load cache for the diff-snapshot path. Keyed on path
    // (typically the worker's restore source — same across all
    // cycle-snapshot calls in a pool). Capped at one entry: the
    // base CompactSnapshot is ~100 MiB on rust:1-slim, holding
    // multiple is wasteful and the cycle-snapshot path always
    // hits the same path. Invalidated by file mtime change so
    // a re-baked base correctly forces a reload.
    struct DiffBaseCache {
        path: String,
        modified: Option<SystemTime>,
        snapshot: std::sync::Arc<crate::vmm::snapshot::CompactSnapshot>,
    }
    let mut diff_base_cache: Option<DiffBaseCache> = None;

    // Auto-skip-blob-on-cycle-restore state. The GIC blob restore is
    // ~1.4 ms — the single fattest chunk in the per-cycle restore
    // path. For typical OCI workloads (sh, python, node, nginx, etc.)
    // the kernel programs the GIC distributor at boot and never
    // touches it after — meaning the live blob at end-of-cycle is
    // byte-identical to the snapshot blob, and re-applying it is an
    // expensive idempotent no-op.
    //
    // Strategy: on the FIRST cycle-restore for this worker process,
    // capture the live GIC blob and compare it byte-for-byte to the
    // snapshot blob. If equal, mark this worker "drift-clean" and
    // skip the blob restore on every subsequent cycle. If different,
    // mark "drift-detected" and always restore for the rest of this
    // worker's lifetime.
    //
    // Cost: one capture (~1.4 ms) on first cycle, paid once per
    // worker process. Saves ~1.4 ms × every subsequent cycle for
    // typical workloads. Pays a 1.4 ms one-time cost with no
    // ongoing penalty for exotic workloads (custom kernels, runtime
    // IRQ-affinity changes, etc.).
    //
    // Embedders can force the always-restore path with the
    // SUPERMACHINE_FORCE_GIC_RESTORE=1 env (escape hatch for
    // workloads where the auto-detect's first-cycle sample misses
    // late-emerging drift). The existing
    // SUPERMACHINE_SKIP_WARM_GIC_RESTORE=1 env still forces skip
    // (test-only; bypasses the safety check).
    let force_gic_restore = std::env::var_os("SUPERMACHINE_FORCE_GIC_RESTORE").is_some();
    let mut gic_drift_check_done = false;
    let mut gic_drift_detected = false;

    // Outer pool-worker loop. Without pool-worker mode this runs once.
    loop {
        let dispatch_exit = worker::dispatch_vcpu(
            0,
            &vmm.vm.vcpu,
            &vmm.bus,
            &vmm.coord,
            &vmm.all_mmio,
            &vmm.posix_fs,
            &vmm.dax_sessions,
            &vmm.fs_devices,
            &vmm.vsock,
            &vmm.vm,
            DispatchSnapshot {
                after_ms: snapshot_after_ms.or(if bake_then_pool { Some(10_000) } else { None }),
                at_heartbeat: snapshot_at,
                // bake_then_pool defaults to ALSO listening for the
                // listener-up trigger (preserves existing service-
                // image bake behavior). Pre-exec is opt-in via
                // resources flag — set by the bake driver only for
                // skip_warm_snapshot=true.
                on_listener: snapshot_on_listener || bake_then_pool,
                // Pre-exec is one-shot: after bake-ready fires, the
                // PRE_EXEC_READY atomic stays true forever (init-oci
                // already passed that point). Re-entering dispatch
                // with on_pre_exec=true would re-trigger the branch
                // and waste another quiesce window. Gate it on
                // !bake_ready_signaled so it only fires before the
                // first BakeReady return.
                on_pre_exec: snapshot_on_pre_exec && !bake_ready_signaled,
                quiesce_ms,
                out_path: snapshot_out,
                stop_requested: Some(pool.pause_flag()),
                bake_ready_signal: bake_then_pool && !bake_ready_signaled,
                vsock_exec_path,
                serial: &vmm.serial,
            },
        )?;
        // BakeReady is the bake-then-pool init-done signal. Fire
        // BAKE_READY on the supervisor socket and re-enter dispatch
        // (the host now drives via SNAPSHOT_ASYNC / SNAPSHOT / QUIT).
        if dispatch_exit == worker::DispatchExit::BakeReady {
            bake_ready_signaled = true;
            pool.signal_bake_ready();
            continue;
        }
        if dispatch_exit != worker::DispatchExit::Stopped
            && dispatch_exit != worker::DispatchExit::Canceled
        {
            break;
        }
        if pool.should_quit() {
            break;
        }
        if !pool.pause_requested() {
            if pool_mode && dispatch_exit == worker::DispatchExit::Canceled {
                continue;
            }
            break;
        }
        pool.clear_pause();
        // Snapshot requests are handled in-place: capture, save,
        // post result, then loop back to dispatch (the workload
        // is still alive). The pool host typically drops the Vm
        // immediately after — pool.shutdown then breaks us out
        // of the outer loop.
        if let Some(snap_req) = pool.take_snapshot_request() {
            let cap_t0 = std::time::Instant::now();
            // Quiesce the guest to WFI before capture. Mirrors
            // the bake-time snapshot path so the runner's snap
            // is taken from a clean kernel-idle state rather
            // than mid-syscall.
            //
            // Multi-vCPU exception: skip quiesce when vcpus > 1.
            // The host-side snapshot driver (`PooledVm::snapshot`
            // and the pipelined-bake's warm-snapshot RPC) issues
            // `smpark_park` BEFORE sending SNAPSHOT, which already
            // drove vCPU 0 + secondaries through a kernel-side
            // ioctl that ends with secondaries in WFI and vCPU 0
            // back in the agent's accept loop (also kernel-idle).
            // Layering quiesce_to_wfi on top added a second 100 ms
            // canceller-driven step that empirically left the
            // virtio-vsock RX path desynced — subsequent CONTROL
            // RPCs (the post-capture `smpark_unpark`) timed out.
            // Single-vCPU still runs quiesce_to_wfi; smpark only
            // applies for vcpus > 1 by design.
            let n_vcpus_total = vmm.coord.secondary_handles_snapshot().len() + 1;
            if n_vcpus_total <= 1 {
                if let Err(e) = worker::quiesce_to_wfi(
                    &vmm.vm.vcpu,
                    &vmm.bus,
                    &vmm.coord,
                    &vmm.dax_sessions,
                    100,
                ) {
                    pool.post_snapshot_result(Err(format!("quiesce: {e:?}")));
                    continue;
                }
            }
            // Multi-vCPU rendezvous: ask secondaries to exit
            // hv_vcpu_run, capture their own register state on
            // their owning threads, deposit, and wait. We
            // (vcpu0's thread) then read vcpu0 inline and splice
            // the secondaries' states into the snapshot. Without
            // this, multi-vCPU snapshots would have stale or
            // missing per-vcpu state for indices >0 and the
            // restored guest's secondaries would resume from
            // garbage.
            let secondary_handles = vmm.coord.secondary_handles_snapshot();
            if !secondary_handles.is_empty() {
                vmm.coord.request_snapshot_pause(&secondary_handles);
            }
            // Park the virtio-fs I/O workers for the capture window:
            // they write FUSE replies + used-ring updates + IRQ state
            // from non-vCPU threads, so the vCPU rendezvous above no
            // longer implies quiescent guest RAM / device state. Each
            // worker parks at a FUSE-chain boundary (a request is never
            // captured popped-but-unanswered). Dropped at the end of
            // this iteration (incl. every error-path `continue`), just
            // after the guest itself resumes — deferred doorbells then
            // replay.
            let _fs_pause = crate::devices::virtio::fs::FsIoPauseGuard::new(&vmm.fs_devices);
            let virtio = snapshot::VirtioSnapshot {
                // Unified device records: kind from the virtio device id, mmio
                // from the transport, backing None (HVF takes device backing from
                // the Image on restore; restore is positional). 7c step 3.
                devices: vmm
                    .all_mmio
                    .iter()
                    .map(|m| crate::snapshot_frame::DeviceRecord {
                        kind: crate::snapshot_frame::DeviceKind::from_virtio_id(m.device_id()),
                        mmio: m.capture_state(),
                        backing: crate::snapshot_frame::DeviceBacking::None,
                    })
                    .collect(),
                vsock_listeners: vmm.vsock.muxer().capture_tsi_listeners(),
            };
            let secondary_states = if !secondary_handles.is_empty() {
                vmm.coord.take_secondary_states()
            } else {
                Vec::new()
            };
            // Note: we DO NOT prune finished async-save entries
            // here. Pruning at this point dropped the
            // Arc<CompactSnapshot> for completed base saves,
            // which then forced the warm SNAPSHOT-with-base call
            // to fall through to the slow lazy-load-from-disk
            // path (~500 ms penalty). Retention is bounded in
            // practice (the bake-then-pool flow has at most
            // base async + warm sync = 2 entries); QUIT drains
            // them all. If a long-running pool worker eventually
            // accumulates entries we'll add a bounded LRU here.

            // Diff-snapshot path (Sync + base_path provided).
            // Two ways to get the base CompactSnapshot:
            //
            //   1. In-flight async save matches base_path — join
            //      the save thread, reuse its in-memory copy
            //      (zero extra I/O). Hot path for the bake's
            //      pipelined warm-after-base flow.
            //
            //   2. No matching in-flight save — load base from
            //      disk lazily via mmap-walk. Adds ~150 ms but
            //      enables the diff path on the cycle-snapshot
            //      flow (`pooled.snapshot()` from a worker that
            //      was restored via `--restore-from`, not via
            //      SNAPSHOT_ASYNC). Net win is still ~100 ms vs
            //      the plain streaming sync save.
            let diff_base_arc: Option<(String, std::sync::Arc<snapshot::CompactSnapshot>)> =
                if matches!(snap_req.mode, crate::vmm::pool::SnapshotMode::Sync) {
                    snap_req.base_path.as_deref().and_then(|bp| {
                        // Try in-flight first.
                        let idx = in_flight_async_saves.iter().position(|s| s.path == bp);
                        if let Some(i) = idx {
                            let entry = in_flight_async_saves.remove(i);
                            let _ = entry.handle.join();
                            return Some((entry.path, entry.snapshot));
                        }
                        // Cache hit on the lazy-loaded base?
                        // The cycle-snapshot path repeats the
                        // same `bp` for every call, so this hit
                        // rate is ~100% in practice.
                        let cur_modified =
                            std::fs::metadata(bp).ok().and_then(|m| m.modified().ok());
                        if let Some(c) = diff_base_cache.as_ref() {
                            if c.path == bp && c.modified == cur_modified {
                                return Some((bp.to_owned(), std::sync::Arc::clone(&c.snapshot)));
                            }
                        }
                        // Lazy-load from disk and populate cache.
                        match snapshot::load_compact_from_file(bp) {
                            Ok(loaded) => {
                                let arc = std::sync::Arc::new(loaded);
                                diff_base_cache = Some(DiffBaseCache {
                                    path: bp.to_owned(),
                                    modified: cur_modified,
                                    snapshot: std::sync::Arc::clone(&arc),
                                });
                                Some((bp.to_owned(), arc))
                            }
                            Err(e) => {
                                eprintln!(
                                    "[snapshot] diff base lazy-load from {bp} failed: {e:?}; \
                                     falling back to plain save"
                                );
                                None
                            }
                        }
                    })
                } else {
                    None
                };

            match snap_req.mode {
                crate::vmm::pool::SnapshotMode::Sync => {
                    if let Some((base_path, base_arc)) = diff_base_arc {
                        // Capture compact (in pause window), resume
                        // guest, then save via clonefile + diff
                        // pwrite.
                        let cap_only_t0 = std::time::Instant::now();
                        let compact =
                            match snapshot::capture_compact(&vmm.vm, virtio, secondary_states) {
                                Ok(c) => c,
                                Err(e) => {
                                    if !secondary_handles.is_empty() {
                                        vmm.coord.release_after_snapshot();
                                    }
                                    pool.post_snapshot_result(Err(format!(
                                        "diff snapshot capture: {e}"
                                    )));
                                    continue;
                                }
                            };
                        if !secondary_handles.is_empty() {
                            vmm.coord.release_after_snapshot();
                        }
                        let capture_us = cap_only_t0.elapsed().as_micros();

                        // Spawn the save in a background thread so the
                        // runner can immediately resume the dispatch
                        // loop (vCPU 0 re-enters hv_vcpu_run). Prior
                        // to 0.7.32 this save ran on the runner thread
                        // — for an 8 GiB warm snapshot it blocked
                        // vCPU 0 for ~20 s. Linux RCU on vCPU 0 then
                        // detected a `rcu_sched kthread starved for
                        // 5732 jiffies` self-stall, and the in-guest
                        // vsock kernel state (which RCU-protects)
                        // wedged — the agent's `accept()` no longer
                        // serviced incoming vsock connections, and
                        // the first post-acquire `vm.exec` died with
                        // `agent closed connection before sending
                        // EXIT`. Field-report fix.
                        //
                        // The supervisor's `done_rx.recv_timeout(60s)`
                        // returns as soon as we post the result here
                        // (capture only — `save_us=0`). The save
                        // thread runs to completion, posts
                        // `SAVE_DONE <path>` on the supervisor
                        // channel so the bake driver can wait for
                        // the file to land before declaring success.
                        let out_path = snap_req.out_path.clone();
                        let snap_for_thread = std::sync::Arc::new(compact);
                        let snap_for_inflight = std::sync::Arc::clone(&snap_for_thread);
                        let out_path_for_save = out_path.clone();
                        let base_arc_for_thread = base_arc;
                        let base_path_for_thread = base_path;
                        let posix_fs_blobs_for_thread: Vec<Option<Vec<u8>>> = vmm
                            .posix_fs
                            .iter()
                            .map(|pfs| Some(pfs.snapshot_state()))
                            .collect();
                        // DAX sidecar capture happens here (sync) so
                        // the metadata reflects the guest-paused
                        // instant. The actual file write happens on
                        // the background save thread, after the
                        // snapshot file lands. Order matches `mounts`
                        // exactly (and therefore matches the posix-fs
                        // blob indexing above).
                        let dax_blobs_for_thread: Vec<Option<Vec<u8>>> = vmm
                            .dax_sessions
                            .iter()
                            .map(|s| Some(s.snapshot_state()))
                            .collect();
                        let supervisor_writer_for_save = pool.supervisor_writer_clone();
                        // Per-VM smpark unpark GPA, captured for the
                        // async sidecar write (the save thread can't
                        // borrow `vmm`); 0 = not announced → sidecar
                        // skipped.
                        let smpark_gpa_for_save = vmm
                            .serial
                            .smpark_state_gpa
                            .load(std::sync::atomic::Ordering::Acquire);
                        let join = match std::thread::Builder::new()
                            .name("supermachine-snapshot-diff-save".into())
                            .spawn(move || {
                                // Try clone+diff. On failure (clonefile
                                // EXDEV, meta overflow, etc.) fall
                                // back to plain compact save.
                                let save_result = snapshot::save_compact_to_file_via_clone(
                                    &snap_for_thread,
                                    &base_arc_for_thread,
                                    &base_path_for_thread,
                                    &out_path_for_save,
                                );
                                let final_result = match save_result {
                                    Ok(s) => Ok(s),
                                    Err(diff_err) => {
                                        eprintln!(
                                            "[snapshot] diff path failed ({diff_err:?}); \
                                             falling back to plain compact save"
                                        );
                                        snapshot::save_compact_to_file(
                                            &snap_for_thread,
                                            &out_path_for_save,
                                        )
                                    }
                                };
                                // Posix-fs sidecar — same lifecycle
                                // as the Async path. Best-effort.
                                if final_result.is_ok() && !posix_fs_blobs_for_thread.is_empty() {
                                    let sidecar = snapshot::posix_fs_sidecar_path(&out_path_for_save);
                                    if let Err(e) = snapshot::write_posix_fs_sidecar(
                                        &sidecar,
                                        &posix_fs_blobs_for_thread,
                                    ) {
                                        eprintln!(
                                            "[snapshot-async] posix-fs sidecar {sidecar} write failed: \
                                             {e:?} — warm restores will fall back to lazy LOOKUP"
                                        );
                                    }
                                }
                                // DAX sidecar — same best-effort
                                // contract. Missing/failed sidecar
                                // means the restored guest SIGBUSes
                                // on the first DAX page touch not
                                // already paged-in pre-snapshot.
                                if final_result.is_ok() && !dax_blobs_for_thread.is_empty() {
                                    let sidecar = snapshot::dax_sidecar_path(&out_path_for_save);
                                    if let Err(e) = snapshot::write_dax_sidecar(
                                        &sidecar,
                                        &dax_blobs_for_thread,
                                    ) {
                                        eprintln!(
                                            "[snapshot-async] dax sidecar {sidecar} write failed: \
                                             {e:?} — cycle-restored guests will SIGBUS on first \
                                             touch of un-paged-in DAX pages"
                                        );
                                    }
                                }
                                // smpark sidecar — snapshot-scoped
                                // unpark-page GPA for host-direct
                                // multi-vCPU unpark on restore.
                                if final_result.is_ok() {
                                    snapshot::capture_and_write_smpark_sidecar(
                                        smpark_gpa_for_save,
                                        &out_path_for_save,
                                    );
                                }
                                // Notify supervisor so the bake
                                // driver / pool spawn_one can wait
                                // for the file rather than poll.
                                if let Some(w) = supervisor_writer_for_save {
                                    if let Ok(mut g) = w.lock() {
                                        if let Some(s) = g.as_mut() {
                                            use std::io::Write;
                                            let _ = match &final_result {
                                                Ok(_) => writeln!(s, "SAVE_DONE {out_path_for_save}"),
                                                Err(e) => writeln!(
                                                    s,
                                                    "SAVE_FAIL {out_path_for_save} {}",
                                                    format!("{e:?}").replace('\n', " ")
                                                ),
                                            };
                                            let _ = s.flush();
                                        }
                                    }
                                }
                            }) {
                            Ok(j) => j,
                            Err(e) => {
                                pool.post_snapshot_result(Err(format!(
                                    "spawn diff save thread: {e}"
                                )));
                                continue;
                            }
                        };
                        // Read length now; the Arc is about to be
                        // moved into the in-flight tracker.
                        let n_pages_for_acct = snap_for_inflight.pages.len();
                        // Inline-diff (the in-VM builder's per-layer chain):
                        // wait for the diff save to land before reporting DONE
                        // so the caller can read the file immediately — but
                        // keep the snapshot's RAM retained (with a trivial
                        // placeholder handle) so the next layer diffs against it
                        // in memory. The save still ran on its own thread, so
                        // there's no runner-thread RCU concern. Default path
                        // (inline_diff=false) is unchanged: fire-and-forget.
                        let save_us = if snap_req.inline_diff {
                            let _ = join.join();
                            in_flight_async_saves.push(InFlightAsyncSave {
                                path: out_path,
                                handle: std::thread::spawn(|| {}),
                                snapshot: snap_for_inflight,
                            });
                            cap_only_t0.elapsed().as_micros().saturating_sub(capture_us)
                        } else {
                            in_flight_async_saves.push(InFlightAsyncSave {
                                path: out_path,
                                handle: join,
                                snapshot: snap_for_inflight,
                            });
                            0
                        };
                        // Report DONE. For the bg path the save lands later on
                        // its own thread (bake.rs waits for SAVE_DONE); for the
                        // inline path it's already on disk.
                        pool.post_snapshot_result(Ok(crate::vmm::pool::SnapshotResult {
                            // The "logical" snapshot byte size; the
                            // background save may write fewer (diff
                            // path) or more (compact pad) — but
                            // this is the customer-visible
                            // accounting we already had pre-0.7.32.
                            bytes_written: (n_pages_for_acct as u64) * 4096,
                            capture_us,
                            save_us,
                        }));
                    } else {
                        // Plain streaming capture+save under guest
                        // pause.
                        let stream_t0 = std::time::Instant::now();
                        let save_stats = match snapshot::capture_and_save_streaming(
                            &vmm.vm,
                            &virtio,
                            &secondary_states,
                            &snap_req.out_path,
                        ) {
                            Ok(s) => s,
                            Err(e) => {
                                if !secondary_handles.is_empty() {
                                    vmm.coord.release_after_snapshot();
                                }
                                pool.post_snapshot_result(Err(format!("snapshot: {e}")));
                                continue;
                            }
                        };
                        if !secondary_handles.is_empty() {
                            vmm.coord.release_after_snapshot();
                        }
                        let total_us = stream_t0.elapsed().as_micros();
                        let _ = cap_t0;
                        // Posix-fs sidecar — see Sync+diff branch.
                        snapshot::capture_and_write_posix_fs_sidecar(
                            &snap_req.out_path,
                            &vmm.posix_fs,
                        );
                        // DAX sidecar — see Sync+diff branch.
                        snapshot::capture_and_write_dax_sidecar(
                            &snap_req.out_path,
                            &vmm.dax_sessions,
                        );
                        // smpark sidecar — snapshot-scoped unpark-page GPA.
                        snapshot::capture_and_write_smpark_sidecar(
                            vmm.serial
                                .smpark_state_gpa
                                .load(std::sync::atomic::Ordering::Acquire),
                            &snap_req.out_path,
                        );
                        pool.post_snapshot_result(Ok(crate::vmm::pool::SnapshotResult {
                            bytes_written: save_stats.ram_bytes + save_stats.ram_data_bytes,
                            capture_us: 0,
                            save_us: total_us,
                        }));
                    }
                }
                crate::vmm::pool::SnapshotMode::Async => {
                    // Pipelined: capture into a compact in-memory
                    // buffer (~50 ms for ~100 MiB of non-zero pages
                    // on M-series), resume guest immediately, write
                    // to disk in a background thread.
                    let cap_only_t0 = std::time::Instant::now();
                    let compact = match snapshot::capture_compact(&vmm.vm, virtio, secondary_states)
                    {
                        Ok(c) => c,
                        Err(e) => {
                            if !secondary_handles.is_empty() {
                                vmm.coord.release_after_snapshot();
                            }
                            pool.post_snapshot_result(Err(format!("snapshot: {e}")));
                            continue;
                        }
                    };
                    if !secondary_handles.is_empty() {
                        vmm.coord.release_after_snapshot();
                    }
                    let capture_us = cap_only_t0.elapsed().as_micros();
                    let out_path = snap_req.out_path.clone();
                    let n_pages = compact.pages.len();
                    let ram_size = compact.ram_size as u64;
                    // Wrap in Arc so a follow-up SNAPSHOT-with-base
                    // call can borrow the in-memory base for the
                    // diff path.
                    let compact_arc = std::sync::Arc::new(compact);
                    let snap_for_thread = std::sync::Arc::clone(&compact_arc);
                    let out_path_for_save = out_path.clone();
                    // Capture posix-fs state synchronously HERE
                    // (while the guest is still paused for this
                    // snapshot capture's secondary rendezvous) so
                    // the sidecar reflects the same instant the RAM
                    // image does. The actual sidecar write happens
                    // on the background thread after the snap save
                    // completes — non-blocking from the runner's
                    // POV.
                    let posix_fs_blobs_for_thread: Vec<Option<Vec<u8>>> = vmm
                        .posix_fs
                        .iter()
                        .map(|pfs| Some(pfs.snapshot_state()))
                        .collect();
                    // DAX sidecar capture — same instant-of-pause
                    // invariant as the posix-fs blobs above.
                    let dax_blobs_for_thread: Vec<Option<Vec<u8>>> = vmm
                        .dax_sessions
                        .iter()
                        .map(|s| Some(s.snapshot_state()))
                        .collect();
                    // Clone the supervisor writer Arc so the bg save
                    // thread can post `SAVE_DONE <path>` when its
                    // write completes. This is what enables a host-
                    // side observer (e.g. a fresh `Pool::spawn_one`
                    // wanting to restore from this snapshot file) to
                    // wait for the save to finish without polling
                    // the filesystem.
                    let supervisor_writer_for_save = pool.supervisor_writer_clone();
                    // Per-VM smpark unpark GPA, captured for the async
                    // sidecar write (the save thread can't borrow
                    // `vmm`); 0 = not announced → sidecar skipped.
                    let smpark_gpa_for_save = vmm
                        .serial
                        .smpark_state_gpa
                        .load(std::sync::atomic::Ordering::Acquire);
                    let join = match std::thread::Builder::new()
                        .name("supermachine-snapshot-async-save".into())
                        .spawn(move || {
                            let result = snapshot::save_compact_to_file(
                                &snap_for_thread,
                                &out_path_for_save,
                            );
                            if let Err(ref e) = result {
                                eprintln!(
                                    "[snapshot-async] save {out_path_for_save} failed: {e:?}"
                                );
                            }
                            // Write the posix-fs sidecar only after
                            // the main snapshot landed — if the save
                            // failed, a stale sidecar would be
                            // worse than no sidecar.
                            if result.is_ok() && !posix_fs_blobs_for_thread.is_empty() {
                                let sidecar = snapshot::posix_fs_sidecar_path(&out_path_for_save);
                                if let Err(e) = snapshot::write_posix_fs_sidecar(
                                    &sidecar,
                                    &posix_fs_blobs_for_thread,
                                ) {
                                    eprintln!(
                                        "[snapshot-async] posix-fs sidecar {sidecar} write failed: \
                                         {e:?} — warm restores will fall back to lazy LOOKUP"
                                    );
                                }
                            }
                            // DAX sidecar — same after-save invariant.
                            if result.is_ok() && !dax_blobs_for_thread.is_empty() {
                                let sidecar = snapshot::dax_sidecar_path(&out_path_for_save);
                                if let Err(e) =
                                    snapshot::write_dax_sidecar(&sidecar, &dax_blobs_for_thread)
                                {
                                    eprintln!(
                                        "[snapshot-async] dax sidecar {sidecar} write failed: \
                                         {e:?} — cycle-restored guests will SIGBUS on first \
                                         touch of un-paged-in DAX pages"
                                    );
                                }
                            }
                            // smpark sidecar — snapshot-scoped unpark-page GPA.
                            if result.is_ok() {
                                snapshot::capture_and_write_smpark_sidecar(
                                    smpark_gpa_for_save,
                                    &out_path_for_save,
                                );
                            }
                            // Notify supervisor. Best-effort — if the
                            // host-side reader has gone away or the
                            // writer is dropped, the message is just
                            // lost (no panic, no deadlock).
                            if let Some(w) = supervisor_writer_for_save {
                                if let Ok(mut g) = w.lock() {
                                    if let Some(s) = g.as_mut() {
                                        use std::io::Write;
                                        let _ = match &result {
                                            Ok(_) => writeln!(s, "SAVE_DONE {out_path_for_save}"),
                                            Err(e) => writeln!(
                                                s,
                                                "SAVE_FAIL {out_path_for_save} {}",
                                                format!("{e:?}").replace('\n', " ")
                                            ),
                                        };
                                        let _ = s.flush();
                                    }
                                }
                            }
                        }) {
                        Ok(j) => j,
                        Err(e) => {
                            pool.post_snapshot_result(Err(format!("spawn async save thread: {e}")));
                            continue;
                        }
                    };
                    in_flight_async_saves.push(InFlightAsyncSave {
                        path: out_path,
                        handle: join,
                        snapshot: compact_arc,
                    });
                    // Report immediately. `bytes_written` is the
                    // captured working set (≈ what'll hit disk).
                    pool.post_snapshot_result(Ok(crate::vmm::pool::SnapshotResult {
                        bytes_written: ram_size + (n_pages as u64) * 4096,
                        capture_us,
                        save_us: 0,
                    }));
                }
            }
            // 0.7.44+ pre-exec sync unblock. If init-oci is paused
            // inside `read(0)` at the pre-exec pause point (it
            // signaled readiness by writing the
            // "[SUPERMACHINE-PRE-EXEC] ready" marker on PL011), push
            // 'P\n' now so the guest unblocks the moment we resume
            // the vCPU. The byte queues in RX_QUEUE during the
            // capture window — pushing here, BEFORE the resume,
            // guarantees the guest sees it on the first UART poll
            // after resumption with zero added latency.
            //
            // No-op if PRE_EXEC_SYNC_READY is false (legacy
            // nanosleep mode, or post-pivot snapshots where this
            // marker has long since fired).
            //
            // One-shot: clear the atomic so subsequent snapshots
            // in the same worker (e.g. pool-cycle snapshots) don't
            // push spurious bytes that the workload TTY would
            // consume.
            if vmm
                .serial
                .pre_exec_sync_ready
                .swap(false, std::sync::atomic::Ordering::SeqCst)
            {
                vmm.serial.push_rx_byte(b'P');
                vmm.serial.push_rx_byte(b'\n');
            }
            // Resume dispatch — the workload is still alive.
            continue;
        }
        let Some(req) = pool.take_restore_request() else {
            break;
        };
        if let Some(p) = req.egress_policy {
            crate::vmm::egress_policy::set(&p);
        }
        let t0 = std::time::Instant::now();
        // Park the virtio-fs I/O workers across the whole cycle-restore:
        // `remap_cow_from_file*` below rewrites guest RAM wholesale, and
        // a worker mid-FUSE-chain would write reply bytes / used-ring
        // updates into it from a non-vCPU thread. (Nests with the guard
        // inside `restore_snapshot_timed_with_options`.) Dropped at the
        // end of this iteration, after the restored state is coherent.
        let _fs_pause = crate::devices::virtio::fs::FsIoPauseGuard::new(&vmm.fs_devices);
        // Reset all per-dispatch state.
        let phase_t0 = std::time::Instant::now();
        vmm.reset_vsock_transport();
        // Serial counterpart: drop stale console bytes / bake-detection
        // state so they don't leak from the prior cycle. The serial
        // state is per-VM (0.7.56+), but ONE worker's `Vmm` is reused
        // across many cycle-restores, so this reset is still required
        // between cycles.
        vmm.serial.reset_for_restore();
        // Balloon accounting (num_pages/actual + the process-global
        // INFLATED_PAGES gauge) isn't in the snapshot; zero it so it
        // matches the fresh CoW RAM remap and doesn't drift across
        // pooled clones.
        vmm.balloon.reset_for_restore();
        // virtio-fs notif_pool holds guest-offered notification chains
        // that aren't in the snapshot; clear them so a notification this
        // cycle can't target a stale descriptor from the prior cycle
        // (guest memory corruption + used-ring desync otherwise).
        for fs in &vmm.fs_devices {
            fs.reset_for_restore();
        }
        let reset_vsock_us = phase_t0.elapsed().as_micros();
        let phase_t0 = std::time::Instant::now();
        let file_meta = std::fs::metadata(&req.path).map_err(|source| RunError::MmapCow {
            path: req.path.clone(),
            source,
        })?;
        let modified = file_meta.modified().ok();
        let cache_hit = warm_snapshot_cache.as_ref().is_some_and(|cached| {
            cached.path == req.path
                && cached.file_len == file_meta.len()
                && cached.modified == modified
        });
        if !cache_hit {
            let file = std::fs::File::open(&req.path).map_err(|source| RunError::MmapCow {
                path: req.path.clone(),
                source,
            })?;
            let (snap, ram_offset, memory_bytes) =
                snapshot::load_meta(&req.path).map_err(|source| RunError::SnapshotLoad {
                    path: req.path.clone(),
                    source,
                })?;
            warm_snapshot_cache = Some(WarmSnapshotCache {
                path: req.path.clone(),
                file_len: file_meta.len(),
                modified,
                file,
                snap,
                ram_offset,
                memory_bytes,
            });
        }
        let load_meta_us = phase_t0.elapsed().as_micros();
        let cached = warm_snapshot_cache
            .as_ref()
            .expect("warm snapshot cache populated");
        // Multi-vCPU cycle-restore — phase 1: park secondaries in
        // `maybe_apply_restore` so they're out of `hv_vcpu_run`
        // before we remap RAM and rewrite GIC + vcpu0 state. Each
        // secondary spins waiting for its target state to be
        // published; we publish AFTER GIC + vcpu0 are done, in
        // phase 2 below.
        //
        // Without this, secondaries kept whatever state they had
        // accumulated during the previous pool cycle — the freshly
        // restored vcpu0/GIC state mismatched their stale ICH LRs
        // and redist bits, and the next softirq/IPI panicked.
        let secondary_handles = vmm.coord.secondary_handles_snapshot();
        if !secondary_handles.is_empty() {
            vmm.coord.pause_secondaries_for_restore(&secondary_handles);
        }

        // Re-mmap RAM from the (potentially new) snapshot path.
        // SAFETY: vCPU 0 has just exited dispatch; secondaries are
        // now parked in `maybe_apply_restore` (phase 1 above) so
        // the RAM remap is safe across all vCPUs.
        let phase_t0 = std::time::Instant::now();
        unsafe {
            if fixed_warm_ram_remap {
                vmm.vm.remap_cow_from_file_fixed(
                    &cached.file,
                    cached.ram_offset,
                    cached.memory_bytes,
                )?;
            } else {
                vmm.vm
                    .remap_cow_from_file(&cached.file, cached.ram_offset, cached.memory_bytes)?;
            }
        }
        let remap_cow_us = phase_t0.elapsed().as_micros();
        // Auto-skip-blob-on-cycle-restore: on the first cycle for
        // this worker, capture the live blob and compare to the
        // snapshot blob. Equal → distributor didn't drift during
        // last workload run → safe to skip on every cycle for this
        // worker. Different → drift; always restore.
        //
        // Cost is paid in `gic_capture_us`; net win starts on
        // cycle 2 (saves ~1.4 ms/cycle for typical workloads).
        let mut gic_capture_us: u128 = 0;
        if !gic_drift_check_done && !force_gic_restore && !skip_warm_gic_restore {
            let cap_t0 = std::time::Instant::now();
            match crate::hypervisor::HypervisorVm::capture_intc(&*vmm.vm.vm) {
                Ok(live_blob) => {
                    if live_blob.as_slice() != cached.snap.intc_blob.as_slice() {
                        gic_drift_detected = true;
                        eprintln!(
                            "  [warm-restore] GIC drift detected on first cycle \
                             (live={} bytes, snap={} bytes); always-restore for \
                             this worker. Set SUPERMACHINE_FORCE_GIC_RESTORE=1 \
                             to disable the auto-skip optimization globally.",
                            live_blob.len(),
                            cached.snap.intc_blob.len(),
                        );
                    } else {
                        eprintln!(
                            "  [warm-restore] GIC blob stable across cycles; \
                             auto-skipping blob restore (saves ~1.4 ms/cycle)."
                        );
                    }
                }
                Err(e) => {
                    // Capture failed — be conservative, fall back to
                    // always-restore. Probably HVF in a state where
                    // capture isn't allowed; restore will likely fail
                    // too, but that's the existing behavior.
                    gic_drift_detected = true;
                    eprintln!(
                        "  [warm-restore] GIC capture failed ({e:?}); \
                         falling back to always-restore."
                    );
                }
            }
            gic_drift_check_done = true;
            gic_capture_us = cap_t0.elapsed().as_micros();
        }
        // Decide whether to skip on THIS cycle. `skip_warm_gic_restore`
        // is the test-only env override (force-skip, bypassing the
        // safety check). Otherwise we skip iff we've completed the
        // drift check AND it found no drift, AND the embedder hasn't
        // forced restore.
        let skip_blob_this_cycle = if skip_warm_gic_restore {
            true // explicit override
        } else if force_gic_restore {
            false // embedder escape hatch
        } else {
            gic_drift_check_done && !gic_drift_detected
        };
        // Restore vcpu0 + GIC blob from the already-loaded metadata.
        // The GIC blob restore inside this is GLOBAL (distributor +
        // redistributor) — must precede secondaries' per-vcpu state
        // application so their ICH LRs reference a coherent INTID
        // space.
        let phase_t0 = std::time::Instant::now();
        let restore_timings = vmm.restore_snapshot_timed_with_options(
            &cached.snap,
            snapshot::SnapshotRestoreOptions {
                skip_intc_blob: skip_blob_this_cycle,
            },
        )?;
        // Phase 2: publish per-secondary states; each secondary's
        // spin-wait in `maybe_apply_restore` resolves, applies its
        // state on its OWNING thread, and parks at the resume
        // rendezvous. We block here until all running secondaries
        // have applied.
        if !secondary_handles.is_empty() {
            let secondary_states: Vec<Option<crate::vmm::snapshot::PerVcpuState>> =
                (1..vmm.coord.n_vcpus as usize)
                    .map(|i| cached.snap.per_vcpu.get(i).cloned())
                    .collect();
            vmm.coord
                .publish_and_wait_secondary_restore(&secondary_states);
            // Bump the resume generation — secondaries return from
            // `maybe_apply_restore` and re-enter `hv_vcpu_run` with
            // the snapshot's coherent state.
            vmm.coord.release_after_restore();
        }
        let restore_snapshot_us = phase_t0.elapsed().as_micros();
        let us = t0.elapsed().as_micros();
        let timings = WarmRestoreTimings {
            reset_vsock_us,
            remap_cow_us,
            load_meta_us,
            restore_snapshot_us,
            ram_copy_us: restore_timings.ram_copy_us,
            gic_restore_us: restore_timings.gic_restore_us,
            vcpu_restore_us: restore_timings.vcpu_restore_us,
            vtimer_offset_us: restore_timings.vtimer_offset_us,
            mmio_restore_us: restore_timings.mmio_restore_us,
            listener_restore_us: restore_timings.listener_restore_us,
        };
        eprintln!(
            "  warm restore from {} in {us} us (reset={} remap={} load_meta={} gic_check={} restore={} ram={} gic={} vcpu={} vtimer={} mmio={} listener={}{})",
            req.path,
            timings.reset_vsock_us,
            timings.remap_cow_us,
            timings.load_meta_us,
            gic_capture_us,
            timings.restore_snapshot_us,
            timings.ram_copy_us,
            timings.gic_restore_us,
            timings.vcpu_restore_us,
            timings.vtimer_offset_us,
            timings.mmio_restore_us,
            timings.listener_restore_us,
            if skip_blob_this_cycle { " [gic-skip]" } else { "" },
        );
        report.warm_restores += 1;
        pool.complete_restore(us, vmm.vsock.muxer().first_host_port(), timings);
    }
    // Drain any in-flight async saves before returning. Without
    // this, the worker process exits, the OS reaps the save
    // thread mid-write, and we leave a `<path>.partial` instead
    // of the canonical file. Critical for the bake-then-pool
    // flow's base snapshot, where the save runs in parallel
    // with the warmup workload and the warm SNAPSHOT+QUIT might
    // arrive before the base save completes.
    for entry in in_flight_async_saves.drain(..) {
        let _ = entry.handle.join();
    }
    // 0.7.44+ short-circuit for standalone workers: skip the
    // natural unwind path entirely. The drops of `vmm` (MicroVm)
    // and `vsock` (VsockMuxer) would call `hv_vcpu_destroy` +
    // `pthread_mutex_destroy` from C++ static destructors while
    // secondary vCPU threads are still in `hv_vcpu_run` and the
    // muxer-io thread is still in `_platform_memmove` → SIGSEGV
    // → "supervisor closed mid-protocol" cascade under FORKS=15.
    //
    // Confirmed reproduced by SIGSEGV crash reports at
    // ~/Library/Logs/DiagnosticReports/supermachine-worker-2026-05-21-{161738,164410,172409}.ips:
    //   main thread in hv_vcpu_destroy / pthread_mutex_destroy,
    //   vsock-muxer-io thread faulting on freed memory.
    //
    // Kernel handles all per-process cleanup (fds, mmaps, HVF
    // state) at process death. Async saves have just been
    // drained → no data loss.
    if exit_after_run {
        unsafe { libc::_exit(0) };
    }
    Ok(report)
}
