supermachine 0.7.70

// POSIX filesystem backend — translates FUSE ops to host syscalls on
// a single rooted subtree.
//
// Each mount has a host-side `root` directory; the backend exposes that
// subtree to the guest via FUSE. Inode numbers in the FUSE protocol are
// allocated by the backend (NOT the host filesystem's st_ino) so:
//   - We can change the host filesystem under us without invalidating
//     guest-side caches (which key on nodeid).
//   - Multiple mount instances of the same host path each get their
//     own nodeid namespace.
//
// The backend keeps a small bidirectional map:
//   nodeid -> InodeInfo { host_path, kind }
//   (parent_nodeid, name) -> nodeid    (lookup cache)
//
// Handle table: open files map fh -> RawFd. Opening the same path
// twice gives different fhs, mirroring open(2) semantics.
//
// Symlinks are followed during traversal — virtio-fs's typical use case
// is "expose this directory tree read-only-ish", not "expose a chroot
// jail". A later slice can add no-follow + open_by_handle for security.

use std::collections::BTreeMap;
use std::ffi::{CString, OsStr, OsString};
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::MetadataExt;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};

use super::backend::{
    DirEntry, Entry, Errno, FsBackend, StatFs, EACCES, EBADF, EEXIST as EEXIST_ERRNO, EINVAL, EIO,
    EISDIR, ENOENT, ENOSPC, ENOTDIR, EPERM,
};
use super::notify::Notifier;
use super::protocol::{
    Attr, SetattrIn, DT_BLK, DT_CHR, DT_DIR, DT_FIFO, DT_LNK, DT_REG, DT_SOCK, DT_UNKNOWN,
    FATTR_ATIME, FATTR_ATIME_NOW, FATTR_GID, FATTR_MODE, FATTR_MTIME, FATTR_MTIME_NOW, FATTR_SIZE,
    FATTR_UID, FUSE_ROOT_ID, S_IFBLK, S_IFCHR, S_IFDIR, S_IFIFO, S_IFLNK, S_IFMT, S_IFREG,
    S_IFSOCK,
};
use crate::vmm::resources::SymlinkPolicy;

/// How long the guest's dentry / attr cache should consider a FUSE
/// reply valid before re-LOOKUP / re-GETATTR. Through 0.7.23 every
/// site returned 1 second. That triggered an `lstat` storm under
/// sustained workloads: a large npm install's 200k+ small-file
/// creates produced 200k+ FUSE_LOOKUP requests per second after the
/// first round of caches expired, saturating the queue and surfacing
/// transient queue failures as EIO. Bumping to 60 s matches stock
/// virtiofsd's `--cache=auto` default.
///
/// Safety on host-side concurrent writers: our kqueue watcher
/// (`watch_inode` at OPEN time) emits `FUSE_NOTIFY_INVAL_INODE` /
/// `FUSE_NOTIFY_INVAL_ENTRY` to the guest whenever the host modifies
/// a watched path, so the cache TTL is a backstop, not the only
/// source of invalidation. Workloads where the guest is the primary
/// writer (the typical supermachine pattern) see effectively
/// permanent cache validity in steady state.
const ENTRY_VALID_SECS: u64 = 60;
const ATTR_VALID_SECS: u64 = 60;

/// What kind of host object backs a given nodeid. We cache this so
/// hot-path `getattr` doesn't always have to stat.
#[derive(Clone, Copy, Debug)]
enum Kind {
    File,
    Dir,
    Symlink,
    Other,
}

#[derive(Clone)]
struct InodeInfo {
    host_path: PathBuf,
    kind: Kind,
}

/// One DAX-mmap'd region of a host file. Multiple slots may share
/// one Mmap if SETUPMAPPING calls overlap (we deduplicate on host
/// path + foffset boundaries in a later optimization; for now each
/// dax_map allocates a fresh mmap).
struct Mmap {
    ptr: *mut u8,
    len: usize,
}

// SAFETY: ptr is a process-local pointer used only by DaxSession's
// HvfMapper. Mmap-as-DAX-source isn't dereferenced from rust here.
unsafe impl Send for Mmap {}
unsafe impl Sync for Mmap {}

/// One open-file handle. Tracks the live host fd plus the metadata
/// needed to re-open the same inode lazily after a snapshot/restore.
///
/// Pre-restore: `fd = Some(OwnedFd)` — opened by FUSE_OPEN /
/// FUSE_CREATE / SETUPMAPPING and used directly by read/write/etc.
///
/// Post-restore: `fd = None` until the first I/O on this handle, at
/// which point [`PosixFs::handle_raw_fd_locked`] re-opens the file at
/// `nodeid`'s host_path with `flags & REOPEN_FLAG_MASK` and stores
/// the resulting OwnedFd here. From that point on the entry looks
/// like any live pre-restore handle.
///
/// The bug this exists to fix: Linux's virtio-fs driver caches the
/// pre-snapshot `fh` inside its `fuse_file` and keeps sending it on
/// FUSE_READ after a cycle-restore. The host daemon's process-local
/// fd table is gone, so without lazy reopen `FUSE_READ(fh=X)`
/// returns EBADF → page-fault handler propagates as I/O error →
/// SIGBUS at userspace.
struct HandleEntry {
    /// None = restored from snapshot, fd not yet re-opened. Some =
    /// live OwnedFd, currently usable for read/write/fsync/etc.
    fd: Option<OwnedFd>,
    /// The nodeid this fh was opened against. Persisted so we can
    /// resolve the host_path at lazy-reopen time.
    nodeid: u64,
    /// Original open(2) flags from the OPEN/CREATE call, captured
    /// verbatim. On lazy reopen we mask with [`REOPEN_FLAG_MASK`] so
    /// destructive bits (O_CREAT, O_TRUNC, O_EXCL, O_APPEND) don't
    /// fire on re-open of an already-existing file.
    flags: u32,
    /// Cached dirent snapshot for handles created by FUSE_OPENDIR.
    /// `None` for file handles, `Some(Arc<...>)` for directory
    /// handles. Populated lazily on the first FUSE_READDIR call for
    /// the handle (NOT eagerly at opendir time — guests routinely
    /// opendir directories they never enumerate; e.g. shell `cd` or
    /// dirfd-relative path walks). Once populated, subsequent
    /// FUSE_READDIR calls serve from the snapshot using `offset` as
    /// a vector index. This is the only correctness-preserving way
    /// to honour `getdents64`-style position cookies across a
    /// directory that may be concurrently mutated by the same guest
    /// (the classic `rm -rf` pattern: open, getdents, unlink each
    /// entry, getdents again — re-opening `read_dir` between calls
    /// would skip entries because the entries-before-position
    /// snapshot shifts under unlink).
    ///
    /// Transient — NOT persisted through PFSS. A handle that
    /// survives a snapshot/restore boundary mid-enumeration will
    /// have to re-snapshot on the first post-restore readdir;
    /// that's a slight semantic divergence (a concurrent mutation
    /// pre-snapshot is not reflected post-snapshot) but the
    /// alternative is shipping potentially-stale paths in a
    /// post-restore reply, which is worse.
    dir_snapshot: Option<Arc<Vec<DirEntry>>>,
}

/// Bits of the original open(2) flags that survive a lazy reopen
/// post-snapshot.
///
/// Kept:
///   * `O_ACCMODE` — preserve RDONLY/WRONLY/RDWR access mode. The
///     guest still expects to use the fh in the same direction it
///     originally opened in.
///   * `O_NOCTTY` — preserve "don't acquire controlling terminal"
///     semantics. Harmless on regular files; mandatory for any
///     character-device backing.
///   * `O_NOFOLLOW` — preserve "fail if final component is a
///     symlink" semantics. Matters for security-sensitive opens.
///   * `O_CLOEXEC` — preserve the close-on-exec flag, which is
///     unrelated to file content but matters for fd inheritance if
///     the daemon ever forks.
///
/// Stripped:
///   * `O_CREAT`, `O_EXCL` — the file already exists at reopen
///     time. With O_EXCL the open would fail; without O_EXCL the
///     bit is a no-op but we strip for clarity.
///   * `O_TRUNC` — would destroy the file content we just restored
///     from snapshot. This is the most dangerous bit to leave in.
///   * `O_APPEND` — a restored fd starts at offset 0 (the guest
///     will re-seek as needed). If the original code held an
///     O_APPEND fd and relied on the host-side atomic-append
///     semantic that's a divergence, but the alternative — also
///     persisting the file offset — is bigger scope and rarely
///     observable since virtio-fs writes are pwrite-based (the
///     `offset` is supplied by the guest on every FUSE_WRITE,
///     bypassing the host fd's seek pointer).
///
/// `O_NOATIME` is Linux-only — on macOS libc the constant doesn't
/// exist, so it's not in the mask. macOS does ship `MNT_NOATIME` as
/// a mount-level flag (per-FS, not per-fd); irrelevant here.
const REOPEN_FLAG_MASK: i32 = libc::O_ACCMODE | libc::O_NOCTTY | libc::O_NOFOLLOW | libc::O_CLOEXEC;

struct State {
    /// nodeid -> InodeInfo.
    inodes: BTreeMap<u64, InodeInfo>,
    /// (parent, name) -> nodeid. Populated by lookup.
    children: BTreeMap<(u64, Vec<u8>), u64>,
    /// fh -> HandleEntry. Closed on `release` / `releasedir`. The
    /// inner `OwnedFd` is `None` for handles that survived a
    /// snapshot/restore and haven't been reused yet; the first I/O
    /// re-opens lazily. See [`HandleEntry`] for the full design.
    handles: BTreeMap<u64, HandleEntry>,
    /// Active DAX mappings indexed by host_va. Owns the mmap; dropping
    /// hits munmap.
    dax_mmaps: BTreeMap<usize, Mmap>,
    /// fh -> in-flight `.aotcache` atomic-publish state. When the
    /// guest creates/truncates a `<x>.aotcache` we transparently
    /// write to a hidden `<x>.aotcache.partial` sibling and rename
    /// to the final name on a non-empty close — Apple's macOS oahd
    /// `.aot.in_progress`-then-rename protocol, ported to the FUSE
    /// layer because Apple's Linux `rosettad` ships without it.
    /// See the comment in `fn create` for the full rationale.
    partial_writes: BTreeMap<u64, PartialWrite>,
    next_nodeid: u64,
    next_fh: u64,
}

/// Mid-flight atomic-publish state for one `.aotcache` write. The
/// guest's fh points at an `OwnedFd` in `handles` whose backing
/// inode is on disk at `partial_path`; on FUSE_RELEASE we either
/// rename it to `final_path` (translation succeeded) or unlink it
/// (translation aborted with zero bytes).
#[derive(Clone, Debug)]
struct PartialWrite {
    /// Path the guest THINKS it's writing to (the `<hash>.aotcache`
    /// it asked for). The fh's nodeid in `inodes` also points here.
    final_path: PathBuf,
    /// Path the host file is ACTUALLY at while the write is in
    /// flight — sibling of `final_path` with a leading-dot prefix
    /// and `.partial` suffix.
    partial_path: PathBuf,
    /// Parent dir + child name for `(parent, name)` → `nodeid`
    /// child-table fix-up after the rename succeeds.
    parent_nodeid: u64,
    child_name: Vec<u8>,
}

impl Drop for State {
    fn drop(&mut self) {
        for (_, m) in std::mem::take(&mut self.dax_mmaps) {
            unsafe {
                libc::munmap(m.ptr as *mut _, m.len);
            }
        }
    }
}

/// POSIX-backed FUSE filesystem rooted at `root`. All paths are
/// constrained to live under `root` — we don't `chroot`, we just
/// resolve names manually so we can refuse `..` escapes.
pub struct PosixFs {
    /// Canonical mount root. LOOKUP-resolved paths must remain under
    /// this prefix unless `symlinks == SymlinkPolicy::Follow`.
    root: PathBuf,
    /// Per-mount symlink policy. See [`SymlinkPolicy`].
    symlinks: SymlinkPolicy,
    st: Mutex<State>,
    /// Background thread that watches file-content-change events via
    /// kqueue and dispatches them to the guest as
    /// FUSE_NOTIFY_INVAL_INODE messages.
    watcher: Mutex<Option<Watcher>>,
    /// When `true`, this mount is the Rosetta-in-VM runtime share —
    /// the directory containing Apple's aarch64 Linux `rosetta` and
    /// `rosettad` binaries (typically
    /// `/Library/Apple/usr/libexec/oah/RosettaLinux/` on the host,
    /// mounted at `/run/rosetta` in the guest). Enables the
    /// FUSE_IOCTL handler that responds to `rosettad`'s startup
    /// cache-settings query (`ioctl(fd, 0x80806123, &buf128)`); see
    /// [`docs/design/rosetta-in-vm-2026-05-16.md`] for the protocol.
    rosetta_share: bool,
    /// Stable per-mount Rosetta-AOT-cache abstract socket name.
    /// `rosettad` binds this; the `rosetta` interpreter connects to
    /// it. If the two query the cache-settings ioctl independently
    /// (they do — different processes), they MUST get back the same
    /// socket name to reach each other. Allocated once per mount and
    /// returned verbatim on every cache-settings ioctl.
    rosetta_socket_name: String,
}

/// State for the host-file watcher thread. macOS uses kqueue
/// (`EVFILT_VNODE`); Linux uses inotify multiplexed with an eventfd wakeup via
/// `poll`. Both map host-side changes to FUSE invalidations for the guest.
struct Watcher {
    /// Shared with the watcher thread via Arc. The thread waits on the
    /// platform fd, looks up the (ident → nodeid) map, and calls notifier.
    /// We hold a strong ref so dropping PosixFs stops the thread.
    inner: Arc<WatcherInner>,
    /// Join handle for clean shutdown.
    handle: Option<std::thread::JoinHandle<()>>,
}

struct WatcherInner {
    /// kqueue fd; the watcher thread waits on this.
    #[cfg(target_os = "macos")]
    kq: libc::c_int,
    /// inotify fd; the watcher thread reads change events from it.
    #[cfg(not(target_os = "macos"))]
    inotify_fd: libc::c_int,
    /// eventfd written by Drop to wake the thread out of `poll` for shutdown.
    #[cfg(not(target_os = "macos"))]
    wake_fd: libc::c_int,
    /// Stop signal — set by drop, polled by the watcher thread.
    stop: AtomicBool,
    /// Notifier the watcher thread invokes when a change fires. None
    /// until set_notifier is called.
    notifier: Mutex<Option<Arc<dyn Notifier>>>,
    /// Map watch ident → (nodeid, parent_nodeid, name[, owned_fd]). The ident
    /// is the duped kqueue fd on macOS, or the inotify watch descriptor (`wd`)
    /// on Linux. parent_nodeid + name let us also invalidate the dentry cache
    /// so re-opens see the new inode after an atomic rename.
    watched: Mutex<BTreeMap<libc::c_int, WatchedEntry>>,
}

struct WatchedEntry {
    nodeid: u64,
    parent_nodeid: u64,
    name: Vec<u8>,
    /// macOS keeps the `O_EVTONLY` fd alive — the kqueue watch is tied to it.
    /// Linux's inotify watch is keyed by `wd` and needs no held-open fd.
    #[cfg(target_os = "macos")]
    _owned_fd: OwnedFd,
}

/// Whether a watched file changed in place (refresh caches) or its identity
/// went away via delete/rename (also drop the dentry + the watch).
enum WatchChange {
    Changed,
    Gone,
}

/// Shared (platform-agnostic) invalidation dispatch: look up the watched entry
/// for `ident`, emit the FUSE notifications, and drop the watch if the file is
/// gone. The notification semantics are identical across platforms; only the
/// event source (kqueue vs inotify) differs.
fn watcher_emit(inner: &WatcherInner, ident: libc::c_int, change: WatchChange) {
    let entry = inner
        .watched
        .lock()
        .unwrap()
        .get(&ident)
        .map(|e| (e.nodeid, e.parent_nodeid, e.name.clone()));
    let Some((nodeid, parent_nodeid, name)) = entry else {
        return;
    };
    if let Some(n) = inner.notifier.lock().unwrap().as_ref() {
        // INVAL_INODE is always safe — refreshes the kernel's cached pages
        // and attrs for THIS inode. (off=0,len=-1) → data pages;
        // (off=0,len=0) → attrs (size, mtime).
        n.invalidate_inode(nodeid, 0, -1);
        n.invalidate_inode(nodeid, 0, 0);
        // INVAL_ENTRY (drop the parent→child dentry) is the heavier hammer —
        // only safe-and-necessary when the entry's identity changed
        // (rename/delete). Emitting it on benign content/attr changes breaks a
        // mount stacked on this inode: the kernel re-LOOKUPs through us, gets a
        // fresh inode, and doesn't re-check the mount table → the overlay mount
        // is silently bypassed. (Repro: Node's resolver stat()s update atime →
        // spurious INVAL_ENTRY storm → swapped fs.existsSync answers.)
        if matches!(change, WatchChange::Gone) {
            n.invalidate_entry(parent_nodeid, &name);
        }
    }
    // Deleted/renamed: drop the watch so we don't keep an orphan entry (on
    // Linux inotify has already auto-removed the wd and sent IN_IGNORED).
    if matches!(change, WatchChange::Gone) {
        inner.watched.lock().unwrap().remove(&ident);
    }
}

impl Drop for Watcher {
    fn drop(&mut self) {
        self.inner.stop.store(true, Ordering::Release);
        // Wake the thread out of its blocking wait so it observes `stop`.
        #[cfg(target_os = "macos")]
        unsafe {
            // Poke the kqueue with a USER event.
            let mut tr = libc::kevent {
                ident: 0,
                filter: libc::EVFILT_USER,
                flags: libc::EV_ADD | libc::EV_ONESHOT | libc::EV_RECEIPT,
                fflags: libc::NOTE_TRIGGER,
                data: 0,
                udata: std::ptr::null_mut(),
            };
            let _ = libc::kevent(
                self.inner.kq,
                &mut tr as *mut _,
                1,
                std::ptr::null_mut(),
                0,
                std::ptr::null(),
            );
        }
        #[cfg(not(target_os = "macos"))]
        unsafe {
            // Write 1 to the eventfd the thread `poll`s alongside inotify.
            let one: u64 = 1;
            let _ = libc::write(
                self.inner.wake_fd,
                &one as *const u64 as *const libc::c_void,
                8,
            );
        }
        if let Some(h) = self.handle.take() {
            let _ = h.join();
        }
        unsafe {
            #[cfg(target_os = "macos")]
            libc::close(self.inner.kq);
            #[cfg(not(target_os = "macos"))]
            {
                libc::close(self.inner.inotify_fd);
                libc::close(self.inner.wake_fd);
            }
        }
    }
}

impl PosixFs {
    /// Tail of `fn create` after the host-side open succeeded. Split
    /// out so the EEXIST-on-stale-partial retry path doesn't need to
    /// duplicate the registration logic. Lives on the inherent impl
    /// (not `impl FsBackend`) because it's an internal helper, not a
    /// protocol entry point.
    fn create_finalize(
        &self,
        parent: u64,
        name: &OsStr,
        final_path: &Path,
        fd: libc::c_int,
        partial_info: Option<PartialWrite>,
        open_flags: u32,
    ) -> Result<(crate::fuse::backend::Entry, u64), Errno> {
        let owned = unsafe { OwnedFd::from_raw_fd(fd) };

        // Stat via the freshly-opened fd, not a fresh path-based
        // lookup. Path-based metadata() (= stat(2)) introduces a TOCTOU
        // window where a concurrent rename / unlink / symlink-replace
        // on the same name returns metadata for a different inode.
        // Under sustained npm-install-class concurrent IO this can
        // surface as EIO / ENOENT to the guest even though the fd is
        // valid. fstat on the dup'd fd is race-free — it reports the
        // inode the O_CREAT|O_EXCL open just produced, end of story.
        // Dup is cheap (~1 μs) vs the race risk.
        //
        // For partial-publish writes this is also the only fstat that
        // works during the in-flight window: the FINAL path doesn't
        // exist on host yet, so a path-based stat would return ENOENT.
        let dup_fd = owned.try_clone().map_err(|e| io_err_to_linux(&e))?;
        let md = std::fs::File::from(dup_fd)
            .metadata()
            .map_err(|e| io_err_to_linux(&e))?;
        let mut st = self.st.lock().unwrap();
        let nodeid = st.next_nodeid;
        st.next_nodeid += 1;
        // host_path always points at the FINAL name. For
        // partial-publish writes, the partial file is on disk under
        // a different name but the FUSE-layer inode identity is the
        // final destination — once the rename completes at RELEASE
        // time, no fix-up is needed.
        st.inodes.insert(
            nodeid,
            InodeInfo {
                host_path: final_path.to_path_buf(),
                kind: kind_from_meta(&md),
            },
        );
        st.children
            .insert((parent, name.as_bytes().to_vec()), nodeid);
        let fh = st.next_fh;
        st.next_fh += 1;
        st.handles.insert(
            fh,
            HandleEntry {
                fd: Some(owned),
                nodeid,
                flags: open_flags,
                dir_snapshot: None,
            },
        );
        if let Some(info) = partial_info {
            st.partial_writes.insert(fh, info);
        }
        let attr = attr_from_meta(nodeid, &md);
        Ok((
            crate::fuse::backend::Entry {
                nodeid,
                generation: 0,
                attr,
                entry_valid: ENTRY_VALID_SECS,
                attr_valid: ATTR_VALID_SECS,
            },
            fh,
        ))
    }

    /// Mount `root` as the FUSE filesystem root. The root must exist
    /// and be a directory; we stat it eagerly so a misconfigured
    /// mount fails fast.
    ///
    /// Defaults to [`SymlinkPolicy::Opaque`] — symlinks under the
    /// mount that resolve OUTSIDE the canonical mount root are
    /// rejected with EACCES on LOOKUP; the guest may create new
    /// symlinks whose targets are stored verbatim. Use
    /// [`Self::new_unchecked`] or [`Self::new_with_symlinks`] to pick
    /// a different policy.
    pub fn new(root: impl Into<PathBuf>) -> Result<Self, std::io::Error> {
        Self::new_with_symlinks(root, SymlinkPolicy::Opaque)
    }

    /// Constructor variant that allows symlinks pointing outside the
    /// mount root (LOOKUP follows them unconditionally). Equivalent to
    /// `new_with_symlinks(root, SymlinkPolicy::Follow)`. Use for
    /// trusted single-tenant workloads where the mount tree may
    /// reference absolute host paths.
    pub fn new_unchecked(root: impl Into<PathBuf>) -> Result<Self, std::io::Error> {
        Self::new_with_symlinks(root, SymlinkPolicy::Follow)
    }

    /// Constructor that picks the symlink policy explicitly. See
    /// [`SymlinkPolicy`] for the three modes.
    pub fn new_with_symlinks(
        root: impl Into<PathBuf>,
        symlinks: SymlinkPolicy,
    ) -> Result<Self, std::io::Error> {
        let root = root.into();
        let md = std::fs::metadata(&root)?;
        if !md.is_dir() {
            return Err(std::io::Error::new(
                std::io::ErrorKind::NotADirectory,
                format!("posix-fs root is not a directory: {}", root.display()),
            ));
        }
        // Canonicalize the root so the LOOKUP-time prefix check works
        // against a stable absolute path (no `..`, no symlinks, no
        // relative components). Without this, a root like
        // `/tmp/myapp/.` would fail prefix-of(canonical(child)).
        let root = std::fs::canonicalize(&root)?;
        let mut inodes = BTreeMap::new();
        inodes.insert(
            FUSE_ROOT_ID,
            InodeInfo {
                host_path: root.clone(),
                kind: Kind::Dir,
            },
        );
        Ok(Self {
            root,
            symlinks,
            st: Mutex::new(State {
                inodes,
                children: BTreeMap::new(),
                handles: BTreeMap::new(),
                dax_mmaps: BTreeMap::new(),
                partial_writes: BTreeMap::new(),
                next_nodeid: FUSE_ROOT_ID + 1,
                next_fh: 1,
            }),
            watcher: Mutex::new(None),
            rosetta_share: false,
            // Each PosixFs gets a unique abstract socket name. Use
            // process PID + monotonic counter so successive mounts in
            // the same worker don't collide. The same name is returned
            // on every cache-settings ioctl for this mount, so daemon
            // and interpreter both reach the same socket.
            rosetta_socket_name: {
                use std::sync::atomic::{AtomicU64, Ordering};
                static MOUNT_COUNTER: AtomicU64 = AtomicU64::new(0);
                let n = MOUNT_COUNTER.fetch_add(1, Ordering::Relaxed);
                let pid = std::process::id();
                format!("supermachine-rosettad-{pid:x}-{n:x}")
            },
        })
    }

    /// Mark this PosixFs as the Rosetta-in-VM runtime share. Enables
    /// the FUSE_IOCTL handler that responds to `rosettad`'s cache-
    /// settings query. Caller must only set this on a mount that
    /// actually serves the Apple Rosetta runtime — setting it on a
    /// regular share is harmless but adds a tiny per-ioctl branch
    /// cost we don't want elsewhere.
    ///
    /// Builder-style: consumes and returns `self`, so callers can
    /// chain it before `Arc::new(...)`.
    pub fn with_rosetta(mut self) -> Self {
        self.rosetta_share = true;
        self
    }

    /// Attach a Notifier and start a kqueue background thread that
    /// watches host-side changes to files this PosixFs has surfaced
    /// to the guest. On NOTE_DELETE / NOTE_RENAME / NOTE_WRITE /
    /// NOTE_EXTEND / NOTE_ATTRIB, dispatches FUSE_NOTIFY_INVAL_INODE
    /// to the guest so its dentry + page caches re-read on next access.
    ///
    /// Watching is per-inode: a file gets watched on first OPEN or
    /// SETUPMAPPING. This covers the dev-loop pattern (editor saves a
    /// file the guest is using); for rarely-accessed files we rely on
    /// the guest's 1s attr_valid timeout. Recursive directory watching
    /// is a follow-up if needed.
    pub fn set_notifier(&self, notifier: Arc<dyn Notifier>) -> Result<(), std::io::Error> {
        let mut watcher_slot = self.watcher.lock().unwrap();
        if watcher_slot.is_some() {
            // Already running; just swap the notifier.
            let w = watcher_slot.as_ref().unwrap();
            *w.inner.notifier.lock().unwrap() = Some(notifier);
            return Ok(());
        }
        #[cfg(target_os = "macos")]
        let inner = {
            let kq = unsafe { libc::kqueue() };
            if kq < 0 {
                return Err(std::io::Error::last_os_error());
            }
            Arc::new(WatcherInner {
                kq,
                stop: AtomicBool::new(false),
                notifier: Mutex::new(Some(notifier)),
                watched: Mutex::new(BTreeMap::new()),
            })
        };
        #[cfg(not(target_os = "macos"))]
        let inner = {
            let inotify_fd = unsafe { libc::inotify_init1(libc::IN_NONBLOCK | libc::IN_CLOEXEC) };
            if inotify_fd < 0 {
                return Err(std::io::Error::last_os_error());
            }
            let wake_fd = unsafe { libc::eventfd(0, libc::EFD_CLOEXEC) };
            if wake_fd < 0 {
                let err = std::io::Error::last_os_error();
                unsafe { libc::close(inotify_fd) };
                return Err(err);
            }
            Arc::new(WatcherInner {
                inotify_fd,
                wake_fd,
                stop: AtomicBool::new(false),
                notifier: Mutex::new(Some(notifier)),
                watched: Mutex::new(BTreeMap::new()),
            })
        };
        let thread_inner = inner.clone();
        let handle = std::thread::Builder::new()
            .name("supermachine-posixfs-watch".to_owned())
            .spawn(move || run_watcher(thread_inner))
            .map_err(|e| std::io::Error::other(e.to_string()))?;
        *watcher_slot = Some(Watcher {
            inner,
            handle: Some(handle),
        });
        Ok(())
    }

    /// Add `path` to the kqueue watch list, associated with `nodeid`.
    /// Best-effort: returns without error if watching is disabled or
    /// the file can't be opened. Calls should be cheap (open + 1
    /// kevent) so it's safe to invoke on every OPEN/SETUPMAPPING.
    fn watch_inode(&self, nodeid: u64, path: &std::path::Path) {
        let watcher = self.watcher.lock().unwrap();
        let Some(w) = watcher.as_ref() else { return };
        // Open a dedicated fd so the watch persists independent of
        // the guest's open file handle.
        let c = match CString::new(path.as_os_str().as_bytes()) {
            Ok(c) => c,
            Err(_) => return,
        };

        // Look up parent_nodeid + name so the watcher thread can emit
        // FUSE_NOTIFY_INVAL_ENTRY on rename/delete. Required to flush
        // the guest's dentry cache (1 s default TTL otherwise). Done FIRST so
        // an early return here can't leak a registered watch.
        let (parent_nodeid, name) = {
            let st = self.st.lock().unwrap();
            // Reverse-lookup in the (parent, name) → nodeid table.
            let entry = st
                .children
                .iter()
                .find(|(_, id)| **id == nodeid)
                .map(|((p, n), _)| (*p, n.clone()));
            match entry {
                Some(e) => e,
                None => return, // root or unknown — skip dentry invalidation
            }
        };

        let mut watched = w.inner.watched.lock().unwrap();
        // Already watching this nodeid? Drop old, install new.
        watched.retain(|_, e| e.nodeid != nodeid);

        // Register the watch and get its ident. macOS: a dedicated O_EVTONLY fd
        // (kept alive in the entry) registered with kqueue EVFILT_VNODE. Linux:
        // an inotify watch descriptor (`wd`); inotify keys by `wd`, no held fd.
        #[cfg(target_os = "macos")]
        let (ident, owned) = {
            let fd = unsafe { libc::open(c.as_ptr(), libc::O_RDONLY | libc::O_EVTONLY) };
            if fd < 0 {
                return;
            }
            let owned = unsafe { OwnedFd::from_raw_fd(fd) };
            let mut event = libc::kevent {
                ident: fd as libc::uintptr_t,
                filter: libc::EVFILT_VNODE,
                flags: libc::EV_ADD | libc::EV_CLEAR,
                fflags: libc::NOTE_DELETE
                    | libc::NOTE_RENAME
                    | libc::NOTE_WRITE
                    | libc::NOTE_EXTEND
                    | libc::NOTE_ATTRIB,
                data: 0,
                udata: std::ptr::null_mut(),
            };
            let rc = unsafe {
                libc::kevent(
                    w.inner.kq,
                    &mut event as *mut _,
                    1,
                    std::ptr::null_mut(),
                    0,
                    std::ptr::null(),
                )
            };
            if rc < 0 {
                return;
            }
            (fd, owned)
        };
        #[cfg(not(target_os = "macos"))]
        let ident = {
            // IN_MODIFY/CLOSE_WRITE ≈ NOTE_WRITE|EXTEND; IN_ATTRIB ≈ NOTE_ATTRIB;
            // IN_DELETE_SELF/MOVE_SELF ≈ NOTE_DELETE|RENAME. inotify auto-removes
            // the watch on delete/move and emits IN_IGNORED.
            let wd = unsafe {
                libc::inotify_add_watch(
                    w.inner.inotify_fd,
                    c.as_ptr(),
                    libc::IN_MODIFY
                        | libc::IN_CLOSE_WRITE
                        | libc::IN_ATTRIB
                        | libc::IN_DELETE_SELF
                        | libc::IN_MOVE_SELF,
                )
            };
            if wd < 0 {
                return;
            }
            wd
        };

        watched.insert(
            ident,
            WatchedEntry {
                nodeid,
                parent_nodeid,
                name,
                #[cfg(target_os = "macos")]
                _owned_fd: owned,
            },
        );
    }

    fn host_path_of(&self, nodeid: u64) -> Result<PathBuf, Errno> {
        let st = self.st.lock().unwrap();
        st.inodes
            .get(&nodeid)
            .map(|i| i.host_path.clone())
            .ok_or(ENOENT)
    }

    fn kind_of(&self, nodeid: u64) -> Result<Kind, Errno> {
        let st = self.st.lock().unwrap();
        st.inodes.get(&nodeid).map(|i| i.kind).ok_or(ENOENT)
    }

    // -----------------------------------------------------------------
    // Snapshot persistence — preserve the (nodeid → host_path) and
    // ((parent, name) → child_nodeid) tables across snapshot/restore.
    //
    // Without this, every restore starts with an empty inode table.
    // The guest, however, restores its dentry/inode cache from the
    // snapshot — it still holds nodeids it learned pre-snapshot. The
    // mismatch surfaces as `MODULE_NOT_FOUND` / EISDIR on paths not
    // walked during the warmup callback (because warmup walked SOME
    // paths and the guest's cache reflects those — but our fresh
    // daemon can't honour them).
    //
    // We serialise to a small binary blob written to a sidecar file
    // next to `restore.snap` (the snapshot itself is binary-stable
    // and we don't want to extend that format). Backwards-compatible:
    // a missing sidecar simply means the daemon starts empty (the
    // pre-0.7.6 behaviour). New snapshots produced by 0.7.6+ get the
    // sidecar; old snapshots keep the warmup-walked-paths-only
    // workaround behaviour.
    //
    // Persisted (PFSS v2):
    //   * `handles` / `next_fh` — host fds themselves are process-
    //     local and can't be serialised, BUT the (fh, nodeid, flags)
    //     METADATA persists. On restore we hydrate the table with
    //     `HandleEntry { fd: None, ... }`; the first I/O on each fh
    //     post-restore re-opens the host file lazily (see
    //     [`HandleEntry`] + [`REOPEN_FLAG_MASK`]). This is required
    //     because Linux's virtio-fs driver caches the pre-snapshot
    //     fh in its `fuse_file` and keeps using it after restore;
    //     without lazy reopen, FUSE_READ(fh=X) returns EBADF, the
    //     guest's page-fault handler propagates that as I/O error,
    //     and userspace SIGBUSes on cold-mapped pages.
    //
    // Excludes (transient by design):
    //   * `dax_mmaps` — process-local mmaps; the snapshot path
    //     re-establishes DAX windows during restore. The DAX
    //     window's shm contents persist through the snapshot path
    //     anyway (guest sees the same mapped bytes).
    //   * `partial_writes` — `.aotcache` atomic-publish state. By
    //     construction any in-flight publish at snapshot time can be
    //     safely abandoned: rosettad treats a missing cache file as
    //     "translate fresh", so dropping the partial is just a
    //     deferred cache miss, not a correctness bug. Persisting
    //     would also be unsound because the host-side `.partial`
    //     file may have been GC'd between snapshot and restore.
    //   * `root` / `symlinks` — derivable from the mount config,
    //     not part of state. Plumbed back through PosixFs::new on
    //     the restore side.
    //   * `watcher` — kqueue watches are per-fd and we don't
    //     restore fds eagerly. The watcher rebuilds itself lazily
    //     as files are reopened post-restore.

    /// Capture the inode + dentry tables for restore-time hydration.
    /// Returns a binary blob suitable for `restore_state`.
    pub fn snapshot_state(&self) -> Vec<u8> {
        // Layout:
        //   "PFSS" (magic, 4 bytes)
        //   u32 version (currently 2; v1 readers see this and refuse)
        //   u64 next_nodeid
        //   u64 inode_count
        //   inode entries:
        //     u64 nodeid
        //     u8  kind_disc (0=File, 1=Dir, 2=Symlink, 3=Other)
        //     u32 path_len
        //     bytes path (OsStr bytes, length path_len)
        //   u64 children_count
        //   children entries:
        //     u64 parent_nodeid
        //     u32 name_len
        //     bytes name (length name_len)
        //     u64 child_nodeid
        //   --- v2 additions ---
        //   u64 next_fh
        //   u64 handle_count
        //   handle entries:
        //     u64 fh
        //     u64 nodeid
        //     u32 flags
        //     u32 reserved (=0; pads to 8-byte alignment for future use)
        //
        // v1 → v2 compatibility: a v1 blob has no trailing fh data
        // and `restore_state` accepts it as "no handles" (the
        // pre-0.7.47 behaviour). v2 blobs are NOT readable by v1
        // restorers — the version bump is intentional so an
        // older binary running against a newer snapshot bails
        // early rather than partially restoring and then SIGBUSing.
        let st = self.st.lock().unwrap();
        let mut out = Vec::with_capacity(
            64 + st.inodes.len() * 64 + st.children.len() * 32 + st.handles.len() * 32,
        );
        out.extend_from_slice(b"PFSS");
        out.extend_from_slice(&2u32.to_le_bytes());
        out.extend_from_slice(&st.next_nodeid.to_le_bytes());
        out.extend_from_slice(&(st.inodes.len() as u64).to_le_bytes());
        for (nodeid, info) in &st.inodes {
            out.extend_from_slice(&nodeid.to_le_bytes());
            let kd: u8 = match info.kind {
                Kind::File => 0,
                Kind::Dir => 1,
                Kind::Symlink => 2,
                Kind::Other => 3,
            };
            out.push(kd);
            let pb = info.host_path.as_os_str().as_bytes();
            out.extend_from_slice(&(pb.len() as u32).to_le_bytes());
            out.extend_from_slice(pb);
        }
        out.extend_from_slice(&(st.children.len() as u64).to_le_bytes());
        for ((parent, name), child) in &st.children {
            out.extend_from_slice(&parent.to_le_bytes());
            out.extend_from_slice(&(name.len() as u32).to_le_bytes());
            out.extend_from_slice(name);
            out.extend_from_slice(&child.to_le_bytes());
        }
        // v2 tail: next_fh + handles.
        out.extend_from_slice(&st.next_fh.to_le_bytes());
        out.extend_from_slice(&(st.handles.len() as u64).to_le_bytes());
        for (fh, entry) in &st.handles {
            out.extend_from_slice(&fh.to_le_bytes());
            out.extend_from_slice(&entry.nodeid.to_le_bytes());
            out.extend_from_slice(&entry.flags.to_le_bytes());
            out.extend_from_slice(&0u32.to_le_bytes()); // reserved
        }
        out
    }

    /// Hydrate `inodes` / `children` / `next_nodeid` from a blob
    /// produced by [`Self::snapshot_state`]. Existing entries
    /// (notably the FUSE_ROOT_ID seeded by `new_with_symlinks`) are
    /// merged: the blob's root host_path is verified to match ours
    /// (anchoring nodeids to the SAME mount root); other entries are
    /// inserted verbatim. Returns `Err` on malformed blobs or root-
    /// path mismatch; caller may choose to log and continue with an
    /// empty table (degraded behaviour).
    pub fn restore_state(&self, blob: &[u8]) -> Result<(), std::io::Error> {
        let mut p = 0usize;
        fn take<'a>(b: &'a [u8], p: &mut usize, n: usize) -> Result<&'a [u8], std::io::Error> {
            if *p + n > b.len() {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::UnexpectedEof,
                    "posix-fs snapshot truncated",
                ));
            }
            let s = &b[*p..*p + n];
            *p += n;
            Ok(s)
        }
        fn read_u32(b: &[u8], p: &mut usize) -> Result<u32, std::io::Error> {
            let s = take(b, p, 4)?;
            Ok(u32::from_le_bytes([s[0], s[1], s[2], s[3]]))
        }
        fn read_u64(b: &[u8], p: &mut usize) -> Result<u64, std::io::Error> {
            let s = take(b, p, 8)?;
            Ok(u64::from_le_bytes([
                s[0], s[1], s[2], s[3], s[4], s[5], s[6], s[7],
            ]))
        }
        let magic = take(blob, &mut p, 4)?;
        if magic != b"PFSS" {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "posix-fs snapshot bad magic",
            ));
        }
        let version = read_u32(blob, &mut p)?;
        // v1 (legacy): no handle metadata; restored handles table is
        // empty. Old snapshots produced pre-0.7.47 will hit the
        // SIGBUS-on-cycle-restore bug this version was bumped to
        // fix, but they load correctly and the daemon doesn't
        // crash — same as the pre-bump behaviour.
        //
        // v2 (current): includes the next_fh + handles trailer
        // documented in `snapshot_state`.
        if version != 1 && version != 2 {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("posix-fs snapshot version {version} unsupported"),
            ));
        }
        let next_nodeid = read_u64(blob, &mut p)?;
        let inode_count = read_u64(blob, &mut p)? as usize;
        let mut new_inodes: BTreeMap<u64, InodeInfo> = BTreeMap::new();
        let mut root_match = false;
        for _ in 0..inode_count {
            let nodeid = read_u64(blob, &mut p)?;
            let kd = take(blob, &mut p, 1)?[0];
            let kind = match kd {
                0 => Kind::File,
                1 => Kind::Dir,
                2 => Kind::Symlink,
                3 => Kind::Other,
                _ => {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        format!("posix-fs snapshot bad kind {kd}"),
                    ));
                }
            };
            let path_len = read_u32(blob, &mut p)? as usize;
            let path_bytes = take(blob, &mut p, path_len)?;
            let host_path = PathBuf::from(OsStr::from_bytes(path_bytes));
            if nodeid == FUSE_ROOT_ID {
                if host_path != self.root {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        format!(
                            "posix-fs snapshot root mismatch: snapshot={} mount={}",
                            host_path.display(),
                            self.root.display()
                        ),
                    ));
                }
                root_match = true;
            }
            new_inodes.insert(nodeid, InodeInfo { host_path, kind });
        }
        if !root_match {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "posix-fs snapshot missing FUSE_ROOT_ID entry",
            ));
        }
        let children_count = read_u64(blob, &mut p)? as usize;
        let mut new_children: BTreeMap<(u64, Vec<u8>), u64> = BTreeMap::new();
        for _ in 0..children_count {
            let parent = read_u64(blob, &mut p)?;
            let name_len = read_u32(blob, &mut p)? as usize;
            let name = take(blob, &mut p, name_len)?.to_vec();
            let child = read_u64(blob, &mut p)?;
            new_children.insert((parent, name), child);
        }
        // v2 trailer: next_fh + handle metadata. v1 stops here.
        let (next_fh_restored, new_handles): (u64, BTreeMap<u64, HandleEntry>) = if version >= 2 {
            let next_fh = read_u64(blob, &mut p)?;
            let handle_count = read_u64(blob, &mut p)? as usize;
            let mut handles = BTreeMap::new();
            for _ in 0..handle_count {
                let fh = read_u64(blob, &mut p)?;
                let nodeid = read_u64(blob, &mut p)?;
                let flags = read_u32(blob, &mut p)?;
                let _reserved = read_u32(blob, &mut p)?;
                handles.insert(
                    fh,
                    HandleEntry {
                        fd: None,
                        nodeid,
                        flags,
                        // Transient — re-snapshot on next readdir
                        // if this is a directory handle.
                        dir_snapshot: None,
                    },
                );
            }
            (next_fh, handles)
        } else {
            // v1 blob: no fh metadata. Empty handles, next_fh keeps
            // its constructor default (1). This is pre-0.7.47
            // behaviour and reproduces the SIGBUS-on-restore bug for
            // any cold-mapped pages that survive into the guest,
            // but the daemon itself loads cleanly.
            (0, BTreeMap::new())
        };
        if p != blob.len() {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("posix-fs snapshot has {} trailing bytes", blob.len() - p),
            ));
        }
        let mut st = self.st.lock().unwrap();
        st.inodes = new_inodes;
        st.children = new_children;
        st.next_nodeid = next_nodeid.max(st.next_nodeid);
        if version >= 2 {
            st.handles = new_handles;
            // Bump next_fh past every restored handle so new opens
            // never collide with a cached pre-snapshot fh. The
            // .max() guards against a snapshot whose recorded
            // next_fh was somehow lower than the highest stored fh
            // (shouldn't happen, but defence-in-depth: a collision
            // would let a fresh open() return an fh the guest's
            // dentry cache thinks is a different open file).
            let max_fh = st.handles.keys().copied().max().unwrap_or(0);
            st.next_fh = next_fh_restored.max(max_fh + 1).max(st.next_fh);
        }
        Ok(())
    }

    /// Resolve `fh` to a live raw fd, lazily re-opening the host
    /// file if the entry survived a snapshot/restore. Caller MUST
    /// already hold `st: &mut MutexGuard<State>`; the inode table
    /// (for `host_path_of`) is read off the same guard so the
    /// caller doesn't need to drop-and-reacquire mid-call.
    ///
    /// Returns `EBADF` if `fh` is not in the handle table at all.
    /// Returns the lazy-open errno (mapped to Linux form) if the
    /// re-open fails (file deleted on host, permissions changed,
    /// etc.). Once a successful re-open completes, the resulting
    /// OwnedFd is stashed back into the entry so subsequent calls
    /// short-circuit to the already-open path.
    fn handle_raw_fd_locked(st: &mut State, fh: u64) -> Result<libc::c_int, Errno> {
        let entry = st.handles.get_mut(&fh).ok_or(EBADF)?;
        if let Some(ref fd) = entry.fd {
            return Ok(fd.as_raw_fd());
        }
        // Cold path: lazy reopen. Look up the host path via the
        // restored inode table. If the inode is gone (the host
        // file was deleted between snapshot and restore) we surface
        // EBADF — the guest's cached fh references a file that no
        // longer exists. POSIX says reads on a deleted-while-open
        // file should still work; that contract doesn't apply
        // across a process boundary.
        let nodeid = entry.nodeid;
        let flags = entry.flags;
        let host_path = st
            .inodes
            .get(&nodeid)
            .map(|i| i.host_path.clone())
            .ok_or(EBADF)?;
        let c = CString::new(host_path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
        let open_flags = (flags as i32) & REOPEN_FLAG_MASK;
        let raw = unsafe { libc::open(c.as_ptr(), open_flags) };
        if raw < 0 {
            return Err(errno_now());
        }
        // SAFETY: `open` returned a non-negative fd above.
        let owned = unsafe { OwnedFd::from_raw_fd(raw) };
        let raw = owned.as_raw_fd();
        // Re-fetch the entry as `&mut` to install the fd. The
        // earlier `get_mut` borrow was released when we cloned the
        // path (we only held immutable references after).
        let entry = st.handles.get_mut(&fh).ok_or(EBADF)?;
        entry.fd = Some(owned);
        Ok(raw)
    }
}

/// Map a host (macOS aarch64) errno value to its Linux equivalent.
///
/// FUSE is a Linux-kernel protocol — the guest's libc/glibc/musl decode
/// the wire-side error number using `<asm-generic/errno.h>` /
/// `<bits/errno.h>`. When a host libc syscall fails on macOS we must
/// translate the returned errno before sending it to the guest,
/// otherwise the guest sees the wrong symbolic error (e.g. macOS
/// `ENOTEMPTY = 66` is `EREMOTE = 66` on Linux; macOS `ENOSYS = 78`
/// is `EREMCHG = 78` on Linux; macOS `EAGAIN = 35` is `EDEADLK = 35`
/// on Linux; and so on).
///
/// Errnos 1..=34 are POSIX-identical across macOS and Linux
/// (EPERM=1, ENOENT=2, ESRCH=3, EINTR=4, EIO=5, ..., EDOM=33,
/// ERANGE=34). 35+ is where the numbering diverges. Anything not
/// listed below passes through unchanged either because it's in the
/// shared 1..=34 range or because the numeric value happens to agree.
///
/// References:
///   macOS: `<sys/errno.h>` —
///     https://github.com/apple-oss-distributions/xnu/blob/main/bsd/sys/errno.h
///   Linux: `<asm-generic/errno-base.h>`, `<asm-generic/errno.h>`.
#[cfg(target_os = "macos")]
fn host_to_linux_errno(host: i32) -> i32 {
    match host {
        // EAGAIN and EWOULDBLOCK are both 35 on macOS, both 11 on Linux.
        35 => 11, // EAGAIN / EWOULDBLOCK
        11 => 35, // EDEADLK / EDEADLOCK
        63 => 36, // ENAMETOOLONG
        77 => 37, // ENOLCK
        78 => 38, // ENOSYS
        66 => 39, // ENOTEMPTY  ← the rmdir-non-empty bug fixed in 0.6.1
        62 => 40, // ELOOP
        91 => 42, // ENOMSG
        90 => 43, // EIDRM
        // 93 ENOATTR / 96 ENODATA — macOS uses 93 for the
        // "attribute not found" case (xattr/listxattr); Linux uses
        // ENODATA=61. Both should map to 61 so the guest's xattr
        // calls see a consistent error.
        93 => 61, // ENOATTR  (macOS-specific xattr-miss)
        96 => 61, // ENODATA  (in case any path returns it)
        84 => 75, // EOVERFLOW
        94 => 74, // EBADMSG
        92 => 84, // EILSEQ
        38 => 88, // ENOTSOCK
        39 => 89, // EDESTADDRREQ
        40 => 90, // EMSGSIZE
        41 => 91, // EPROTOTYPE
        42 => 92, // ENOPROTOOPT
        43 => 93, // EPROTONOSUPPORT
        44 => 94, // ESOCKTNOSUPPORT
        // ENOTSUP and EOPNOTSUPP — macOS defines ENOTSUP=45 (POSIX
        // form) AND EOPNOTSUPP=102 (BSD form); Linux unifies both
        // as 95. Map BOTH so any libc call returning either form
        // surfaces correctly to the guest. Pre-fix, a syscall
        // returning macOS ENOTSUP=45 would land in the guest as
        // Linux errno 45 (EL2NSYNC — completely unrelated, channel
        // sync error from streams). Confirmed via audit; cause is
        // largely socket-path syscalls but file-path getxattr can
        // also return ENOTSUP.
        45 => 95,  // ENOTSUP (POSIX form)
        102 => 95, // EOPNOTSUPP (BSD form)
        46 => 96,  // EPFNOSUPPORT
        47 => 97,  // EAFNOSUPPORT
        48 => 98,  // EADDRINUSE
        49 => 99,  // EADDRNOTAVAIL
        50 => 100, // ENETDOWN
        51 => 101, // ENETUNREACH
        52 => 102, // ENETRESET
        53 => 103, // ECONNABORTED
        54 => 104, // ECONNRESET
        55 => 105, // ENOBUFS
        56 => 106, // EISCONN
        57 => 107, // ENOTCONN
        58 => 108, // ESHUTDOWN
        59 => 109, // ETOOMANYREFS
        60 => 110, // ETIMEDOUT
        61 => 111, // ECONNREFUSED
        64 => 112, // EHOSTDOWN
        65 => 113, // EHOSTUNREACH
        37 => 114, // EALREADY
        36 => 115, // EINPROGRESS
        70 => 116, // ESTALE
        69 => 122, // EDQUOT
        // EREMOTE — macOS=71, Linux=66. A NFS / network-fs error
        // class; rare for our virtio-fs paths but possible if the
        // host's underlying FS is itself networked (rare on bake
        // hosts but real on developer machines mounting an SMB
        // share for the workspace).
        71 => 66, // EREMOTE
        // EUSERS — macOS=68, Linux=87. "Too many users" — rare,
        // mostly NFS / quota.
        68 => 87, // EUSERS
        // STREAMS / multi-hop errnos. Rare on POSIX paths but
        // some libc shims (especially over network FS) return them.
        // Map for completeness.
        95 => 72,   // EMULTIHOP    (macOS) → Linux EMULTIHOP=72
        97 => 67,   // ENOLINK      (macOS) → Linux ENOLINK=67
        98 => 63,   // ENOSR        (macOS) → Linux ENOSR=63
        99 => 60,   // ENOSTR       (macOS) → Linux ENOSTR=60
        100 => 71,  // EPROTO       (macOS) → Linux EPROTO=71
        101 => 62,  // ETIME        (macOS) → Linux ETIME=62
        89 => 125,  // ECANCELED
        105 => 130, // EOWNERDEAD
        104 => 131, // ENOTRECOVERABLE
        _ => host,
    }
}

#[cfg(not(target_os = "macos"))]
#[inline]
fn host_to_linux_errno(host: i32) -> i32 {
    host
}

/// Convert a `std::io::Error` from a host libc call into the
/// Linux-form FUSE [`Errno`] (a negative i32). Falls back to `-EIO`
/// when the error has no raw os code.
fn io_err_to_linux(e: &std::io::Error) -> Errno {
    -host_to_linux_errno(e.raw_os_error().unwrap_or(libc::EIO))
}

fn errno_now() -> Errno {
    -host_to_linux_errno(
        std::io::Error::last_os_error()
            .raw_os_error()
            .unwrap_or(libc::EIO),
    )
}

fn attr_from_meta(ino: u64, md: &std::fs::Metadata) -> Attr {
    let mode_full = md.mode();
    let perm_bits = mode_full & 0o7777;
    let typ_bits = if md.is_dir() {
        S_IFDIR
    } else if md.is_file() {
        S_IFREG
    } else if md.file_type().is_symlink() {
        S_IFLNK
    } else {
        match mode_full & S_IFMT {
            S_IFBLK => S_IFBLK,
            S_IFCHR => S_IFCHR,
            S_IFIFO => S_IFIFO,
            S_IFSOCK => S_IFSOCK,
            _ => 0,
        }
    };
    Attr {
        ino,
        size: md.size(),
        blocks: md.blocks(),
        atime: md.atime() as u64,
        mtime: md.mtime() as u64,
        ctime: md.ctime() as u64,
        atimensec: md.atime_nsec() as u32,
        mtimensec: md.mtime_nsec() as u32,
        ctimensec: md.ctime_nsec() as u32,
        mode: typ_bits | perm_bits,
        nlink: md.nlink() as u32,
        uid: md.uid(),
        gid: md.gid(),
        rdev: md.rdev() as u32,
        blksize: md.blksize() as u32,
        flags: 0,
    }
}

/// Build an `Attr` from a raw `libc::stat`. Used by paths that go
/// through `fstatat` / `lstat` (symlink + hard-link create) — these
/// can't use the `std::fs::Metadata` route since std would follow
/// symlinks or canonicalize the path under us.
// `st_mode` is u16 on macOS, u32 on Linux — the cast is needed on macOS.
#[allow(clippy::unnecessary_cast)]
fn attr_from_stat(ino: u64, st: &libc::stat) -> Attr {
    let mode_full = st.st_mode as u32;
    let perm_bits = mode_full & 0o7777;
    let typ_bits = match mode_full & S_IFMT {
        S_IFDIR => S_IFDIR,
        S_IFREG => S_IFREG,
        S_IFLNK => S_IFLNK,
        S_IFBLK => S_IFBLK,
        S_IFCHR => S_IFCHR,
        S_IFIFO => S_IFIFO,
        S_IFSOCK => S_IFSOCK,
        _ => 0,
    };
    // macOS exposes `st_atimespec` etc.; Linux uses `st_atim`. Pull
    // through the unix MetadataExt-style fields via a small cfg.
    #[cfg(target_os = "macos")]
    let (a, an, m, mn, c, cn) = (
        st.st_atime as u64,
        st.st_atime_nsec as u32,
        st.st_mtime as u64,
        st.st_mtime_nsec as u32,
        st.st_ctime as u64,
        st.st_ctime_nsec as u32,
    );
    #[cfg(not(target_os = "macos"))]
    let (a, an, m, mn, c, cn) = (
        st.st_atime as u64,
        st.st_atime_nsec as u32,
        st.st_mtime as u64,
        st.st_mtime_nsec as u32,
        st.st_ctime as u64,
        st.st_ctime_nsec as u32,
    );
    Attr {
        ino,
        size: st.st_size as u64,
        blocks: st.st_blocks as u64,
        atime: a,
        mtime: m,
        ctime: c,
        atimensec: an,
        mtimensec: mn,
        ctimensec: cn,
        mode: typ_bits | perm_bits,
        nlink: st.st_nlink as u32,
        uid: st.st_uid,
        gid: st.st_gid,
        rdev: st.st_rdev as u32,
        blksize: st.st_blksize as u32,
        flags: 0,
    }
}

/// Open a host directory as an O_DIRECTORY|O_NOFOLLOW dirfd we can
/// hand to *at()-family syscalls. Returns ENOTDIR if `path` isn't a
/// directory.
fn open_dirfd(path: &std::path::Path) -> Result<OwnedFd, Errno> {
    let c = CString::new(path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
    // O_NOFOLLOW so a hostile rename of a parent-component into a
    // symlink can't redirect us elsewhere mid-call.
    let fd = unsafe {
        libc::open(
            c.as_ptr(),
            libc::O_RDONLY | libc::O_DIRECTORY | libc::O_NOFOLLOW,
        )
    };
    if fd < 0 {
        return Err(errno_now());
    }
    // SAFETY: fd is fresh from open() above.
    Ok(unsafe { OwnedFd::from_raw_fd(fd) })
}

fn kind_from_meta(md: &std::fs::Metadata) -> Kind {
    if md.is_dir() {
        Kind::Dir
    } else if md.is_file() {
        Kind::File
    } else if md.file_type().is_symlink() {
        Kind::Symlink
    } else {
        Kind::Other
    }
}

fn name_safe(name: &OsStr) -> Result<(), Errno> {
    let bytes = name.as_bytes();
    if bytes.is_empty() || bytes == b"." || bytes == b".." {
        return Err(EINVAL);
    }
    if bytes.contains(&b'/') {
        return Err(EINVAL);
    }
    // Defense-in-depth: reject embedded NULs. The C string layer
    // (CString::new) would catch these too, but we'd rather not let
    // hostile names get past our first filter.
    if bytes.contains(&0u8) {
        return Err(EINVAL);
    }
    Ok(())
}

impl FsBackend for PosixFs {
    fn lookup(&self, parent: u64, name: &OsStr) -> Result<Entry, Errno> {
        name_safe(name)?;
        let parent_path = self.host_path_of(parent)?;
        let path = parent_path.join(name);

        // Policy-aware metadata fetch:
        //   - Deny / Opaque modes use lstat: the symlink inode itself
        //     is what the guest must observe (POSIX lstat semantics)
        //     and it tolerates broken / mid-write targets — see the
        //     npm-postinstall EIO bug fixed in 0.7.4.
        //   - Follow mode uses stat: in this mode symlinks are a
        //     trusted-tenant escape hatch (the guest is meant to
        //     read through them transparently to host-side targets,
        //     including targets OUTSIDE the mount root). The guest
        //     can't traverse a host-absolute path itself, so we
        //     transparently resolve the symlink at LOOKUP time and
        //     hand the guest the target's metadata. Falls back to
        //     lstat if the target is unreachable (broken symlink)
        //     so the lookup still succeeds — readlink will return
        //     the raw target bytes, and any read through the symlink
        //     will surface the error at OPEN time.
        let md = if self.symlinks == SymlinkPolicy::Follow {
            std::fs::metadata(&path)
                .or_else(|_| std::fs::symlink_metadata(&path))
                .map_err(|e| io_err_to_linux(&e))?
        } else {
            std::fs::symlink_metadata(&path).map_err(|e| io_err_to_linux(&e))?
        };

        // Symlink containment check. Under `SymlinkPolicy::Deny` and
        // `::Opaque` (default), canonicalize the path and verify it
        // lives under our mount root. Protects against a hostile guest
        // planting symlinks like `etc -> /etc` in their mount and
        // reading host secrets. `Follow` skips the check.
        //
        // We only need to canonicalize when a symlink is involved AND
        // its target is reachable. For a broken symlink (target
        // doesn't exist), the symlink itself can't be used for
        // anything — `open()` through it would fail at the guest's
        // own resolution step — so we let the lookup succeed and rely
        // on the open-time check (which goes through our own
        // host_path_of + open(O_NOFOLLOW) path) to catch escapes.
        if self.symlinks != SymlinkPolicy::Follow && md.file_type().is_symlink() {
            // canonicalize follows symlinks; if it succeeds, verify
            // containment. If it errors (e.g. broken symlink), allow
            // the lookup — readlink will still return the target
            // bytes verbatim (which is POSIX symlink semantics), and
            // any actual use of the symlink to access data goes
            // through our O_NOFOLLOW-guarded open path.
            if let Ok(canonical) = std::fs::canonicalize(&path) {
                if !canonical.starts_with(&self.root) {
                    return Err(EACCES);
                }
            }
        }

        let mut st = self.st.lock().unwrap();
        // Reuse existing nodeid if we've looked this child up before.
        let key = (parent, name.as_bytes().to_vec());
        let nodeid = match st.children.get(&key) {
            Some(&id) => id,
            None => {
                let id = st.next_nodeid;
                st.next_nodeid += 1;
                st.inodes.insert(
                    id,
                    InodeInfo {
                        host_path: path.clone(),
                        kind: kind_from_meta(&md),
                    },
                );
                st.children.insert(key, id);
                id
            }
        };
        let attr = attr_from_meta(nodeid, &md);
        Ok(Entry {
            nodeid,
            generation: 0,
            attr,
            entry_valid: ENTRY_VALID_SECS,
            attr_valid: ATTR_VALID_SECS,
        })
    }

    fn forget(&self, _nodeid: u64, _nlookup: u64) {
        // We retain inode entries indefinitely for path stability.
        // Real production would reference-count and prune. Tests
        // don't depend on prune so we no-op.
    }

    fn getattr(&self, nodeid: u64, fh: Option<u64>) -> Result<Attr, Errno> {
        // If the caller passed a valid fh AND that fh names a
        // `.aotcache` write that's still mid-publish, stat via fstat
        // on the partial fd instead of the final path. The final
        // path doesn't exist on disk yet (we wrote to a `.partial`
        // sibling — see `fn create`), so a path-based stat would
        // return ENOENT and break the writer's own fstat() calls.
        if let Some(fh) = fh {
            let mut st = self.st.lock().unwrap();
            if st.partial_writes.contains_key(&fh) {
                // Lazy-resolve in case the handle survived a
                // snapshot/restore — see HandleEntry. For a partial-
                // publish fh this shouldn't happen in practice (we
                // don't persist partial_writes), but checking the
                // contains_key first means the lookup only matters
                // for live partial-publish state.
                let raw = Self::handle_raw_fd_locked(&mut st, fh)?;
                drop(st);
                let mut st_buf = std::mem::MaybeUninit::<libc::stat>::uninit();
                // SAFETY: raw is a valid fd from an OwnedFd; st_buf
                // has the right size/alignment for libc::stat.
                let rc = unsafe { libc::fstat(raw, st_buf.as_mut_ptr()) };
                if rc != 0 {
                    return Err(errno_now());
                }
                let s = unsafe { st_buf.assume_init() };
                return Ok(attr_from_stat(nodeid, &s));
            }
        }
        let path = self.host_path_of(nodeid)?;
        // Mirror lookup's policy-aware metadata fetch (see lookup for
        // rationale). For non-symlink inodes the two are equivalent;
        // the distinction only matters when the inode IS a symlink.
        let md = if self.symlinks == SymlinkPolicy::Follow {
            std::fs::metadata(&path)
                .or_else(|_| std::fs::symlink_metadata(&path))
                .map_err(|e| io_err_to_linux(&e))?
        } else {
            std::fs::symlink_metadata(&path).map_err(|e| io_err_to_linux(&e))?
        };
        Ok(attr_from_meta(nodeid, &md))
    }

    fn open(&self, nodeid: u64, flags: u32) -> Result<u64, Errno> {
        let path = self.host_path_of(nodeid)?;
        if let Kind::Dir = self.kind_of(nodeid)? {
            return Err(EISDIR);
        }
        // Best-effort: register a kqueue watch on this inode so host
        // changes propagate to the guest as FUSE_NOTIFY_INVAL_INODE.
        self.watch_inode(nodeid, &path);

        // Atomic-publish on re-open with O_TRUNC. rosettad's write
        // path is `openat(path, O_RDWR|O_CREAT|O_TRUNC, mode)` — if
        // the cache file from a prior successful translation exists,
        // the kernel routes this to FUSE_OPEN with O_TRUNC (the
        // FUSE_CREATE path is only hit when the final file is
        // missing). Truncating the existing valid file in-place and
        // then aborting the rewrite leaves a 0-byte cache that
        // breaks every subsequent rosetta interpreter load (see
        // `fn create` for the full failure-mode rundown).
        //
        // Defense: when O_TRUNC fires on a `<x>.aotcache`, treat it
        // the same as CREATE — redirect writes to a `.partial`
        // sibling, leave the existing valid file intact in the
        // meantime, and `rename()` only on a non-empty close. If
        // the rewrite is aborted, the original cache survives. If
        // it succeeds, the rename atomically swaps in the new
        // content. macOS oahd does this same dance with
        // `.aot.in_progress` for the same reason — protect a
        // previously-working cache entry from being clobbered by
        // a torn rewrite.
        let want_trunc = (flags as i32 & libc::O_TRUNC) != 0;
        let is_aotcache_path = path.extension().and_then(|s| s.to_str()) == Some("aotcache");
        if want_trunc && is_aotcache_path {
            // Get the parent + filename for the partial path.
            let (parent_path, name) = match (path.parent(), path.file_name()) {
                (Some(p), Some(n)) => (p, n),
                _ => return Err(EINVAL),
            };
            let mut bytes = Vec::with_capacity(name.as_bytes().len() + 10);
            bytes.push(b'.');
            bytes.extend_from_slice(name.as_bytes());
            bytes.extend_from_slice(b".partial");
            let partial_path = parent_path.join(OsStr::from_bytes(&bytes));
            let partial_c =
                CString::new(partial_path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
            let access = flags as i32 & libc::O_ACCMODE;
            // O_CREAT|O_EXCL — same retry-on-EEXIST treatment as in
            // `fn create`. A stale `.partial` from a prior crashed
            // write would block this open until we unlink and retry.
            let mode: libc::c_uint = 0o644;
            let mut fd = unsafe {
                libc::open(
                    partial_c.as_ptr(),
                    access | libc::O_CREAT | libc::O_EXCL,
                    mode,
                )
            };
            if fd < 0 && errno_now() == EEXIST_ERRNO {
                unsafe { libc::unlink(partial_c.as_ptr()) };
                fd = unsafe {
                    libc::open(
                        partial_c.as_ptr(),
                        access | libc::O_CREAT | libc::O_EXCL,
                        mode,
                    )
                };
            }
            if fd < 0 {
                return Err(errno_now());
            }
            let owned = unsafe { OwnedFd::from_raw_fd(fd) };
            let mut st = self.st.lock().unwrap();
            let fh = st.next_fh;
            st.next_fh += 1;
            // Capture the flags actually passed to open() so a
            // post-snapshot lazy reopen lands on the SAME partial
            // path. (In practice, partial_writes isn't persisted,
            // so this fh won't survive — but the HandleEntry
            // contract is "flags == what open() saw", so record it.)
            let aotcache_flags = (access | libc::O_CREAT | libc::O_EXCL) as u32;
            st.handles.insert(
                fh,
                HandleEntry {
                    fd: Some(owned),
                    nodeid,
                    flags: aotcache_flags,
                    dir_snapshot: None,
                },
            );
            // Find the (parent_nodeid, name) child mapping so the
            // `release` path can refresh the children table after
            // rename. The find is O(N) but child counts on cache
            // dirs are tiny.
            let mut parent_nodeid = 0u64;
            let mut child_name = Vec::new();
            for (&(pn, ref cn), &nid) in &st.children {
                if nid == nodeid {
                    parent_nodeid = pn;
                    child_name = cn.clone();
                    break;
                }
            }
            st.partial_writes.insert(
                fh,
                PartialWrite {
                    final_path: path.clone(),
                    partial_path,
                    parent_nodeid,
                    child_name,
                },
            );
            return Ok(fh);
        }

        let c = CString::new(path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
        // Mask out CREAT/EXCL: virtio-fs always issues a LOOKUP first;
        // OPEN should not create. We honor RDONLY/RDWR/WRONLY + DIRECT.
        let access = flags as i32 & libc::O_ACCMODE;
        let fd = unsafe { libc::open(c.as_ptr(), access) };
        if fd < 0 {
            return Err(errno_now());
        }
        // SAFETY: open returned a valid fd above.
        let owned = unsafe { OwnedFd::from_raw_fd(fd) };
        let mut st = self.st.lock().unwrap();
        let fh = st.next_fh;
        st.next_fh += 1;
        st.handles.insert(
            fh,
            HandleEntry {
                fd: Some(owned),
                nodeid,
                flags: access as u32,
                dir_snapshot: None,
            },
        );
        Ok(fh)
    }

    fn read(&self, _nodeid: u64, fh: u64, offset: u64, size: u32) -> Result<Vec<u8>, Errno> {
        let mut st = self.st.lock().unwrap();
        // Lazy-reopen on first I/O post-restore. See HandleEntry.
        let raw = Self::handle_raw_fd_locked(&mut st, fh)?;
        drop(st);
        let mut buf = vec![0u8; size as usize];
        let n = unsafe {
            libc::pread(
                raw,
                buf.as_mut_ptr() as *mut _,
                buf.len(),
                offset as libc::off_t,
            )
        };
        if n < 0 {
            return Err(errno_now());
        }
        buf.truncate(n as usize);
        Ok(buf)
    }

    fn release(&self, _nodeid: u64, fh: u64) -> Result<(), Errno> {
        // Pull the fd AND any partial-publish state out of state in
        // one lock-and-release. Holding the lock across the rename or
        // unlink syscall would block all FUSE traffic on this device
        // for the duration; pulling first lets the syscall happen
        // unlocked.
        //
        // The HandleEntry may have `fd: None` if the handle was
        // restored from a snapshot and never had I/O issued against
        // it post-restore. In that case there's nothing to close —
        // just drop the metadata entry. For partial-publish handles
        // this can't normally happen (we don't persist
        // `partial_writes`), but the entry-removal contract is
        // "release always frees the slot, with or without a live
        // fd" so the lookup still works.
        let (entry, partial) = {
            let mut st = self.st.lock().unwrap();
            let entry = st.handles.remove(&fh).ok_or(EBADF)?;
            let partial = st.partial_writes.remove(&fh);
            (entry, partial)
        };

        // Non-aotcache files: trivial close (drop the fd, which is a
        // no-op if fd was None — the OwnedFd's Drop closes only on Some).
        let Some(info) = partial else {
            drop(entry.fd);
            return Ok(());
        };

        // `.aotcache` partial-publish path. Atomic-rename completes
        // Apple's macOS oahd `.aot.in_progress`→rename protocol that
        // we're emulating here at the FUSE layer; see the comment in
        // `fn create` for why this is necessary on Linux.
        //
        // Steps:
        //   1. fstat the writer's fd to see if anything was written.
        //   2. If size > 0: renameat(partial → final). The cache
        //      entry appears atomically with content; concurrent
        //      lookups of the final name see ENOENT or the complete
        //      file, never an empty / partial one.
        //   3. If size == 0: unlink the partial. The aborted
        //      translation leaves nothing on disk; rosettad's next
        //      attempt for the same build-id sees "no cache" and
        //      re-translates instead of loading garbage.
        //   4. On rename failure (ENOSPC, cross-device, etc.):
        //      unlink the partial anyway and surface the error to
        //      the guest. Better to lose the cache entry than to
        //      keep a stray partial that will eventually break the
        //      interpreter.
        //
        // All file-state-bearing syscalls (fstat, renameat, unlink)
        // happen with the fd still open so the host can resolve
        // the partial path even if the directory is concurrently
        // mutated. The fd is dropped at the end.
        //
        // If `fd` is None (partial-write fh that survived a snapshot
        // somehow — defensive only since we don't persist
        // partial_writes), fall back to fstat-via-path-on-final. The
        // final path won't exist on host yet because the partial
        // hasn't been renamed — treat as size=0 and just unlink the
        // partial.
        let owned = match entry.fd {
            Some(fd) => fd,
            None => {
                // Best-effort: unlink the partial path if it exists,
                // and report success. Nothing was written through
                // this restored fh.
                let partial_c =
                    CString::new(info.partial_path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
                unsafe { libc::unlink(partial_c.as_ptr()) };
                return Ok(());
            }
        };
        let raw = owned.as_raw_fd();
        let mut st_buf = std::mem::MaybeUninit::<libc::stat>::uninit();
        let st_rc = unsafe { libc::fstat(raw, st_buf.as_mut_ptr()) };
        let size = if st_rc == 0 {
            // SAFETY: fstat returned 0, so st_buf is initialized.
            unsafe { st_buf.assume_init().st_size }
        } else {
            // fstat failure is exceptional but not actionable here —
            // treat as "empty / aborted" so we clean up the partial.
            0
        };

        let final_c = match CString::new(info.final_path.as_os_str().as_bytes()) {
            Ok(s) => s,
            Err(_) => {
                drop(owned);
                return Err(EINVAL);
            }
        };
        let partial_c = match CString::new(info.partial_path.as_os_str().as_bytes()) {
            Ok(s) => s,
            Err(_) => {
                drop(owned);
                return Err(EINVAL);
            }
        };

        let result = if size > 0 {
            // POSIX rename(2) atomically replaces the destination if
            // it exists; that's what we want — last successful write
            // wins. rosettad is content-addressable by build-id so
            // any two writers producing the same final name are
            // producing equivalent content.
            let rc = unsafe { libc::rename(partial_c.as_ptr(), final_c.as_ptr()) };
            if rc != 0 {
                let e = errno_now();
                // Rename failed. Don't leave the partial sitting on
                // disk where the acquire-time scrub would have to
                // mop it up later — best-effort unlink now.
                unsafe { libc::unlink(partial_c.as_ptr()) };
                Err(e)
            } else {
                Ok(())
            }
        } else {
            unsafe { libc::unlink(partial_c.as_ptr()) };
            Ok(())
        };

        drop(owned);
        result
    }

    fn write(&self, _nodeid: u64, fh: u64, offset: u64, data: &[u8]) -> Result<u32, Errno> {
        let mut st = self.st.lock().unwrap();
        // Lazy-reopen on first I/O post-restore. See HandleEntry.
        let raw = Self::handle_raw_fd_locked(&mut st, fh)?;
        drop(st);
        let n = unsafe {
            libc::pwrite(
                raw,
                data.as_ptr() as *const _,
                data.len(),
                offset as libc::off_t,
            )
        };
        if n < 0 {
            return Err(errno_now());
        }
        Ok(n as u32)
    }

    fn fsync(&self, _nodeid: u64, fh: u64, _datasync: bool) -> Result<(), Errno> {
        let mut st = self.st.lock().unwrap();
        // Lazy-reopen on first I/O post-restore. See HandleEntry.
        let raw = Self::handle_raw_fd_locked(&mut st, fh)?;
        drop(st);
        // Durability semantics: the guest's libc `fsync(2)` promises
        // "data is on durable storage" — Linux's contract. macOS's
        // plain `fsync(2)` does NOT honor that contract; it only
        // flushes from the kernel's page cache into the disk's
        // hardware buffer. If the drive's volatile cache is hit by a
        // power loss between the `fsync` return and the cache flush,
        // the data is gone — opposite of what the guest's caller
        // (SQLite WAL, npm install's atomic-rename, postgres WAL,
        // git's `commit -m`) expects.
        //
        // macOS exposes the Linux-equivalent guarantee through
        // `fcntl(F_FULLFSYNC)`, which forces the drive controller to
        // commit its cache to the medium before returning. That's
        // the right behaviour to surface to the guest.
        //
        // Cost: F_FULLFSYNC is genuinely slower than plain fsync
        // (~10-100×). For our typical bake-then-pool flow that's
        // acceptable — fsync is rare on the hot path. For
        // workloads where it isn't, users can opt out via
        // `SUPERMACHINE_FSYNC_WEAK=1`, falling back to plain fsync.
        //
        // We don't differentiate `datasync` (FUSE_FSYNCDIR's data-
        // only mode). macOS doesn't offer a data-only barrier; on
        // Linux the difference between fsync/fdatasync is a metadata
        // optimisation that we'd lose to the host syscall layer
        // anyway. Always use the strongest form.
        let weak_fsync = std::env::var_os("SUPERMACHINE_FSYNC_WEAK").is_some();
        let rc = if weak_fsync {
            unsafe { libc::fsync(raw) }
        } else {
            // Strongest durability the host offers. On macOS plain `fsync`
            // does NOT flush the drive's write cache — `fcntl(F_FULLFSYNC)`
            // does (falling back to `fsync` on ENOTSUP/EINVAL for FSes
            // without barrier control). On Linux `fsync` already issues the
            // device cache flush, so it IS the strong form.
            #[cfg(target_os = "macos")]
            {
                let rc = unsafe { libc::fcntl(raw, libc::F_FULLFSYNC) };
                if rc != 0 {
                    let e = std::io::Error::last_os_error();
                    if e.raw_os_error() == Some(libc::ENOTSUP)
                        || e.raw_os_error() == Some(libc::EINVAL)
                    {
                        unsafe { libc::fsync(raw) }
                    } else {
                        rc
                    }
                } else {
                    0
                }
            }
            #[cfg(not(target_os = "macos"))]
            {
                unsafe { libc::fsync(raw) }
            }
        };
        if rc != 0 {
            return Err(errno_now());
        }
        Ok(())
    }

    fn opendir(&self, nodeid: u64, _flags: u32) -> Result<u64, Errno> {
        let path = self.host_path_of(nodeid)?;
        let c = CString::new(path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
        let dir_flags = libc::O_RDONLY | libc::O_DIRECTORY;
        let fd = unsafe { libc::open(c.as_ptr(), dir_flags) };
        if fd < 0 {
            return Err(errno_now());
        }
        let owned = unsafe { OwnedFd::from_raw_fd(fd) };
        let mut st = self.st.lock().unwrap();
        let fh = st.next_fh;
        st.next_fh += 1;
        st.handles.insert(
            fh,
            HandleEntry {
                fd: Some(owned),
                nodeid,
                flags: dir_flags as u32,
                // dir_snapshot stays None — populated on first
                // FUSE_READDIR (lazy: avoids a host read_dir for
                // opendirs that never enumerate, e.g. shell `cd`
                // or dirfd-relative path walks).
                dir_snapshot: None,
            },
        );
        Ok(fh)
    }

    fn readdir(
        &self,
        nodeid: u64,
        fh: u64,
        offset: u64,
        _size: u32,
    ) -> Result<Vec<DirEntry>, Errno> {
        // Fast path: if we already snapshotted the entries on a
        // prior call against this fh, slice from the snapshot using
        // `offset` as a position cookie.
        //
        // Why this matters (the rm -rf-over-virtio-fs bug):
        // FUSE's READDIR contract is iterator-like. The kernel
        // calls READDIR repeatedly, passing the LAST entry's `off`
        // as the next call's `offset`. The cookie is opaque to the
        // kernel — we define what it means. Pre-fix we re-ran
        // `std::fs::read_dir(host_path)` on every call and used
        // the iteration index as the cookie; that breaks badly when
        // the directory is concurrently mutated by the same guest:
        //
        //   1. Guest: opendir(fh)
        //   2. Guest: getdents64 → host serves entries 0..49
        //      (host iter pos 0..49 of e.g. 1000-entry dir),
        //      cookie 50 returned
        //   3. Guest: unlink(entry_0), unlink(entry_1), ..., unlink(entry_49)
        //   4. Guest: getdents64(offset=50) → host re-runs read_dir
        //      on a now-950-entry dir, SKIPS the first 50, returns
        //      entries that USED TO BE at positions 100..149.
        //      Entries 50..99 of the original list are NEVER
        //      returned to the kernel — rm -rf doesn't know they
        //      exist and rmdir() of the parent later returns
        //      ENOTEMPTY.
        //
        // Fix: snapshot the entries on the FIRST readdir call for
        // the fh, cache them in HandleEntry::dir_snapshot, and
        // serve every subsequent call from the cache. The cookie
        // becomes a stable index into the snapshot — concurrent
        // unlinks don't shift it. Trade-off documented on
        // HandleEntry::dir_snapshot: entries created AFTER the
        // first readdir won't appear, which is POSIX-permitted
        // (Linux's getdents64(2) explicitly says "the behaviour
        // of readdir when files are added or removed from the
        // directory after the call to opendir is undefined").
        {
            let st = self.st.lock().unwrap();
            if let Some(entry) = st.handles.get(&fh) {
                if let Some(snap) = entry.dir_snapshot.as_ref() {
                    let start = (offset as usize).min(snap.len());
                    return Ok(snap[start..].to_vec());
                }
            }
        }

        // Slow path: first readdir on this fh. Capture the full
        // directory listing, stash it on the handle, then serve.
        let path = self.host_path_of(nodeid)?;
        let rd = std::fs::read_dir(&path).map_err(|e| io_err_to_linux(&e))?;
        let mut entries: Vec<DirEntry> = Vec::new();
        for entry_res in rd {
            let entry = match entry_res {
                Ok(e) => e,
                Err(_) => continue,
            };
            // Hide in-flight `.aotcache` writes from the guest's
            // directory view. They live on disk as
            // `.<name>.aotcache.partial` (leading dot + .partial
            // suffix) — see `fn create` for the atomic-publish
            // mechanism. Filtering at READDIR keeps the guest's
            // view consistent with what it can resolve by name:
            // LOOKUP of `<x>.aotcache` returns ENOENT during the
            // write window (the final path doesn't exist yet), so
            // showing the partial in `readdir` would just give the
            // guest a name it can't open.
            let name_bytes_owned = entry.file_name().as_bytes().to_vec();
            if name_bytes_owned.starts_with(b".")
                && name_bytes_owned.ends_with(b".aotcache.partial")
            {
                continue;
            }
            let typ = match entry.file_type() {
                Ok(t) if t.is_dir() => DT_DIR,
                Ok(t) if t.is_file() => DT_REG,
                Ok(t) if t.is_symlink() => DT_LNK,
                Ok(t) => match t {
                    t if t.is_block_device() => DT_BLK,
                    t if t.is_char_device() => DT_CHR,
                    t if t.is_fifo() => DT_FIFO,
                    t if t.is_socket() => DT_SOCK,
                    _ => DT_UNKNOWN,
                },
                Err(_) => DT_UNKNOWN,
            };
            // Inode number: we can't allocate a nodeid until LOOKUP
            // runs (the guest will issue a LOOKUP for any entry it
            // wants to use). Send the host's st_ino so directory
            // listings show stable values; the guest only relies on
            // the name+type for the readdir surface.
            let ino = entry.metadata().map(|m| m.ino()).unwrap_or(0);
            entries.push(DirEntry {
                ino,
                name: name_bytes_owned,
                typ,
            });
        }
        // Cache for subsequent readdir calls on the same fh.
        // Guard the cache install behind a handle lookup — if the
        // handle was released between the slow path's start and
        // here (caller error, but possible) we just don't cache.
        // A subsequent readdir on a now-stale fh would return
        // EBADF in any sane caller.
        let cached = Arc::new(entries);
        {
            let mut st = self.st.lock().unwrap();
            if let Some(h) = st.handles.get_mut(&fh) {
                // Only install if still empty. If a concurrent
                // readdir on the same fh raced and installed
                // already (the std::Mutex serialises but we
                // dropped it above to do the host read_dir), we
                // honour the first install — both snapshots are
                // valid; using the same one across all callers
                // makes the cookie semantics deterministic.
                if h.dir_snapshot.is_none() {
                    h.dir_snapshot = Some(Arc::clone(&cached));
                }
            }
        }
        let start = (offset as usize).min(cached.len());
        Ok(cached[start..].to_vec())
    }

    fn releasedir(&self, _nodeid: u64, fh: u64) -> Result<(), Errno> {
        let mut st = self.st.lock().unwrap();
        st.handles.remove(&fh).ok_or(EBADF).map(|_| ())
    }

    // `f_bsize` is u32 on macOS but i64 (c_long) on Linux — the cast narrows on
    // Linux and is a no-op on macOS.
    #[allow(clippy::unnecessary_cast)]
    fn statfs(&self, nodeid: u64) -> Result<StatFs, Errno> {
        let path = self.host_path_of(nodeid)?;
        let c = CString::new(path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
        let mut s: libc::statfs = unsafe { std::mem::zeroed() };
        if unsafe { libc::statfs(c.as_ptr(), &mut s) } < 0 {
            return Err(errno_now());
        }
        Ok(StatFs {
            blocks: s.f_blocks,
            bfree: s.f_bfree,
            bavail: s.f_bavail,
            files: s.f_files,
            ffree: s.f_ffree,
            // f_bsize is u32 on macOS but i64 (c_long) on Linux — narrow it.
            bsize: s.f_bsize as u32,
            namelen: 255,
            frsize: s.f_bsize as u32,
        })
    }

    fn create(
        &self,
        parent: u64,
        name: &OsStr,
        mode: u32,
        flags: u32,
    ) -> Result<(crate::fuse::backend::Entry, u64), Errno> {
        name_safe(name)?;
        let parent_path = self.host_path_of(parent)?;
        let final_path = parent_path.join(name);

        // Atomic-publish for `.aotcache` files. Apple's macOS `oahd`
        // protects its translation cache by writing to a
        // `<x>.aot.in_progress` file then `rename()`ing it to
        // `<x>.aot` — guaranteeing the final name only ever names a
        // fully-written cache entry. The Linux `rosettad` Apple ships
        // for Virtualization.framework does NOT do this: it
        // `openat(..., O_CREAT|O_TRUNC)` the final path and writes in
        // place (verified by disassembly — no rename/linkat/fsync
        // syscalls anywhere in the binary). Any abort / OOM / SIGKILL
        // between create and the final pwrite leaves a 0-byte or
        // partial file. A subsequent rosetta interpreter run for the
        // same build-id loads that file, pread returns 0 bytes, and
        // the interpreter either trips its
        // `aot_hdr->segment_count <= kMaxSegments` assertion
        // (ImageInfo.cpp:71 — stack garbage at offset 8 parsed as
        // segment_count, almost always > kMaxSegments=8 → SIGTRAP) or
        // SIGBUSes on mmap-past-EOF. Either kills the workload child
        // immediately on amd64 exec.
        //
        // Reverse-engineered context:
        //   - macOS protocol: https://ffri.github.io/ProjectChampollion/part1/
        //   - Linux deficiency: confirmed by `llvm-objdump -d` of
        //     `/Library/Apple/usr/libexec/oah/RosettaLinux/rosettad`
        //   - Bug structurally impossible in Apple's `container` CLI
        //     because they don't use persistent host-side caching;
        //     it's our 0.7.23+ design choice (cross-VM-lifetime cache)
        //     that exposes it.
        //
        // Recipe (matches macOS oahd's pattern at the FUSE layer):
        //   1. Guest's CREATE of `<name>.aotcache` lands here.
        //   2. We open `.<name>.partial` (dotfile, sibling of final)
        //      with O_CREAT|O_EXCL instead, and stash a `PartialWrite`
        //      against the fh.
        //   3. The guest writes through the same fh — the kernel
        //      doesn't know about the rename, so it just routes
        //      pwrite-class calls to the same handle.
        //   4. On FUSE_RELEASE (close), if the file has content,
        //      `renameat()` partial → final. The cache entry appears
        //      atomically with content. If size == 0, unlink the
        //      partial; nothing ever named `<x>.aotcache` exists.
        //   5. Concurrent lookups of `<x>.aotcache` find nothing
        //      (the host doesn't have a final file yet) → guest sees
        //      "not cached" → rosettad re-translates instead of
        //      trying to load garbage.
        //   6. `.partial` files are filtered from READDIR so the
        //      guest's directory view stays consistent with what it
        //      can resolve by name.
        //
        // The `inodes` entry for the new nodeid points at the FINAL
        // path even while the write is in flight — so once the
        // rename completes, no fix-up is needed for the host path
        // mapping. GETATTR on the writer's own fh redirects to the
        // partial via the fh→PartialWrite lookup; see `fn getattr`.
        let is_aotcache = name.as_bytes().ends_with(b".aotcache");
        let (open_path, partial_info) = if is_aotcache {
            // `.<name>.partial`. The leading dot is what hides it
            // from `READDIR` filtering (we check for `.partial`
            // suffix) and from any guest tool that walks the cache
            // dir.
            let mut bytes = Vec::with_capacity(name.as_bytes().len() + 10);
            bytes.push(b'.');
            bytes.extend_from_slice(name.as_bytes());
            bytes.extend_from_slice(b".partial");
            let partial_name = OsStr::from_bytes(&bytes);
            let partial_path = parent_path.join(partial_name);
            let info = PartialWrite {
                final_path: final_path.clone(),
                partial_path: partial_path.clone(),
                parent_nodeid: parent,
                child_name: name.as_bytes().to_vec(),
            };
            (partial_path, Some(info))
        } else {
            (final_path.clone(), None)
        };

        let c = CString::new(open_path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
        let access = flags as i32 & libc::O_ACCMODE;
        // If we're publishing via partial, force O_CREAT|O_EXCL even
        // if a stale partial exists from a prior crashed write —
        // `O_EXCL` failure means another writer is mid-flight for
        // the same build-id; that's a real concurrency case we
        // surface as EEXIST so rosettad's existing retry path can
        // handle it.
        let fd = unsafe {
            libc::open(
                c.as_ptr(),
                access | libc::O_CREAT | libc::O_EXCL,
                mode as libc::c_uint,
            )
        };
        if fd < 0 {
            let e = errno_now();
            // For the partial path, a prior-crash leftover would
            // trigger EEXIST on every retry until acquire-time
            // cleanup runs. Treat EEXIST as "stale orphan from a
            // previous lifetime": unlink and retry once. This is
            // safe because the partial is, by construction,
            // pid-tagless and only ever written by the in-flight
            // CREATE — nothing else should own it.
            if partial_info.is_some() && e == EEXIST_ERRNO {
                unsafe { libc::unlink(c.as_ptr()) };
                let fd2 = unsafe {
                    libc::open(
                        c.as_ptr(),
                        access | libc::O_CREAT | libc::O_EXCL,
                        mode as libc::c_uint,
                    )
                };
                if fd2 < 0 {
                    return Err(errno_now());
                }
                return self.create_finalize(
                    parent,
                    name,
                    &final_path,
                    fd2,
                    partial_info,
                    (access | libc::O_CREAT | libc::O_EXCL) as u32,
                );
            }
            return Err(e);
        }
        self.create_finalize(
            parent,
            name,
            &final_path,
            fd,
            partial_info,
            (access | libc::O_CREAT | libc::O_EXCL) as u32,
        )
    }

    fn mkdir(
        &self,
        parent: u64,
        name: &OsStr,
        mode: u32,
    ) -> Result<crate::fuse::backend::Entry, Errno> {
        name_safe(name)?;
        let parent_path = self.host_path_of(parent)?;
        let full = parent_path.join(name);
        let c = CString::new(full.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
        let rc = unsafe { libc::mkdir(c.as_ptr(), mode as libc::mode_t) };
        if rc != 0 {
            return Err(errno_now());
        }
        // Same TOCTOU concern as create() above, but mkdir doesn't
        // return an fd. Reach for the parent's dirfd + fstatat with
        // AT_SYMLINK_NOFOLLOW so any concurrent symlink-replace on
        // the just-created directory name still yields the actual
        // directory's stat (or a clean ENOENT if it raced through
        // mkdir + rmdir in the gap). Avoids the bare path-based
        // stat() race that's been seen as transient EIO under load.
        let md = match open_dirfd(&parent_path) {
            Ok(parent_dirfd) => {
                let name_c = CString::new(name.as_bytes()).map_err(|_| EINVAL)?;
                let mut sb: libc::stat = unsafe { std::mem::zeroed() };
                let rc = unsafe {
                    libc::fstatat(
                        parent_dirfd.as_raw_fd(),
                        name_c.as_ptr(),
                        &mut sb,
                        libc::AT_SYMLINK_NOFOLLOW,
                    )
                };
                if rc != 0 {
                    return Err(errno_now());
                }
                // Convert libc::stat → std::fs::Metadata for kind/attr
                // helpers. Going via path-based symlink_metadata after
                // the AT_SYMLINK_NOFOLLOW probe keeps the existing
                // conversion code working — the probe above is the
                // race-guard; the conversion now CAN race but at worst
                // returns ENOENT (we propagate), never EIO from a stale
                // stat.
                std::fs::symlink_metadata(&full).map_err(|e| io_err_to_linux(&e))?
            }
            Err(_) => std::fs::symlink_metadata(&full).map_err(|e| io_err_to_linux(&e))?,
        };
        let mut st = self.st.lock().unwrap();
        let nodeid = st.next_nodeid;
        st.next_nodeid += 1;
        st.inodes.insert(
            nodeid,
            InodeInfo {
                host_path: full,
                kind: Kind::Dir,
            },
        );
        st.children
            .insert((parent, name.as_bytes().to_vec()), nodeid);
        Ok(crate::fuse::backend::Entry {
            nodeid,
            generation: 0,
            attr: attr_from_meta(nodeid, &md),
            entry_valid: ENTRY_VALID_SECS,
            attr_valid: ATTR_VALID_SECS,
        })
    }

    fn unlink(&self, parent: u64, name: &OsStr) -> Result<(), Errno> {
        name_safe(name)?;
        let parent_path = self.host_path_of(parent)?;
        let full = parent_path.join(name);
        let c = CString::new(full.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
        let rc = unsafe { libc::unlink(c.as_ptr()) };
        if rc != 0 {
            return Err(errno_now());
        }
        let mut st = self.st.lock().unwrap();
        st.children.remove(&(parent, name.as_bytes().to_vec()));
        Ok(())
    }

    fn rmdir(&self, parent: u64, name: &OsStr) -> Result<(), Errno> {
        name_safe(name)?;
        let parent_path = self.host_path_of(parent)?;
        let full = parent_path.join(name);
        let c = CString::new(full.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
        let rc = unsafe { libc::rmdir(c.as_ptr()) };
        if rc != 0 {
            return Err(errno_now());
        }
        let mut st = self.st.lock().unwrap();
        st.children.remove(&(parent, name.as_bytes().to_vec()));
        Ok(())
    }

    fn symlink(
        &self,
        parent: u64,
        name: &OsStr,
        target: &OsStr,
    ) -> Result<crate::fuse::backend::Entry, Errno> {
        // Deny mode: refuse outright with EPERM. POSIX symlink(2)
        // returns EPERM when the filesystem doesn't support symlinks,
        // which is the closest match to "policy denies symlinks here".
        if self.symlinks == SymlinkPolicy::Deny {
            return Err(EPERM);
        }
        name_safe(name)?;
        // The `target` is opaque bytes per POSIX. Only constraint:
        // CString rejects embedded NULs (the kernel does too via
        // symlinkat). Targets containing `..` or absolute paths are
        // legal — the host doesn't resolve them, the guest's kernel
        // does at lookup time.
        if target.as_bytes().is_empty() || target.as_bytes().contains(&0u8) {
            return Err(EINVAL);
        }
        let parent_path = self.host_path_of(parent)?;
        // Open parent as O_PATH|O_DIRECTORY|O_NOFOLLOW dirfd so we use
        // symlinkat — no race against a parent rename. macOS supports
        // O_NOFOLLOW + O_DIRECTORY. (Linux additionally needs O_PATH
        // which we'll add behind a cfg when we port; macOS doesn't
        // have O_PATH but accepts plain O_DIRECTORY|O_RDONLY for an
        // *at-only dirfd.)
        let parent_dirfd = open_dirfd(&parent_path)?;
        let target_c = CString::new(target.as_bytes()).map_err(|_| EINVAL)?;
        let name_c = CString::new(name.as_bytes()).map_err(|_| EINVAL)?;
        let rc = unsafe {
            libc::symlinkat(target_c.as_ptr(), parent_dirfd.as_raw_fd(), name_c.as_ptr())
        };
        if rc != 0 {
            return Err(errno_now());
        }
        // Stat the new symlink itself (do NOT follow).
        let mut stb: libc::stat = unsafe { std::mem::zeroed() };
        let rc = unsafe {
            libc::fstatat(
                parent_dirfd.as_raw_fd(),
                name_c.as_ptr(),
                &mut stb,
                libc::AT_SYMLINK_NOFOLLOW,
            )
        };
        if rc != 0 {
            return Err(errno_now());
        }
        let full = parent_path.join(name);
        let mut st = self.st.lock().unwrap();
        let nodeid = st.next_nodeid;
        st.next_nodeid += 1;
        st.inodes.insert(
            nodeid,
            InodeInfo {
                host_path: full,
                kind: Kind::Symlink,
            },
        );
        st.children
            .insert((parent, name.as_bytes().to_vec()), nodeid);
        Ok(crate::fuse::backend::Entry {
            nodeid,
            generation: 0,
            attr: attr_from_stat(nodeid, &stb),
            entry_valid: ENTRY_VALID_SECS,
            attr_valid: ATTR_VALID_SECS,
        })
    }

    fn readlink(&self, nodeid: u64) -> Result<Vec<u8>, Errno> {
        let path = self.host_path_of(nodeid)?;
        let c = CString::new(path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
        // PATH_MAX is 1024 on Darwin and 4096 on Linux. Use 4096 as a
        // safe upper bound for both; if the link is longer the
        // truncation is silent per POSIX (readlink returns the
        // truncated length and we honor it).
        let mut buf = vec![0u8; 4096];
        let n = unsafe { libc::readlink(c.as_ptr(), buf.as_mut_ptr() as *mut _, buf.len()) };
        if n < 0 {
            return Err(errno_now());
        }
        buf.truncate(n as usize);
        Ok(buf)
    }

    // `st_mode` is u16 on macOS, u32 on Linux — the casts below are needed on macOS.
    #[allow(clippy::unnecessary_cast)]
    fn link(
        &self,
        nodeid: u64,
        new_parent: u64,
        new_name: &OsStr,
    ) -> Result<crate::fuse::backend::Entry, Errno> {
        // Symmetric with symlink under Deny — "no metadata surprises"
        // means no new hard links either. Hard links can't ESCAPE the
        // mount (link(2) returns EXDEV across filesystems and our
        // mount is one host FS), but `Deny` is about predictability
        // not just escape resistance.
        if self.symlinks == SymlinkPolicy::Deny {
            return Err(EPERM);
        }
        name_safe(new_name)?;
        let src_path = self.host_path_of(nodeid)?;
        let new_parent_path = self.host_path_of(new_parent)?;
        let dst_path = new_parent_path.join(new_name);
        let src_c = CString::new(src_path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
        let dst_c = CString::new(dst_path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
        // POSIX leaves link(2)'s symlink-handling implementation-
        // defined. macOS's `link(2)` FOLLOWS symbolic links by default
        // (per Apple's man page); Linux's `link(2)` does NOT follow
        // (since 2.6.18 — the SUSv3 behaviour). Without this branch,
        // a guest that calls `link("/work/sym", "/work/alias")` where
        // sym → target.txt would get a hardlink to target.txt's inode
        // instead of a hardlink to the symlink inode. That diverges
        // from Linux and breaks any tool that relies on link's no-
        // follow semantics (npm's atomic-rename pattern doesn't, but
        // some build systems / git's index logic do).
        //
        // Use linkat() with flags=0 — both macOS and Linux interpret
        // that as "do NOT follow symlinks on the source." (AT_SYMLINK_-
        // FOLLOW is the opt-in for following on Linux; macOS uses the
        // same flag with the same semantics.)
        let rc = unsafe {
            libc::linkat(
                libc::AT_FDCWD,
                src_c.as_ptr(),
                libc::AT_FDCWD,
                dst_c.as_ptr(),
                0, // no AT_SYMLINK_FOLLOW: hardlink the symlink itself if src is one
            )
        };
        if rc != 0 {
            return Err(errno_now());
        }
        // Stat the destination without following — we want the file's
        // own attrs (and a hard link can't be a symlink anyway, but
        // belt-and-braces).
        let dst_c2 = CString::new(dst_path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
        let mut stb: libc::stat = unsafe { std::mem::zeroed() };
        let rc = unsafe { libc::lstat(dst_c2.as_ptr(), &mut stb) };
        if rc != 0 {
            return Err(errno_now());
        }
        let mut st = self.st.lock().unwrap();
        let new_nodeid = st.next_nodeid;
        st.next_nodeid += 1;
        // The new dentry gets a fresh nodeid pointing at the same
        // underlying file. Two nodeids mapping to one host path is
        // legal — open()/stat() on either both hit the same inode.
        let kind = if (stb.st_mode as u32 & S_IFMT) == S_IFDIR {
            Kind::Dir
        } else if (stb.st_mode as u32 & S_IFMT) == S_IFREG {
            Kind::File
        } else if (stb.st_mode as u32 & S_IFMT) == S_IFLNK {
            Kind::Symlink
        } else {
            Kind::Other
        };
        st.inodes.insert(
            new_nodeid,
            InodeInfo {
                host_path: dst_path,
                kind,
            },
        );
        st.children
            .insert((new_parent, new_name.as_bytes().to_vec()), new_nodeid);
        Ok(crate::fuse::backend::Entry {
            nodeid: new_nodeid,
            generation: 0,
            attr: attr_from_stat(new_nodeid, &stb),
            entry_valid: ENTRY_VALID_SECS,
            attr_valid: ATTR_VALID_SECS,
        })
    }

    fn setattr(&self, nodeid: u64, fh: Option<u64>, attr: SetattrIn) -> Result<Attr, Errno> {
        let path = self.host_path_of(nodeid)?;
        // Resolve the open fd once if the kernel passed an FH — lets us
        // ftruncate / fchmod without reopening, which is faster and
        // also safer (no parent-component race). Lazy-reopens the
        // host fd if the handle survived a snapshot/restore — see
        // HandleEntry.
        let fd_raw: Option<libc::c_int> = if let Some(handle) = fh {
            let mut st = self.st.lock().unwrap();
            Some(Self::handle_raw_fd_locked(&mut st, handle)?)
        } else {
            None
        };
        let path_c = CString::new(path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;

        // SIZE — apply first so growth + chmod ordering matches POSIX.
        if attr.valid & FATTR_SIZE != 0 {
            let rc = unsafe {
                if let Some(fd) = fd_raw {
                    libc::ftruncate(fd, attr.size as libc::off_t)
                } else {
                    libc::truncate(path_c.as_ptr(), attr.size as libc::off_t)
                }
            };
            if rc != 0 {
                return Err(errno_now());
            }
        }

        // MODE — chmod via fd if available, else path. The mode bits
        // carried in SetattrIn are the permission bits only (no type
        // bits) per the FUSE spec; we mask defensively.
        if attr.valid & FATTR_MODE != 0 {
            let mode_bits = (attr.mode & 0o7777) as libc::mode_t;
            let rc = unsafe {
                if let Some(fd) = fd_raw {
                    libc::fchmod(fd, mode_bits)
                } else {
                    libc::chmod(path_c.as_ptr(), mode_bits)
                }
            };
            if rc != 0 {
                return Err(errno_now());
            }
        }

        // UID / GID — at most one chown call. lchown so we don't
        // chase symlinks (the symlink itself is the target inode).
        if attr.valid & (FATTR_UID | FATTR_GID) != 0 {
            let uid = if attr.valid & FATTR_UID != 0 {
                attr.uid as libc::uid_t
            } else {
                u32::MAX as libc::uid_t
            };
            let gid = if attr.valid & FATTR_GID != 0 {
                attr.gid as libc::gid_t
            } else {
                u32::MAX as libc::gid_t
            };
            let rc = unsafe {
                if let Some(fd) = fd_raw {
                    libc::fchown(fd, uid, gid)
                } else {
                    libc::lchown(path_c.as_ptr(), uid, gid)
                }
            };
            if rc != 0 {
                return Err(errno_now());
            }
        }

        // ATIME / MTIME via utimensat. UTIME_NOW lets the kernel sentinel
        // "set to current time" pass through (FATTR_ATIME_NOW /
        // FATTR_MTIME_NOW). UTIME_OMIT skips the field.
        let want_times =
            attr.valid & (FATTR_ATIME | FATTR_MTIME | FATTR_ATIME_NOW | FATTR_MTIME_NOW) != 0;
        if want_times {
            let ts_atime = if attr.valid & FATTR_ATIME_NOW != 0 {
                libc::timespec {
                    tv_sec: 0,
                    tv_nsec: libc::UTIME_NOW,
                }
            } else if attr.valid & FATTR_ATIME != 0 {
                libc::timespec {
                    tv_sec: attr.atime as libc::time_t,
                    tv_nsec: attr.atimensec as libc::c_long,
                }
            } else {
                libc::timespec {
                    tv_sec: 0,
                    tv_nsec: libc::UTIME_OMIT,
                }
            };
            let ts_mtime = if attr.valid & FATTR_MTIME_NOW != 0 {
                libc::timespec {
                    tv_sec: 0,
                    tv_nsec: libc::UTIME_NOW,
                }
            } else if attr.valid & FATTR_MTIME != 0 {
                libc::timespec {
                    tv_sec: attr.mtime as libc::time_t,
                    tv_nsec: attr.mtimensec as libc::c_long,
                }
            } else {
                libc::timespec {
                    tv_sec: 0,
                    tv_nsec: libc::UTIME_OMIT,
                }
            };
            let times = [ts_atime, ts_mtime];
            // AT_FDCWD + absolute path is fine; AT_SYMLINK_NOFOLLOW so
            // we touch the inode itself, not the symlink target.
            let rc = unsafe {
                libc::utimensat(
                    libc::AT_FDCWD,
                    path_c.as_ptr(),
                    times.as_ptr(),
                    libc::AT_SYMLINK_NOFOLLOW,
                )
            };
            if rc != 0 {
                return Err(errno_now());
            }
        }

        // Restat to compute the post-change Attr. lstat so we report
        // the symlink itself (parity with `attr_from_stat` callers).
        let mut stb: libc::stat = unsafe { std::mem::zeroed() };
        let rc = unsafe { libc::lstat(path_c.as_ptr(), &mut stb) };
        if rc != 0 {
            return Err(errno_now());
        }
        Ok(attr_from_stat(nodeid, &stb))
    }

    fn rename(
        &self,
        old_parent: u64,
        old_name: &OsStr,
        new_parent: u64,
        new_name: &OsStr,
        flags: u32,
    ) -> Result<(), Errno> {
        name_safe(old_name)?;
        name_safe(new_name)?;
        // We don't currently advertise Rename2 flags through FUSE_INIT,
        // so the kernel shouldn't send non-zero flags. If it does (e.g.
        // a forward-compatible newer kernel) reject loudly rather than
        // silently performing a vanilla rename, which could clobber a
        // file the caller wanted RENAME_NOREPLACE to protect.
        if flags != 0 {
            return Err(EINVAL);
        }
        let old_parent_path = self.host_path_of(old_parent)?;
        let new_parent_path = self.host_path_of(new_parent)?;
        let old_full = old_parent_path.join(old_name);
        let new_full = new_parent_path.join(new_name);

        // renameat with parent dirfds: no race on parent path components
        // (a hostile rename of an ancestor mid-call can't redirect us).
        let old_dirfd = open_dirfd(&old_parent_path)?;
        let new_dirfd = open_dirfd(&new_parent_path)?;
        let old_c = CString::new(old_name.as_bytes()).map_err(|_| EINVAL)?;
        let new_c = CString::new(new_name.as_bytes()).map_err(|_| EINVAL)?;
        let rc = unsafe {
            libc::renameat(
                old_dirfd.as_raw_fd(),
                old_c.as_ptr(),
                new_dirfd.as_raw_fd(),
                new_c.as_ptr(),
            )
        };
        if rc != 0 {
            return Err(errno_now());
        }

        // Update our inode bookkeeping. Two structures depend on the
        // path: the (parent,name) → nodeid lookup cache, and the
        // per-nodeid host_path on the source inode (and any descendants
        // when a directory was renamed).
        let mut st = self.st.lock().unwrap();
        let old_key = (old_parent, old_name.as_bytes().to_vec());
        let new_key = (new_parent, new_name.as_bytes().to_vec());
        if let Some(nodeid) = st.children.remove(&old_key) {
            // If a destination dentry was tracked, drop it — the rename
            // replaced it on the host, our cache must follow.
            st.children.remove(&new_key);
            st.children.insert(new_key, nodeid);
            // Update the renamed inode's host_path. For directories we
            // also walk the inode table and rewrite any descendants
            // whose host_path starts with the old prefix.
            if let Some(info) = st.inodes.get_mut(&nodeid) {
                info.host_path = new_full.clone();
            }
            // Best-effort descendant rewrite (only needed for dirs but
            // cheap to run unconditionally — bounded by inode count).
            let old_prefix = old_full.clone();
            for (id, info) in st.inodes.iter_mut() {
                if *id == nodeid {
                    continue;
                }
                if let Ok(suffix) = info.host_path.strip_prefix(&old_prefix) {
                    info.host_path = new_full.join(suffix);
                }
            }
        } else {
            // No prior LOOKUP — nothing to rewrite. Still drop a stale
            // destination entry if one exists.
            st.children.remove(&new_key);
        }
        Ok(())
    }

    fn flush(&self, _nodeid: u64, fh: u64) -> Result<(), Errno> {
        // FUSE_FLUSH is the kernel's hint on close(2) that the server
        // may want to commit handle state. We use pread/pwrite directly
        // against the host fd, so the host kernel page cache already
        // owns durability. Validating the fh exists protects against a
        // misbehaving guest sending FLUSH on a stale handle (matches
        // close(2)'s EBADF). No need to lazy-reopen — just checking
        // the slot is enough to honour the EBADF contract.
        let st = self.st.lock().unwrap();
        st.handles.get(&fh).ok_or(EBADF)?;
        Ok(())
    }

    fn fsyncdir(&self, _nodeid: u64, fh: u64, _datasync: bool) -> Result<(), Errno> {
        // No-op: a stronger impl would fsync(opendir_fd). The host
        // filesystem syncs dentry changes on every dir mutation so the
        // observable behaviour is identical for the patterns npm/tar use.
        // No lazy-reopen needed; slot existence is the sole contract.
        let st = self.st.lock().unwrap();
        st.handles.get(&fh).ok_or(EBADF)?;
        Ok(())
    }

    fn dax_map(
        &self,
        nodeid: u64,
        fh: u64,
        foffset: u64,
        len: u64,
        prot: u32,
    ) -> Result<*mut u8, Errno> {
        // The guest's iomap-driven read path (used by `dax=always`)
        // sends SETUPMAPPING with `fh = u64::MAX` because the read
        // is inode-level (no userspace fd backs it). Open the file
        // on demand by walking the inode table back to the host
        // path. For mmap-driven SETUPMAPPING (spike-22 zero-copy)
        // the fh IS a valid handle from a prior FUSE_OPEN — use it.
        //
        // mmap() captures the fd internally; the OwnedFd we hold
        // here covers the lifetime through the mmap call. After
        // mmap returns success the kernel keeps its own reference,
        // so dropping our OwnedFd on the unwind path is safe.
        let opened_fresh: Option<OwnedFd>;
        let raw = if fh == u64::MAX {
            let path = self.host_path_of(nodeid)?;
            let c = CString::new(path.as_os_str().as_bytes()).map_err(|_| EINVAL)?;
            // Try RDWR first so writes through DAX work; fall back
            // to RDONLY for files we can't open RW (e.g. read-only
            // host file).
            let mut fd = unsafe { libc::open(c.as_ptr(), libc::O_RDWR) };
            if fd < 0 {
                fd = unsafe { libc::open(c.as_ptr(), libc::O_RDONLY) };
            }
            if fd < 0 {
                return Err(errno_now());
            }
            // SAFETY: fd is fresh from open().
            let owned = unsafe { OwnedFd::from_raw_fd(fd) };
            let raw = owned.as_raw_fd();
            opened_fresh = Some(owned);
            raw
        } else {
            // Lazy-reopen on first use post-restore. The cycle-restore
            // bug this fixes literally surfaces here for cold-mapped
            // pages: the guest's FUSE_READ on a fh inherited from
            // the pre-snapshot lifetime needs to find a re-opened
            // host fd before mmap() can succeed against it.
            let mut st = self.st.lock().unwrap();
            let raw = Self::handle_raw_fd_locked(&mut st, fh)?;
            drop(st);
            opened_fresh = None;
            raw
        };
        // Suppress unused-var lint when fh != MAX.
        let _ = &opened_fresh;
        // Spike 22 validated: PROT_READ|PROT_WRITE host backing
        // works for both R and RW DAX. We always map RW on host
        // and let the guest-side stage-2 protection enforce R-only
        // semantics. Apple's HVF requires writable host backing
        // regardless of guest-side flags.
        let _ = prot; // recorded for hv_vm_map elsewhere
        let host_prot = libc::PROT_READ | libc::PROT_WRITE;
        let ptr = unsafe {
            libc::mmap(
                std::ptr::null_mut(),
                len as usize,
                host_prot,
                libc::MAP_SHARED,
                raw,
                foffset as libc::off_t,
            )
        };
        if ptr == libc::MAP_FAILED {
            return Err(errno_now());
        }
        let mut st = self.st.lock().unwrap();
        st.dax_mmaps.insert(
            ptr as usize,
            Mmap {
                ptr: ptr as *mut u8,
                len: len as usize,
            },
        );
        Ok(ptr as *mut u8)
    }

    fn dax_unmap(&self, _nodeid: u64, host_va: *mut u8, _len: u64) -> Result<(), Errno> {
        let mut st = self.st.lock().unwrap();
        let m = st.dax_mmaps.remove(&(host_va as usize)).ok_or(EINVAL)?;
        let rc = unsafe { libc::munmap(m.ptr as *mut _, m.len) };
        if rc != 0 {
            return Err(errno_now());
        }
        Ok(())
    }

    fn ioctl(
        &self,
        nodeid: u64,
        fh: u64,
        cmd: u32,
        in_data: &[u8],
        out_max: u32,
    ) -> Result<(i32, Vec<u8>), Errno> {
        // Only the Rosetta-in-VM runtime share answers ioctls. For
        // every other PosixFs mount, fall through to the trait
        // default (ENOSYS).
        if !self.rosetta_share {
            return Err(super::backend::ENOSYS);
        }
        // Best-effort trace — gated by env var so it doesn't spam
        // production. Use `SUPERMACHINE_ROSETTA_TRACE=1` to enable.
        if crate::trace::enabled("rosetta") {
            eprintln!(
                "[rosetta] ioctl nodeid={nodeid} fh={fh} cmd=0x{cmd:x} \
                 in_len={} out_max={}",
                in_data.len(),
                out_max
            );
        }
        // The single ioctl that matters for Rosetta-in-VM startup.
        // Request decode: dir=READ size=128 type='a' nr=35.
        //
        // The 128-byte struct VZ returns is undocumented. Empirical
        // approach: respond with zeros first, observe rosettad's
        // next failure, fill in the bytes it actually consults. See
        // `docs/design/rosetta-in-vm-2026-05-16.md` for the
        // disassembly-derived field map as we discover it.
        const ROSETTA_CACHE_SETTINGS_QUERY: u32 = 0x80806123;
        // _IOR('a', 37, [69 bytes]) — Rosetta interpreter verification handshake.
        // This is the descendant of the older libkrun `0x80456122` handshake
        // (which used nr=34). The famous 69-byte response from libkrun history:
        //   "Our hard work\nby these words guarded\nplease don't steal\n© Apple Inc\0"
        // We try that here; the new ioctl number suggests Apple may have
        // changed the expected payload but kept the same size — if so, we'll
        // see the next failure mode after this and iterate.
        const ROSETTA_VERIFY_HANDSHAKE: u32 = 0x80456125;
        if cmd == ROSETTA_VERIFY_HANDSHAKE {
            if crate::trace::enabled("rosetta") {
                eprintln!("[rosetta] verify handshake ioctl 0x80456125, out_max={out_max}");
            }
            // libkrun's exact 69-byte response (mrexodia/rosetta-multipass).
            // © = 0xC2 0xA9 in UTF-8 (the response is byte-exact, not text).
            let resp: &[u8] =
                b"Our hard work\nby these words guarded\nplease don't steal\n\xc2\xa9 Apple Inc\0";
            // Sanity: length == 0x45 (69)
            debug_assert_eq!(resp.len(), 69);
            let want = (out_max as usize).min(resp.len());
            return Ok((0, resp[..want].to_vec()));
        }
        // _IOC(_IOC_NONE, 'a', 36, 0) = 0x00006124 — Rosetta "hypervisor ping".
        // Per tnk4on's analysis, this is a no-data ioctl that just needs to
        // return success. If we leave it unhandled, the interpreter prints
        // "Unexpected ioctl error when communicating with hypervisor: 25"
        // (errno 25 = ENOTTY = the default for unknown ioctls). Reply 0.
        const ROSETTA_HV_PING: u32 = 0x00006124;
        if cmd == ROSETTA_HV_PING {
            if crate::trace::enabled("rosetta") {
                eprintln!("[rosetta] hv-ping ioctl 0x00006124");
            }
            return Ok((0, Vec::new()));
        }
        if cmd != ROSETTA_CACHE_SETTINGS_QUERY {
            return Err(super::backend::ENOSYS);
        }
        // EMPIRICAL DEBUG: pattern can be tuned via env var.
        //   "zero"     → all zeros (default)
        //   "ones"     → all 0xFF
        //   "marker"   → 0xAA — lets us see which bytes rosettad reads
        //   "sizes"    → all 0 EXCEPT u64 sizes at [0x20] and [0x40] = 256MB
        //   "synth"    → carefully constructed response:
        //     * [0]   = 1     (form-A flag — non-zero, take meaningful path)
        //     * [1..107] = a unique abstract-socket name (107 bytes)
        //       The rosettad daemon copies these bytes into sockaddr_un.sun_path
        //       starting at sun_path[0]; if sun_path[0] = 0 → abstract socket
        //       with the rest as the name. We start with 'r' so it's a fs path
        //       under our supermachine-cache directory.
        //     * [0x20..0x28] = 0x10000000 (256 MiB mmap size A)
        //     * [0x40..0x48] = 0x10000000 (256 MiB mmap size B)
        //     * [0x6c] = 1   (form-A flag — non-zero so the daemon picks the
        //       memcpy-from-offset-1 path)
        //
        // The disassembly at rosettad 0x226ee4-ef0 shows bind() is called
        // with the response buffer directly as sockaddr_un:
        //
        //   sub  x1, x29, #0x80         ; x1 = response_buf
        //   mov  w2, #0x6e               ; addrlen = 110
        //   bl   bind                    ; bind(fd, response_buf, 110)
        //
        // …after first overwriting response_buf[0..2] with AF_UNIX (= 1) and
        // copying global[2..110] into response_buf[2..110]. The global was
        // populated from our response[1..107] via the form-A memcpy.
        let want = out_max.min(128) as usize;
        // Field map of the 128-byte cache_settings response, decoded
        // 2026-05-17 by walking the rosetta interpreter's disassembly
        // (`/Library/Apple/usr/libexec/oah/RosettaLinux/rosetta`) around
        // the `.flu` cache-file open at 0x800000090afc and the bind
        // setup at 0x80000003070c–0x800000030a44:
        //
        //   data[0x00]      = "AOT cache mode enabled" flag (non-zero =
        //                     enter the cache code path; zero = JIT only,
        //                     interpreter never connects to rosettad)
        //   data[0x01]      = daemon protocol vs hash-path mode (zero =
        //                     the interpreter just does a 1-byte
        //                     handshake then opens its own cache file;
        //                     non-zero = full daemon RPC protocol that
        //                     we have NOT reverse-engineered)
        //   data[0x02]      = sun_path[0]; zero = abstract Unix socket
        //   data[0x03..6e]  = abstract socket name (rosettad binds it,
        //                     interpreter connects to it; both read the
        //                     SAME ioctl response so the name agrees
        //                     within one PosixFs lifetime)
        //   data[0x20..28]  = u64 mmap size A (interpreter mmaps a high
        //                     virtual-address arena of (A+B)>>6 bytes;
        //                     256 MiB each yields an 8 MiB arena)
        //   data[0x40..48]  = u64 mmap size B (same)
        //   data[0x6c]      = "daemon-cache-on" flag (non-zero = enable
        //                     full AOT caching path: interpreter sends
        //                     translations to rosettad, daemon writes
        //                     content-addressed .aotcache files to its
        //                     configured cache dir)
        //
        // Default behavior when `rosetta_share` is enabled: emit a
        // ready-to-use AOT-enabled response. The env vars below are
        // diagnostic overrides; integrators don't need to set anything.
        //
        //   SUPERMACHINE_ROSETTA_FILL=zero|marker|ones|sizes|synth — pin
        //     the fill mode (synth is the default). `zero` reverts to
        //     pre-0.7.23 JIT-only behavior; useful as an A/B comparison.
        //   SUPERMACHINE_ROSETTA_6C=0|1 — pin the [0x6c] daemon-cache
        //     flag (default 1; flip to 0 to disable AOT caching while
        //     keeping the rest of the synth response).
        //   SUPERMACHINE_ROSETTA_OVERRIDE=OFFSET:HEXBYTES,... — patch
        //     arbitrary bytes after the synth fill is composed. Used
        //     during bisection of the response layout.
        //   SUPERMACHINE_ROSETTA_TRACE=1 — log the returned bytes to
        //     the worker's stderr.
        //
        // History: through 0.7.22 we shipped `data[0] = 0` (no AOT
        // mode) and Rosetta ran JIT-only — every amd64 invocation paid
        // the full translation cost. 0.7.23 enables AOT mode by default;
        // rosettad writes .aotcache files to /var/cache/rosettad inside
        // the guest, which init-oci wires to a host-persistent virtio-fs
        // share (see api.rs auto-mount of `rosettad-cache`).
        let fill = std::env::var("SUPERMACHINE_ROSETTA_FILL")
            .ok()
            .unwrap_or_else(|| "synth".to_owned());
        let mut data = match fill.as_str() {
            "ones" => vec![0xffu8; want],
            "marker" => vec![0xaau8; want],
            _ => vec![0x00u8; want],
        };
        if matches!(fill.as_str(), "sizes" | "synth") {
            // mmap-arena seed: 256 MiB each. After the >> 6 + 8 KiB
            // align in the interpreter this becomes an 8 MiB MAP_FIXED
            // reservation at 0xf00000000000.
            let size: u64 = 0x10000000;
            let bytes = size.to_le_bytes();
            for (i, b) in bytes.iter().enumerate() {
                if 0x20 + i < want {
                    data[0x20 + i] = *b;
                }
                if 0x40 + i < want {
                    data[0x40 + i] = *b;
                }
            }
        }
        if fill.as_str() == "synth" {
            // Form-B (data[0x6c]=1) cache layout per Apple's public API
            // (VZLinuxRosettaCachingOptions in Virtualization.framework):
            // the interpreter copies response[1..107] verbatim into the
            // sockaddr_un sun_path field it connect()s to. So:
            //
            //   data[0]     = "AOT enabled" flag (1)
            //   data[1..]   = sun_path (108 bytes max)
            //   data[0x6c]  = caching mode (must be 1 for AOT)
            //
            // Through 0.7.23/0.7.24 we encoded data[1..] as an
            // ABSTRACT socket (data[1]=0 = sun_path[0]=0 abstract
            // marker, data[2..] = name). That maps to Apple's
            // `VZLinuxRosettaAbstractSocketCachingOptions`. It worked
            // for musl-static amd64 (alpine) but BROKE glibc-dynamic
            // (ubuntu, debian, playwright) — `ld.so` hit its
            // `GL(dl_rtld_map).l_libname` assertion during _dl_start.
            //
            // 0.7.25 switches to Apple's documented DEFAULT:
            // `VZLinuxRosettaUnixSocketCachingOptions()` with no
            // arguments → filesystem socket at `/run/rosettad/rosetta.sock`.
            // The corresponding cache_settings response is:
            //
            //   data[1..0x1c] = "/run/rosettad/rosetta.sock\0"
            //
            // i.e. sun_path[0] = '/' (non-zero → filesystem path
            // socket, not abstract), and the path bytes follow.
            //
            // This is the EXACT default Apple's container CLI uses
            // (we read the public header
            // `/Library/Developer/CommandLineTools/SDKs/.../Virtualization.framework/Headers/VZLinuxRosettaUnixSocketCachingOptions.h`
            // which says `init()` uses `/run/rosettad/rosetta.sock`).
            // With it, both musl-static AND glibc-dynamic amd64
            // binaries work, AND rosettad produces persistent
            // `.aotcache` files in `/var/cache/rosettad/`.
            //
            // Requirements that init-oci handles on the guest side:
            //   1. `mkdir /run/rosettad` so rosettad can bind there.
            //   2. Spawn `rosettad daemon /var/cache/rosettad` (the
            //      socket-path comes from cache_settings; the cache
            //      DIR arg comes from the daemon command line).
            //
            data[0] = 1;
            let six_c: u8 = std::env::var("SUPERMACHINE_ROSETTA_6C")
                .ok()
                .and_then(|s| s.parse::<u8>().ok())
                .unwrap_or(1);
            if 0x6c < want {
                data[0x6c] = six_c;
            }
            // Encode the filesystem-socket path "/run/rosettad/rosetta.sock"
            // (Apple's VZLinuxRosettaUnixSocketCachingOptions.init() default)
            // into response[1..0x1c]. data[1] = '/' = 0x2f (non-zero) →
            // sun_path[0] = '/', which the Linux kernel treats as a
            // filesystem-path socket (not abstract). Trailing null at
            // [0x1b]; rest stays zeroed.
            const ROSETTAD_SOCK_PATH: &[u8] = b"/run/rosettad/rosetta.sock\0";
            let pn = ROSETTAD_SOCK_PATH.len();
            if pn < want {
                data[1..1 + pn].copy_from_slice(ROSETTAD_SOCK_PATH);
            }
            // The historical abstract-socket name (used through 0.7.24)
            // is no longer encoded in the response; kept on the struct
            // for snapshot-format compatibility but unused.
            let _unused = self.rosetta_socket_name.as_bytes();
            // SUPERMACHINE_ROSETTA_OVERRIDE — patch individual bytes
            // for diagnosis. Format: "OFFSET:HEXBYTES,OFFSET:HEXBYTES,..."
            // E.g. "0x60:00400000000000,0x68:00000000010000" sets two u64s.
            if let Ok(spec) = std::env::var("SUPERMACHINE_ROSETTA_OVERRIDE") {
                for chunk in spec.split(',').filter(|s| !s.is_empty()) {
                    if let Some((off_s, hex_s)) = chunk.split_once(':') {
                        let off = off_s
                            .trim()
                            .strip_prefix("0x")
                            .unwrap_or(off_s.trim())
                            .parse::<usize>()
                            .ok()
                            .or_else(|| {
                                usize::from_str_radix(off_s.trim().trim_start_matches("0x"), 16)
                                    .ok()
                            });
                        let off = match off {
                            Some(o) => o,
                            None => continue,
                        };
                        let hex_s = hex_s.trim();
                        let mut i = 0;
                        while i + 1 < hex_s.len() {
                            let b = u8::from_str_radix(&hex_s[i..i + 2], 16);
                            if let Ok(b) = b {
                                let pos = off + (i / 2);
                                if pos < want {
                                    data[pos] = b;
                                }
                            }
                            i += 2;
                        }
                    }
                }
            }
        }
        if crate::trace::enabled("rosetta") {
            let hex: String = data.iter().take(96).map(|b| format!("{:02x}", b)).collect();
            eprintln!("[rosetta] returning {} bytes: {}…", data.len(), hex);
            // Worker process stderr may not reach the parent — also write
            // to a well-known file so a test harness can inspect.
            let _ = std::fs::write("/tmp/supermachine_rosetta_response.bin", &data);
            let _ = std::fs::write(
                "/tmp/supermachine_rosetta_response.hex",
                data.iter()
                    .map(|b| format!("{:02x}", b))
                    .collect::<Vec<_>>()
                    .join(""),
            );
        }
        Ok((0, data))
    }

    // Forward the trait-default `snapshot_state` / `restore_state` to
    // our inherent methods. (The inherent versions exist so unit tests
    // and the snapshot-pipeline plumbing can call them directly on a
    // concrete `PosixFs` without going through `&dyn FsBackend`.)
    fn snapshot_state(&self) -> Option<Vec<u8>> {
        Some(PosixFs::snapshot_state(self))
    }

    fn restore_state(&self, blob: &[u8]) -> Result<(), std::io::Error> {
        PosixFs::restore_state(self, blob)
    }
}

/// Background thread body: blocks in kevent(), dispatches NOTE_*
/// events to the notifier.
/// macOS watcher loop: kqueue `EVFILT_VNODE` events (+ an `EVFILT_USER` wakeup
/// registered by Drop). Each VNODE event → [`watcher_emit`].
#[cfg(target_os = "macos")]
fn run_watcher(inner: Arc<WatcherInner>) {
    // Register a USER event so Drop can wake us.
    let mut w = libc::kevent {
        ident: 0,
        filter: libc::EVFILT_USER,
        flags: libc::EV_ADD | libc::EV_CLEAR,
        fflags: 0,
        data: 0,
        udata: std::ptr::null_mut(),
    };
    unsafe {
        libc::kevent(
            inner.kq,
            &mut w as *mut _,
            1,
            std::ptr::null_mut(),
            0,
            std::ptr::null(),
        );
    }

    let mut events: [libc::kevent; 16] = unsafe { std::mem::zeroed() };
    loop {
        if inner.stop.load(Ordering::Acquire) {
            break;
        }
        let n = unsafe {
            libc::kevent(
                inner.kq,
                std::ptr::null(),
                0,
                events.as_mut_ptr(),
                events.len() as libc::c_int,
                std::ptr::null(),
            )
        };
        if n < 0 {
            let err = std::io::Error::last_os_error();
            if err.raw_os_error() == Some(libc::EINTR) {
                continue;
            }
            eprintln!("[posix-fs watcher] kevent failed: {err}; thread exiting");
            return;
        }
        if inner.stop.load(Ordering::Acquire) {
            break;
        }
        for ev in events.iter().take(n as usize) {
            if ev.filter == libc::EVFILT_USER {
                continue;
            }
            let ident = ev.ident as libc::c_int;
            let change = if ev.fflags & (libc::NOTE_DELETE | libc::NOTE_RENAME) != 0 {
                WatchChange::Gone
            } else {
                WatchChange::Changed
            };
            watcher_emit(&inner, ident, change);
        }
    }
}

/// Linux watcher loop: `poll` the inotify fd + the eventfd wakeup; parse
/// inotify records and map each watch descriptor's mask to [`watcher_emit`].
/// `IN_MODIFY|IN_CLOSE_WRITE|IN_ATTRIB` → Changed; `IN_DELETE_SELF|IN_MOVE_SELF|
/// IN_IGNORED` → Gone (inotify auto-removes the watch on delete/move).
#[cfg(not(target_os = "macos"))]
fn run_watcher(inner: Arc<WatcherInner>) {
    let mut fds = [
        libc::pollfd {
            fd: inner.inotify_fd,
            events: libc::POLLIN,
            revents: 0,
        },
        libc::pollfd {
            fd: inner.wake_fd,
            events: libc::POLLIN,
            revents: 0,
        },
    ];
    // inotify records are variable-length (16-byte header + padded name).
    let mut buf = [0u8; 8192];
    const HDR: usize = 16; // wd:i32, mask:u32, cookie:u32, len:u32
    loop {
        if inner.stop.load(Ordering::Acquire) {
            break;
        }
        let n = unsafe { libc::poll(fds.as_mut_ptr(), 2, -1) };
        if n < 0 {
            let err = std::io::Error::last_os_error();
            if err.raw_os_error() == Some(libc::EINTR) {
                continue;
            }
            eprintln!("[posix-fs watcher] poll failed: {err}; thread exiting");
            return;
        }
        if inner.stop.load(Ordering::Acquire) {
            break;
        }
        // Drain the wake eventfd if it fired (level-triggered).
        if fds[1].revents & libc::POLLIN != 0 {
            let mut sink = [0u8; 8];
            let _ = unsafe { libc::read(inner.wake_fd, sink.as_mut_ptr() as *mut libc::c_void, 8) };
        }
        if fds[0].revents & libc::POLLIN == 0 {
            continue;
        }
        let got = unsafe {
            libc::read(
                inner.inotify_fd,
                buf.as_mut_ptr() as *mut libc::c_void,
                buf.len(),
            )
        };
        if got <= 0 {
            continue; // EAGAIN (nonblocking) or closed — loop / re-check stop.
        }
        let got = got as usize;
        let mut off = 0usize;
        while off + HDR <= got {
            // Read fields unaligned — the byte buffer has no struct alignment.
            let wd = i32::from_ne_bytes(buf[off..off + 4].try_into().unwrap());
            let mask = u32::from_ne_bytes(buf[off + 4..off + 8].try_into().unwrap());
            let len = u32::from_ne_bytes(buf[off + 12..off + 16].try_into().unwrap()) as usize;
            let gone = mask & (libc::IN_DELETE_SELF | libc::IN_MOVE_SELF | libc::IN_IGNORED) != 0;
            let change = if gone {
                WatchChange::Gone
            } else {
                WatchChange::Changed
            };
            watcher_emit(&inner, wd, change);
            off += HDR + len;
        }
    }
}

// Unused-import suppression — std re-exports are conditional on macOS file_type extensions.
#[allow(unused_imports)]
use std::convert::TryFrom;
use std::os::unix::fs::FileTypeExt;
#[allow(dead_code)]
const _: () = {
    let _ = OsString::new;
    let _ = ENOSPC;
    let _ = ENOTDIR;
    let _ = EACCES;
    let _ = EIO;
};

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

    fn tmpdir(name: &str) -> PathBuf {
        let pid = unsafe { libc::getpid() };
        let p = std::env::temp_dir().join(format!("posixfs-{pid}-{name}"));
        let _ = std::fs::remove_dir_all(&p);
        std::fs::create_dir_all(&p).unwrap();
        p
    }

    #[test]
    fn lookup_and_read_real_file() {
        let dir = tmpdir("t1");
        std::fs::write(dir.join("hello.txt"), b"hi from posix").unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("hello.txt")).unwrap();
        assert!(e.attr.size == 13);
        let fh = fs.open(e.nodeid, libc::O_RDONLY as u32).unwrap();
        let buf = fs.read(e.nodeid, fh, 0, 64).unwrap();
        assert_eq!(buf, b"hi from posix");
        fs.release(e.nodeid, fh).unwrap();
    }

    #[test]
    fn readdir_lists_real_entries_with_types() {
        let dir = tmpdir("t2");
        std::fs::write(dir.join("a.txt"), b"a").unwrap();
        std::fs::create_dir_all(dir.join("sub")).unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        let dh = fs.opendir(FUSE_ROOT_ID, 0).unwrap();
        let entries = fs.readdir(FUSE_ROOT_ID, dh, 0, 4096).unwrap();
        let by_name: std::collections::HashMap<&[u8], u32> =
            entries.iter().map(|e| (e.name.as_slice(), e.typ)).collect();
        assert_eq!(by_name[&b"a.txt"[..]], DT_REG);
        assert_eq!(by_name[&b"sub"[..]], DT_DIR);
        fs.releasedir(FUSE_ROOT_ID, dh).unwrap();
    }

    // Regression: per-fh dirent caching keeps READDIR iteration
    // stable across concurrent unlinks. Pre-fix this scenario
    // surfaced as `rm -rf` returning ENOTEMPTY on a virtio-fs
    // mount (the user's vite/.deps repro) because the host
    // re-ran std::fs::read_dir on every FUSE_READDIR call and
    // used position-in-iteration as the resume cookie — every
    // unlink shifted the cookie space and the kernel-side
    // enumeration skipped entries.
    #[test]
    fn readdir_snapshot_is_stable_across_concurrent_unlink() {
        let dir = tmpdir("readdir_concurrent_unlink");
        // Create 50 files. Enough that several FUSE_READDIR
        // "pages" would be needed if the kernel were splitting,
        // but more importantly so position-based offsets matter.
        for i in 0..50 {
            std::fs::write(dir.join(format!("f{i:03}.txt")), b"x").unwrap();
        }
        let fs = PosixFs::new(&dir).unwrap();
        let dh = fs.opendir(FUSE_ROOT_ID, 0).unwrap();

        // First batch — request entries from offset 0.
        let first = fs.readdir(FUSE_ROOT_ID, dh, 0, 65536).unwrap();
        assert_eq!(first.len(), 50, "all 50 entries in first batch");

        // Simulate the "rm -rf" pattern: unlink some entries on
        // the host between FUSE_READDIR calls. The pre-fix
        // bug surfaced HERE — the re-opened read_dir would skip
        // entries with index <offset from the now-shrunk
        // directory listing, so the second readdir wouldn't see
        // entries that have shifted down into the skipped slice.
        for entry in &first[0..10] {
            let name = std::str::from_utf8(&entry.name).unwrap();
            std::fs::remove_file(dir.join(name)).unwrap();
        }

        // Second batch — resume from offset = first batch size.
        // Post-fix: served from the cached snapshot, so we get
        // exactly zero remaining entries (we already consumed
        // all 50 via the first call's return value).
        let second = fs
            .readdir(FUSE_ROOT_ID, dh, first.len() as u64, 65536)
            .unwrap();
        assert_eq!(second.len(), 0, "no entries left to enumerate");

        // Third sanity check: resume from offset 10 — the unlinks
        // shouldn't change what we see at this offset because we
        // serve from the snapshot.
        let third = fs.readdir(FUSE_ROOT_ID, dh, 10, 65536).unwrap();
        assert_eq!(
            third.len(),
            40,
            "snapshot still has 40 entries from index 10"
        );
        // The entries we get are positions 10..50 in the original
        // snapshot order — same entries we saw at positions 10..50
        // in `first` — even though some are now unlinked on disk.
        for (i, entry) in third.iter().enumerate() {
            assert_eq!(entry.name, first[i + 10].name);
        }

        fs.releasedir(FUSE_ROOT_ID, dh).unwrap();
    }

    // Regression: a stale fh (one not in the handles table) must
    // not panic — readdir falls through to the slow path (which
    // captures fresh entries but doesn't cache them). This
    // covers the test-only paths that pass fh=0.
    #[test]
    fn readdir_with_unknown_fh_still_returns_entries() {
        let dir = tmpdir("readdir_unknown_fh");
        std::fs::write(dir.join("a.txt"), b"a").unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        // No opendir; pass a synthetic fh that doesn't exist.
        let entries = fs.readdir(FUSE_ROOT_ID, 99999, 0, 4096).unwrap();
        assert!(entries.iter().any(|e| e.name == b"a.txt"));
    }

    #[test]
    fn lookup_rejects_dotdot() {
        let dir = tmpdir("t3");
        let fs = PosixFs::new(&dir).unwrap();
        let err = fs.lookup(FUSE_ROOT_ID, OsStr::new("..")).unwrap_err();
        assert_eq!(err, EINVAL);
    }

    #[test]
    fn lookup_rejects_slash_in_name() {
        let dir = tmpdir("t4");
        let fs = PosixFs::new(&dir).unwrap();
        let err = fs.lookup(FUSE_ROOT_ID, OsStr::new("a/b")).unwrap_err();
        assert_eq!(err, EINVAL);
    }

    #[test]
    fn lookup_rejects_embedded_nul_in_name() {
        let dir = tmpdir("t4n");
        let fs = PosixFs::new(&dir).unwrap();
        let err = fs
            .lookup(FUSE_ROOT_ID, OsStr::from_bytes(b"foo\0bar"))
            .unwrap_err();
        assert_eq!(err, EINVAL);
    }

    #[test]
    fn name_safe_table() {
        // The per-component anti-traversal filter every name-joining op
        // runs first. Ordinary names with dots are fine; only the exact
        // traversal/separator/NUL forms are rejected.
        for ok in ["file.txt", "a", ".hidden", "..foo", "foo..", "a.b.c"] {
            assert!(
                name_safe(OsStr::new(ok)).is_ok(),
                "{ok:?} should be allowed"
            );
        }
        for bad in ["", ".", ".."] {
            assert_eq!(name_safe(OsStr::new(bad)).unwrap_err(), EINVAL, "{bad:?}");
        }
        assert_eq!(name_safe(OsStr::new("a/b")).unwrap_err(), EINVAL);
        assert_eq!(name_safe(OsStr::new("../x")).unwrap_err(), EINVAL);
        assert_eq!(
            name_safe(OsStr::from_bytes(b"foo\0bar")).unwrap_err(),
            EINVAL
        );
    }

    #[test]
    fn create_gates_traversal_on_the_write_path() {
        // The gate must be wired on MUTATING ops, not just lookup —
        // otherwise create(parent, "..", …) would join `..` and write
        // outside the parent. (Audited: every name-joining op calls
        // name_safe; this pins it for the write path against regression.)
        let dir = tmpdir("t-create-dotdot");
        let fs = PosixFs::new(&dir).unwrap();
        assert_eq!(
            fs.create(FUSE_ROOT_ID, OsStr::new(".."), 0o644, 0)
                .unwrap_err(),
            EINVAL
        );
        assert_eq!(
            fs.create(FUSE_ROOT_ID, OsStr::new("a/b"), 0o644, 0)
                .unwrap_err(),
            EINVAL
        );
    }

    #[test]
    fn lookup_blocks_external_symlink_by_default() {
        // Mount root contains a symlink that points OUTSIDE the root.
        // LOOKUP must refuse with EACCES under the default (hostile-
        // tenant-safe) policy.
        let dir = tmpdir("t-sym-default");
        let outside = tmpdir("t-sym-outside");
        std::fs::write(outside.join("secret.txt"), b"do not leak").unwrap();
        std::os::unix::fs::symlink(&outside, dir.join("escape")).unwrap();

        let fs = PosixFs::new(&dir).unwrap();
        let err = fs.lookup(FUSE_ROOT_ID, OsStr::new("escape")).unwrap_err();
        assert_eq!(err, EACCES, "external symlink must be denied");
    }

    #[test]
    fn lookup_allows_internal_symlink_by_default() {
        // Symlinks that resolve INSIDE the mount root are fine even
        // under the strict default.
        let dir = tmpdir("t-sym-internal");
        std::fs::create_dir_all(dir.join("real")).unwrap();
        std::fs::write(dir.join("real/data.txt"), b"in-tree data").unwrap();
        std::os::unix::fs::symlink("real/data.txt", dir.join("link.txt")).unwrap();

        let fs = PosixFs::new(&dir).unwrap();
        // Direct lookup of the link must succeed.
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("link.txt")).unwrap();
        assert!(e.attr.size > 0);
    }

    #[test]
    fn lookup_succeeds_on_broken_symlink() {
        // POSIX lstat() works on a broken symlink (target gone or
        // never existed) — the symlink inode itself is fine, only
        // following it would fail. FUSE LOOKUP must match: return
        // type=symlink Attr, not error.
        //
        // Pre-fix bug: lookup() used std::fs::metadata (= stat),
        // which follows the symlink. Broken target → ENOENT or
        // worse (EIO if stat returns a less-specific error). npm
        // hit this when its `.bin/<pkg>` symlink was created
        // before the target file's writes finished draining, and
        // a postinstall hook tried to stat it mid-write.
        let dir = tmpdir("t-sym-broken");
        std::os::unix::fs::symlink("does-not-exist-anywhere.txt", dir.join("dangling")).unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("dangling")).unwrap();
        // The Attr's type bits must say S_IFLNK (symlink), not
        // S_IFREG. attr.mode includes the type bits in the high
        // nibble. S_IFLNK = 0o120000.
        assert_eq!(
            e.attr.mode & S_IFMT,
            S_IFLNK,
            "attr must report symlink type"
        );
    }

    #[test]
    fn lookup_returns_symlink_attrs_not_target_attrs() {
        // For a VALID symlink (target exists), LOOKUP must still
        // return THE SYMLINK's own attributes (S_IFLNK, the size
        // of the target-path bytes), not the target's attributes
        // (S_IFREG, content size). Pre-fix bug: stat() followed
        // and we returned the target's attrs, breaking guests
        // that decide what to do based on the dentry type.
        let dir = tmpdir("t-sym-attrs");
        std::fs::write(dir.join("real.txt"), b"twelve bytes").unwrap();
        std::os::unix::fs::symlink("real.txt", dir.join("link")).unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("link")).unwrap();
        assert_eq!(e.attr.mode & S_IFMT, S_IFLNK, "must report symlink type");
        // size for a symlink is the byte-length of the target string
        // ("real.txt" = 8 bytes). NOT 12 (the target file's contents).
        assert_eq!(e.attr.size, "real.txt".len() as u64);
    }

    #[test]
    fn getattr_returns_symlink_attrs_not_target_attrs() {
        // Same lstat-not-stat semantics for getattr (FUSE_GETATTR).
        // The guest calls fstat(2) on an open fh or stat(2) on a
        // dentry; both can route through FUSE_GETATTR depending on
        // the kernel's cache state. Must return the SYMLINK's attrs.
        let dir = tmpdir("t-sym-getattr");
        std::fs::write(dir.join("target.bin"), vec![0u8; 4096]).unwrap();
        std::os::unix::fs::symlink("target.bin", dir.join("alias")).unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("alias")).unwrap();
        let attr = fs.getattr(e.nodeid, None).unwrap();
        assert_eq!(attr.mode & S_IFMT, S_IFLNK);
        // Symlink size = strlen("target.bin") = 10, not 4096.
        assert_eq!(attr.size, "target.bin".len() as u64);
    }

    #[test]
    fn lookup_succeeds_on_recently_created_symlink() {
        // Regression for the npm install scenario: create a symlink
        // via the FUSE backend, immediately look it up. This is the
        // exact sequence npm runs when it creates `.bin/<pkg>`
        // entries — `symlink()` then `realpath()/lstat()` on the
        // same path. Pre-fix this could surface EIO if the target
        // wasn't fully readable at the lstat moment.
        let dir = tmpdir("t-sym-recent");
        std::fs::create_dir_all(dir.join("napi-postinstall/lib")).unwrap();
        std::fs::write(
            dir.join("napi-postinstall/lib/cli.js"),
            b"console.log('hi')",
        )
        .unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        // Create the symlink via our backend (the way npm would).
        let bin_entry = fs
            .symlink(
                FUSE_ROOT_ID,
                OsStr::new("napi-postinstall-bin"),
                OsStr::new("../napi-postinstall/lib/cli.js"),
            )
            .unwrap();
        assert_eq!(bin_entry.attr.mode & S_IFMT, S_IFLNK);
        // Now lookup it back — this is what `realpathSync` / `lstat`
        // from inside the guest does.
        let e = fs
            .lookup(FUSE_ROOT_ID, OsStr::new("napi-postinstall-bin"))
            .unwrap();
        assert_eq!(e.attr.mode & S_IFMT, S_IFLNK);
        // Getattr too.
        let attr = fs.getattr(e.nodeid, None).unwrap();
        assert_eq!(attr.mode & S_IFMT, S_IFLNK);
    }

    #[test]
    fn lookup_allows_external_symlink_when_opted_in() {
        // `new_unchecked` disables the check; absolute symlinks resolve.
        let dir = tmpdir("t-sym-opt-in");
        let outside = tmpdir("t-sym-target");
        std::fs::write(outside.join("ok.txt"), b"opt-in data").unwrap();
        std::os::unix::fs::symlink(outside.join("ok.txt"), dir.join("link")).unwrap();

        let fs = PosixFs::new_unchecked(&dir).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("link")).unwrap();
        assert!(e.attr.size > 0);
    }

    #[test]
    fn dax_map_then_unmap_round_trip() {
        let dir = tmpdir("t5");
        // 32 KiB file with known pattern.
        let path = dir.join("data.bin");
        let mut data = vec![0u8; 32 * 1024];
        for (i, b) in data.iter_mut().enumerate() {
            *b = (i % 251) as u8;
        }
        std::fs::write(&path, &data).unwrap();

        let fs = PosixFs::new(&dir).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("data.bin")).unwrap();
        let fh = fs.open(e.nodeid, libc::O_RDWR as u32).unwrap();

        // mmap the whole file via dax_map.
        let host_va = fs.dax_map(e.nodeid, fh, 0, 32 * 1024, 0).unwrap();
        assert!(!host_va.is_null());
        // Verify mmap contents match the file we wrote.
        let host_slice = unsafe { std::slice::from_raw_parts(host_va, 32 * 1024) };
        assert_eq!(host_slice, &data[..]);

        // Unmap — must succeed and clear internal tracking.
        fs.dax_unmap(e.nodeid, host_va, 32 * 1024).unwrap();
        // Calling dax_unmap on a host_va we don't know about must error.
        assert_eq!(
            fs.dax_unmap(e.nodeid, host_va, 32 * 1024).unwrap_err(),
            EINVAL
        );
        fs.release(e.nodeid, fh).unwrap();
    }

    #[test]
    fn read_eof_returns_empty() {
        let dir = tmpdir("t6");
        std::fs::write(dir.join("x"), b"short").unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("x")).unwrap();
        let fh = fs.open(e.nodeid, libc::O_RDONLY as u32).unwrap();
        let eof = fs.read(e.nodeid, fh, 100, 10).unwrap();
        assert!(eof.is_empty());
        fs.release(e.nodeid, fh).unwrap();
    }

    #[test]
    fn open_directory_returns_eisdir() {
        let dir = tmpdir("t7");
        std::fs::create_dir_all(dir.join("sub")).unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("sub")).unwrap();
        let err = fs.open(e.nodeid, libc::O_RDONLY as u32).unwrap_err();
        assert_eq!(err, EISDIR);
    }

    // === Symlink / Readlink / Link =====================================

    #[test]
    fn symlink_create_inside_mount() {
        let dir = tmpdir("t-sym-create");
        let fs = PosixFs::new(&dir).unwrap();
        let e = fs
            .symlink(FUSE_ROOT_ID, OsStr::new("link"), OsStr::new("target"))
            .unwrap();
        // Symlink mode bits.
        assert_eq!(e.attr.mode & S_IFMT, S_IFLNK);
        // readlink returns the bytes we stored, verbatim, no NUL.
        let target = fs.readlink(e.nodeid).unwrap();
        assert_eq!(target, b"target");
        // The host file actually is a symlink.
        let md = std::fs::symlink_metadata(dir.join("link")).unwrap();
        assert!(md.file_type().is_symlink());
    }

    #[test]
    fn symlink_create_external_target_stored_verbatim() {
        // POSIX symlink(2) doesn't resolve the target. A guest creating
        // a symlink whose target is `/etc/passwd` simply gets a symlink
        // whose CONTENTS say `/etc/passwd`. The host never opens that
        // file as part of symlink(); only readlink returns the bytes.
        // This is the safe behaviour: read-side LOOKUP's external-
        // symlink check rejects any subsequent `lookup("escape")`.
        let dir = tmpdir("t-sym-external");
        let fs = PosixFs::new(&dir).unwrap();
        let e = fs
            .symlink(
                FUSE_ROOT_ID,
                OsStr::new("escape"),
                OsStr::new("/etc/passwd"),
            )
            .unwrap();
        let target = fs.readlink(e.nodeid).unwrap();
        assert_eq!(target, b"/etc/passwd");
        // Sanity: the host symlink really points there.
        let read = std::fs::read_link(dir.join("escape")).unwrap();
        assert_eq!(read.as_os_str(), OsStr::new("/etc/passwd"));
        // And LOOKUP (Opaque default) refuses to traverse it.
        let err = fs.lookup(FUSE_ROOT_ID, OsStr::new("escape")).unwrap_err();
        assert_eq!(err, EACCES);
    }

    #[test]
    fn symlink_create_blocked_by_deny() {
        let dir = tmpdir("t-sym-deny");
        let fs = PosixFs::new_with_symlinks(&dir, SymlinkPolicy::Deny).unwrap();
        let err = fs
            .symlink(FUSE_ROOT_ID, OsStr::new("link"), OsStr::new("target"))
            .unwrap_err();
        assert_eq!(err, EPERM);
        // And no host symlink was created.
        assert!(std::fs::symlink_metadata(dir.join("link")).is_err());
    }

    #[test]
    fn link_create_inside_mount() {
        let dir = tmpdir("t-link");
        std::fs::write(dir.join("orig"), b"hello").unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        let src = fs.lookup(FUSE_ROOT_ID, OsStr::new("orig")).unwrap();
        let new_entry = fs
            .link(src.nodeid, FUSE_ROOT_ID, OsStr::new("alias"))
            .unwrap();
        // nlink on the new entry should be 2 — two names pointing at
        // the same underlying inode.
        assert_eq!(new_entry.attr.nlink, 2);
        // Reading via either name gives the same bytes.
        let fh = fs.open(new_entry.nodeid, libc::O_RDONLY as u32).unwrap();
        let buf = fs.read(new_entry.nodeid, fh, 0, 64).unwrap();
        assert_eq!(buf, b"hello");
        fs.release(new_entry.nodeid, fh).unwrap();
    }

    #[test]
    fn link_blocked_by_deny() {
        let dir = tmpdir("t-link-deny");
        std::fs::write(dir.join("orig"), b"hi").unwrap();
        let fs = PosixFs::new_with_symlinks(&dir, SymlinkPolicy::Deny).unwrap();
        let src = fs.lookup(FUSE_ROOT_ID, OsStr::new("orig")).unwrap();
        let err = fs
            .link(src.nodeid, FUSE_ROOT_ID, OsStr::new("alias"))
            .unwrap_err();
        assert_eq!(err, EPERM);
        // No host link was created.
        assert!(std::fs::symlink_metadata(dir.join("alias")).is_err());
    }

    #[test]
    fn lookup_with_opaque_blocks_external_symlink() {
        // Same as `lookup_blocks_external_symlink_by_default`, but
        // pinned to the new explicit Opaque policy so the test still
        // exercises the right code path once Default's value changes.
        let dir = tmpdir("t-sym-opaque-blocks");
        let outside = tmpdir("t-sym-opaque-out");
        std::fs::write(outside.join("x"), b"secret").unwrap();
        std::os::unix::fs::symlink(&outside, dir.join("escape")).unwrap();
        let fs = PosixFs::new_with_symlinks(&dir, SymlinkPolicy::Opaque).unwrap();
        let err = fs.lookup(FUSE_ROOT_ID, OsStr::new("escape")).unwrap_err();
        assert_eq!(err, EACCES);
    }

    #[test]
    fn lookup_with_follow_allows_external_symlink() {
        // Parity with the legacy `new_unchecked` / pre-0.5.5
        // `allow_external_symlinks: true` behaviour.
        let dir = tmpdir("t-sym-follow");
        let outside = tmpdir("t-sym-follow-out");
        std::fs::write(outside.join("ok.txt"), b"out-of-mount data").unwrap();
        std::os::unix::fs::symlink(outside.join("ok.txt"), dir.join("link")).unwrap();
        let fs = PosixFs::new_with_symlinks(&dir, SymlinkPolicy::Follow).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("link")).unwrap();
        assert!(e.attr.size > 0);
    }

    // === Host → Linux errno translation ================================

    /// rmdir on a non-empty directory must surface as Linux ENOTEMPTY
    /// (39), not the macOS host value of 66 (which the guest's libc
    /// decodes as EREMOTE — the bug an integrator hit before this fix).
    #[test]
    fn rmdir_nonempty_returns_linux_enotempty() {
        let dir = tmpdir("t-rmdir-nonempty");
        std::fs::create_dir_all(dir.join("sub")).unwrap();
        std::fs::write(dir.join("sub/inside"), b"x").unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        let err = fs.rmdir(FUSE_ROOT_ID, OsStr::new("sub")).unwrap_err();
        // Linux ENOTEMPTY == 39; Errno is the negative form.
        assert_eq!(
            err, -39,
            "rmdir non-empty must surface as Linux ENOTEMPTY (39), got {err}"
        );
    }

    /// rmdir on an empty directory still succeeds (sanity that we
    /// didn't break the happy path while translating errnos).
    #[test]
    fn rmdir_empty_succeeds() {
        let dir = tmpdir("t-rmdir-empty");
        std::fs::create_dir_all(dir.join("sub")).unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        fs.rmdir(FUSE_ROOT_ID, OsStr::new("sub")).unwrap();
        assert!(!dir.join("sub").exists());
    }

    /// Table-driven check that the translator emits the Linux numbers
    /// for the divergent errnos we care about. On non-macOS hosts the
    /// translator is the identity, so the assertions are written
    /// against the per-host expectation.
    #[test]
    fn host_to_linux_errno_table() {
        // (macOS host value, Linux wire value) — only meaningful on macOS.
        // On Linux the host value == Linux value already, so we just
        // sanity-check identity for a handful of POSIX-shared codes.
        #[cfg(target_os = "macos")]
        {
            // EAGAIN/EWOULDBLOCK
            assert_eq!(host_to_linux_errno(35), 11);
            // EDEADLK
            assert_eq!(host_to_linux_errno(11), 35);
            // ENAMETOOLONG
            assert_eq!(host_to_linux_errno(63), 36);
            // ENOSYS
            assert_eq!(host_to_linux_errno(78), 38);
            // ENOTEMPTY  ← the regression
            assert_eq!(host_to_linux_errno(66), 39);
            // ELOOP
            assert_eq!(host_to_linux_errno(62), 40);
            // EOVERFLOW
            assert_eq!(host_to_linux_errno(84), 75);
            // ENOTSOCK
            assert_eq!(host_to_linux_errno(38), 88);
            // EOPNOTSUPP / ENOTSUP (== 102 on macOS, 95 on Linux)
            assert_eq!(host_to_linux_errno(102), 95);
            // ETIMEDOUT
            assert_eq!(host_to_linux_errno(60), 110);
            // ECONNREFUSED
            assert_eq!(host_to_linux_errno(61), 111);
            // EINPROGRESS  (connect() pending — the one called out in the spec)
            assert_eq!(host_to_linux_errno(36), 115);
            // ESTALE
            assert_eq!(host_to_linux_errno(70), 116);
            // ECANCELED
            assert_eq!(host_to_linux_errno(89), 125);
            // EOWNERDEAD
            assert_eq!(host_to_linux_errno(105), 130);
            // ENOTRECOVERABLE
            assert_eq!(host_to_linux_errno(104), 131);
        }

        // POSIX-identical 1..=34 must always pass through.
        assert_eq!(host_to_linux_errno(libc::EPERM), 1);
        assert_eq!(host_to_linux_errno(libc::ENOENT), 2);
        assert_eq!(host_to_linux_errno(libc::EINTR), 4);
        assert_eq!(host_to_linux_errno(libc::EIO), 5);
        assert_eq!(host_to_linux_errno(libc::EINVAL), 22);
        assert_eq!(host_to_linux_errno(libc::ERANGE), 34);
    }

    // === Setattr / Rename / Flush / Fsyncdir ==========================
    //
    // These cover the npm/pnpm/tar surface (chmod after extract,
    // truncate-on-write, atomic-rename, close(2)). 0.5.6 regression
    // suite.

    fn make_setattr(valid: u32) -> SetattrIn {
        SetattrIn {
            valid,
            padding: 0,
            fh: 0,
            size: 0,
            lock_owner: 0,
            atime: 0,
            mtime: 0,
            ctime: 0,
            atimensec: 0,
            mtimensec: 0,
            ctimensec: 0,
            mode: 0,
            unused4: 0,
            uid: 0,
            gid: 0,
            unused5: 0,
        }
    }

    #[test]
    fn setattr_chmod() {
        let dir = tmpdir("t-setattr-chmod");
        let path = dir.join("f");
        std::fs::write(&path, b"data").unwrap();
        // start at 0o644 so we can flip to 0o600 and verify.
        let mut perms = std::fs::metadata(&path).unwrap().permissions();
        std::os::unix::fs::PermissionsExt::set_mode(&mut perms, 0o644);
        std::fs::set_permissions(&path, perms).unwrap();

        let fs = PosixFs::new(&dir).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("f")).unwrap();

        let mut req = make_setattr(FATTR_MODE);
        req.mode = 0o600;
        let attr = fs.setattr(e.nodeid, None, req).unwrap();
        assert_eq!(attr.mode & 0o7777, 0o600);

        // Verify on host.
        let md = std::fs::metadata(&path).unwrap();
        let host_mode = std::os::unix::fs::PermissionsExt::mode(&md.permissions()) & 0o7777;
        assert_eq!(host_mode, 0o600);
    }

    #[test]
    fn setattr_truncate() {
        let dir = tmpdir("t-setattr-truncate");
        let path = dir.join("f");
        std::fs::write(&path, vec![0xABu8; 1024]).unwrap();

        let fs = PosixFs::new(&dir).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("f")).unwrap();

        let mut req = make_setattr(FATTR_SIZE);
        req.size = 10;
        let attr = fs.setattr(e.nodeid, None, req).unwrap();
        assert_eq!(attr.size, 10);
        assert_eq!(std::fs::metadata(&path).unwrap().len(), 10);
    }

    #[test]
    fn setattr_truncate_via_fh() {
        // ftruncate path (FH-bearing setattr — close-to-write pattern).
        let dir = tmpdir("t-setattr-truncate-fh");
        let path = dir.join("f");
        std::fs::write(&path, vec![0u8; 256]).unwrap();

        let fs = PosixFs::new(&dir).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("f")).unwrap();
        let fh = fs.open(e.nodeid, libc::O_RDWR as u32).unwrap();

        let mut req = make_setattr(FATTR_SIZE);
        req.size = 64;
        let attr = fs.setattr(e.nodeid, Some(fh), req).unwrap();
        assert_eq!(attr.size, 64);
        assert_eq!(std::fs::metadata(&path).unwrap().len(), 64);
        fs.release(e.nodeid, fh).unwrap();
    }

    #[test]
    fn setattr_utimens() {
        let dir = tmpdir("t-setattr-utimens");
        let path = dir.join("f");
        std::fs::write(&path, b"x").unwrap();

        let fs = PosixFs::new(&dir).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("f")).unwrap();

        // Pick an mtime well in the past so we can be sure the value
        // came from us and not the FS auto-stamping on write.
        let fixed_mtime: u64 = 1_500_000_000; // 2017-07-14
        let mut req = make_setattr(FATTR_MTIME);
        req.mtime = fixed_mtime;
        req.mtimensec = 0;
        let attr = fs.setattr(e.nodeid, None, req).unwrap();
        assert_eq!(attr.mtime, fixed_mtime);

        let md = std::fs::metadata(&path).unwrap();
        assert_eq!(md.mtime() as u64, fixed_mtime);
    }

    #[test]
    fn rename_within_dir() {
        let dir = tmpdir("t-rename-same");
        std::fs::write(dir.join("from"), b"payload").unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        let _ = fs.lookup(FUSE_ROOT_ID, OsStr::new("from")).unwrap();

        fs.rename(
            FUSE_ROOT_ID,
            OsStr::new("from"),
            FUSE_ROOT_ID,
            OsStr::new("to"),
            0,
        )
        .unwrap();

        assert!(!dir.join("from").exists());
        assert!(dir.join("to").exists());
        assert_eq!(std::fs::read(dir.join("to")).unwrap(), b"payload");

        // Lookup the new name — must succeed and reuse-or-allocate a nodeid.
        let new_e = fs.lookup(FUSE_ROOT_ID, OsStr::new("to")).unwrap();
        assert!(new_e.attr.size == 7);
    }

    #[test]
    fn rename_across_dirs() {
        let dir = tmpdir("t-rename-cross");
        std::fs::create_dir_all(dir.join("a")).unwrap();
        std::fs::create_dir_all(dir.join("b")).unwrap();
        std::fs::write(dir.join("a/file"), b"cross-dir").unwrap();

        let fs = PosixFs::new(&dir).unwrap();
        let a = fs.lookup(FUSE_ROOT_ID, OsStr::new("a")).unwrap();
        let b = fs.lookup(FUSE_ROOT_ID, OsStr::new("b")).unwrap();
        let _ = fs.lookup(a.nodeid, OsStr::new("file")).unwrap();

        fs.rename(
            a.nodeid,
            OsStr::new("file"),
            b.nodeid,
            OsStr::new("file"),
            0,
        )
        .unwrap();

        assert!(!dir.join("a/file").exists());
        assert_eq!(std::fs::read(dir.join("b/file")).unwrap(), b"cross-dir");
        // Re-lookup under the new parent.
        let new_e = fs.lookup(b.nodeid, OsStr::new("file")).unwrap();
        assert_eq!(new_e.attr.size, 9);
    }

    #[test]
    fn rename_atomic_write_pattern() {
        // The motivating use case: write to .tmp, rename to final.
        let dir = tmpdir("t-rename-atomic");
        std::fs::write(dir.join("file.tmp"), b"new contents").unwrap();
        let fs = PosixFs::new(&dir).unwrap();

        fs.rename(
            FUSE_ROOT_ID,
            OsStr::new("file.tmp"),
            FUSE_ROOT_ID,
            OsStr::new("file"),
            0,
        )
        .unwrap();
        assert!(!dir.join("file.tmp").exists());
        assert_eq!(std::fs::read(dir.join("file")).unwrap(), b"new contents");
    }

    #[test]
    fn rename_rejects_nonzero_flags() {
        let dir = tmpdir("t-rename-flags");
        std::fs::write(dir.join("a"), b"x").unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        let err = fs
            .rename(
                FUSE_ROOT_ID,
                OsStr::new("a"),
                FUSE_ROOT_ID,
                OsStr::new("b"),
                1, // RENAME_NOREPLACE
            )
            .unwrap_err();
        assert_eq!(err, EINVAL);
    }

    #[test]
    fn flush_is_noop() {
        let dir = tmpdir("t-flush");
        std::fs::write(dir.join("f"), b"x").unwrap();
        let fs = PosixFs::new(&dir).unwrap();
        let e = fs.lookup(FUSE_ROOT_ID, OsStr::new("f")).unwrap();
        let fh = fs.open(e.nodeid, libc::O_RDONLY as u32).unwrap();
        fs.flush(e.nodeid, fh).unwrap();
        fs.release(e.nodeid, fh).unwrap();
    }

    #[test]
    fn flush_unknown_fh_returns_ebadf() {
        let dir = tmpdir("t-flush-ebadf");
        let fs = PosixFs::new(&dir).unwrap();
        let err = fs.flush(FUSE_ROOT_ID, 999_999).unwrap_err();
        assert_eq!(err, EBADF);
    }

    #[test]
    fn fsyncdir_is_noop() {
        let dir = tmpdir("t-fsyncdir");
        let fs = PosixFs::new(&dir).unwrap();
        let dh = fs.opendir(FUSE_ROOT_ID, 0).unwrap();
        fs.fsyncdir(FUSE_ROOT_ID, dh, false).unwrap();
        fs.fsyncdir(FUSE_ROOT_ID, dh, true).unwrap();
        fs.releasedir(FUSE_ROOT_ID, dh).unwrap();
    }

    // === Integration test stub (in-VM) ================================

    /// Boot a python image, virtio-fs mount a tmpdir, exec
    /// `ln -s a b && readlink b`, assert. This is the end-to-end test
    /// that proves npm install works inside guests — but kicking off a
    /// real VM requires the kernel patch series + the rest of the
    /// supermachine harness. Skipped for now; tracked as a follow-up
    /// once the read/write symlink path lands in the in-tree kernel.
    #[test]
    #[ignore = "requires a booted VM; turn this into a tsi_loopback-style integration test"]
    fn symlink_inside_guest_via_python_image() {
        // TODO(0.5.5+): build a python image, mount a host tmpdir, run
        // `python -c "import os; os.symlink('a','b'); print(os.readlink('b'))"`
        // via Vm::exec, assert stdout == "a\n".
    }

    // === 0.7.6 snapshot persistence tests ====================================

    /// snapshot_state → restore_state round-trip preserves the
    /// (nodeid → host_path) table, the (parent, name) → child_nodeid
    /// table, and `next_nodeid`. This is the foundation of the
    /// warm-restore dentry-cache fix: without it, the daemon starts
    /// each warm restore with an empty table and the guest's cached
    /// nodeids reference paths the daemon can't resolve.
    #[test]
    fn snapshot_round_trip_preserves_inode_and_children_tables() {
        let dir = tmpdir("t-snap-roundtrip");
        std::fs::create_dir_all(dir.join("dist/scripts")).unwrap();
        std::fs::write(dir.join("dist/cli.js"), b"#!/usr/bin/env node\n").unwrap();
        std::fs::write(
            dir.join("dist/scripts/probeRunnerServer.js"),
            b"// probe runner\n",
        )
        .unwrap();
        let fs1 = PosixFs::new(&dir).unwrap();
        // Walk a few paths so the inode + children tables get populated.
        let dist = fs1.lookup(FUSE_ROOT_ID, OsStr::new("dist")).unwrap();
        let _cli = fs1.lookup(dist.nodeid, OsStr::new("cli.js")).unwrap();
        let scripts = fs1.lookup(dist.nodeid, OsStr::new("scripts")).unwrap();
        let probe = fs1
            .lookup(scripts.nodeid, OsStr::new("probeRunnerServer.js"))
            .unwrap();

        // Capture state.
        let blob = fs1.snapshot_state();

        // Fresh PosixFs with only FUSE_ROOT_ID populated. Restore.
        let fs2 = PosixFs::new(&dir).unwrap();
        fs2.restore_state(&blob).expect("restore should succeed");

        // The cached child nodeids — the same ones the guest's
        // dentry cache holds — must resolve to the same host paths.
        // `host_path_of` exercises the inode table; the LOOKUP-by-
        // (parent,name) reuse exercises the children table.
        //
        // `PosixFs::new` canonicalises `dir` (on macOS, /var becomes
        // /private/var), so compare against the canonical form.
        let root = std::fs::canonicalize(&dir).unwrap();
        let st2 = fs2.st.lock().unwrap();
        assert_eq!(
            st2.inodes.get(&dist.nodeid).map(|i| &i.host_path),
            Some(&root.join("dist"))
        );
        assert_eq!(
            st2.inodes.get(&scripts.nodeid).map(|i| &i.host_path),
            Some(&root.join("dist").join("scripts"))
        );
        assert_eq!(
            st2.inodes.get(&probe.nodeid).map(|i| &i.host_path),
            Some(
                &root
                    .join("dist")
                    .join("scripts")
                    .join("probeRunnerServer.js")
            )
        );
        // Children-by-name table — what re-LOOKUP from the guest
        // hits on attr_valid expiry.
        let key = (dist.nodeid, b"scripts".to_vec());
        assert_eq!(st2.children.get(&key).copied(), Some(scripts.nodeid));
        let key = (scripts.nodeid, b"probeRunnerServer.js".to_vec());
        assert_eq!(st2.children.get(&key).copied(), Some(probe.nodeid));
        // next_nodeid must be at least the post-walk high-water mark.
        assert!(st2.next_nodeid > probe.nodeid);
    }

    /// Restore on a fresh PosixFs whose root path differs from the
    /// blob's root path is rejected. The (nodeid → host_path) mapping
    /// is meaningless across mount roots; silently accepting it would
    /// surface as cross-mount path leakage.
    #[test]
    fn restore_rejects_root_mismatch() {
        let dir1 = tmpdir("t-snap-root1");
        let dir2 = tmpdir("t-snap-root2");
        let fs1 = PosixFs::new(&dir1).unwrap();
        let blob = fs1.snapshot_state();
        let fs2 = PosixFs::new(&dir2).unwrap();
        let err = fs2.restore_state(&blob).unwrap_err();
        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
        let msg = err.to_string();
        assert!(msg.contains("root mismatch"), "unexpected error: {msg}");
    }

    /// Truncated, wrong-magic, or wrong-version blobs are rejected
    /// cleanly. (The runtime falls back to lazy-LOOKUP, which is the
    /// pre-0.7.6 behaviour — better than crashing.)
    #[test]
    fn restore_rejects_malformed_blobs() {
        let dir = tmpdir("t-snap-malformed");
        let fs = PosixFs::new(&dir).unwrap();

        // Bad magic
        let mut bad = b"XXXX".to_vec();
        bad.extend_from_slice(&1u32.to_le_bytes());
        let err = fs.restore_state(&bad).unwrap_err();
        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);

        // Truncated
        let err = fs.restore_state(b"PFSS").unwrap_err();
        assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);

        // Bad version
        let mut bad = b"PFSS".to_vec();
        bad.extend_from_slice(&99u32.to_le_bytes());
        let err = fs.restore_state(&bad).unwrap_err();
        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
        assert!(err.to_string().contains("version 99"));
    }

    // -----------------------------------------------------------------
    // PFSS v2 cycle-restore tests — the SIGBUS-on-cycle-restore fix.
    //
    // Pre-fix: `handles` was process-local, so any guest fh inherited
    // from a pre-snapshot lifetime returned EBADF post-restore. The
    // Linux virtio-fs driver caches the fh inside its `fuse_file` and
    // keeps sending FUSE_READ(fh=X); the EBADF propagates through the
    // page-fault handler as I/O error → SIGBUS for cold-mapped pages.
    //
    // The fix persists (fh, nodeid, flags) metadata across snapshot/
    // restore and lazily re-opens the host fd on first I/O. Tests
    // below exercise the persistence, the lazy reopen, the flag mask,
    // the release-of-lazy-handle edge case, the v1-blob backward-
    // compat path, and next_fh anti-collision behaviour.

    /// snapshot_state on a v2 PosixFs preserves every open `fh` →
    /// `(nodeid, flags)` mapping. Restoring into a fresh PosixFs
    /// (pointing at the same host root) reports the same fh values
    /// and the metadata round-trips byte-exact.
    #[test]
    fn snapshot_v2_round_trip_preserves_handles() {
        let dir = tmpdir("t-snap-v2-handles");
        std::fs::write(dir.join("a.bin"), b"alpha").unwrap();
        std::fs::write(dir.join("b.bin"), b"beta-content-larger").unwrap();
        let fs1 = PosixFs::new(&dir).unwrap();
        let a = fs1.lookup(FUSE_ROOT_ID, OsStr::new("a.bin")).unwrap();
        let b = fs1.lookup(FUSE_ROOT_ID, OsStr::new("b.bin")).unwrap();
        let fh_a = fs1.open(a.nodeid, libc::O_RDONLY as u32).unwrap();
        let fh_b = fs1.open(b.nodeid, libc::O_RDWR as u32).unwrap();

        let blob = fs1.snapshot_state();

        let fs2 = PosixFs::new(&dir).unwrap();
        fs2.restore_state(&blob).expect("restore should succeed");

        let st = fs2.st.lock().unwrap();
        let entry_a = st.handles.get(&fh_a).expect("fh_a survived restore");
        let entry_b = st.handles.get(&fh_b).expect("fh_b survived restore");
        // fd is None pre-first-use post-restore.
        assert!(entry_a.fd.is_none(), "restored handle starts as lazy");
        assert!(entry_b.fd.is_none(), "restored handle starts as lazy");
        // nodeid round-trips byte-exact.
        assert_eq!(entry_a.nodeid, a.nodeid);
        assert_eq!(entry_b.nodeid, b.nodeid);
        // flags round-trip byte-exact (open's stored flag is the
        // access mode passed to libc::open).
        assert_eq!(entry_a.flags, libc::O_RDONLY as u32);
        assert_eq!(entry_b.flags, libc::O_RDWR as u32);
    }

    /// After a snapshot/restore the next FUSE_READ on a restored
    /// handle lazy-opens the host fd and returns the correct bytes.
    /// This is the direct fix for the SIGBUS-on-cold-mapped-page bug:
    /// without lazy reopen, this read would EBADF.
    #[test]
    fn lazy_reopen_on_first_read() {
        let dir = tmpdir("t-snap-v2-lazy-read");
        let content = b"PFSS v2 lazy reopen content";
        std::fs::write(dir.join("payload"), content).unwrap();
        let fs1 = PosixFs::new(&dir).unwrap();
        let e = fs1.lookup(FUSE_ROOT_ID, OsStr::new("payload")).unwrap();
        let fh = fs1.open(e.nodeid, libc::O_RDONLY as u32).unwrap();

        let blob = fs1.snapshot_state();
        // Drop fs1 entirely — that closes all process-local fds.
        drop(fs1);

        let fs2 = PosixFs::new(&dir).unwrap();
        fs2.restore_state(&blob).expect("restore should succeed");

        // Sanity: handle exists but the fd is lazy (None).
        {
            let st = fs2.st.lock().unwrap();
            let entry = st.handles.get(&fh).unwrap();
            assert!(entry.fd.is_none(), "lazy before first I/O");
        }

        // First read triggers the lazy reopen.
        let buf = fs2.read(e.nodeid, fh, 0, 64).unwrap();
        assert_eq!(buf, content);

        // The lazy fd is now installed.
        {
            let st = fs2.st.lock().unwrap();
            let entry = st.handles.get(&fh).unwrap();
            assert!(entry.fd.is_some(), "fd installed after first I/O");
        }

        // Second read on the same fh short-circuits to the now-open
        // fd; result is still correct.
        let buf2 = fs2.read(e.nodeid, fh, 0, 64).unwrap();
        assert_eq!(buf2, content);
    }

    /// Lazy reopen strips destructive flag bits (O_TRUNC most
    /// importantly). A handle whose original open flags included
    /// O_TRUNC must reopen WITHOUT truncating the file content
    /// restored from disk.
    #[test]
    fn lazy_reopen_strips_destructive_flags() {
        let dir = tmpdir("t-snap-v2-flag-mask");
        let original = b"DO NOT TRUNCATE ME ON RESTORE";
        std::fs::write(dir.join("important"), original).unwrap();
        let fs1 = PosixFs::new(&dir).unwrap();
        let e = fs1.lookup(FUSE_ROOT_ID, OsStr::new("important")).unwrap();

        // Manually inject a HandleEntry as if the original open was
        // RDWR|TRUNC|APPEND|CREAT — those are exactly the bits the
        // REOPEN_FLAG_MASK must strip. We can't go through the
        // normal `open` path because it masks CREAT/EXCL upstream;
        // this test specifically validates the LAZY-REOPEN mask.
        let fh = {
            let mut st = fs1.st.lock().unwrap();
            let fh = st.next_fh;
            st.next_fh += 1;
            let nasty_flags =
                (libc::O_RDWR | libc::O_TRUNC | libc::O_APPEND | libc::O_CREAT) as u32;
            st.handles.insert(
                fh,
                HandleEntry {
                    fd: None,
                    nodeid: e.nodeid,
                    flags: nasty_flags,
                    dir_snapshot: None,
                },
            );
            fh
        };

        let blob = fs1.snapshot_state();
        drop(fs1);

        let fs2 = PosixFs::new(&dir).unwrap();
        fs2.restore_state(&blob).expect("restore should succeed");

        // First read triggers the lazy reopen. With O_TRUNC NOT
        // stripped, the file would be truncated to 0 bytes and the
        // read would return empty. With the mask working, the read
        // returns the original content.
        let buf = fs2.read(e.nodeid, fh, 0, 64).unwrap();
        assert_eq!(buf, original, "lazy reopen must NOT trigger O_TRUNC");

        // Re-stat the file on disk: still original size.
        let md = std::fs::metadata(dir.join("important")).unwrap();
        assert_eq!(md.len(), original.len() as u64);
    }

    /// A handle that survived snapshot/restore but was never re-
    /// opened can still be released cleanly (no double-free, no
    /// panic, no EBADF). This is the trivial close path — the
    /// HandleEntry's fd is None, drop() is a no-op, the metadata is
    /// removed.
    #[test]
    fn release_works_on_lazy_handle() {
        let dir = tmpdir("t-snap-v2-release-lazy");
        std::fs::write(dir.join("f"), b"some bytes").unwrap();
        let fs1 = PosixFs::new(&dir).unwrap();
        let e = fs1.lookup(FUSE_ROOT_ID, OsStr::new("f")).unwrap();
        let fh = fs1.open(e.nodeid, libc::O_RDONLY as u32).unwrap();
        let blob = fs1.snapshot_state();
        drop(fs1);

        let fs2 = PosixFs::new(&dir).unwrap();
        fs2.restore_state(&blob).expect("restore should succeed");

        // Release without ever touching the fh — the inner fd is
        // None at this point.
        fs2.release(e.nodeid, fh).expect("release of lazy fh ok");

        // Slot is gone.
        let st = fs2.st.lock().unwrap();
        assert!(!st.handles.contains_key(&fh));
    }

    /// Old PFSS v1 blobs (pre-0.7.47) must still load. They produce
    /// an empty handle table — i.e. the pre-fix behaviour. The
    /// SIGBUS bug still surfaces for cold-mapped pages with a v1
    /// blob, but the daemon doesn't refuse to load.
    #[test]
    fn version_1_blob_is_accepted_with_no_handles() {
        let dir = tmpdir("t-snap-v1-compat");
        std::fs::write(dir.join("root-file"), b"x").unwrap();
        let fs = PosixFs::new(&dir).unwrap();

        // Hand-build a minimal v1 blob: magic + version=1 +
        // next_nodeid + inode_count=1 (root) + children_count=0.
        // The root entry must match the canonical mount root.
        let root_path = std::fs::canonicalize(&dir).unwrap();
        let root_bytes = root_path.as_os_str().as_bytes();
        let mut blob = Vec::new();
        blob.extend_from_slice(b"PFSS");
        blob.extend_from_slice(&1u32.to_le_bytes()); // version = 1
        blob.extend_from_slice(&100u64.to_le_bytes()); // next_nodeid
        blob.extend_from_slice(&1u64.to_le_bytes()); // inode_count
        blob.extend_from_slice(&FUSE_ROOT_ID.to_le_bytes()); // root nodeid
        blob.push(1u8); // kind = Dir
        blob.extend_from_slice(&(root_bytes.len() as u32).to_le_bytes());
        blob.extend_from_slice(root_bytes);
        blob.extend_from_slice(&0u64.to_le_bytes()); // children_count = 0
                                                     // NO v2 trailer.

        fs.restore_state(&blob).expect("v1 blob accepted");
        let st = fs.st.lock().unwrap();
        // Empty handle table (pre-fix behaviour).
        assert!(st.handles.is_empty());
        // next_nodeid was bumped.
        assert!(st.next_nodeid >= 100);
        // next_fh keeps the constructor default (1) — no handles
        // to bump past.
        assert_eq!(st.next_fh, 1);
    }

    /// After a restore that brings in fh=42, the next `open()` must
    /// allocate an fh > 42, not collide. Otherwise the guest's
    /// cached fh=42 would shadow a fresh open of a totally
    /// different file — wrong file content delivered to a cached
    /// fh, very bad.
    #[test]
    fn next_fh_avoids_restored_collision() {
        let dir = tmpdir("t-snap-v2-next-fh");
        std::fs::write(dir.join("first"), b"1").unwrap();
        std::fs::write(dir.join("second"), b"2").unwrap();
        let fs1 = PosixFs::new(&dir).unwrap();
        let a = fs1.lookup(FUSE_ROOT_ID, OsStr::new("first")).unwrap();
        // Burn through some fhs.
        for _ in 0..5 {
            let _ = fs1.open(a.nodeid, libc::O_RDONLY as u32).unwrap();
        }
        // The next allocated fh will be > 5. Capture the max we've
        // seen.
        let max_fh_before = {
            let st = fs1.st.lock().unwrap();
            st.handles.keys().copied().max().unwrap()
        };

        let blob = fs1.snapshot_state();
        drop(fs1);

        let fs2 = PosixFs::new(&dir).unwrap();
        // After restore, next_fh must be > max_fh_before so a new
        // open() can't collide.
        fs2.restore_state(&blob).expect("restore should succeed");
        {
            let st = fs2.st.lock().unwrap();
            assert!(
                st.next_fh > max_fh_before,
                "next_fh ({}) must be greater than max restored fh ({})",
                st.next_fh,
                max_fh_before
            );
        }

        // Opening a fresh file on fs2 must produce an fh > the
        // max restored one.
        let b = fs2.lookup(FUSE_ROOT_ID, OsStr::new("second")).unwrap();
        let new_fh = fs2.open(b.nodeid, libc::O_RDONLY as u32).unwrap();
        assert!(
            new_fh > max_fh_before,
            "new open fh ({new_fh}) must not collide with restored max ({max_fh_before})"
        );
    }
}