// 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::unix::ffi::OsStrExt;
use std::os::unix::fs::MetadataExt;
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
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, ENOENT, ENOSPC, ENOTDIR,
    EISDIR, EIO, EPERM,
};
use crate::vmm::resources::SymlinkPolicy;
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,
};

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

struct State {
    /// nodeid -> InodeInfo.
    inodes: BTreeMap<u64, InodeInfo>,
    /// (parent, name) -> nodeid. Populated by lookup.
    children: BTreeMap<(u64, Vec<u8>), u64>,
    /// fh -> OwnedFd. Closed on `release` / `releasedir`.
    handles: BTreeMap<u64, OwnedFd>,
    /// 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 kqueue watcher thread.
struct Watcher {
    /// Shared with the watcher thread via Arc. The thread reads kq,
    /// looks up the (kq_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.
    kq: libc::c_int,
    /// Stop signal — set by drop, polled by the watcher thread.
    stop: AtomicBool,
    /// Notifier the watcher thread invokes when a kevent fires. None
    /// until set_notifier is called.
    notifier: Mutex<Option<Arc<dyn Notifier>>>,
    /// Map kqueue ident (an fd we duped specifically for kqueue) →
    /// (nodeid, parent_nodeid, name, owned_fd). 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>,
    _owned_fd: OwnedFd,
}

impl Drop for Watcher {
    fn drop(&mut self) {
        self.inner.stop.store(true, Ordering::Release);
        // Poke the kqueue with a USER event so the thread wakes from kevent.
        let trigger = 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 mut tr = trigger;
        unsafe {
            let _ = libc::kevent(self.inner.kq, &mut tr as *mut _, 1, std::ptr::null_mut(), 0, std::ptr::null());
        }
        if let Some(h) = self.handle.take() {
            let _ = h.join();
        }
        unsafe {
            libc::close(self.inner.kq);
        }
    }
}

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>,
    ) -> 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, owned);
        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(());
        }
        let kq = unsafe { libc::kqueue() };
        if kq < 0 {
            return Err(std::io::Error::last_os_error());
        }
        let inner = Arc::new(WatcherInner {
            kq,
            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,
        };
        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) };

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

        let ev = 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 mut event = ev;
        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;
        }
        watched.insert(
            fd,
            WatchedEntry {
                nodeid,
                parent_nodeid,
                name,
                _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.
    //
    // Excludes:
    //   * `handles` / `next_fh` — open file handles can't survive
    //     daemon-process restart (host fds are process-local). Any
    //     guest fd open at snapshot time is invalid post-restore;
    //     the guest must reopen. In practice nothing user-facing
    //     keeps fds across a pool snapshot/restore boundary.
    //   * `dax_mmaps` — process-local mmaps; same reason. The DAX
    //     window's shm contents persist through the snapshot path
    //     anyway (guest sees the same mapped bytes).
    //   * `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. 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 1)
        //   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
        let st = self.st.lock().unwrap();
        let mut out = Vec::with_capacity(64 + st.inodes.len() * 64 + st.children.len() * 32);
        out.extend_from_slice(b"PFSS");
        out.extend_from_slice(&1u32.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());
        }
        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)?;
        if version != 1 {
            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);
        }
        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);
        Ok(())
    }
}

/// 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.
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 st = self.st.lock().unwrap();
            if st.partial_writes.contains_key(&fh) {
                let raw = match st.handles.get(&fh) {
                    Some(o) => o.as_raw_fd(),
                    None => return Err(EBADF),
                };
                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)?;
        match self.kind_of(nodeid)? {
            Kind::Dir => 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;
            st.handles.insert(fh, owned);
            // 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, owned);
        Ok(fh)
    }

    fn read(&self, _nodeid: u64, fh: u64, offset: u64, size: u32) -> Result<Vec<u8>, Errno> {
        let st = self.st.lock().unwrap();
        let raw = st.handles.get(&fh).ok_or(EBADF)?.as_raw_fd();
        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.
        let (owned, partial) = {
            let mut st = self.st.lock().unwrap();
            let owned = st.handles.remove(&fh).ok_or(EBADF)?;
            let partial = st.partial_writes.remove(&fh);
            (owned, partial)
        };

        // Non-aotcache files: trivial close (drop the fd).
        let Some(info) = partial else {
            drop(owned);
            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.
        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 st = self.st.lock().unwrap();
        let raw = st.handles.get(&fh).ok_or(EBADF)?.as_raw_fd();
        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 st = self.st.lock().unwrap();
        let raw = st.handles.get(&fh).ok_or(EBADF)?.as_raw_fd();
        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 {
            // F_FULLFSYNC may return ENOTSUP on some filesystem
            // types (e.g. some network FSes that don't expose
            // barrier control). Fall back to plain fsync in that
            // case — best-effort beats hard-failing the guest's
            // FUSE_FSYNC.
            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
            }
        };
        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 fd = unsafe { libc::open(c.as_ptr(), libc::O_RDONLY | libc::O_DIRECTORY) };
        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, owned);
        Ok(fh)
    }

    fn readdir(
        &self,
        nodeid: u64,
        _fh: u64,
        offset: u64,
        _size: u32,
    ) -> Result<Vec<DirEntry>, Errno> {
        // We rebuild the dirent list each call rather than caching
        // (real production should cache per-fh between consecutive
        // offsets). Use std::fs::read_dir for portability.
        let path = self.host_path_of(nodeid)?;
        let rd = std::fs::read_dir(&path).map_err(|e| io_err_to_linux(&e))?;
        let mut out = Vec::new();
        for (i, entry_res) in rd.enumerate() {
            if (i as u64) < offset {
                continue;
            }
            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);
            out.push(DirEntry {
                ino,
                name: name_bytes_owned,
                typ,
            });
        }
        Ok(out)
    }

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

    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,
            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,
                );
            }
            return Err(e);
        }
        self.create_finalize(parent, name, &final_path, fd, partial_info)
    }

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

    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).
        let fd_raw: Option<libc::c_int> = if let Some(handle) = fh {
            let st = self.st.lock().unwrap();
            Some(st.handles.get(&handle).ok_or(EBADF)?.as_raw_fd())
        } 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).
        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.
        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 {
            let st = self.st.lock().unwrap();
            let raw = st.handles.get(&fh).ok_or(EBADF)?.as_raw_fd();
            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 std::env::var_os("SUPERMACHINE_ROSETTA_TRACE").is_some() {
            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 std::env::var_os("SUPERMACHINE_ROSETTA_TRACE").is_some() {
                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 std::env::var_os("SUPERMACHINE_ROSETTA_TRACE").is_some() {
                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 1 + 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 std::env::var_os("SUPERMACHINE_ROSETTA_TRACE").is_some() {
            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.
fn run_watcher(inner: Arc<WatcherInner>) {
    // Register a USER event so Drop can wake us.
    let wakeup = libc::kevent {
        ident: 0,
        filter: libc::EVFILT_USER,
        flags: libc::EV_ADD | libc::EV_CLEAR,
        fflags: 0,
        data: 0,
        udata: std::ptr::null_mut(),
    };
    let mut w = wakeup;
    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;
            }
            // Look up the watched entry for the kqueue ident.
            let fd = ev.ident as libc::c_int;
            let entry = inner
                .watched
                .lock()
                .unwrap()
                .get(&fd)
                .map(|e| (e.nodeid, e.parent_nodeid, e.name.clone()));
            let Some((nodeid, parent_nodeid, name)) = entry else { continue };
            if let Some(n) = inner.notifier.lock().unwrap().as_ref() {
                // INVAL_INODE is always safe to emit — it just refreshes
                // the kernel's cached pages and attrs for THIS inode.
                //   (off=0, len=-1) → invalidates data pages
                //   (off=0, len=0)  → invalidates attrs (size, mtime).
                //     The (0,0) sentinel is FUSE protocol specific.
                n.invalidate_inode(nodeid, 0, -1);
                n.invalidate_inode(nodeid, 0, 0);
                // INVAL_ENTRY is the heavier hammer — drops the kernel's
                // CHILD-OF-PARENT dentry mapping so the next access
                // re-LOOKUPs the name. Only safe-and-necessary when the
                // identity of the entry has changed (rename/delete) —
                // emitting it on benign attribute or content changes
                // breaks any mount stacked on top of this inode: the
                // kernel re-LOOKUPs the dentry through OUR FUSE backend,
                // gets a fresh inode, and does NOT re-check the mount
                // table for the new inode (the mount was on the OLD
                // inode's mount-point dentry). Result: subsequent
                // accesses bypass the overlay mount entirely.
                //
                // Concrete repro pre-fix: a guest mounts mountB on top
                // of /workspace/node_modules. Some process does a stat
                // that updates atime on a file under node_modules. macOS
                // kqueue fires NOTE_ATTRIB on the watched file. We emit
                // INVAL_ENTRY for that file. The KERNEL invalidates the
                // dentry. Next access re-LOOKUPs the file → goes through
                // OUR mount (workspace), NOT mountB. Mount overlay
                // silently bypassed. Integrator hit this as "fs.existsSync
                // returns swapped answers after require.resolve" — Node's
                // module resolver does stat()s that trigger atime updates
                // which trigger our spurious INVAL_ENTRY storm.
                if ev.fflags & (libc::NOTE_DELETE | libc::NOTE_RENAME) != 0 {
                    n.invalidate_entry(parent_nodeid, &name);
                }
            }
            // If the file was deleted or renamed, drop the watch so
            // we don't keep an orphan fd.
            if ev.fflags & (libc::NOTE_DELETE | libc::NOTE_RENAME) != 0 {
                inner.watched.lock().unwrap().remove(&fd);
            }
        }
    }
}

// Unused-import suppression — std re-exports are conditional on macOS file_type extensions.
use std::os::unix::fs::FileTypeExt;
#[allow(unused_imports)]
use std::convert::TryFrom;
#[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();
    }

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