heddle-mount 0.15.5

// SPDX-License-Identifier: Apache-2.0
//! Unit tests for the content-addressed mount core.
//!
//! These exercise [`ContentAddressedMount`] directly, without going
//! through any platform shell. They use the same `Repository` fixture
//! that `crates/repo` uses in its own tests: an init-default repo in
//! a tempdir, with files written into the worktree and snapshotted
//! to advance `main`.

use std::{
    ffi::OsStr,
    fs,
    path::Path,
    sync::{
        Arc,
        atomic::{AtomicUsize, Ordering},
    },
};

use objects::{
    object::{
        Action, ActionId, Attribution, Blob, ContentHash, Principal, State, StateId, ThreadName,
        Tree, TreeEntry,
    },
    store::{ObjectStore, ShallowInfo},
    util::gitlink_placeholder_bytes,
};
use oplog::OpLog;
use refs::RefManager;
use repo::{RepoConfig, Repository};
use sley::ObjectId as GitObjectId;
use tempfile::TempDir;

use crate::{
    core::{ContentAddressedMount, MAX_MOUNT_HOT_FILE_SIZE},
    error::MountError,
    shell::{NodeId, NodeKind, PlatformShell},
};

/// Shared test mocks. Lives under `tests::mocks` so the per-platform
/// adapter unit tests (FUSE on Linux, FSKit on macOS, ProjFS on
/// Windows) can reuse the same in-memory shell without duplicating
/// it inline.
///
/// Gated on the features that actually consume the mocks so OSS-only
/// builds (no `fuse` / `fskit` / `projfs`) don't trip clippy's
/// `-D warnings` on dead test code.
#[cfg(any(
    all(target_os = "linux", feature = "fuse"),
    all(target_os = "macos", feature = "fskit"),
    all(target_os = "windows", feature = "projfs"),
))]
pub(crate) mod mocks {
    use std::{
        ffi::OsStr,
        sync::{
            Arc,
            atomic::{AtomicUsize, Ordering},
        },
        time::UNIX_EPOCH,
    };

    use crate::{
        error::{MountError, Result},
        shell::{Attrs, DIR_UNIX_MODE, Entry, NodeId, NodeKind, PlatformShell},
    };

    /// Trivial in-memory shell that lets adapter unit tests validate
    /// the session construct-and-drop lifecycle without needing a real
    /// `ContentAddressedMount` (which requires a `Repository`).
    ///
    /// Increments `drops` exactly once when the shell is finally
    /// dropped, so a test that boxes the shell into a C ABI handle can
    /// assert the box was reclaimed.
    ///
    /// `dead_code` is silenced because this is only consumed by the
    /// FSKit / ProjFS unit tests — the FUSE shell tests use
    /// [`PanicShell`] but never [`CountingShell`], and clippy's
    /// `-D warnings` would otherwise fail the Linux+fuse build.
    #[allow(dead_code)]
    pub struct CountingShell {
        pub drops: Arc<AtomicUsize>,
    }

    #[allow(dead_code)]
    impl CountingShell {
        pub fn new() -> (Self, Arc<AtomicUsize>) {
            let drops = Arc::new(AtomicUsize::new(0));
            (
                Self {
                    drops: Arc::clone(&drops),
                },
                drops,
            )
        }
    }

    impl Drop for CountingShell {
        fn drop(&mut self) {
            self.drops.fetch_add(1, Ordering::SeqCst);
        }
    }

    impl PlatformShell for CountingShell {
        fn lookup(&self, _parent: NodeId, _name: &OsStr) -> Result<Option<Entry>> {
            Ok(None)
        }
        fn read(&self, _node: NodeId, _offset: u64, _buf: &mut [u8]) -> Result<usize> {
            Ok(0)
        }
        fn write(&self, _node: NodeId, _offset: u64, _data: &[u8]) -> Result<usize> {
            Err(MountError::ReadOnly)
        }
        fn enumerate(&self, _dir: NodeId) -> Result<Vec<Entry>> {
            Ok(vec![])
        }
        fn attrs(&self, node: NodeId) -> Result<Attrs> {
            Ok(Attrs {
                node,
                kind: NodeKind::Directory,
                size: 0,
                unix_mode: DIR_UNIX_MODE,
                nlink: 2,
                mtime: UNIX_EPOCH,
            })
        }
        fn invalidate(&self, _node: NodeId) -> Result<()> {
            Ok(())
        }
    }

    /// Shell that panics on every PlatformShell call. Used by the
    /// FFI panic-resilience tests to drive an unwind into a
    /// trampoline body — the `catch_unwind` wrappers in
    /// `fskit::guarded_c_int` and `projfs::guarded_hresult` must
    /// translate the panic into `EIO` / a Win32 I/O HRESULT instead
    /// of letting the unwind cross the C ABI boundary (which Rust
    /// ≥1.81 turns into an abort that would crash the host process
    /// and every projected/materialised volume with it).
    ///
    /// `dead_code` is silenced because not every feature flag pulls
    /// in a consumer: the Linux+fuse build uses it via
    /// `fuse::tests::guard_call_translates_panic_to_eio`, but a
    /// Windows+projfs-only build wires `guarded_hresult` through its
    /// closure tests (no PanicShell required), so the type sits
    /// dormant on that feature set. clippy's `-D warnings` would
    /// otherwise break the Windows ProjFS clippy gate.
    #[allow(dead_code)]
    pub struct PanicShell;

    impl PlatformShell for PanicShell {
        fn lookup(&self, _parent: NodeId, _name: &OsStr) -> Result<Option<Entry>> {
            panic!("panic-shell: lookup intentionally panics")
        }
        fn read(&self, _node: NodeId, _offset: u64, _buf: &mut [u8]) -> Result<usize> {
            panic!("panic-shell: read intentionally panics")
        }
        fn write(&self, _node: NodeId, _offset: u64, _data: &[u8]) -> Result<usize> {
            panic!("panic-shell: write intentionally panics")
        }
        fn enumerate(&self, _dir: NodeId) -> Result<Vec<Entry>> {
            panic!("panic-shell: enumerate intentionally panics")
        }
        fn attrs(&self, _node: NodeId) -> Result<Attrs> {
            panic!("panic-shell: attrs intentionally panics")
        }
        fn invalidate(&self, _node: NodeId) -> Result<()> {
            panic!("panic-shell: invalidate intentionally panics")
        }
    }
}

/// Build a repository with a small, deterministic tree:
///
/// ```text
/// hello.txt    "world"
/// nested/
///   inner.txt  "deep"
///   note.md    "# heading\nbody\n"
/// run.sh       executable, "#!/bin/sh\n"
/// ```
fn fixture() -> (TempDir, Repository) {
    let temp = TempDir::new().unwrap();
    let repo = Repository::init_default(temp.path()).unwrap();
    fs::write(temp.path().join("hello.txt"), b"world").unwrap();
    fs::create_dir_all(temp.path().join("nested")).unwrap();
    fs::write(temp.path().join("nested/inner.txt"), b"deep").unwrap();
    fs::write(temp.path().join("nested/note.md"), b"# heading\nbody\n").unwrap();
    let run_path = temp.path().join("run.sh");
    fs::write(&run_path, b"#!/bin/sh\n").unwrap();
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        let mut perms = fs::metadata(&run_path).unwrap().permissions();
        perms.set_mode(0o755);
        fs::set_permissions(&run_path, perms).unwrap();
    }
    repo.snapshot(Some("fixture".into()), None).unwrap();
    (temp, repo)
}

fn open_mount_test_repo_with_store<S: ObjectStore>(
    heddle_dir: impl AsRef<Path>,
    store: S,
) -> Repository<RefManager, OpLog, S> {
    let heddle_dir = heddle_dir.as_ref().to_path_buf();
    let root = heddle_dir
        .parent()
        .expect("test heddle dir should live under a worktree root")
        .to_path_buf();
    let config = RepoConfig::load(&heddle_dir.join("config.toml")).unwrap();
    let refs = RefManager::new(&heddle_dir);
    let oplog = OpLog::new_unattributed(&heddle_dir);
    let shallow = ShallowInfo::load(&heddle_dir).unwrap();
    Repository::from_parts(root, heddle_dir, store, refs, oplog, config, shallow)
}

fn open_mount() -> (TempDir, ContentAddressedMount) {
    let (temp, repo) = fixture();
    let mount = ContentAddressedMount::new(repo, "main").unwrap();
    (temp, mount)
}

fn mount_with_gitlink() -> (TempDir, ContentAddressedMount, GitObjectId) {
    let temp = TempDir::new().unwrap();
    let repo = Repository::init_default(temp.path()).unwrap();
    let target: GitObjectId = "0303030303030303030303030303030303030303"
        .parse()
        .expect("git oid");
    let tree = Tree::from_entries(vec![
        TreeEntry::gitlink("vendor", target).expect("gitlink entry"),
    ]);
    let tree_hash = repo.store().put_tree(&tree).unwrap();
    let state = State::new(
        tree_hash,
        Vec::new(),
        Attribution::human(Principal::new("Gitlink Tester", "gitlink@example.test")),
    );
    repo.store().put_state(&state).unwrap();
    repo.refs()
        .set_thread(&ThreadName::new("main"), &state.state_id)
        .unwrap();
    let mount = ContentAddressedMount::new(repo, "main").unwrap();
    (temp, mount, target)
}

#[test]
fn lookup_hits_root_entry() {
    let (_temp, mount) = open_mount();
    let entry = mount
        .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
        .unwrap()
        .expect("hello.txt should exist");
    assert_eq!(entry.kind, NodeKind::File);
    assert_eq!(entry.size, 5);
    assert_eq!(entry.unix_mode & 0o777, 0o644);
}

#[test]
fn lookup_misses_return_none() {
    let (_temp, mount) = open_mount();
    let missing = mount
        .lookup(NodeId::ROOT, OsStr::new("does-not-exist"))
        .unwrap();
    assert!(missing.is_none());
}

#[test]
fn read_full_file() {
    let (_temp, mount) = open_mount();
    let node = mount.lookup_path("hello.txt").unwrap();
    let mut buf = vec![0u8; 64];
    let n = mount.read(node, 0, &mut buf).unwrap();
    assert_eq!(&buf[..n], b"world");
}

#[test]
fn read_with_offset_returns_tail() {
    let (_temp, mount) = open_mount();
    let node = mount.lookup_path("nested/note.md").unwrap();
    let mut buf = vec![0u8; 64];
    let n = mount.read(node, 10, &mut buf).unwrap();
    assert_eq!(&buf[..n], b"body\n");
}

#[test]
fn read_past_eof_yields_zero() {
    let (_temp, mount) = open_mount();
    let node = mount.lookup_path("hello.txt").unwrap();
    let mut buf = vec![0u8; 16];
    let n = mount.read(node, 9_999, &mut buf).unwrap();
    assert_eq!(n, 0);
}

#[test]
fn enumerate_root_lists_all_entries() {
    let (_temp, mount) = open_mount();
    let entries = mount.enumerate(NodeId::ROOT).unwrap();
    let names: Vec<String> = entries
        .iter()
        .map(|e| e.name.to_string_lossy().into_owned())
        .collect();
    assert!(names.contains(&"hello.txt".to_string()));
    assert!(names.contains(&"nested".to_string()));
    assert!(names.contains(&"run.sh".to_string()));
}

#[test]
fn enumerate_nested_lists_subdir_entries() {
    let (_temp, mount) = open_mount();
    let nested = mount.lookup_path("nested").unwrap();
    let entries = mount.enumerate(nested).unwrap();
    let names: Vec<_> = entries
        .iter()
        .map(|e| e.name.to_string_lossy().into_owned())
        .collect();
    assert_eq!(
        {
            let mut sorted = names.clone();
            sorted.sort();
            sorted
        },
        vec!["inner.txt".to_string(), "note.md".to_string()]
    );
}

#[test]
fn attrs_distinguish_file_and_directory() {
    let (_temp, mount) = open_mount();

    let root_attrs = mount.attrs(NodeId::ROOT).unwrap();
    assert_eq!(root_attrs.kind, NodeKind::Directory);
    assert_eq!(root_attrs.unix_mode & 0o170000, 0o040000);
    assert!(root_attrs.size >= 3); // at least the three top-level entries

    let file = mount.lookup_path("hello.txt").unwrap();
    let file_attrs = mount.attrs(file).unwrap();
    assert_eq!(file_attrs.kind, NodeKind::File);
    assert_eq!(file_attrs.size, 5);
    assert_eq!(file_attrs.nlink, 1);
}

#[test]
fn attrs_preserve_executable_bit() {
    let (_temp, mount) = open_mount();
    let run = mount.lookup_path("run.sh").unwrap();
    let attrs = mount.attrs(run).unwrap();
    assert_eq!(attrs.unix_mode & 0o111, 0o111);
}

#[test]
fn gitlink_reads_as_read_only_placeholder() {
    let (_temp, mount, target) = mount_with_gitlink();
    let node = mount.lookup_path("vendor").unwrap();
    let attrs = mount.attrs(node).unwrap();
    let placeholder = gitlink_placeholder_bytes(&target);

    assert_eq!(attrs.kind, NodeKind::File);
    assert_eq!(attrs.size, placeholder.len() as u64);

    let mut buf = vec![0u8; placeholder.len() + 16];
    let n = mount.read(node, 0, &mut buf).unwrap();
    assert_eq!(&buf[..n], placeholder.as_slice());

    let err = mount.write(node, 0, b"not-a-gitlink").unwrap_err();
    assert!(matches!(err, MountError::ReadOnly));
}

#[test]
fn write_to_overlay_then_read_back() {
    // Two-tier write: a write against a captured `File` NodeId
    // mints a hot-tier buffer keyed off the file's path. Subsequent
    // reads through that NodeId serve from the buffer (read-after-
    // write consistency in the same FUSE session).
    let (_temp, mount) = open_mount();
    let node = mount.lookup_path("hello.txt").unwrap();
    let written = mount.write(node, 0, b"WORLD").unwrap();
    assert_eq!(written, 5);
    // Hot buffer should be populated; warm tier still empty.
    assert_eq!(mount.hot_buffer_count(), 1);
    assert!(mount.warm_keys().is_empty());

    // Read through the same NodeId serves from the hot buffer.
    let mut buf = vec![0u8; 16];
    let n = mount.read(node, 0, &mut buf).unwrap();
    assert_eq!(&buf[..n], b"WORLD");
}

// ---------------------------------------------------------------------------
// POSIX `pwrite` semantics: partial overwrites preserve the captured
// tail; offsets past EOF zero-fill. These exercise the fix for the
// "fresh hot buffer" bug — without the seed-from-durable-source step,
// any write less than the full file length would silently truncate
// on flush.
// ---------------------------------------------------------------------------

/// Build a mount whose captured tree contains a single file with the
/// given content. Useful for tests that want to assert against the
/// shape of a captured blob after a partial overwrite.
fn mount_with_seed(path: &str, content: &[u8]) -> (TempDir, ContentAddressedMount) {
    let temp = TempDir::new().unwrap();
    let repo = Repository::init_default(temp.path()).unwrap();
    let full = temp.path().join(path);
    if let Some(parent) = full.parent() {
        fs::create_dir_all(parent).unwrap();
    }
    fs::write(&full, content).unwrap();
    repo.snapshot(Some("seed".into()), None).unwrap();
    let mount = ContentAddressedMount::new(repo, "main").unwrap();
    (temp, mount)
}

fn store_contains_blob_with_bytes(mount: &ContentAddressedMount, bytes: &[u8]) -> bool {
    let store = mount.repo_handle().store();
    store.list_blobs().unwrap().into_iter().any(|hash| {
        store
            .get_blob(&hash)
            .unwrap()
            .map(|blob| blob.into_content() == bytes)
            .unwrap_or(false)
    })
}

/// heddle#877 / HEDDLE-DR-5: wire offset+len must be clamped before
/// `Vec::resize` — overflow must not panic and huge offsets must not
/// attempt multi-TiB allocations.
#[test]
fn write_rejects_overflowing_offset_plus_length() {
    let (_temp, mount) = mount_with_seed("big.txt", b"x");
    let node = mount.lookup_path("big.txt").unwrap();
    let err = mount.write(node, u64::MAX, &[0u8]).unwrap_err();
    assert!(matches!(err, MountError::InvalidArgument(_)), "got {err:?}");
    assert_eq!(err.to_errno(), libc::EINVAL);
}

#[test]
fn write_rejects_extent_beyond_max_hot_file_size() {
    let (_temp, mount) = mount_with_seed("cap.txt", b"x");
    let node = mount.lookup_path("cap.txt").unwrap();
    let err = mount
        .write(node, MAX_MOUNT_HOT_FILE_SIZE, &[0u8])
        .unwrap_err();
    assert!(matches!(err, MountError::FileTooLarge(_)), "got {err:?}");
    assert_eq!(err.to_errno(), libc::EFBIG);
}

#[test]
fn write_in_bounds_at_max_hot_file_size_succeeds() {
    let (_temp, mount) = mount_with_seed("edge.txt", b"");
    let node = mount.lookup_path("edge.txt").unwrap();
    let written = mount
        .write(node, MAX_MOUNT_HOT_FILE_SIZE - 1, b"z")
        .unwrap();
    assert_eq!(written, 1);
}

#[test]
fn set_attrs_truncate_rejects_beyond_max_hot_file_size() {
    use crate::shell::AttrUpdate;

    let (_temp, mount) = mount_with_seed("trunc.txt", b"hi");
    let node = mount.lookup_path("trunc.txt").unwrap();
    let err = mount
        .set_attrs(
            node,
            AttrUpdate {
                size: Some(MAX_MOUNT_HOT_FILE_SIZE + 1),
                ..Default::default()
            },
        )
        .unwrap_err();
    assert!(matches!(err, MountError::FileTooLarge(_)), "got {err:?}");
    assert_eq!(err.to_errno(), libc::EFBIG);
}

#[test]
fn write_to_directory_returns_read_only() {
    let (_temp, mount) = open_mount();
    let err = mount.write(NodeId::ROOT, 0, b"x").unwrap_err();
    assert!(matches!(err, MountError::ReadOnly));
}

#[test]
fn unknown_thread_is_enoent_shaped() {
    let (_temp, repo) = fixture();
    let err = match ContentAddressedMount::new(repo, "no-such-thread") {
        Ok(_) => panic!("expected unknown-thread error"),
        Err(err) => err,
    };
    assert!(matches!(err, MountError::UnknownThread(_)));
    assert_eq!(err.to_errno(), libc::ENOENT);
}

#[test]
fn invalidate_drops_the_mapping() {
    let (_temp, mount) = open_mount();
    let node = mount.lookup_path("hello.txt").unwrap();
    mount.invalidate(node).unwrap();
    // Re-lookup should hand back a fresh (potentially equal) NodeId
    // and not be stale. Reading the freshly-handed-out id should
    // succeed.
    let again = mount.lookup_path("hello.txt").unwrap();
    let mut buf = vec![0u8; 16];
    let n = mount.read(again, 0, &mut buf).unwrap();
    assert_eq!(&buf[..n], b"world");
}

// ---------------------------------------------------------------------------
// Part 1: blob_size doesn't load full blobs
// ---------------------------------------------------------------------------

/// Wrapping ObjectStore that counts calls to `get_blob`. Used to
/// prove `enumerate`/`attrs` don't pull blob contents through
/// `get_blob` when only the size is needed.
struct CountingStore {
    inner: objects::store::FsStore,
    get_blob_calls: Arc<AtomicUsize>,
    blob_size_calls: Arc<AtomicUsize>,
}

impl objects::store::SidecarStore for CountingStore {}

impl ObjectStore for CountingStore {
    fn get_blob(&self, hash: &ContentHash) -> objects::store::Result<Option<Blob>> {
        self.get_blob_calls.fetch_add(1, Ordering::Relaxed);
        self.inner.get_blob(hash)
    }
    fn put_blob(&self, blob: &Blob) -> objects::store::Result<ContentHash> {
        self.inner.put_blob(blob)
    }
    fn has_blob(&self, hash: &ContentHash) -> objects::store::Result<bool> {
        self.inner.has_blob(hash)
    }
    fn blob_size(&self, hash: &ContentHash) -> objects::store::Result<Option<u64>> {
        self.blob_size_calls.fetch_add(1, Ordering::Relaxed);
        self.inner.blob_size(hash)
    }
    fn get_tree(&self, hash: &ContentHash) -> objects::store::Result<Option<Tree>> {
        self.inner.get_tree(hash)
    }
    fn put_tree(&self, tree: &Tree) -> objects::store::Result<ContentHash> {
        self.inner.put_tree(tree)
    }
    fn has_tree(&self, hash: &ContentHash) -> objects::store::Result<bool> {
        self.inner.has_tree(hash)
    }
    fn get_state(&self, id: &StateId) -> objects::store::Result<Option<State>> {
        self.inner.get_state(id)
    }
    fn put_state(&self, state: &State) -> objects::store::Result<()> {
        self.inner.put_state(state)
    }
    fn has_state(&self, id: &StateId) -> objects::store::Result<bool> {
        self.inner.has_state(id)
    }
    fn list_states(&self) -> objects::store::Result<Vec<StateId>> {
        self.inner.list_states()
    }
    fn get_action(&self, id: &ActionId) -> objects::store::Result<Option<Action>> {
        self.inner.get_action(id)
    }
    fn put_action(&self, action: &mut Action) -> objects::store::Result<ActionId> {
        self.inner.put_action(action)
    }
    fn list_actions(&self) -> objects::store::Result<Vec<ActionId>> {
        self.inner.list_actions()
    }
    fn list_blobs(&self) -> objects::store::Result<Vec<ContentHash>> {
        self.inner.list_blobs()
    }
    fn list_trees(&self) -> objects::store::Result<Vec<ContentHash>> {
        self.inner.list_trees()
    }
}

#[test]
fn enumerate_serves_size_without_loading_blob_bytes() {
    // Build a fixture, then re-open the repo with a counting store
    // wrapped around the FsStore. enumerate() must return sizes
    // without ever calling get_blob — only blob_size.
    let temp = TempDir::new().unwrap();
    let repo = Repository::init_default(temp.path()).unwrap();
    fs::write(temp.path().join("a.txt"), b"first").unwrap();
    fs::write(temp.path().join("b.txt"), b"second-larger-payload").unwrap();
    fs::write(temp.path().join("c.txt"), vec![0u8; 4096]).unwrap();
    repo.snapshot(Some("fixture".into()), None).unwrap();
    drop(repo);

    let get_blob_calls = Arc::new(AtomicUsize::new(0));
    let blob_size_calls = Arc::new(AtomicUsize::new(0));
    let inner = objects::store::FsStore::new(temp.path().join(".heddle"));
    let store = CountingStore {
        inner,
        get_blob_calls: get_blob_calls.clone(),
        blob_size_calls: blob_size_calls.clone(),
    };
    let repo = open_mount_test_repo_with_store(temp.path().join(".heddle"), store);
    let mount = ContentAddressedMount::new(repo, "main").unwrap();

    let entries = mount.enumerate(NodeId::ROOT).unwrap();
    let names: Vec<_> = entries
        .iter()
        .map(|e| e.name.to_string_lossy().into_owned())
        .collect();
    assert!(names.contains(&"a.txt".to_string()));
    assert!(names.contains(&"b.txt".to_string()));
    assert!(names.contains(&"c.txt".to_string()));
    let a_size = entries
        .iter()
        .find(|e| e.name == "a.txt")
        .map(|e| e.size)
        .unwrap();
    assert_eq!(a_size, 5);
    let c_size = entries
        .iter()
        .find(|e| e.name == "c.txt")
        .map(|e| e.size)
        .unwrap();
    assert_eq!(c_size, 4096);

    // The killer assertion: enumerate() must not have pulled blob
    // bytes. blob_size() should have been called for each blob entry,
    // get_blob() never.
    assert_eq!(
        get_blob_calls.load(Ordering::Relaxed),
        0,
        "enumerate() pulled blob bytes when only size was needed"
    );
    assert!(
        blob_size_calls.load(Ordering::Relaxed) >= 3,
        "expected blob_size to be called at least once per blob entry"
    );

    // Same expectation for attrs().
    let prior_get_blob = get_blob_calls.load(Ordering::Relaxed);
    let node = mount.lookup_path("c.txt").unwrap();
    let _attrs = mount.attrs(node).unwrap();
    assert_eq!(
        get_blob_calls.load(Ordering::Relaxed),
        prior_get_blob,
        "attrs() pulled blob bytes when only size was needed"
    );
}

// ---------------------------------------------------------------------------
// Part 2: two-tier write model
// ---------------------------------------------------------------------------

/// Build a fresh repo + mount pointing at `main`. The repo is empty
/// (no captured state beyond the seeded empty-tree main).
fn fresh_mount() -> (TempDir, ContentAddressedMount) {
    let temp = TempDir::new().unwrap();
    let repo = Repository::init_default(temp.path()).unwrap();
    let mount = ContentAddressedMount::new(repo, "main").unwrap();
    (temp, mount)
}

/// Mint a brand-new file path in the mount via lookup, returning a
/// NodeId we can write to. The mount has no `create()` entrypoint
/// yet (FUSE wires that separately); for tests we install a
/// `PendingFile` record directly. This mirrors what the `create`
/// callback will ultimately do.
fn create_pending_file(
    mount: &ContentAddressedMount,
    name: &str,
    mode: objects::object::FileMode,
) -> NodeId {
    use crate::core::test_helpers::install_pending_file;
    install_pending_file(mount, name, mode)
}

#[test]
fn write_then_read_same_file() {
    let (_temp, mount) = fresh_mount();
    let node = create_pending_file(&mount, "draft.txt", objects::object::FileMode::Normal);
    let written = mount.write(node, 0, b"hello mount").unwrap();
    assert_eq!(written, 11);

    let mut buf = vec![0u8; 64];
    let n = mount.read(node, 0, &mut buf).unwrap();
    assert_eq!(&buf[..n], b"hello mount");
}

#[test]
fn flush_promotes_buffer_to_warm_tier() {
    let (_temp, mount) = fresh_mount();
    let node = create_pending_file(&mount, "out.txt", objects::object::FileMode::Normal);
    mount.write(node, 0, b"promote me").unwrap();
    assert_eq!(mount.hot_buffer_count(), 1);
    assert!(mount.warm_keys().is_empty());

    mount.flush(node).unwrap();
    assert_eq!(mount.hot_buffer_count(), 0, "hot buffer should be drained");
    let warm = mount.warm_keys();
    assert_eq!(warm.len(), 1);
    assert_eq!(warm[0], std::path::PathBuf::from("out.txt"));
}

#[test]
fn test_release_orphaned_node_flushes_pending_writes() {
    let (_temp, mount) = open_mount();
    let node = mount.lookup_path("hello.txt").unwrap();
    let payload = b"release-orphan-cas-bytes";

    mount.on_open(node).expect("open");
    mount.write(node, 0, payload).expect("write hot bytes");
    assert!(
        !store_contains_blob_with_bytes(&mount, payload),
        "sanity: dirty hot bytes must not be in CAS before release"
    );

    mount
        .unlink_entry(NodeId::ROOT, OsStr::new("hello.txt"))
        .expect("unlink while open");
    assert!(
        mount.orphans_contains(node),
        "unlink while open must mark the node orphaned"
    );

    mount.release(node).expect("final release");

    assert!(
        !mount.orphans_contains(node),
        "final release must retire orphan lifecycle state"
    );
    assert!(
        store_contains_blob_with_bytes(&mount, payload),
        "final release of an orphan must persist dirty hot bytes to CAS"
    );
    assert!(
        mount
            .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
            .unwrap()
            .is_none(),
        "orphan CAS persistence must not resurrect the unlinked path"
    );
}

#[test]
fn test_double_release_is_guarded() {
    let (_temp, mount) = open_mount();
    let node = mount.lookup_path("hello.txt").unwrap();
    let payload = b"double-release-cas-bytes";

    mount.on_open(node).expect("open");
    mount.write(node, 0, payload).expect("write hot bytes");
    mount.release(node).expect("first release flushes");
    let blob = mount
        .warm_blob("hello.txt")
        .expect("first release promoted to warm CAS blob");

    mount.release(node).expect("second release is a safe no-op");

    assert_eq!(
        mount.warm_blob("hello.txt"),
        Some(blob),
        "double release must not corrupt or replace the warm blob"
    );
    assert_eq!(
        mount.hot_buffer_count(),
        0,
        "double release must not recreate a hot buffer"
    );
    assert!(
        store_contains_blob_with_bytes(&mount, payload),
        "flushed bytes must remain readable from CAS after double release"
    );
}

#[test]
fn test_release_nonexistent_node_safe_error() {
    let (_temp, mount) = open_mount();
    let err = mount
        .release(NodeId(9_999_999_617))
        .expect_err("release of an unknown node must fail safely");

    assert!(
        matches!(err, MountError::NotFound(_)),
        "unexpected release error: {err:?}"
    );
    assert_eq!(err.to_errno(), libc::ENOENT);
}

#[test]
fn captured_file_read_after_flush_through_same_node_id_serves_overlay() {
    // Regression: the FUSE shell reuses the NodeId the kernel cached
    // for a captured-tree file across the open → write → close →
    // reopen → read cycle (FUSE's dentry TTL keeps the dentry alive
    // for the cache window, so the kernel never re-issues `lookup`).
    // The core's `read` must therefore consult the pending overlay
    // for a `NodeRecord::File`'s path before falling back to the
    // captured blob — otherwise post-flush reads through the same
    // NodeId silently return the *pre*-write bytes and "write through
    // the mount" looks broken from userspace.
    //
    // The companion `lookup_after_write_serves_new_content` test
    // doesn't trip this because it re-resolves via `lookup_path`
    // after the flush, which refreshes the inode record. Real FUSE
    // dispatchers don't do that re-resolution — the kernel does.
    let (_temp, mount) = open_mount();
    let node = mount.lookup_path("hello.txt").unwrap();

    // Sanity: pre-write content from the captured tree.
    let mut buf = vec![0u8; 64];
    let n = mount.read(node, 0, &mut buf).unwrap();
    assert_eq!(&buf[..n], b"world");

    // Write through the *captured* file's NodeId, flush to warm.
    mount.write(node, 0, b"WORLD").unwrap();
    mount.flush(node).unwrap();

    // Re-read via the same NodeId — no fresh `lookup_path` call.
    let mut buf = vec![0u8; 64];
    let n = mount.read(node, 0, &mut buf).unwrap();
    assert_eq!(
        &buf[..n],
        b"WORLD",
        "captured-file read after flush must serve warm tier, not captured blob"
    );
    let attrs = mount.attrs(node).unwrap();
    assert_eq!(
        attrs.size, 5,
        "captured-file attrs after flush must reflect warm-tier size"
    );
}

#[test]
fn lookup_after_write_serves_new_content() {
    // Write a new file via the pending tier, then look it up by
    // path and read through the resulting NodeId. Should return the
    // bytes we just wrote, not whatever the captured tree said.
    let (_temp, mount) = fresh_mount();
    let node = create_pending_file(&mount, "fresh.md", objects::object::FileMode::Normal);
    mount.write(node, 0, b"# fresh\n").unwrap();

    // Hot-tier read-after-write through lookup.
    let looked_up = mount.lookup_path("fresh.md").unwrap();
    let mut buf = vec![0u8; 64];
    let n = mount.read(looked_up, 0, &mut buf).unwrap();
    assert_eq!(&buf[..n], b"# fresh\n");

    // Promote and re-read — now it's in the warm tier.
    mount.flush(node).unwrap();
    let looked_up_warm = mount.lookup_path("fresh.md").unwrap();
    let n = mount.read(looked_up_warm, 0, &mut buf).unwrap();
    assert_eq!(&buf[..n], b"# fresh\n");
}

#[test]
fn cross_thread_blob_dedup() {
    // The killer demo. Two mounts against two different threads of
    // the same repo write identical content to *different* paths.
    // The pending tier promotes both to CAS via put_blob, which is
    // content-addressed: the same bytes hash to the same blob_oid,
    // so the store ends up with exactly one blob — not two.
    let temp = TempDir::new().unwrap();
    let repo_a = Repository::init_default(temp.path()).unwrap();
    // Add a sibling thread by reusing the seeded `main` head.
    let main_id = repo_a
        .refs()
        .get_thread(&ThreadName::new("main"))
        .unwrap()
        .unwrap();
    repo_a
        .refs()
        .set_thread(&ThreadName::new("feature"), &main_id)
        .unwrap();
    drop(repo_a);

    // Open two independent mounts against the same backing store.
    let repo_main = Repository::open(temp.path()).unwrap();
    let mount_main = ContentAddressedMount::new(repo_main, "main").unwrap();
    let repo_feat = Repository::open(temp.path()).unwrap();
    let mount_feat = ContentAddressedMount::new(repo_feat, "feature").unwrap();

    let payload = b"shared module content\n// dedup demo\n";

    let n_main = create_pending_file(&mount_main, "lib.rs", objects::object::FileMode::Normal);
    mount_main.write(n_main, 0, payload).unwrap();
    mount_main.flush(n_main).unwrap();

    let n_feat = create_pending_file(&mount_feat, "module.rs", objects::object::FileMode::Normal);
    mount_feat.write(n_feat, 0, payload).unwrap();
    mount_feat.flush(n_feat).unwrap();

    // Both warm tiers point at the same blob oid — content-addressed
    // dedup falls out for free.
    let oid_a = mount_main.warm_blob("lib.rs").expect("a promoted");
    let oid_b = mount_feat.warm_blob("module.rs").expect("b promoted");
    assert_eq!(
        oid_a, oid_b,
        "identical content must hash to the same blob_oid across threads"
    );

    // Verify only one blob exists in the underlying store with that
    // hash. (list_blobs returns all unique hashes; we just check
    // ours appears once.)
    let repo_check = Repository::open(temp.path()).unwrap();
    let blobs = repo_check.store().list_blobs().unwrap();
    let count = blobs.iter().filter(|h| **h == oid_a).count();
    assert_eq!(
        count, 1,
        "writing the same payload to two threads must yield exactly one blob in the store"
    );
}

// ---------------------------------------------------------------------------
// Part 3: nested-tree fold-up (Task A)
// ---------------------------------------------------------------------------

/// Walk a path component-by-component through `lookup`, mirroring how
/// FUSE actually descends. Catches regressions in the `Dir` parent
/// path-tracking that `lookup_path` (which uses the registry) might
/// hide.
fn lookup_path_via_components(
    mount: &ContentAddressedMount,
    path: &str,
) -> Option<crate::shell::Entry> {
    let mut node = NodeId::ROOT;
    let mut last = None;
    for comp in std::path::Path::new(path).components() {
        let std::path::Component::Normal(name) = comp else {
            continue;
        };
        let entry = mount.lookup(node, name).ok().flatten()?;
        node = entry.node;
        last = Some(entry);
    }
    last
}

#[test]
fn lookup_serves_implicit_pending_dir_before_capture() {
    // Before capture, writing `newdir/foo.rs` should make `newdir`
    // resolvable as an *implicit* directory through component-wise
    // lookup, and the file readable through it.
    let (_temp, mount) = fresh_mount();
    let node = create_pending_file(&mount, "newdir/foo.rs", objects::object::FileMode::Normal);
    mount.write(node, 0, b"hello").unwrap();

    let dir_entry = lookup_path_via_components(&mount, "newdir")
        .expect("newdir resolves as implicit pending dir");
    assert_eq!(dir_entry.kind, NodeKind::Directory);

    let file_entry = lookup_path_via_components(&mount, "newdir/foo.rs")
        .expect("newdir/foo.rs resolves through implicit dir");
    let mut buf = vec![0u8; 16];
    let n = mount.read(file_entry.node, 0, &mut buf).unwrap();
    assert_eq!(&buf[..n], b"hello");
}

// ---------------------------------------------------------------------------
// Part 4: oplog + thread metadata wiring (Task B)
// ---------------------------------------------------------------------------

// ---------------------------------------------------------------------------
// Part 5: clock-driven safety-sweep (Task C)
// ---------------------------------------------------------------------------

#[test]
fn clock_sweep_promotes_idle_buffers() {
    use std::time::Duration;
    let (_temp, repo) = fixture();
    let mount = ContentAddressedMount::new(repo, "main")
        .unwrap()
        .with_promotion_policy(crate::core::PromotionPolicy {
            idle_after: Duration::from_millis(50),
            sweep_interval: Some(Duration::from_millis(80)),
        });

    let node = create_pending_file(&mount, "draft.txt", objects::object::FileMode::Normal);
    mount.write(node, 0, b"sleeping").unwrap();
    assert_eq!(mount.hot_buffer_count(), 1);
    assert!(mount.warm_keys().is_empty());

    // Wait long enough for several sweep iterations to fire after
    // the idle window expires.
    std::thread::sleep(Duration::from_millis(400));
    assert_eq!(
        mount.hot_buffer_count(),
        0,
        "clock sweep should have promoted the idle buffer"
    );
    assert_eq!(mount.warm_keys().len(), 1);
}

#[test]
fn no_sweep_interval_disables_clock_promotion() {
    use std::time::Duration;
    let (_temp, repo) = fixture();
    let mount = ContentAddressedMount::new(repo, "main")
        .unwrap()
        .with_promotion_policy(crate::core::PromotionPolicy {
            idle_after: Duration::from_millis(50),
            sweep_interval: None,
        });

    let node = create_pending_file(&mount, "draft.txt", objects::object::FileMode::Normal);
    mount.write(node, 0, b"sleeping").unwrap();
    std::thread::sleep(Duration::from_millis(250));
    // Without a sweep interval AND without any other mutating call,
    // the hot buffer should still be in the hot tier. (The
    // event-driven sweep only fires on writes; we did exactly one,
    // and that was at t=0 before the idle window even started.)
    assert_eq!(mount.hot_buffer_count(), 1);
    assert!(mount.warm_keys().is_empty());
}

#[test]
fn drop_joins_sweep_thread_cleanly() {
    // Construct a mount with a fast sweep, write a file, then drop
    // the mount. The Drop impl must signal-and-join cleanly without
    // deadlocking. We bound the test with a separate thread + join
    // timeout so a regression doesn't hang CI forever.
    use std::{sync::mpsc::channel, time::Duration};

    let (tx, rx) = channel();
    let join = std::thread::spawn(move || {
        let (_temp, repo) = fixture();
        let mount = ContentAddressedMount::new(repo, "main")
            .unwrap()
            .with_promotion_policy(crate::core::PromotionPolicy {
                idle_after: Duration::from_millis(20),
                sweep_interval: Some(Duration::from_millis(30)),
            });
        let node = create_pending_file(&mount, "x.txt", objects::object::FileMode::Normal);
        mount.write(node, 0, b"k").unwrap();
        std::thread::sleep(Duration::from_millis(80));
        drop(mount);
        let _ = tx.send(());
    });
    let result = rx.recv_timeout(Duration::from_secs(5));
    let join_result = join.join();
    assert!(result.is_ok(), "drop did not complete within 5s");
    assert!(join_result.is_ok(), "test thread panicked");
}

// ---------------------------------------------------------------------------
// Part 6: comprehensive coverage (Task D)
// ---------------------------------------------------------------------------

// D3. Crash recovery test.
#[test]
fn crash_recovery_warm_durable_hot_lost() {
    let temp = TempDir::new().unwrap();
    let repo = Repository::init_default(temp.path()).unwrap();
    {
        let mount = ContentAddressedMount::new(repo, "main")
            .unwrap()
            .with_promotion_policy(crate::core::PromotionPolicy {
                idle_after: std::time::Duration::from_secs(3600),
                sweep_interval: None,
            });
        let n1 = create_pending_file(&mount, "durable.txt", objects::object::FileMode::Normal);
        mount.write(n1, 0, b"durable").unwrap();
        mount.flush(n1).unwrap(); // promote to warm tier
        let n2 = create_pending_file(&mount, "transient.txt", objects::object::FileMode::Normal);
        mount.write(n2, 0, b"gone").unwrap();
        // No flush. Drop simulates a crash.
        // We can verify the durable blob exists in the store.
        let durable_blob = mount.warm_blob("durable.txt").unwrap();
        let blobs_before = mount.repo_handle().store().list_blobs().unwrap();
        assert!(blobs_before.contains(&durable_blob));
    }
    // Re-open the repo and a new mount on the same backing store.
    let repo = Repository::open(temp.path()).unwrap();
    let mount = ContentAddressedMount::new(repo, "main").unwrap();
    // Hot tier was lost — `transient.txt` doesn't exist.
    let lookup = mount
        .lookup(NodeId::ROOT, OsStr::new("transient.txt"))
        .unwrap();
    assert!(
        lookup.is_none(),
        "hot-tier-only file should be gone after crash"
    );
    // Warm-tier blob is durable in the store, but it was never
    // captured into a state (no `capture()` was called), so the
    // mount's *tree* doesn't surface it either. The durability
    // boundary is the blob, not the tree.
    let durable_lookup = mount
        .lookup(NodeId::ROOT, OsStr::new("durable.txt"))
        .unwrap();
    assert!(
        durable_lookup.is_none(),
        "warm-tier-only file (no capture) is not in the captured tree"
    );
}

// D4. Cross-thread blob dedup at scale.
#[test]
fn cross_thread_blob_dedup_at_scale() {
    let temp = TempDir::new().unwrap();
    let repo = Repository::init_default(temp.path()).unwrap();
    let main_id = repo
        .refs()
        .get_thread(&ThreadName::new("main"))
        .unwrap()
        .unwrap();
    // Make 9 sibling threads.
    for i in 0..9 {
        repo.refs()
            .set_thread(&ThreadName::new(format!("feat-{i}")), &main_id)
            .unwrap();
    }
    drop(repo);

    let thread_names: Vec<String> = std::iter::once("main".to_string())
        .chain((0..9).map(|i| format!("feat-{i}")))
        .collect();
    // 10 files per thread; index%2==0 are shared-content, the rest
    // are unique.
    let shared_count = 5; // shared 0..4 across all threads
    for name in &thread_names {
        let r = Repository::open(temp.path()).unwrap();
        let mount = ContentAddressedMount::new(r, name).unwrap();
        for i in 0..10 {
            let path = format!("file{i}.txt");
            let node = create_pending_file(&mount, &path, objects::object::FileMode::Normal);
            let bytes = if i < shared_count {
                format!("shared-content-{i}\n").into_bytes()
            } else {
                format!("unique-{name}-{i}\n").into_bytes()
            };
            mount.write(node, 0, &bytes).unwrap();
        }
        mount.flush_all().unwrap();
    }

    // Now check the blob set: shared_count distinct blobs from shared
    // files, plus 5 unique * 10 threads = 50 unique-content blobs.
    let repo = Repository::open(temp.path()).unwrap();
    let blobs: std::collections::HashSet<_> =
        repo.store().list_blobs().unwrap().into_iter().collect();
    // Empty blob from initial seed may also be present, so we check
    // we have at *least* shared_count + 50 distinct blobs and at
    // most a small constant overhead beyond.
    let expected_unique = shared_count + 5 * 10;
    assert!(
        blobs.len() >= expected_unique && blobs.len() <= expected_unique + 4,
        "expected ~{expected_unique} distinct blobs, got {}",
        blobs.len()
    );
}

// ---------------------------------------------------------------------------
// Part 5: write-side overlay ops — create / mkdir / unlink / rmdir / rename /
// setattr / symlink. Each op exercises the PlatformShell trait method on a
// real ContentAddressedMount, so the test catches both the core implementation
// and the trait dispatch (it's the same path FUSE / FSKit / ProjFS callbacks
// take). Issue: heddle#180 — unblocks `open(O_CREAT)` and friends on the
// Linux FUSE shell.
// ---------------------------------------------------------------------------
mod write_ops {
    use std::path::Path;

    use objects::object::FileMode;

    use super::*;
    use crate::shell::AttrUpdate;

    /// `create_file` mints a fresh PendingFile under root, visible to
    /// subsequent `lookup` / `read` calls. The first `write` against
    /// the returned NodeId seeds an empty buffer; the byte stream
    /// flows through the existing two-tier write model.
    #[test]
    fn create_file_in_root_then_write_and_read_back() {
        let (_temp, mount) = open_mount();
        let entry = mount
            .create_file(
                NodeId::ROOT,
                OsStr::new("Cargo.lock"),
                FileMode::Normal,
                false,
            )
            .expect("create_file");
        assert_eq!(entry.kind, NodeKind::File);
        assert_eq!(entry.name, "Cargo.lock");

        // Lookup must now resolve the same path.
        let looked_up = mount
            .lookup(NodeId::ROOT, OsStr::new("Cargo.lock"))
            .expect("lookup ok")
            .expect("lookup hit");
        assert_eq!(looked_up.node, entry.node);

        // Write + read-back through the freshly minted NodeId.
        mount.write(entry.node, 0, b"[package]\n").expect("write");
        let mut buf = vec![0u8; 32];
        let n = mount.read(entry.node, 0, &mut buf).expect("read");
        assert_eq!(&buf[..n], b"[package]\n");
    }

    /// `create_file` with `exclusive=true` (`O_CREAT|O_EXCL`) against
    /// an already-existing captured path must fail with
    /// `AlreadyExists` (errno `EEXIST`).
    #[test]
    fn create_file_exclusive_against_existing_returns_eexist() {
        let (_temp, mount) = open_mount();
        let err = mount
            .create_file(
                NodeId::ROOT,
                OsStr::new("hello.txt"),
                FileMode::Normal,
                true,
            )
            .expect_err("exclusive create on existing must fail");
        assert!(matches!(err, MountError::AlreadyExists(_)), "got {err:?}");
        assert_eq!(err.to_errno(), libc::EEXIST);
    }

    /// `create_file` with `exclusive=false` (`O_CREAT` without
    /// `O_EXCL`) against an existing captured path returns the
    /// existing entry — that's the POSIX `open(O_CREAT)` shape, and
    /// what cargo / rustc rely on when re-opening an artifact for
    /// rewrite.
    #[test]
    fn create_file_non_exclusive_returns_existing_entry() {
        let (_temp, mount) = open_mount();
        let entry = mount
            .create_file(
                NodeId::ROOT,
                OsStr::new("hello.txt"),
                FileMode::Normal,
                false,
            )
            .expect("non-exclusive create on existing returns entry");
        assert_eq!(entry.name, "hello.txt");
        let captured = mount
            .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
            .unwrap()
            .unwrap();
        assert_eq!(captured.node, entry.node);
    }

    /// Names containing `/` or `\0`, or the reserved `.` / `..`
    /// pseudo-entries, must be rejected at the create boundary with
    /// `EINVAL` — not silently shoved into the overlay.
    #[test]
    fn create_file_rejects_invalid_names() {
        let (_temp, mount) = open_mount();
        for bad in ["", ".", "..", "a/b", "with\0nul"] {
            let err = mount
                .create_file(NodeId::ROOT, OsStr::new(bad), FileMode::Normal, false)
                .expect_err(&format!("name {bad:?} must be rejected"));
            assert!(
                matches!(err, MountError::InvalidArgument(_)),
                "{bad}: {err:?}"
            );
            assert_eq!(err.to_errno(), libc::EINVAL);
        }
    }

    /// `make_dir` creates an empty pending directory under root that
    /// shows up in lookup + enumerate immediately. Subsequent
    /// `create_file` calls under it must work.
    #[test]
    fn make_dir_creates_empty_visible_dir() {
        let (_temp, mount) = open_mount();
        let dir_entry = mount
            .make_dir(NodeId::ROOT, OsStr::new("target"))
            .expect("make_dir");
        assert_eq!(dir_entry.kind, NodeKind::Directory);

        // Visible via lookup.
        let looked = mount
            .lookup(NodeId::ROOT, OsStr::new("target"))
            .unwrap()
            .unwrap();
        assert_eq!(looked.node, dir_entry.node);

        // Root enumerate must include the new directory.
        let root_entries = mount.enumerate(NodeId::ROOT).unwrap();
        assert!(
            root_entries
                .iter()
                .any(|e| e.name == "target" && e.kind == NodeKind::Directory),
            "root enumerate did not include the new dir: {root_entries:?}"
        );

        // Create a file inside it; visible via lookup under the dir.
        let file = mount
            .create_file(
                dir_entry.node,
                OsStr::new("out.bin"),
                FileMode::Normal,
                false,
            )
            .expect("create_file under new dir");
        let from_lookup = mount
            .lookup(dir_entry.node, OsStr::new("out.bin"))
            .unwrap()
            .unwrap();
        assert_eq!(from_lookup.node, file.node);
    }

    /// `make_dir` against an existing path (captured or pending) is
    /// `EEXIST`. POSIX `mkdir(2)` shape.
    #[test]
    fn make_dir_existing_returns_eexist() {
        let (_temp, mount) = open_mount();
        // `nested/` already exists in the fixture's captured tree.
        let err = mount
            .make_dir(NodeId::ROOT, OsStr::new("nested"))
            .expect_err("mkdir on existing must fail");
        assert_eq!(err.to_errno(), libc::EEXIST);
    }

    /// `unlink_entry` against a captured file tombstones it: post-
    /// unlink lookup returns `None`, enumerate skips it, and a
    /// subsequent `create_file` (POSIX `unlink+open(O_CREAT)`) mints
    /// a fresh empty inode at the same path.
    #[test]
    fn unlink_entry_removes_captured_file_and_allows_recreate() {
        let (_temp, mount) = open_mount();
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("hello.txt"))
            .expect("unlink");
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
                .unwrap()
                .is_none(),
            "post-unlink lookup must return None"
        );
        let entries = mount.enumerate(NodeId::ROOT).unwrap();
        assert!(
            !entries.iter().any(|e| e.name == "hello.txt"),
            "enumerate still surfaces unlinked file"
        );

        // Recreate.
        let recreated = mount
            .create_file(
                NodeId::ROOT,
                OsStr::new("hello.txt"),
                FileMode::Normal,
                false,
            )
            .expect("recreate after unlink");
        mount.write(recreated.node, 0, b"REBORN").unwrap();
        let mut buf = vec![0u8; 16];
        let n = mount.read(recreated.node, 0, &mut buf).unwrap();
        assert_eq!(&buf[..n], b"REBORN");
    }

    /// `unlink_entry` on a directory is `EISDIR` (POSIX `unlink(2)`).
    #[test]
    fn unlink_entry_on_directory_returns_eisdir() {
        let (_temp, mount) = open_mount();
        let err = mount
            .unlink_entry(NodeId::ROOT, OsStr::new("nested"))
            .expect_err("unlink on dir must fail");
        assert_eq!(err.to_errno(), libc::EISDIR);
    }

    /// `unlink_entry` on a name that doesn't exist is `ENOENT`.
    #[test]
    fn unlink_entry_missing_returns_enoent() {
        let (_temp, mount) = open_mount();
        let err = mount
            .unlink_entry(NodeId::ROOT, OsStr::new("nonexistent"))
            .expect_err("unlink missing must fail");
        assert_eq!(err.to_errno(), libc::ENOENT);
    }

    /// `rmdir_entry` removes an empty pending directory. Subsequent
    /// lookup returns `None`.
    #[test]
    fn rmdir_entry_removes_empty_pending_dir() {
        let (_temp, mount) = open_mount();
        let dir = mount.make_dir(NodeId::ROOT, OsStr::new("scratch")).unwrap();
        let _ = dir; // keep alive
        mount
            .rmdir_entry(NodeId::ROOT, OsStr::new("scratch"))
            .expect("rmdir");
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("scratch"))
                .unwrap()
                .is_none()
        );
    }

    /// `rmdir_entry` on a directory that has any visible child (pending
    /// or captured) must fail with `ENOTEMPTY`.
    #[test]
    fn rmdir_entry_non_empty_returns_enotempty() {
        let (_temp, mount) = open_mount();
        // The fixture's `nested/` has captured children.
        let err = mount
            .rmdir_entry(NodeId::ROOT, OsStr::new("nested"))
            .expect_err("rmdir on non-empty must fail");
        assert_eq!(err.to_errno(), libc::ENOTEMPTY);
    }

    /// `rmdir_entry` on a regular file is `ENOTDIR`.
    #[test]
    fn rmdir_entry_on_file_returns_enotdir() {
        let (_temp, mount) = open_mount();
        let err = mount
            .rmdir_entry(NodeId::ROOT, OsStr::new("hello.txt"))
            .expect_err("rmdir on file must fail");
        assert_eq!(err.to_errno(), libc::ENOTDIR);
    }

    /// File rename within the same directory: source disappears,
    /// destination resolves to the renamed file with the same bytes.
    /// This is the cargo / git path: write `foo.tmp` then rename to
    /// `foo`.
    #[test]
    fn rename_entry_file_same_dir() {
        let (_temp, mount) = open_mount();
        let src = mount
            .create_file(
                NodeId::ROOT,
                OsStr::new("Cargo.lock.tmp"),
                FileMode::Normal,
                false,
            )
            .unwrap();
        mount.write(src.node, 0, b"[atomic]\n").unwrap();
        mount.flush(src.node).unwrap();

        mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("Cargo.lock.tmp"),
                NodeId::ROOT,
                OsStr::new("Cargo.lock"),
            )
            .expect("rename");

        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("Cargo.lock.tmp"))
                .unwrap()
                .is_none(),
            "source path must be gone after rename"
        );
        let dst = mount
            .lookup(NodeId::ROOT, OsStr::new("Cargo.lock"))
            .unwrap()
            .expect("dst resolves");
        let mut buf = vec![0u8; 16];
        let n = mount.read(dst.node, 0, &mut buf).unwrap();
        assert_eq!(&buf[..n], b"[atomic]\n");
    }

    /// Rename across directories: src in `nested/`, dst in root.
    #[test]
    fn rename_entry_cross_dir() {
        let (_temp, mount) = open_mount();
        // Source lives in the captured tree at `nested/inner.txt`.
        let nested = mount
            .lookup(NodeId::ROOT, OsStr::new("nested"))
            .unwrap()
            .unwrap();
        mount
            .rename_entry(
                nested.node,
                OsStr::new("inner.txt"),
                NodeId::ROOT,
                OsStr::new("moved.txt"),
            )
            .expect("cross-dir rename");

        assert!(
            mount
                .lookup(nested.node, OsStr::new("inner.txt"))
                .unwrap()
                .is_none(),
            "source path must be gone after rename"
        );
        let dst = mount
            .lookup(NodeId::ROOT, OsStr::new("moved.txt"))
            .unwrap()
            .expect("dst resolves");
        let mut buf = vec![0u8; 32];
        let n = mount.read(dst.node, 0, &mut buf).unwrap();
        assert_eq!(&buf[..n], b"deep");
    }

    /// Rename onto an existing file of the same kind: POSIX allows it
    /// (atomic replace). The destination's prior content is gone.
    #[test]
    fn rename_entry_replaces_existing_file() {
        let (_temp, mount) = open_mount();
        let src = mount
            .create_file(NodeId::ROOT, OsStr::new("draft"), FileMode::Normal, false)
            .unwrap();
        mount.write(src.node, 0, b"draft body").unwrap();
        mount.flush(src.node).unwrap();
        // hello.txt exists in the fixture; rename overwrites it.
        mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("draft"),
                NodeId::ROOT,
                OsStr::new("hello.txt"),
            )
            .expect("rename-over");
        let dst = mount
            .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
            .unwrap()
            .unwrap();
        let mut buf = vec![0u8; 32];
        let n = mount.read(dst.node, 0, &mut buf).unwrap();
        assert_eq!(&buf[..n], b"draft body");
    }

    /// After `unlink_entry` the path→inode mapping must be retired so
    /// a subsequent `create_file` at the same name mints a *fresh*
    /// inode. Otherwise a still-open handle to the unlinked file would
    /// silently start resolving to the freshly created replacement —
    /// breaks unlink-then-recreate isolation (POSIX open-unlinked temp
    /// files).
    #[test]
    fn unlink_then_recreate_mints_fresh_inode() {
        let (_temp, mount) = open_mount();
        let original = mount
            .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
            .unwrap()
            .expect("captured hello.txt");
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("hello.txt"))
            .expect("unlink");
        let recreated = mount
            .create_file(
                NodeId::ROOT,
                OsStr::new("hello.txt"),
                FileMode::Normal,
                false,
            )
            .expect("recreate");
        assert_ne!(
            original.node, recreated.node,
            "recreated inode must be distinct from the unlinked one"
        );
    }

    /// POSIX `unlink(2)` semantics: if a file is open when it's
    /// unlinked, the kernel keeps the inode alive behind the open
    /// fd, but the *directory entry* is gone — `lookup` returns
    /// `ENOENT` and `readdir` skips the name. A subsequent write
    /// through the open fd updates the orphaned inode's data, but
    /// it must NOT republish the name. Tools that depend on this:
    /// `mkstemp` + `unlink` for private scratch space, sqlite's WAL
    /// shadow files, cargo's atomic-replace pattern. Without the
    /// guard, the unlinked pathname unexpectedly reappears once a
    /// late write hits the open fd.
    #[test]
    fn write_to_unlinked_open_inode_does_not_resurrect_path() {
        let (_temp, mount) = open_mount();
        // `fd = open("temp", O_CREAT|O_RDWR)` — fresh pending file.
        let entry = mount
            .create_file(NodeId::ROOT, OsStr::new("temp"), FileMode::Normal, false)
            .expect("create");
        // The `O_CREAT|O_RDWR` open above bumps `open_count` to 1 —
        // without this the witness-gated unlink (heddle#209) sees the
        // node as `Released` (no entry) and skips the orphan
        // transition, breaking the open-unlinked POSIX flow this test
        // exercises.
        mount.on_open(entry.node).expect("on_open");
        mount.write(entry.node, 0, b"v1").expect("first write");
        // `unlink("temp")` while the handle is still in use.
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("temp"))
            .expect("unlink");
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("temp"))
                .unwrap()
                .is_none(),
            "post-unlink lookup must return None"
        );

        // Write through the original handle. POSIX says this is
        // legal — the inode survives behind the fd — but the
        // pathname must not come back.
        mount
            .write(entry.node, 0, b"v2-after-unlink")
            .expect("write through unlinked-open fd");

        // The data is accessible through the open handle.
        let mut buf = vec![0u8; 64];
        let n = mount
            .read(entry.node, 0, &mut buf)
            .expect("read via unlinked-open fd");
        assert_eq!(&buf[..n], b"v2-after-unlink");

        // The decisive check: the path must still be gone.
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("temp"))
                .unwrap()
                .is_none(),
            "write after unlink must not resurrect the path"
        );
        let entries = mount.enumerate(NodeId::ROOT).unwrap();
        assert!(
            !entries.iter().any(|e| e.name == "temp"),
            "enumerate must not surface the unlinked path: {entries:?}"
        );
        // Flushing the orphan must not promote it to the warm tier
        // (a subsequent capture would resurrect the path in the
        // captured tree otherwise). Drive the flush explicitly so
        // the test pins the contract; orphaned buffers must drop.
        mount.flush(entry.node).expect("flush orphan");
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("temp"))
                .unwrap()
                .is_none(),
            "post-flush lookup must still be gone (orphan must not warm-promote)"
        );
    }

    /// Companion to the orphan-write test: after an unlink, if a
    /// fresh `create_file` mints a new inode at the same name,
    /// writes through the *new* fd must surface normally. The
    /// orphan-write fix must not regress the unlink-then-recreate
    /// path the kernel actually emits for `open(O_CREAT)`.
    #[test]
    fn write_to_recreated_inode_after_unlink_still_publishes_path() {
        let (_temp, mount) = open_mount();
        let original = mount
            .create_file(NodeId::ROOT, OsStr::new("temp"), FileMode::Normal, false)
            .expect("create v1");
        mount.write(original.node, 0, b"v1").expect("write v1");
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("temp"))
            .expect("unlink");

        let recreated = mount
            .create_file(NodeId::ROOT, OsStr::new("temp"), FileMode::Normal, false)
            .expect("recreate");
        assert_ne!(original.node, recreated.node);
        mount
            .write(recreated.node, 0, b"v2-fresh")
            .expect("write v2");

        // The new inode is the one that owns the path now.
        let hit = mount
            .lookup(NodeId::ROOT, OsStr::new("temp"))
            .unwrap()
            .expect("recreated path must resolve");
        assert_eq!(hit.node, recreated.node);
        let mut buf = vec![0u8; 16];
        let n = mount.read(recreated.node, 0, &mut buf).unwrap();
        assert_eq!(&buf[..n], b"v2-fresh");
    }

    /// Rename-over must keep the replaced destination's inode record
    /// resolvable — only the path→inode link is detached. Without this
    /// any FD still holding the dest inode surfaces as ESTALE on the
    /// next callback. POSIX requires open handles to the replaced file
    /// to remain valid.
    #[test]
    fn rename_over_preserves_replaced_destination_inode() {
        let (_temp, mount) = open_mount();
        let dest_orig = mount
            .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
            .unwrap()
            .expect("captured hello.txt");
        let src = mount
            .create_file(NodeId::ROOT, OsStr::new("draft"), FileMode::Normal, false)
            .unwrap();
        mount.write(src.node, 0, b"replacement").unwrap();
        mount.flush(src.node).unwrap();
        mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("draft"),
                NodeId::ROOT,
                OsStr::new("hello.txt"),
            )
            .expect("rename-over");
        // The old destination inode must still resolve — not ESTALE.
        let attrs = mount
            .attrs(dest_orig.node)
            .expect("orphaned dest inode must remain valid");
        assert_eq!(attrs.kind, NodeKind::File);
    }

    /// A directory rename must rebase the `by_path` mapping for every
    /// already-cached descendant inode (not just the directory's own
    /// record). Otherwise an open handle to `old_dir/leaf` still
    /// references the stale path and post-rename reads return ESTALE
    /// — even though the leaf is reachable through the new directory.
    #[test]
    fn directory_rename_rebases_descendant_inode_paths() {
        let (_temp, mount) = open_mount();
        // Build an overlay-only directory with a leaf so the rename
        // exercises `move_overlay_dir` + the inode rebase pass.
        mount
            .make_dir(NodeId::ROOT, OsStr::new("from_dir"))
            .unwrap();
        let from = mount
            .lookup(NodeId::ROOT, OsStr::new("from_dir"))
            .unwrap()
            .unwrap();
        let leaf = mount
            .create_file(from.node, OsStr::new("leaf.txt"), FileMode::Normal, false)
            .unwrap();
        mount.write(leaf.node, 0, b"payload").unwrap();
        mount.flush(leaf.node).unwrap();
        // Cache the leaf inode by looking it up explicitly — this is
        // what the kernel does for any FD-holding lookup.
        let cached = mount
            .lookup(from.node, OsStr::new("leaf.txt"))
            .unwrap()
            .expect("leaf resolves pre-rename");
        assert_eq!(cached.node, leaf.node);

        mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("from_dir"),
                NodeId::ROOT,
                OsStr::new("to_dir"),
            )
            .expect("dir rename");

        // The cached leaf inode must still resolve — its stored path
        // should now be `to_dir/leaf.txt`.
        let mut buf = vec![0u8; 16];
        let n = mount
            .read(cached.node, 0, &mut buf)
            .expect("read via descendant inode after dir rename");
        assert_eq!(&buf[..n], b"payload");
        // And the new path resolves to the same inode.
        let to_dir = mount
            .lookup(NodeId::ROOT, OsStr::new("to_dir"))
            .unwrap()
            .unwrap();
        let via_new = mount
            .lookup(to_dir.node, OsStr::new("leaf.txt"))
            .unwrap()
            .expect("leaf resolves via new dir");
        assert_eq!(via_new.node, cached.node);
    }

    /// `set_attrs(size=0)` truncates the file's hot buffer to zero,
    /// which is what the kernel issues for `O_TRUNC` before any
    /// `write`. cargo writes use `O_CREAT|O_WRONLY|O_TRUNC`; without
    /// this the second build cycle overlays bytes on top of the
    /// previous artifact.
    #[test]
    fn set_attrs_truncate_zero_clears_buffer() {
        let (_temp, mount) = open_mount();
        let node = mount.lookup_path("hello.txt").unwrap();
        // Seed a buffer first.
        mount.write(node, 0, b"world-plus").unwrap();
        let attrs = mount
            .set_attrs(
                node,
                AttrUpdate {
                    size: Some(0),
                    ..Default::default()
                },
            )
            .expect("setattr size=0");
        assert_eq!(attrs.size, 0);
        // Subsequent read returns nothing.
        let mut buf = vec![0u8; 16];
        let n = mount.read(node, 0, &mut buf).unwrap();
        assert_eq!(n, 0);
    }

    /// `set_attrs(size=N)` larger than the current buffer zero-fills
    /// the gap (POSIX `ftruncate(2)`).
    #[test]
    fn set_attrs_truncate_grow_zero_fills() {
        let (_temp, mount) = open_mount();
        let node = mount.lookup_path("hello.txt").unwrap(); // "world", 5 bytes
        let attrs = mount
            .set_attrs(
                node,
                AttrUpdate {
                    size: Some(8),
                    ..Default::default()
                },
            )
            .expect("setattr grow");
        assert_eq!(attrs.size, 8);
        let mut buf = vec![0u8; 16];
        let n = mount.read(node, 0, &mut buf).unwrap();
        assert_eq!(&buf[..n], b"world\0\0\0");
    }

    /// `set_attrs(mode=0o755)` flips a Normal file to Executable in
    /// the overlay; the change is visible on `attrs` immediately and
    /// surfaces in `capture` output (so a freshly-built binary keeps
    /// its `+x` bit).
    #[test]
    fn set_attrs_mode_sets_executable_bit() {
        let (_temp, mount) = open_mount();
        let node = mount.lookup_path("hello.txt").unwrap();
        let attrs = mount
            .set_attrs(
                node,
                AttrUpdate {
                    mode: Some(0o100755),
                    ..Default::default()
                },
            )
            .expect("setattr chmod");
        assert_eq!(attrs.unix_mode & 0o111, 0o111);

        // Refetched attrs preserve the override.
        let again = mount.attrs(node).unwrap();
        assert_eq!(again.unix_mode & 0o111, 0o111);
    }

    /// `enumerate` surfaces overlay symlinks both as standalone
    /// pending-only children (pass-2 `PendingChildKind::Symlink`)
    /// and as overrides on captured-tree entries (pass-1 hit on
    /// `PendingHit::Symlink`). One enumerate exercises both arms.
    #[test]
    fn enumerate_surfaces_overlay_symlinks_in_both_passes() {
        let (_temp, mount) = open_mount();
        // Pass-2 path: brand-new symlink at a name with no captured
        // counterpart.
        mount
            .create_symlink(NodeId::ROOT, OsStr::new("alias"), Path::new("hello.txt"))
            .expect("fresh symlink");
        // Pass-1 path: overlay symlink replaces a captured file.
        // (`run.sh` is in the fixture as an executable file.)
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("run.sh"))
            .expect("clear captured run.sh");
        mount
            .create_symlink(NodeId::ROOT, OsStr::new("run.sh"), Path::new("hello.txt"))
            .expect("symlink overrides captured file");

        let entries = mount.enumerate(NodeId::ROOT).unwrap();
        let alias = entries
            .iter()
            .find(|e| e.name == "alias")
            .expect("fresh symlink missing from enumerate");
        assert_eq!(alias.kind, NodeKind::Symlink);
        let run = entries
            .iter()
            .find(|e| e.name == "run.sh")
            .expect("overlay symlink override missing from enumerate");
        assert_eq!(run.kind, NodeKind::Symlink);
    }

    /// `create_symlink` records a link in the overlay; `read_link`
    /// returns its target bytes.
    #[test]
    fn create_and_read_symlink() {
        let (_temp, mount) = open_mount();
        let entry = mount
            .create_symlink(
                NodeId::ROOT,
                OsStr::new("alias.txt"),
                Path::new("hello.txt"),
            )
            .expect("symlink");
        assert_eq!(entry.kind, NodeKind::Symlink);
        let target = mount.read_link(entry.node).expect("read_link");
        assert_eq!(target.as_os_str(), OsStr::new("hello.txt"));
    }

    /// `rename_entry` over a symlink in the overlay moves the target
    /// bytes and tombstones the source — `move_symlink`'s overlay-only
    /// path. Existing capture/diff tests don't reach this branch, so
    /// it's the main contributor to uncovered patch lines.
    #[test]
    fn rename_entry_moves_overlay_symlink() {
        let (_temp, mount) = open_mount();
        mount
            .create_symlink(NodeId::ROOT, OsStr::new("alias"), Path::new("hello.txt"))
            .expect("create symlink");
        mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("alias"),
                NodeId::ROOT,
                OsStr::new("alias2"),
            )
            .expect("rename symlink");
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("alias"))
                .unwrap()
                .is_none(),
            "source symlink path must be gone after rename",
        );
        let dst = mount
            .lookup(NodeId::ROOT, OsStr::new("alias2"))
            .unwrap()
            .expect("renamed symlink resolves");
        assert_eq!(dst.kind, NodeKind::Symlink);
        let target = mount.read_link(dst.node).expect("read_link via new path");
        assert_eq!(target.as_os_str(), OsStr::new("hello.txt"));
    }

    /// r11 #4 regression: renaming a regular file over a symlink must
    /// not push an `Orphan { open_count: 0 }` state entry for the
    /// displaced symlink's NodeId. Symlinks have no `open`/`release`
    /// lifecycle, so a state entry there is dead bookkeeping that
    /// nothing will ever reap — it just grows under symlink churn.
    ///
    /// Pre-retrofit (heddle#209) `rename_entry_with_options`'s displaced-
    /// destination branch unconditionally did
    /// `pending.state.insert(displaced_dest, Orphan{ open_count })` even
    /// for non-`Live` nodes (Codex PR #182 r11 finding 3293575541). The
    /// witness-gated retrofit replaces that with a
    /// `BrandedPending::witness_live_nonzero` check whose `None` result
    /// IS the short-circuit — a symlink never enters `state`, so the
    /// witness constructor returns `None` and no transition fires.
    #[test]
    fn rename_over_symlink_does_not_orphan_state() {
        let (_temp, mount) = open_mount();
        let link = mount
            .create_symlink(NodeId::ROOT, OsStr::new("link"), Path::new("hello.txt"))
            .expect("create symlink");
        assert!(
            !mount.orphans_contains(link.node),
            "newly-created symlink must have no Pending state entry",
        );
        mount
            .create_file(NodeId::ROOT, OsStr::new("source"), FileMode::Normal, false)
            .expect("create source file");
        mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("source"),
                NodeId::ROOT,
                OsStr::new("link"),
            )
            .expect("rename file over symlink");
        assert!(
            !mount.orphans_contains(link.node),
            "displaced symlink must not acquire a Pending state entry (r11 #4)",
        );
    }

    /// Cross-tree directory rename — i.e. renaming a captured-tree
    /// directory — is intentionally refused by `move_overlay_dir`;
    /// the overlay would otherwise need to rewrite every descendant
    /// tombstone/warm key. Exercises the error path explicitly so
    /// the refusal isn't accidentally relaxed by a later change.
    #[test]
    fn rename_entry_refuses_captured_directory_rename() {
        let (_temp, mount) = open_mount();
        let err = mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("nested"),
                NodeId::ROOT,
                OsStr::new("nested2"),
            )
            .expect_err("captured-dir rename must be refused");
        assert!(matches!(err, MountError::InvalidArgument(_)), "got {err:?}");
        assert_eq!(err.to_errno(), libc::EINVAL);
    }

    /// `move_overlay_dir` rebases every overlay slot under the source
    /// dir, but slots OUTSIDE the source must survive unchanged. This
    /// hits the `None` arms of each `rebase` match — the largest
    /// untouched cluster of patch-uncovered lines in this PR.
    #[test]
    fn rename_overlay_dir_preserves_sibling_overlay_state() {
        let (_temp, mount) = open_mount();
        // Two overlay dirs side by side; rename one. The other's
        // children/state — explicit_dirs, warm (via flushed write),
        // a symlink, and a tombstone — must all stay put.
        mount
            .make_dir(NodeId::ROOT, OsStr::new("from_dir"))
            .unwrap();
        mount
            .make_dir(NodeId::ROOT, OsStr::new("keep_dir"))
            .unwrap();
        let keep = mount
            .lookup(NodeId::ROOT, OsStr::new("keep_dir"))
            .unwrap()
            .unwrap();
        // Warm (flushed) leaf under `keep_dir/`.
        let kept_file = mount
            .create_file(keep.node, OsStr::new("warm.txt"), FileMode::Normal, false)
            .unwrap();
        mount.write(kept_file.node, 0, b"persist").unwrap();
        mount.flush(kept_file.node).unwrap();
        // Symlink under `keep_dir/`.
        mount
            .create_symlink(keep.node, OsStr::new("alias"), Path::new("warm.txt"))
            .unwrap();
        // Tombstone the captured `hello.txt` so the tombstone-rebase
        // branch is exercised against a non-prefixed path.
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("hello.txt"))
            .expect("unlink captured file");

        // The actual rename.
        mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("from_dir"),
                NodeId::ROOT,
                OsStr::new("to_dir"),
            )
            .expect("dir rename");

        // The sibling's children survive.
        let keep_after = mount
            .lookup(NodeId::ROOT, OsStr::new("keep_dir"))
            .unwrap()
            .unwrap();
        assert_eq!(keep_after.node, keep.node);
        let warm_after = mount
            .lookup(keep_after.node, OsStr::new("warm.txt"))
            .unwrap()
            .expect("warm leaf must remain at sibling dir");
        let mut buf = vec![0u8; 16];
        let n = mount.read(warm_after.node, 0, &mut buf).unwrap();
        assert_eq!(&buf[..n], b"persist");
        let alias_after = mount
            .lookup(keep_after.node, OsStr::new("alias"))
            .unwrap()
            .expect("sibling symlink must remain");
        assert_eq!(alias_after.kind, NodeKind::Symlink);
        // The unrelated tombstone survives.
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
                .unwrap()
                .is_none(),
            "unrelated tombstone must survive the rename pass",
        );
        // And the rename itself landed.
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("from_dir"))
                .unwrap()
                .is_none(),
            "source dir must be gone",
        );
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("to_dir"))
                .unwrap()
                .is_some(),
            "destination dir must be present",
        );
    }

    /// Self-rename (same source and destination) is a POSIX no-op:
    /// returns success without touching any state. Without the early
    /// return, the move + tombstone-source path would actually delete
    /// the file.
    #[test]
    fn rename_entry_self_rename_is_noop() {
        let (_temp, mount) = open_mount();
        mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("hello.txt"),
                NodeId::ROOT,
                OsStr::new("hello.txt"),
            )
            .expect("self-rename succeeds");
        // The file remains addressable + readable.
        let hit = mount
            .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
            .unwrap()
            .expect("self-rename did not delete the file");
        let mut buf = vec![0u8; 16];
        let n = mount.read(hit.node, 0, &mut buf).unwrap();
        assert_eq!(&buf[..n], b"world");
    }

    /// POSIX: renaming a directory over a regular file is `ENOTDIR`
    /// (the destination must also be a directory). Catches the
    /// `(Directory, _)` arm of the kind-mismatch guard.
    #[test]
    fn rename_entry_directory_over_file_returns_enotdir() {
        let (_temp, mount) = open_mount();
        mount.make_dir(NodeId::ROOT, OsStr::new("srcdir")).unwrap();
        let err = mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("srcdir"),
                NodeId::ROOT,
                OsStr::new("hello.txt"),
            )
            .expect_err("dir-over-file must fail");
        assert!(matches!(err, MountError::NotADirectory(_)), "got {err:?}");
        assert_eq!(err.to_errno(), libc::ENOTDIR);
    }

    /// POSIX: renaming a regular file over a directory is `EISDIR`.
    /// Catches the `(_, Directory)` arm of the kind-mismatch guard.
    #[test]
    fn rename_entry_file_over_directory_returns_eisdir() {
        let (_temp, mount) = open_mount();
        let err = mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("hello.txt"),
                NodeId::ROOT,
                OsStr::new("nested"),
            )
            .expect_err("file-over-dir must fail");
        assert!(matches!(err, MountError::IsADirectory(_)), "got {err:?}");
        assert_eq!(err.to_errno(), libc::EISDIR);
    }

    /// POSIX: directory-over-directory rename only succeeds if the
    /// destination is empty; a non-empty destination is `ENOTEMPTY`.
    /// Catches the inner branch of the (Directory, Directory) arm.
    #[test]
    fn rename_entry_directory_over_nonempty_directory_returns_enotempty() {
        let (_temp, mount) = open_mount();
        mount.make_dir(NodeId::ROOT, OsStr::new("srcdir")).unwrap();
        mount.make_dir(NodeId::ROOT, OsStr::new("dstdir")).unwrap();
        let dstdir = mount
            .lookup(NodeId::ROOT, OsStr::new("dstdir"))
            .unwrap()
            .unwrap();
        mount
            .create_file(
                dstdir.node,
                OsStr::new("child.txt"),
                FileMode::Normal,
                false,
            )
            .unwrap();
        let err = mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("srcdir"),
                NodeId::ROOT,
                OsStr::new("dstdir"),
            )
            .expect_err("non-empty dest must fail");
        assert!(matches!(err, MountError::NotEmpty(_)), "got {err:?}");
        assert_eq!(err.to_errno(), libc::ENOTEMPTY);
    }

    /// `invalidate` on an orphaned NodeId (an inode the kernel forgets
    /// after `unlink + release`) must retire the orphan tracking entry
    /// — otherwise the `orphans` set accumulates dead IDs across the
    /// session. Indirectly observed via a follow-up unlink+write cycle
    /// on a new NodeId at the same name: that write must take the
    /// normal (republishing) branch, not the orphan branch.
    #[test]
    fn invalidate_clears_orphan_tracking_for_forgotten_inode() {
        let (_temp, mount) = open_mount();
        // Round 1: create, write, unlink — orphans the inode.
        let v1 = mount
            .create_file(NodeId::ROOT, OsStr::new("scratch"), FileMode::Normal, false)
            .unwrap();
        mount.write(v1.node, 0, b"v1").unwrap();
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("scratch"))
            .unwrap();
        // The kernel issues `release` then `forget`; both flow
        // through `invalidate` in our trait surface.
        mount.invalidate(v1.node).expect("invalidate orphan");

        // Round 2: a fresh inode at the same name. Its writes must
        // republish the path normally — the orphan-cleanup pass in
        // `invalidate` is what keeps the orphan set from making this
        // node mistakenly take the orphan branch.
        let v2 = mount
            .create_file(NodeId::ROOT, OsStr::new("scratch"), FileMode::Normal, false)
            .unwrap();
        assert_ne!(v1.node, v2.node);
        mount.write(v2.node, 0, b"v2-fresh").expect("normal write");
        let hit = mount
            .lookup(NodeId::ROOT, OsStr::new("scratch"))
            .unwrap()
            .expect("recreated path resolves");
        assert_eq!(hit.node, v2.node);
        let mut buf = vec![0u8; 16];
        let n = mount.read(v2.node, 0, &mut buf).unwrap();
        assert_eq!(&buf[..n], b"v2-fresh");
    }

    // --- Codex round 7 findings: orphan-aware write-side ops ----------------
    //
    // r6 fixed `write` so a write through an unlinked-but-still-open fd
    // does not republish the path. Codex r7 surfaced four more write-side
    // ops with the same shape — operations that touch `hot_by_path`,
    // `tombstones`, or path-keyed `warm` entries without consulting the
    // orphan set. Each test below pins one contract from the brief.

    /// `ftruncate` through an unlinked-but-still-open captured fd must
    /// affect only the anonymous open inode. POSIX unlink semantics:
    /// the directory entry stays gone until the last close, and a
    /// flush of the orphan must not warm-promote its buffer. Without
    /// the fix, `apply_truncate`'s tombstone-clear + `hot_by_path`
    /// rebind republished the name to every other observer.
    #[test]
    fn truncate_unlinked_open_doesnt_resurrect_path() {
        let (_temp, mount) = open_mount();
        // `fd = open("hello.txt")` — captured file, no overlay yet.
        let node = mount.lookup_path("hello.txt").unwrap();
        // The `open` above bumps `open_count` to 1 — without this the
        // witness-gated unlink (heddle#209) skips the orphan
        // transition for `Released` (no entry) nodes and the test's
        // open-unlinked POSIX flow doesn't engage.
        mount.on_open(node).expect("on_open");
        // `unlink("hello.txt")` while the handle is still in use.
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("hello.txt"))
            .expect("unlink");
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
                .unwrap()
                .is_none(),
            "post-unlink lookup must return None"
        );
        // `ftruncate(fd, 2)` through the now-orphaned NodeId.
        mount
            .set_attrs(
                node,
                AttrUpdate {
                    size: Some(2),
                    ..Default::default()
                },
            )
            .expect("ftruncate through unlinked-open fd");
        // Decisive check: the path must still be gone.
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
                .unwrap()
                .is_none(),
            "truncate after unlink must not resurrect the path"
        );
        let entries = mount.enumerate(NodeId::ROOT).unwrap();
        assert!(
            !entries.iter().any(|e| e.name == "hello.txt"),
            "enumerate must not surface the unlinked path: {entries:?}"
        );
        // Flushing the orphan must not warm-promote it — a subsequent
        // capture would otherwise resurrect the path in the captured
        // tree.
        mount.flush(node).expect("flush orphan");
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
                .unwrap()
                .is_none(),
            "post-flush lookup must still be gone (orphan must not warm-promote)"
        );
    }

    /// Rename-over with the destination still open + holding a hot
    /// buffer must preserve the buffer. Reads through the original
    /// NodeId must continue to see the pre-rename bytes. Without the
    /// fix, `move_file`'s blind `pending.hot.remove(&dest_id)` dropped
    /// the buffer, and reads through the still-open fd then routed
    /// through the rebased path overlay and observed the replacement.
    #[test]
    fn rename_over_preserves_replaced_open_fd() {
        let (_temp, mount) = open_mount();
        // Open the captured "hello.txt", write through it, do NOT
        // flush. The hot buffer keyed by dest_id holds "ORIG-DATA".
        let dest_id = mount.lookup_path("hello.txt").unwrap();
        mount
            .write(dest_id, 0, b"ORIG-DATA")
            .expect("write to dest");
        // Build a source file with replacement bytes; flush so the
        // rename's start-of-move flush_node is a no-op.
        let src = mount
            .create_file(NodeId::ROOT, OsStr::new("draft"), FileMode::Normal, false)
            .expect("create src");
        mount
            .write(src.node, 0, b"REPLACE-DATA")
            .expect("write src");
        mount.flush(src.node).expect("flush src");
        // Rename src over dest. dest's pathname is rebound to src;
        // dest's open fd must continue to see "ORIG-DATA".
        mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("draft"),
                NodeId::ROOT,
                OsStr::new("hello.txt"),
            )
            .expect("rename-over");
        let mut buf = vec![0u8; 32];
        let n = mount
            .read(dest_id, 0, &mut buf)
            .expect("read via replaced dest fd");
        assert_eq!(
            &buf[..n],
            b"ORIG-DATA",
            "open fd on replaced dest must see pre-rename bytes",
        );
    }

    /// Captured-file rename-over: dest has no hot buffer at the time
    /// of the rename, only the captured-tree blob. Reads/attrs through
    /// the still-open fd must serve the captured bytes, not the
    /// source's replacement. Without the fix, the captured-file branch
    /// of `read` consulted `warm[new_path]` (now the source's data)
    /// before falling through to the captured blob.
    #[test]
    fn rename_over_destination_data_via_old_fd() {
        let (_temp, mount) = open_mount();
        // `fd = open("hello.txt")` — captured file, no overlay.
        let dest_id = mount.lookup_path("hello.txt").unwrap();
        // The `open` above bumps `open_count` to 1 — without this the
        // witness-gated rename-over (heddle#209) skips the orphan
        // transition for `Released` destinations and the
        // captured-bytes-via-old-fd flow doesn't engage.
        mount.on_open(dest_id).expect("on_open");
        // Source file with replacement payload; flush so move_file's
        // flush_node returns immediately.
        let src = mount
            .create_file(NodeId::ROOT, OsStr::new("draft"), FileMode::Normal, false)
            .expect("create src");
        mount
            .write(src.node, 0, b"REPLACE-DATA")
            .expect("write src");
        mount.flush(src.node).expect("flush src");
        mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("draft"),
                NodeId::ROOT,
                OsStr::new("hello.txt"),
            )
            .expect("rename-over");
        // Read via the displaced inode id must serve the original
        // captured bytes ("world", 5 bytes) — not the source's
        // "REPLACE-DATA" (12 bytes).
        let mut buf = vec![0u8; 32];
        let n = mount
            .read(dest_id, 0, &mut buf)
            .expect("read via replaced dest fd");
        assert_eq!(
            &buf[..n],
            b"world",
            "open fd on replaced captured dest must see captured bytes, not replacement",
        );
        // attrs must report the captured size too — a stale size from
        // the path overlay would clip the kernel's read buffer.
        let attrs = mount.attrs(dest_id).expect("attrs via replaced dest fd");
        assert_eq!(
            attrs.size, 5,
            "attrs on replaced captured dest must report captured size, not replacement",
        );
    }

    // --- Codex round 8 findings: 3-axis sweep (warm tier + lifecycle + atomicity)
    //
    // r7 made the write-side ops orphan-aware against `pending.hot` /
    // `hot_by_path` / `tombstones` / `inodes.by_path`. r8 closes the
    // remaining three same-shape misses:
    //
    //   1. `pending.warm` was dropped unconditionally on unlink and
    //      rename-over, even when the inode had surviving open fds. The
    //      bytes those fds want to read disappeared with the path.
    //   2. The orphan marker cleared on the FIRST `flush`. FUSE `flush`
    //      fires on EACH descriptor close (incl. `dup`-derived fds);
    //      only `release` is the last-close-per-inode signal. A
    //      premature clear lets a surviving fd's next write republish
    //      the unlinked path.
    //   3. `RENAME_NOREPLACE` did the existence-check in the FUSE shell
    //      and the rename in the core — separate locks, classic TOCTOU.
    //      Concurrent writers could create the destination between the
    //      two operations and the rename would clobber it.

    /// An open fd to a captured file whose warm-tier bytes are present
    /// must still serve those bytes after the path is unlinked. POSIX
    /// open-unlinked semantics: the inode survives behind the fd, and
    /// `pending.warm[path]` is the most-recently-promoted bytes the fd
    /// owns. r7's `unlink_entry` dropped `pending.warm[path]`
    /// unconditionally — orphan reads through the surviving fd lost
    /// the latest writes.
    #[test]
    fn unlink_then_read_warm_via_open_fd() {
        let (_temp, mount) = open_mount();
        // Open hello.txt and write through it. Flush promotes to warm
        // so the bytes are at `pending.warm["hello.txt"]`, not in any
        // hot buffer.
        let node = mount.lookup_path("hello.txt").unwrap();
        // The `open` above bumps `open_count` to 1 — without this the
        // witness-gated unlink (heddle#209) sees the node as
        // `Released` (no entry) and skips the orphan transition.
        mount.on_open(node).expect("on_open");
        mount.write(node, 0, b"WARM-BYTES").expect("write");
        mount.flush(node).expect("flush — promote to warm");
        // Sanity: warm tier holds the bytes.
        assert!(
            mount.warm_blob("hello.txt").is_some(),
            "warm should hold the flushed bytes"
        );
        // `unlink("hello.txt")` while the fd lives on. POSIX: the
        // inode survives; the directory entry is gone.
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("hello.txt"))
            .expect("unlink");
        // Read via the orphaned fd. The most recent durable bytes for
        // this inode are the warm-tier bytes; they must still be
        // reachable.
        let mut buf = vec![0u8; 32];
        let n = mount
            .read(node, 0, &mut buf)
            .expect("read via orphan fd after warm-promoted unlink");
        assert_eq!(
            &buf[..n],
            b"WARM-BYTES",
            "orphan read must serve preserved warm bytes, not captured fallback"
        );
        // attrs must report the warm size too — a stale captured size
        // would clip the kernel's read buffer.
        let attrs = mount.attrs(node).expect("attrs via orphan fd");
        assert_eq!(
            attrs.size, 10,
            "attrs on orphan must report warm size, not captured"
        );
    }

    /// `ftruncate` through an unlinked-but-still-open fd whose only
    /// durable bytes are warm-tier (no captured-tree predecessor) must
    /// seed the truncated buffer from those warm bytes. Without the
    /// warm-preservation fix, the seed lookup falls through to "empty"
    /// and the surviving fd's `ftruncate` followed by `read` returns
    /// zeros — POSIX requires the truncate to start from the inode's
    /// current bytes.
    #[test]
    fn truncate_unlinked_open_keeps_warm_bytes() {
        let (_temp, mount) = open_mount();
        // Pending file (no captured-tree backing) so the orphan has
        // nothing but warm bytes to fall back on.
        let entry = mount
            .create_file(NodeId::ROOT, OsStr::new("scratch"), FileMode::Normal, false)
            .expect("create");
        // The `create` (`O_CREAT|O_RDWR`) above bumps `open_count` to
        // 1 — without this the witness-gated unlink (heddle#209) sees
        // the node as `Released` (no entry) and skips the orphan
        // transition, breaking the open-unlinked POSIX flow.
        mount.on_open(entry.node).expect("on_open");
        mount.write(entry.node, 0, b"hello-world").expect("write");
        mount.flush(entry.node).expect("flush — promote to warm");
        assert!(
            mount.warm_blob("scratch").is_some(),
            "warm should hold the flushed bytes"
        );
        // Unlink while the fd lives on.
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("scratch"))
            .expect("unlink");
        // ftruncate(fd, 5) through the orphaned NodeId.
        mount
            .set_attrs(
                entry.node,
                AttrUpdate {
                    size: Some(5),
                    ..Default::default()
                },
            )
            .expect("ftruncate orphan");
        // Read the truncated bytes. Without the warm-preservation
        // fix, this returns "\0\0\0\0\0" instead of "hello".
        let mut buf = vec![0u8; 16];
        let n = mount
            .read(entry.node, 0, &mut buf)
            .expect("read truncated orphan");
        assert_eq!(
            &buf[..n],
            b"hello",
            "truncate-then-read on orphan must seed from preserved warm bytes"
        );
        // Path stays gone.
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("scratch"))
                .unwrap()
                .is_none(),
            "truncate after unlink must not resurrect the path"
        );
    }

    /// Rename-over when the displaced destination has warm-tier bytes
    /// (no in-flight hot buffer) must preserve those bytes for the
    /// still-open fd. Without the fix, `move_file`'s blind
    /// `pending.warm.remove(new_path)` dropped them; the surviving fd
    /// either sees the source's replacement bytes or an empty file.
    #[test]
    fn rename_over_preserves_warm_for_open_fd() {
        let (_temp, mount) = open_mount();
        // Open hello.txt and write+flush so its bytes live in warm,
        // not hot. r7's existing rename-over preservation fix is for
        // hot buffers; this one exercises the warm-tier preservation.
        let dest_id = mount.lookup_path("hello.txt").unwrap();
        // The `open` above bumps `open_count` to 1 — without this the
        // witness-gated rename-over (heddle#209) skips the orphan
        // transition for `Released` destinations and the
        // warm-preservation path doesn't engage.
        mount.on_open(dest_id).expect("on_open");
        mount
            .write(dest_id, 0, b"DEST-WARM-BYTES")
            .expect("write dest");
        mount.flush(dest_id).expect("flush dest — promote to warm");
        assert!(
            mount.warm_blob("hello.txt").is_some(),
            "dest must be warm-promoted at rename time"
        );
        // Build a source file with replacement payload and flush.
        let src = mount
            .create_file(NodeId::ROOT, OsStr::new("draft"), FileMode::Normal, false)
            .expect("create src");
        mount
            .write(src.node, 0, b"REPLACE-DATA")
            .expect("write src");
        mount.flush(src.node).expect("flush src");
        // Rename src over dest. dest's pathname is rebound to src.
        // dest's open fd must keep seeing the displaced WARM bytes.
        mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("draft"),
                NodeId::ROOT,
                OsStr::new("hello.txt"),
            )
            .expect("rename-over");
        // Read via the displaced inode id. Without the fix, this
        // returns "REPLACE-DATA" (source's bytes via `warm[new_path]`)
        // or captured "world" (fallback after warm drop).
        let mut buf = vec![0u8; 32];
        let n = mount
            .read(dest_id, 0, &mut buf)
            .expect("read via replaced dest fd");
        assert_eq!(
            &buf[..n],
            b"DEST-WARM-BYTES",
            "open fd on replaced dest must see pre-rename WARM bytes"
        );
        // attrs must match.
        let attrs = mount.attrs(dest_id).expect("attrs via replaced dest fd");
        assert_eq!(
            attrs.size, 15,
            "attrs must report preserved warm size, not replacement or captured"
        );
    }

    /// FUSE `flush` fires on every descriptor close (including dup'd
    /// fds); only `release` is the last-close-per-inode signal. r7
    /// cleared the orphan marker in `flush_node`, so the FIRST close of
    /// a dup'd unlinked fd cleared the marker prematurely. A subsequent
    /// write through the surviving dup would then take the non-orphan
    /// branch and republish the unlinked path.
    ///
    /// We exercise the contract by simulating the two-open lifecycle:
    /// `on_open` twice (representing 2 fds), `unlink`, `flush` (orphan
    /// marker must stay), `release` once (one fd closed; marker stays),
    /// `release` again (last close; marker clears).
    #[test]
    fn flush_keeps_orphan_marker_until_release() {
        let (_temp, mount) = open_mount();
        // Capture-backed file, opened twice (e.g. one fd then dup —
        // FUSE would track these as a single open + multiple flushes
        // + one release; or two opens + two releases. Either way the
        // marker must persist across the non-final close.)
        let node = mount.lookup_path("hello.txt").unwrap();
        // Two opens → refcount 2.
        mount.on_open(node).expect("open 1");
        mount.on_open(node).expect("open 2");
        // Unlink while both fds live on.
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("hello.txt"))
            .expect("unlink");
        assert!(
            mount.orphans_contains(node),
            "unlink-while-open must set the orphan marker"
        );
        // `flush` fires on each close (per FUSE protocol). It MUST NOT
        // clear the orphan marker — the surviving fd needs to keep
        // taking the orphan branch on writes.
        mount.flush(node).expect("flush #1");
        assert!(
            mount.orphans_contains(node),
            "flush must not clear the orphan marker"
        );
        // First `release` — represents one of the two fds closing.
        // Marker still present because the other fd holds the inode.
        mount.release(node).expect("release #1");
        assert!(
            mount.orphans_contains(node),
            "non-final release must not clear orphan marker"
        );
        // Second `release` — the last close. Now the marker must
        // clear, the orphan buffer (if any) drops, and the inode's
        // bookkeeping is freed.
        mount.release(node).expect("release #2");
        assert!(
            !mount.orphans_contains(node),
            "final release must clear the orphan marker"
        );
    }

    /// `RENAME_NOREPLACE` must be honoured atomically by the core. r6/r7
    /// did the existence-check in the FUSE shell and the rename in the
    /// core under separate locks — between the two, a concurrent writer
    /// could create the destination and the rename would silently
    /// clobber it. r8 plumbs the flag into the core's mutation
    /// critical section so the check + rename land under the same
    /// write-side lock.
    ///
    /// Deterministic shape: with the flag set, `rename_entry_with_options`
    /// must return `AlreadyExists` when the destination resolves. The
    /// caller is responsible for not racing — but with the core
    /// honouring the flag inside its own critical section, sequential
    /// callers (FUSE serializes write callbacks per inode anyway) get
    /// strict NOREPLACE semantics.
    #[test]
    fn rename_noreplace_is_atomic() {
        use crate::shell::RenameOptions;
        let (_temp, mount) = open_mount();
        // Build a source file with bytes; flush so the rename works
        // entirely from warm tier.
        let src = mount
            .create_file(NodeId::ROOT, OsStr::new("draft"), FileMode::Normal, false)
            .expect("create src");
        mount.write(src.node, 0, b"draft-bytes").expect("write src");
        mount.flush(src.node).expect("flush src");
        // Destination already exists in the captured tree. With
        // NOREPLACE the rename must refuse with `EEXIST`/`AlreadyExists`
        // BEFORE making any mutation.
        let err = mount
            .rename_entry_with_options(
                NodeId::ROOT,
                OsStr::new("draft"),
                NodeId::ROOT,
                OsStr::new("hello.txt"),
                RenameOptions { no_replace: true },
            )
            .expect_err("NOREPLACE over existing dest must fail");
        assert!(
            matches!(err, MountError::AlreadyExists(_)),
            "got unexpected error: {err:?}"
        );
        assert_eq!(err.to_errno(), libc::EEXIST);
        // The source must still be intact — NOREPLACE failure must not
        // leave the source in a half-renamed state.
        let src_after = mount
            .lookup(NodeId::ROOT, OsStr::new("draft"))
            .unwrap()
            .expect("source intact after NOREPLACE rejection");
        let mut buf = vec![0u8; 16];
        let n = mount.read(src_after.node, 0, &mut buf).unwrap();
        assert_eq!(
            &buf[..n],
            b"draft-bytes",
            "source bytes intact after NOREPLACE rejection"
        );
        // And the destination must still be the captured "world".
        let dst = mount
            .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
            .unwrap()
            .expect("dest still resolves");
        let mut buf = vec![0u8; 16];
        let n = mount.read(dst.node, 0, &mut buf).unwrap();
        assert_eq!(
            &buf[..n],
            b"world",
            "dest bytes unchanged after NOREPLACE rejection"
        );

        // Same call WITHOUT the flag must succeed (and replace).
        mount
            .rename_entry_with_options(
                NodeId::ROOT,
                OsStr::new("draft"),
                NodeId::ROOT,
                OsStr::new("hello.txt"),
                RenameOptions::default(),
            )
            .expect("rename without NOREPLACE replaces");
        let dst2 = mount
            .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
            .unwrap()
            .expect("dst still resolves");
        let mut buf = vec![0u8; 32];
        let n = mount.read(dst2.node, 0, &mut buf).unwrap();
        assert_eq!(
            &buf[..n],
            b"draft-bytes",
            "non-NOREPLACE rename replaces dest"
        );
    }

    // --- Codex round 9 finding: hot bytes lost on unlink-of-open ------------
    //
    // r8 closed the warm-tier / lifecycle / atomicity sweep. r9 surfaced a
    // remaining same-shape regression: `unlink_entry` removes
    // `pending.hot[node_id]` for the orphan branch, so an open fd whose only
    // bytes lived in the hot buffer (write then unlink, no flush in between)
    // loses them. Codex thread 3293307302.
    //
    // Under the post-spike unified NodeId-keyed model, hot[node_id] survives
    // the Live → Orphan transition by construction — bytes follow the NodeId,
    // not the path. This test pins the contract: write to an open fd, unlink
    // the path, read through the surviving fd → original bytes.

    /// `open(path) → write(fd, "DIRTY") → unlink(path) → read(fd)` must
    /// return `"DIRTY"`. POSIX open-unlinked semantics: the inode lives
    /// behind the fd until the last close. The pre-spike code dropped
    /// `pending.hot[node_id]` inside `unlink_entry`, so the read fell
    /// through to the captured blob (`"world"`) instead of the dirty hot
    /// bytes the fd had written.
    #[test]
    fn unlink_open_fd_preserves_unflushed_hot_bytes() {
        let (_temp, mount) = open_mount();
        // `fd = open("hello.txt")` — captured file with bytes "world".
        let node = mount.lookup_path("hello.txt").unwrap();
        mount.on_open(node).expect("on_open");
        // Write through the fd, do NOT flush. The bytes live only in
        // `pending.hot[node]`.
        mount
            .write(node, 0, b"DIRTY-BYTES")
            .expect("write through open fd");
        // `unlink("hello.txt")` while the fd lives on.
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("hello.txt"))
            .expect("unlink while open");
        assert!(
            mount
                .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
                .unwrap()
                .is_none(),
            "post-unlink lookup must be gone"
        );
        // Read via the orphaned fd. The bytes the fd wrote must survive
        // the unlink — POSIX open-unlinked. Pre-fix, hot[node] was
        // dropped at unlink and this returned "world" (captured blob).
        let mut buf = vec![0u8; 32];
        let n = mount.read(node, 0, &mut buf).expect("read via orphan fd");
        assert_eq!(
            &buf[..n],
            b"DIRTY-BYTES",
            "open fd → write → unlink → read must return the unflushed bytes \
             (regression: pre-spike code dropped hot[node_id] in unlink_entry)"
        );
        // attrs must report the dirty-buffer size, not the captured size.
        let attrs = mount.attrs(node).expect("attrs via orphan fd");
        assert_eq!(
            attrs.size, 11,
            "attrs on orphan must report hot-buffer size, not captured"
        );
    }

    // --- Codex round 12 findings: capture / invalidate / forget / rmdir / -----
    // --- symlinks / read_link ------------------------------------------------
    //
    // Six P1 + one P3 findings on the post-spike base (`58a30b2`). Each test
    // pins the contract from one Codex review thread on PR #182.

    /// Codex PR #182 r11 finding #3 (P1; heddle#211). `invalidate`
    /// (FUSE `forget`) drops `pending.hot[node]` unconditionally,
    /// without first checking whether the inode is still referenced.
    /// For an `Orphan { open_count >= 1 }` node (open-unlinked POSIX
    /// flow), the kernel can issue `forget` for the dentry-side
    /// reference while a userspace fd still holds the inode; dropping
    /// `hot[node]` strands the surviving fd with no readable bytes.
    /// Post-retrofit (heddle#211) the witness-gated
    /// `BrandedPending::kernel_forget_inode` rejects any `Orphan`
    /// state, and the missing witness short-circuits the entire
    /// forget path in `MountInner::invalidate` — leaving `hot[node]`,
    /// `state[node]`, and the inode record intact until the final
    /// `release` retires them.
    #[test]
    fn invalidate_preserves_hot_bytes_for_orphan_with_open_fd() {
        let (_temp, mount) = open_mount();
        // Create + open a fresh file, write through the fd. Bytes
        // stay in `hot[node]` — no flush, so warm is untouched.
        let entry = mount
            .create_file(NodeId::ROOT, OsStr::new("scratch"), FileMode::Normal, false)
            .expect("create");
        mount.on_open(entry.node).expect("on_open");
        mount
            .write(entry.node, 0, b"HOT-BYTES")
            .expect("write through live fd populates hot");
        // Unlink while the fd lives on. POSIX: dentry gone, inode
        // survives behind the fd; FSM transitions to
        // `Orphan { open_count: 1 }`.
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("scratch"))
            .expect("unlink");
        assert!(
            mount.orphans_contains(entry.node),
            "pre-invalidate sanity: unlink-while-open must orphan the inode"
        );

        // Kernel forget arrives for the dentry-side reference while
        // the fd is still in userspace's hands. Pre-retrofit
        // `invalidate` blindly removed `hot[node]`; post-retrofit
        // the witness rejects `Orphan` and the forget path
        // short-circuits.
        mount
            .invalidate(entry.node)
            .expect("invalidate (kernel forget) on orphan with open fd");

        // Read via the surviving fd. The hot-tier bytes must still
        // be served — the open-unlinked POSIX contract demands that
        // the fd's view of the inode outlives the dentry.
        let mut buf = vec![0u8; 32];
        let n = mount
            .read(entry.node, 0, &mut buf)
            .expect("read via orphan fd after kernel forget");
        assert_eq!(
            &buf[..n],
            b"HOT-BYTES",
            "kernel forget racing an open Orphan fd must not drop \
             hot[node] — the surviving fd needs the bytes (r11 #3)"
        );
    }

    /// Codex thread 3293510310 (P1). `rmdir_entry` plants a
    /// `dir_tombstones` entry but leaves `inodes.by_path[child_path]`
    /// intact. A subsequent `create_file` / `make_dir` at the same path
    /// then coalesces onto the removed directory's NodeId via
    /// `Inodes::intern`'s path-keyed reverse index — that's a stale-handle
    /// class identical to the one `unlink_entry` already guards against
    /// by retiring `by_path`.
    #[test]
    fn rmdir_retires_path_to_inode_binding() {
        let (_temp, mount) = open_mount();
        let dir = mount
            .make_dir(NodeId::ROOT, OsStr::new("scratch"))
            .expect("mkdir");
        mount
            .rmdir_entry(NodeId::ROOT, OsStr::new("scratch"))
            .expect("rmdir");
        // Recreate the name as a regular file. The fresh inode must NOT
        // be the removed directory's NodeId — POSIX remove-and-recreate
        // isolation forbids rebinding a cached dir inode to a different
        // object type.
        let file = mount
            .create_file(NodeId::ROOT, OsStr::new("scratch"), FileMode::Normal, false)
            .expect("recreate as file");
        assert_ne!(
            file.node, dir.node,
            "remove-and-recreate at same path must mint a fresh inode \
             (rmdir must retire inodes.by_path so intern does not coalesce)"
        );
    }

    /// Codex thread 3293510317 (P3). `unlink_entry` always transitions
    /// the removed node into `NodeState::Orphan`, but symlinks are not
    /// openable for IO — they never receive `open`/`release` lifecycle
    /// events to clear the state. The entry accumulates in `pending.state`
    /// until capture/invalidate. The fix gates the orphan transition on
    /// `entry.kind != Symlink`.
    #[test]
    fn unlink_of_symlink_does_not_create_orphan_state() {
        let (_temp, mount) = open_mount();
        let link = mount
            .create_symlink(NodeId::ROOT, OsStr::new("link"), Path::new("hello.txt"))
            .expect("create_symlink");
        mount
            .unlink_entry(NodeId::ROOT, OsStr::new("link"))
            .expect("unlink symlink");
        assert!(
            !mount.orphans_contains(link.node),
            "symlinks have no open/release lifecycle; unlink must not orphan them"
        );
    }

    // --- Codex round 13 findings: setattr lock discipline + name/mode pickiness
    //
    // r12 closed the post-spike NodeId-keyed refactor's residual surgical
    // gaps. r13 surfaces three more same-class issues that all live in the
    // write-side `set_attrs` / `validate_entry_name` surface:
    //
    //   1. `set_attrs(size=...)` routes into `apply_truncate` without taking
    //      `write_mu`. Concurrent with `rename`, the truncate's final
    //      bookkeeping uses the pre-rename pathname — republishing
    //      `hot_by_path[old]` and clearing the rename's tombstone, so the
    //      old name resurrects. (Codex thread 3293733165, P1.)
    //   2. `validate_entry_name` rejects only NUL and `/`, but the tree
    //      serializer additionally rejects `\` and control bytes
    //      (0x01..=0x1F, 0x7F). Names that pass the FUSE-side check then
    //      fail at capture with a confusing "invalid object" error rather
    //      than a clean EINVAL at write time. (Codex thread 3293733163, P2.)
    //   3. `set_attrs`'s Normal↔Executable fold treats any of the three
    //      execute bits as executable (`mode & 0o111 != 0`). The contract
    //      is "user execute only" — a `chmod 0o010` (group execute) must
    //      stay Normal. (Codex thread 3293733164, P2.)

    /// Codex r13 thread 3293733165 (P1). `set_attrs(size=...)` racing
    /// with `rename_entry` must not republish the pre-rename pathname.
    ///
    /// Mechanism without `write_mu` in `set_attrs`: `apply_truncate`
    /// captures `path` at the top via `record_for`, then drops every
    /// lock for the seed `load_blob_bytes` call. A concurrent rename
    /// fits entirely in that lock-free window — it inserts
    /// `tombstones[old]`, removes `hot_by_path[old]`, and rebases the
    /// inode's stored path. When `apply_truncate`'s phase-2 mutation
    /// re-acquires the pending lock and runs
    /// `tombstones.remove(&path) + hot_by_path.insert(path, node)`
    /// with the stale `path = old`, the rename's tombstone is wiped
    /// and `lookup(old)` resurrects the file.
    ///
    /// Stress shape: 200 trials, two threads sync on a barrier and
    /// then run their op concurrently. Even a single observation of
    /// `lookup(old).is_some()` after both complete fails the test.
    /// With the fix (`write_mu` held around the mutating `set_attrs`
    /// paths), the race is structurally impossible.
    #[test]
    fn setattr_truncate_serializes_against_rename() {
        use std::{
            sync::{Arc, Barrier},
            thread,
        };
        const TRIALS: usize = 200;
        let mut resurrected: Vec<usize> = Vec::new();
        for trial in 0..TRIALS {
            let (_temp, mount) = open_mount();
            let mount = Arc::new(mount);
            // `hello.txt` is a captured 5-byte file; `apply_truncate`'s
            // NeedSeed branch will fetch the blob and widen the
            // lock-free window during the race.
            let node = mount.lookup_path("hello.txt").unwrap();
            let barrier = Arc::new(Barrier::new(2));

            let b_a = barrier.clone();
            let m_a = mount.clone();
            let h_a = thread::spawn(move || {
                b_a.wait();
                m_a.set_attrs(
                    node,
                    AttrUpdate {
                        size: Some(2),
                        ..Default::default()
                    },
                )
                .expect("setattr(size=2)");
            });
            let b_b = barrier.clone();
            let m_b = mount.clone();
            let h_b = thread::spawn(move || {
                b_b.wait();
                m_b.rename_entry(
                    NodeId::ROOT,
                    OsStr::new("hello.txt"),
                    NodeId::ROOT,
                    OsStr::new("renamed.txt"),
                )
                .expect("rename");
            });
            h_a.join().unwrap();
            h_b.join().unwrap();

            // Invariant: regardless of interleaving, the pre-rename
            // path must not resolve after both ops complete.
            if mount
                .lookup(NodeId::ROOT, OsStr::new("hello.txt"))
                .unwrap()
                .is_some()
            {
                resurrected.push(trial);
            }
        }
        assert!(
            resurrected.is_empty(),
            "setattr(size)+rename race resurrected pre-rename path \
             in {} of {} trials (trials: {:?})",
            resurrected.len(),
            TRIALS,
            resurrected,
        );
    }

    /// Codex r13 thread 3293733163 (P2). `validate_entry_name` must
    /// reject every name the tree serializer would reject — otherwise
    /// the overlay accepts a create/rename that later blows up at
    /// `capture` with a confusing "invalid object" error.
    ///
    /// The tree serializer's rule (objects::object::tree_entry::
    /// validate_name) rejects: empty, `.`, `..`, anything containing
    /// `/` or `\`, anything with a byte < 0x20 or == 0x7F. The mount's
    /// `validate_entry_name` previously rejected only NUL + `/`.
    ///
    /// Each name in this list pins one rule the mount must enforce up
    /// front. Without the fix, `create_file` / `rename` accept these
    /// and the failure surfaces later (or not at all, leaking a stale
    /// pending entry).
    #[test]
    fn create_rejects_names_tree_cannot_serialize() {
        let (_temp, mount) = open_mount();
        // Each (name, why) tuple is rejected by the tree serializer.
        let cases: &[(&[u8], &str)] = &[
            (b"with\\backslash", "backslash (path separator on Windows)"),
            (b"bel\x07", "BEL (0x07) is a control char"),
            (b"esc\x1b", "ESC (0x1B) is a control char"),
            (b"tab\there", "TAB (0x09) is a control char"),
            (b"newline\nhere", "LF (0x0A) is a control char"),
            (b"cr\rhere", "CR (0x0D) is a control char"),
            (b"del\x7f", "DEL (0x7F) is the upper control char"),
        ];
        for (bytes, why) in cases {
            // Build the OsStr from raw bytes so we can include non-UTF-8
            // tricks (the safe path here is all-ASCII, but the helper
            // is byte-oriented to match the tree serializer's reject
            // set).
            #[cfg(unix)]
            let name = {
                use std::os::unix::ffi::OsStrExt;
                std::ffi::OsString::from(OsStr::from_bytes(bytes))
            };
            #[cfg(not(unix))]
            let name =
                std::ffi::OsString::from(std::str::from_utf8(bytes).expect("ascii test inputs"));

            let err = mount
                .create_file(NodeId::ROOT, &name, FileMode::Normal, false)
                .expect_err(&format!(
                    "create_file must reject {name:?} ({why}) up front"
                ));
            assert!(
                matches!(err, MountError::InvalidArgument(_)),
                "expected InvalidArgument for {name:?} ({why}), got {err:?}"
            );
        }
    }

    /// Codex r13 thread 3293733164 (P2). The Normal↔Executable mode
    /// fold must trigger on the user execute bit (S_IXUSR = 0o100)
    /// only, not on any of the three execute bits (0o111). A
    /// `chmod 0o010` (group execute only) must NOT promote a Normal
    /// file to Executable — that would unexpectedly grant owner
    /// execute at capture time.
    #[test]
    fn chmod_group_execute_alone_doesnt_promote_to_executable() {
        let (_temp, mount) = open_mount();
        let node = mount.lookup_path("hello.txt").unwrap();
        // chmod 0o010: group execute only. Per the contract, this
        // must NOT promote to Executable.
        let attrs = mount
            .set_attrs(
                node,
                AttrUpdate {
                    mode: Some(0o100_010),
                    ..Default::default()
                },
            )
            .expect("chmod 0o010");
        // FileMode::Normal serializes to 0o644 (no execute anywhere);
        // FileMode::Executable serializes to 0o755 (all execute bits).
        // The fix gates the fold on S_IXUSR (0o100), so 0o010 leaves
        // the record as Normal → unix_mode has no execute bits at all.
        assert_eq!(
            attrs.unix_mode & 0o111,
            0,
            "chmod 0o010 must NOT promote to Executable (got unix_mode={:o})",
            attrs.unix_mode
        );

        // Companion assertion (anti-regression): chmod 0o100 (user
        // execute only) DOES promote to Executable.
        let attrs = mount
            .set_attrs(
                node,
                AttrUpdate {
                    mode: Some(0o100_100),
                    ..Default::default()
                },
            )
            .expect("chmod 0o100");
        assert_eq!(
            attrs.unix_mode & 0o111,
            0o111,
            "chmod 0o100 must promote to Executable (got unix_mode={:o})",
            attrs.unix_mode
        );

        // And chmod 0o755 (the canonical Executable path) still
        // works — this is the path the r12 baseline already covers
        // and we don't want to regress.
        let attrs = mount
            .set_attrs(
                node,
                AttrUpdate {
                    mode: Some(0o100_755),
                    ..Default::default()
                },
            )
            .expect("chmod 0o755");
        assert_eq!(
            attrs.unix_mode & 0o111,
            0o111,
            "chmod 0o755 must keep Executable (got unix_mode={:o})",
            attrs.unix_mode
        );
    }

    fn assert_pending_index_matches_scan(mount: &ContentAddressedMount) {
        mount
            .pending_index_matches_full_scan()
            .expect("pending child index diverged from full-scan oracle");
    }

    #[test]
    fn pending_child_index_matches_full_scan_across_mutations() {
        let (_temp, mount) = open_mount();
        assert_pending_index_matches_scan(&mount);

        let alpha = mount
            .make_dir(NodeId::ROOT, OsStr::new("alpha"))
            .expect("mkdir alpha");
        let beta = mount
            .make_dir(NodeId::ROOT, OsStr::new("beta"))
            .expect("mkdir beta");
        assert_pending_index_matches_scan(&mount);

        let warm = mount
            .create_file(alpha.node, OsStr::new("warm.txt"), FileMode::Normal, false)
            .expect("create warm file");
        mount
            .write(warm.node, 0, b"pending")
            .expect("write warm file");
        assert_pending_index_matches_scan(&mount);
        mount.flush(warm.node).expect("promote warm file");
        assert_pending_index_matches_scan(&mount);

        let nested = mount
            .make_dir(alpha.node, OsStr::new("nested"))
            .expect("mkdir nested");
        let deep = mount
            .create_file(
                nested.node,
                OsStr::new("deep.rs"),
                FileMode::Executable,
                false,
            )
            .expect("create nested file");
        mount
            .write(deep.node, 0, b"fn main() {}")
            .expect("write nested file");
        mount
            .create_symlink(alpha.node, OsStr::new("link"), Path::new("warm.txt"))
            .expect("create symlink");
        mount
            .create_file(alpha.node, OsStr::new("Case"), FileMode::Normal, false)
            .expect("create case-sensitive upper name");
        mount
            .create_file(alpha.node, OsStr::new("case"), FileMode::Normal, false)
            .expect("create case-sensitive lower name");
        assert_pending_index_matches_scan(&mount);

        mount
            .rename_entry(
                alpha.node,
                OsStr::new("warm.txt"),
                beta.node,
                OsStr::new("moved.txt"),
            )
            .expect("rename warm file across directories");
        assert_pending_index_matches_scan(&mount);
        mount
            .unlink_entry(beta.node, OsStr::new("moved.txt"))
            .expect("unlink moved file");
        assert_pending_index_matches_scan(&mount);

        mount
            .rename_entry(
                NodeId::ROOT,
                OsStr::new("alpha"),
                NodeId::ROOT,
                OsStr::new("gamma"),
            )
            .expect("rename overlay directory");
        assert_pending_index_matches_scan(&mount);

        let gamma = mount
            .lookup(NodeId::ROOT, OsStr::new("gamma"))
            .unwrap()
            .expect("renamed directory resolves");
        mount
            .unlink_entry(gamma.node, OsStr::new("link"))
            .expect("unlink rebased symlink");
        assert_pending_index_matches_scan(&mount);
    }

    fn mount_with_pending_burst(count: usize) -> (TempDir, ContentAddressedMount) {
        let (temp, mount) = open_mount();
        let target = mount
            .make_dir(NodeId::ROOT, OsStr::new("target"))
            .expect("mkdir target");
        mount
            .create_file(target.node, OsStr::new("only.txt"), FileMode::Normal, false)
            .expect("create target child");
        for index in 0..count {
            mount
                .create_file(
                    NodeId::ROOT,
                    OsStr::new(&format!("burst-{index:05}.tmp")),
                    FileMode::Normal,
                    false,
                )
                .expect("create burst file");
        }
        (temp, mount)
    }

    #[test]
    fn pending_child_index_bounds_work_and_full_scan_is_negative_control() {
        const PENDING_FILES: usize = 2_048;
        let (_small_temp, small) = mount_with_pending_burst(32);
        let (_temp, mount) = mount_with_pending_burst(PENDING_FILES);
        assert_pending_index_matches_scan(&mount);

        let (_, _, small_scan_exists_work, small_scan_children_work) =
            small.pending_index_work(Path::new("missing"));

        let (exists_work, children_work, scan_exists_work, scan_children_work) =
            mount.pending_index_work(Path::new("missing"));
        assert_eq!(exists_work, 1, "indexed exists is one directory lookup");
        assert_eq!(children_work, 0, "missing directory has no direct children");
        assert!(
            scan_exists_work >= PENDING_FILES,
            "negative-control exists scan inspected only {scan_exists_work} entries"
        );
        assert!(
            scan_children_work >= PENDING_FILES,
            "negative-control listing scan inspected only {scan_children_work} entries"
        );
        assert!(
            scan_exists_work > small_scan_exists_work * 32,
            "negative-control exists work did not grow with N: 32 files={small_scan_exists_work}, {PENDING_FILES} files={scan_exists_work}"
        );
        assert!(
            scan_children_work > small_scan_children_work * 32,
            "negative-control listing work did not grow with N: 32 files={small_scan_children_work}, {PENDING_FILES} files={scan_children_work}"
        );

        let (_, target_children_work, _, target_scan_work) =
            mount.pending_index_work(Path::new("target"));
        assert_eq!(
            target_children_work, 1,
            "indexed listing work must equal target-directory fanout"
        );
        assert!(target_scan_work >= PENDING_FILES);
        eprintln!(
            "pending-index counters: indexed exists={exists_work}, missing children={children_work}, target children={target_children_work}; full-scan missing exists 32→{PENDING_FILES}={small_scan_exists_work}→{scan_exists_work}, missing children={small_scan_children_work}→{scan_children_work}, target children={target_scan_work}"
        );
    }

    #[test]
    fn pending_child_index_burst_latency_stays_flat() {
        let (_small_temp, small) = mount_with_pending_burst(64);
        let (_large_temp, large) = mount_with_pending_burst(4_096);
        let iterations = 20_000;
        let small_elapsed = small.time_pending_index(Path::new("target"), iterations);
        let large_elapsed = large.time_pending_index(Path::new("target"), iterations);
        eprintln!(
            "pending-index burst latency: 64 files={small_elapsed:?}, 4096 files={large_elapsed:?}, iterations={iterations}"
        );
        assert!(
            large_elapsed <= small_elapsed.saturating_mul(6) + std::time::Duration::from_millis(2),
            "pending lookup/listing latency scaled with total pending entries: small={small_elapsed:?}, large={large_elapsed:?}"
        );
    }
}

// D2. Real-FUSE Linux integration test.
#[cfg(all(target_os = "linux", feature = "fuse"))]
mod fuse_smoke {
    use std::time::Duration;

    use super::*;
    use crate::FuseShell;

    /// Wait for `path` to come up, bounded by `deadline`.
    fn wait_until_exists(path: &std::path::Path, deadline: Duration) {
        let start = std::time::Instant::now();
        while !path.exists() && start.elapsed() < deadline {
            std::thread::sleep(Duration::from_millis(20));
        }
    }

    #[test]
    #[ignore = "requires /dev/fuse; exercised by the FUSE mount smoke CI lane"]
    fn fuse_open_close_read_round_trip() {
        assert!(
            std::path::Path::new("/dev/fuse").exists(),
            "fuse_open_close_read_round_trip requires /dev/fuse"
        );
        let repo_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(repo_dir.path()).unwrap();
        std::fs::write(repo_dir.path().join("seed.txt"), b"hello").unwrap();
        repo.snapshot(Some("seed".into()), None).unwrap();

        let mount = ContentAddressedMount::new(repo, "main").unwrap();
        let mountpoint = TempDir::new().unwrap();
        let session = FuseShell::new(mount)
            .mount_background(mountpoint.path())
            .expect("mount FUSE session");
        wait_until_exists(&mountpoint.path().join("seed.txt"), Duration::from_secs(5));
        let read = std::fs::read_to_string(mountpoint.path().join("seed.txt")).unwrap();
        assert_eq!(read, "hello");
        drop(session);
    }
}

// ---------------------------------------------------------------------------
// PlatformShell default-impl coverage.
//
// Every write-side method on the trait has a default body that returns
// `MountError::ReadOnly`. A read-only adapter (e.g. a future "snapshot
// browser" shell that only implements the six required methods) inherits
// those defaults. These tests exercise the default bodies so the contract
// stays observable: read-only shells uniformly surface `ReadOnly` rather
// than panicking or silently succeeding.
// ---------------------------------------------------------------------------
#[cfg(test)]
mod platform_shell_defaults {
    use std::{
        ffi::{OsStr, OsString},
        path::Path,
        time::SystemTime,
    };

    use objects::object::FileMode;

    use crate::{
        error::{MountError, Result},
        shell::{AttrUpdate, Attrs, Entry, NodeId, NodeKind, PlatformShell},
    };

    /// Minimal read-only shell that implements only the required
    /// methods. Every write-side default kicks in.
    struct ReadOnlyStub;

    impl PlatformShell for ReadOnlyStub {
        fn lookup(&self, _parent: NodeId, _name: &OsStr) -> Result<Option<Entry>> {
            Ok(None)
        }
        fn read(&self, _node: NodeId, _offset: u64, _buf: &mut [u8]) -> Result<usize> {
            Ok(0)
        }
        fn write(&self, _node: NodeId, _offset: u64, _data: &[u8]) -> Result<usize> {
            Err(MountError::ReadOnly)
        }
        fn enumerate(&self, _dir: NodeId) -> Result<Vec<Entry>> {
            Ok(vec![])
        }
        fn attrs(&self, _node: NodeId) -> Result<Attrs> {
            Ok(Attrs {
                node: NodeId::ROOT,
                kind: NodeKind::Directory,
                size: 0,
                unix_mode: 0o040755,
                nlink: 1,
                mtime: SystemTime::UNIX_EPOCH,
            })
        }
        fn invalidate(&self, _node: NodeId) -> Result<()> {
            Ok(())
        }
    }

    fn is_readonly<T>(r: Result<T>) -> bool {
        matches!(r, Err(MountError::ReadOnly))
    }

    #[test]
    fn defaults_uniformly_surface_readonly() {
        let s = ReadOnlyStub;
        assert!(is_readonly(s.create_file(
            NodeId::ROOT,
            OsStr::new("x"),
            FileMode::Normal,
            false,
        )));
        assert!(is_readonly(s.make_dir(NodeId::ROOT, OsStr::new("d"))));
        assert!(is_readonly(s.unlink_entry(NodeId::ROOT, OsStr::new("x"))));
        assert!(is_readonly(s.rmdir_entry(NodeId::ROOT, OsStr::new("d"))));
        assert!(is_readonly(s.rename_entry(
            NodeId::ROOT,
            OsStr::new("a"),
            NodeId::ROOT,
            OsStr::new("b"),
        )));
        assert!(is_readonly(
            s.set_attrs(NodeId::ROOT, AttrUpdate::default())
        ));
        assert!(is_readonly(s.create_symlink(
            NodeId::ROOT,
            OsStr::new("ln"),
            Path::new("target"),
        )));
        let read_link_result: Result<OsString> = s.read_link(NodeId::ROOT);
        assert!(is_readonly(read_link_result));
    }

    /// The default `release` body delegates to `flush`. A shell that
    /// overrides only `flush` should see `release` follow through.
    #[test]
    fn release_default_delegates_to_flush() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        struct CountFlush(AtomicUsize);
        impl PlatformShell for CountFlush {
            fn lookup(&self, _p: NodeId, _n: &OsStr) -> Result<Option<Entry>> {
                Ok(None)
            }
            fn read(&self, _n: NodeId, _o: u64, _b: &mut [u8]) -> Result<usize> {
                Ok(0)
            }
            fn write(&self, _n: NodeId, _o: u64, _b: &[u8]) -> Result<usize> {
                Ok(0)
            }
            fn enumerate(&self, _d: NodeId) -> Result<Vec<Entry>> {
                Ok(vec![])
            }
            fn attrs(&self, _n: NodeId) -> Result<Attrs> {
                Ok(Attrs {
                    node: NodeId::ROOT,
                    kind: NodeKind::Directory,
                    size: 0,
                    unix_mode: 0o040755,
                    nlink: 1,
                    mtime: SystemTime::UNIX_EPOCH,
                })
            }
            fn invalidate(&self, _n: NodeId) -> Result<()> {
                Ok(())
            }
            fn flush(&self, _n: NodeId) -> Result<()> {
                self.0.fetch_add(1, Ordering::SeqCst);
                Ok(())
            }
        }
        let s = CountFlush(AtomicUsize::new(0));
        s.release(NodeId::ROOT).unwrap();
        assert_eq!(s.0.load(Ordering::SeqCst), 1);
    }
}