use std::ffi::CStr;
use std::mem::MaybeUninit;
use std::os::fd::{AsRawFd, OwnedFd};
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, TryLockError};

use crate::engine::{BackendEntry, EntryKind, FilesystemBackend};
use crate::interfaces::{FileRead, FileWrite, SymlinkInspect};
use crate::{AccessMode, ResolvedEntry, ResourceLimits};
use rustix::fs::{
    CWD, FileType, FlockOperation, Mode as FsMode, OFlags, RawDir, flock, fstat, ftruncate, openat,
    readlinkat,
};
use rustix::io::{pread, pwrite};
use wip_host::{
    EffectGuarantee, ExecutionContext, HostError, OperationAttempt, ProviderResult, Validator,
    WipObject,
};
use wip_interface as wip;

fn entry_kind(mode: u32) -> EntryKind {
    match FileType::from_raw_mode(mode) {
        FileType::Directory => EntryKind::Directory,
        FileType::RegularFile => EntryKind::Regular,
        FileType::Symlink => EntryKind::Symlink,
        _ => EntryKind::Unsupported,
    }
}

fn entry_description(kind: EntryKind) -> &'static str {
    match kind {
        EntryKind::Directory => "Filesystem directory",
        EntryKind::Regular => "Regular filesystem file",
        EntryKind::Symlink => "Symbolic link (not followed)",
        EntryKind::Unsupported => "Unsupported filesystem entry",
    }
}

#[derive(Debug)]
struct HandleCounter {
    current: AtomicUsize,
    maximum: usize,
}

impl HandleCounter {
    fn acquire(self: &Arc<Self>) -> Result<HandlePermit, ProjectionError> {
        let mut current = self.current.load(Ordering::Acquire);
        loop {
            if current >= self.maximum {
                return Err(ProjectionError::Resource);
            }
            match self.current.compare_exchange_weak(
                current,
                current + 1,
                Ordering::AcqRel,
                Ordering::Acquire,
            ) {
                Ok(_) => {
                    return Ok(HandlePermit {
                        counter: Arc::clone(self),
                    });
                }
                Err(actual) => current = actual,
            }
        }
    }
}

#[derive(Debug)]
struct HandlePermit {
    counter: Arc<HandleCounter>,
}

impl Drop for HandlePermit {
    fn drop(&mut self) {
        self.counter.current.fetch_sub(1, Ordering::AcqRel);
    }
}

#[derive(Debug)]
struct EntryInner {
    fd: OwnedFd,
    execution: Mutex<()>,
    _permit: HandlePermit,
}

/// A pinned Linux filesystem entry with bounded request-time capabilities.
///
/// Cloning retains the same open descriptor and handle permit. Use capabilities
/// within [`FilesystemBackend::execute`] for serialized validator checks and
/// dispatch; direct calls still enforce entry kind, access, and request limits.
#[derive(Clone)]
pub struct LinuxEntry {
    name: String,
    kind: EntryKind,
    validator: Option<Validator>,
    initial_validator: Option<Vec<u8>>,
    inner: Arc<EntryInner>,
    context: ExecutionContext,
    access: AccessMode,
    limits: ResourceLimits,
    injection: FailureInjection,
}

impl LinuxEntry {
    /// Returns the access mode used to open this entry.
    #[must_use]
    pub fn access_mode(&self) -> AccessMode {
        self.access
    }

    /// Returns whether this entry supports regular-file reads.
    #[must_use]
    pub fn supports_file_read(&self) -> bool {
        self.kind == EntryKind::Regular
    }

    /// Returns whether this entry supports regular-file replacement.
    #[must_use]
    pub fn supports_file_write(&self) -> bool {
        self.kind == EntryKind::Regular && self.access == AccessMode::ReadWrite
    }

    /// Returns whether this entry supports inspecting a symbolic link itself.
    #[must_use]
    pub fn supports_symlink_inspect(&self) -> bool {
        self.kind == EntryKind::Symlink
    }

    fn refresh_validator(&mut self) -> Result<(), ProjectionError> {
        let stat = fstat(self.fd()).map_err(|_| ProjectionError::Internal)?;
        self.kind = entry_kind(stat.st_mode);
        let validator = matches!(self.kind, EntryKind::Regular | EntryKind::Symlink)
            .then(|| validator_from_stat(&stat));
        self.initial_validator = validator.clone();
        self.validator = validator.map(Validator::new);
        Ok(())
    }

    fn fd(&self) -> &OwnedFd {
        &self.inner.fd
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProjectionError {
    NotFound,
    Resource,
    Internal,
}

impl ProjectionError {
    fn host<T>(self) -> ProviderResult<T> {
        match self {
            Self::NotFound => ProviderResult::NotFound,
            error => ProviderResult::Failure(error.host_error()),
        }
    }

    fn host_error(self) -> HostError {
        match self {
            Self::Resource => HostError::resource_limit(
                "filesystem projection resource limit exceeded",
                EffectGuarantee::NotCommitted,
            ),
            Self::NotFound | Self::Internal => HostError::internal(
                "filesystem projection failed",
                EffectGuarantee::NotCommitted,
            ),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FailureInjection {
    None,
    BeforeWrite,
    AfterTruncate,
}

/// Linux filesystem backend using pinned, no-follow handles.
///
/// Execution holds advisory exclusive `flock` guards on directories in
/// root-to-parent order and on directory or regular-file targets across lineage
/// validation, validator refresh, dispatch, and post-operation observation.
/// Serialization only covers actors that participate in this locking protocol;
/// external writers can still mutate entries without taking the locks. Symlinks
/// are inspected, not followed.
pub struct LinuxBackend {
    root: OwnedFd,
    access: AccessMode,
    limits: ResourceLimits,
    handles: Arc<HandleCounter>,
    injection: Mutex<FailureInjection>,
}

impl LinuxBackend {
    /// Opens and pins a root directory without following symlink components.
    pub fn open(
        root: &Path,
        access: AccessMode,
        limits: ResourceLimits,
    ) -> Result<Self, std::io::Error> {
        Ok(Self {
            root: open_root(root)?,
            access,
            limits,
            handles: Arc::new(HandleCounter {
                current: AtomicUsize::new(0),
                maximum: limits.max_open_handles,
            }),
            injection: Mutex::new(FailureInjection::None),
        })
    }

    /// Returns the current filesystem path of the pinned root directory.
    pub fn effective_root(&self) -> Result<PathBuf, std::io::Error> {
        std::fs::read_link(format!("/proc/self/fd/{}", self.root.as_raw_fd()))
    }

    fn resolve_entry(
        &self,
        context: &ExecutionContext,
        path: &str,
    ) -> Result<ResolvedEntry<LinuxEntry>, ProjectionError> {
        check_context(context)?;
        wip_protocol::validate_path(path).map_err(|_| ProjectionError::NotFound)?;
        let root = self.entry(
            context,
            String::new(),
            reopen(&self.root, OFlags::RDONLY | OFlags::DIRECTORY)?,
        )?;
        if path == "/" {
            return Ok(ResolvedEntry::new(Vec::new(), root));
        }
        // Retain the exact directories traversed, rather than resolving prefixes again.
        let mut ancestors = vec![root];
        let mut segments = path[1..].split('/').peekable();
        while let Some(segment) = segments.next() {
            check_context(context)?;
            if segment.as_bytes().contains(&0) {
                return Err(ProjectionError::NotFound);
            }
            let parent = ancestors.last().ok_or(ProjectionError::Internal)?;
            let probe = openat(
                parent.fd(),
                segment,
                OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC,
                FsMode::empty(),
            )
            .map_err(classify_lookup)?;
            let stat = fstat(&probe).map_err(|_| ProjectionError::Internal)?;
            let kind = entry_kind(stat.st_mode);
            if segments.peek().is_some() && kind != EntryKind::Directory {
                return Err(ProjectionError::NotFound);
            }
            let fd = self.reopen_final(probe, kind)?;
            let entry = self.entry(context, segment.to_owned(), fd)?;
            if segments.peek().is_some() {
                ancestors.push(entry);
            } else {
                return Ok(ResolvedEntry::new(ancestors, entry));
            }
        }
        Err(ProjectionError::NotFound)
    }

    fn open_child(
        &self,
        context: &ExecutionContext,
        parent: &OwnedFd,
        name: &str,
    ) -> Result<LinuxEntry, ProjectionError> {
        check_context(context)?;
        let probe = openat(
            parent,
            name,
            OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC,
            FsMode::empty(),
        )
        .map_err(classify_lookup)?;
        let stat = fstat(&probe).map_err(|_| ProjectionError::Internal)?;
        let kind = entry_kind(stat.st_mode);
        let fd = self.reopen_final(probe, kind)?;
        self.entry(context, name.to_owned(), fd)
    }

    fn reopen_final(&self, probe: OwnedFd, kind: EntryKind) -> Result<OwnedFd, ProjectionError> {
        match kind {
            EntryKind::Directory => reopen(&probe, OFlags::RDONLY | OFlags::DIRECTORY),
            EntryKind::Regular => {
                let access = if self.access == AccessMode::ReadWrite {
                    OFlags::RDWR
                } else {
                    OFlags::RDONLY
                };
                reopen(&probe, access)
            }
            EntryKind::Symlink | EntryKind::Unsupported => Ok(probe),
        }
    }

    fn entry(
        &self,
        context: &ExecutionContext,
        name: String,
        fd: OwnedFd,
    ) -> Result<LinuxEntry, ProjectionError> {
        check_context(context)?;
        let permit = self.handles.acquire()?;
        let stat = fstat(&fd).map_err(|_| ProjectionError::Internal)?;
        let kind = entry_kind(stat.st_mode);
        let validator = matches!(kind, EntryKind::Regular | EntryKind::Symlink)
            .then(|| validator_from_stat(&stat));
        Ok(LinuxEntry {
            name,
            kind,
            initial_validator: validator.clone(),
            validator: validator.map(Validator::new),
            context: context.clone(),
            access: self.access,
            limits: self.limits,
            injection: *self
                .injection
                .lock()
                .unwrap_or_else(|error| error.into_inner()),
            inner: Arc::new(EntryInner {
                fd,
                execution: Mutex::new(()),
                _permit: permit,
            }),
        })
    }

    fn validate_ancestry(
        &self,
        context: &ExecutionContext,
        ancestors: &[LinuxEntry],
        target: &LinuxEntry,
    ) -> Result<(), ProjectionError> {
        check_context(context)?;
        let root = ancestors.first().unwrap_or(target);
        let configured = fstat(&self.root).map_err(|_| ProjectionError::Internal)?;
        let retained = fstat(root.fd()).map_err(|_| ProjectionError::Internal)?;
        // The configured root is a capability, not its possibly renamed textual path.
        if entry_kind(retained.st_mode) != EntryKind::Directory
            || retained.st_dev != configured.st_dev
            || retained.st_ino != configured.st_ino
        {
            return Err(ProjectionError::NotFound);
        }
        for pair in ancestors.windows(2) {
            check_context(context)?;
            let parent = &pair[0];
            let child = &pair[1];
            let visible = openat(
                parent.fd(),
                child.name.as_str(),
                OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC,
                FsMode::empty(),
            )
            .map_err(classify_lookup)?;
            let visible = fstat(&visible).map_err(|_| ProjectionError::Internal)?;
            let retained = fstat(child.fd()).map_err(|_| ProjectionError::Internal)?;
            if entry_kind(visible.st_mode) != EntryKind::Directory
                || entry_kind(retained.st_mode) != EntryKind::Directory
                || visible.st_dev != retained.st_dev
                || visible.st_ino != retained.st_ino
            {
                return Err(ProjectionError::NotFound);
            }
        }
        // Deliberately do not look up the target name: leaf replacement must not
        // redirect execution away from its original pinned inode.
        check_context(context)
    }

    #[cfg(test)]
    fn set_injection(&self, injection: FailureInjection) {
        *self
            .injection
            .lock()
            .unwrap_or_else(|error| error.into_inner()) = injection;
    }
}

impl BackendEntry for LinuxEntry {
    fn kind(&self) -> EntryKind {
        self.kind
    }
}

impl FilesystemBackend for LinuxBackend {
    type Entry = LinuxEntry;

    fn resolve(
        &self,
        context: &ExecutionContext,
        path: &str,
    ) -> ProviderResult<ResolvedEntry<LinuxEntry>> {
        match self.resolve_entry(context, path) {
            Ok(entry) => ProviderResult::Found(entry),
            Err(error) => error.host(),
        }
    }

    fn children(
        &self,
        context: &ExecutionContext,
        entry: &LinuxEntry,
        maximum: usize,
    ) -> Result<Vec<LinuxEntry>, HostError> {
        check_context(context).map_err(resource_host)?;
        if entry.kind != EntryKind::Directory {
            return Ok(Vec::new());
        }
        let directory_fd = reopen(entry.fd(), OFlags::RDONLY | OFlags::DIRECTORY)
            .map_err(ProjectionError::host_error)?;
        let mut buffer = [MaybeUninit::<u8>::uninit(); 8192];
        let mut directory = RawDir::new(directory_fd, &mut buffer);
        let mut children = Vec::new();
        while let Some(item) = directory.next() {
            check_context(context).map_err(resource_host)?;
            let item = item.map_err(|_| internal_not_committed())?;
            let name = item.file_name();
            if name == cstr_dot() || name == cstr_dot_dot() {
                continue;
            }
            let Ok(name) = name.to_str() else {
                // WIP names are Unicode. Non-Unicode backend names are not published.
                continue;
            };
            if name.contains('/') || name.is_empty() || name == "." || name == ".." {
                continue;
            }
            let child = match self.open_child(context, entry.fd(), name) {
                Ok(child) => child,
                Err(ProjectionError::NotFound) => continue,
                Err(error) => return Err(error.host_error()),
            };
            if children.len() >= maximum {
                return Err(resource_not_committed());
            }
            children.push(child);
        }
        Ok(children)
    }

    fn execute(
        &self,
        context: &ExecutionContext,
        resolved: ResolvedEntry<LinuxEntry>,
        callback: impl FnOnce(ResolvedEntry<LinuxEntry>) -> OperationAttempt,
    ) -> ProviderResult<OperationAttempt> {
        if let Err(error) = check_context(context) {
            return error.host();
        }
        let (mut ancestors, mut entry) = resolved.into_parts();
        // Independent resolutions have independent open-file descriptions; clones
        // also need the per-entry mutex because flock alone treats them as one owner.
        // Always acquire directories root-to-parent, then the target (if lockable).
        let mut lock_entries: Vec<_> = ancestors
            .iter()
            .map(|ancestor| Arc::clone(&ancestor.inner))
            .collect();
        if matches!(entry.kind, EntryKind::Directory | EntryKind::Regular) {
            lock_entries.push(Arc::clone(&entry.inner));
        }
        let started = std::time::Instant::now();
        let mut locks = Vec::with_capacity(lock_entries.len());
        for inner in &lock_entries {
            let remaining = self
                .limits
                .request_timeout
                .saturating_sub(started.elapsed());
            match lock_with_context(inner, context, remaining) {
                Ok(lock) => locks.push(lock),
                Err(ProjectionError::Resource) => {
                    return ProviderResult::Failure(HostError::resource_limit(
                        "filesystem lock deadline exceeded",
                        EffectGuarantee::NotCommitted,
                    ));
                }
                Err(ProjectionError::Internal | ProjectionError::NotFound) => {
                    return ProviderResult::Failure(HostError::internal(
                        "filesystem object could not be locked",
                        EffectGuarantee::NotCommitted,
                    ));
                }
            }
        }
        if let Err(error) = self.validate_ancestry(context, &ancestors, &entry) {
            return error.host();
        }
        for ancestor in &mut ancestors {
            if ancestor.refresh_validator().is_err() {
                return ProviderResult::Failure(internal_not_committed());
            }
            ancestor.context = context.clone();
        }
        // Both snapshots must describe the state under the locks, not at resolve time.
        if entry.refresh_validator().is_err() {
            return ProviderResult::Failure(internal_not_committed());
        }
        entry.context = context.clone();
        let mut attempt = callback(ResolvedEntry::new(ancestors, entry.clone()));
        if let OperationAttempt::Started(Ok(result)) = &mut attempt {
            if entry.refresh_validator().is_err() {
                return ProviderResult::Failure(HostError::internal(
                    "filesystem state could not be observed after dispatch",
                    EffectGuarantee::Unknown,
                ));
            }
            result.validator = entry.initial_validator.clone();
        }
        ProviderResult::Found(attempt)
    }
}

// Unlock before releasing clone serialization, including on refresh/dispatch failure.
struct EntryLock<'a> {
    inner: &'a EntryInner,
    _guard: MutexGuard<'a, ()>,
}

impl Drop for EntryLock<'_> {
    fn drop(&mut self) {
        let _ = flock(&self.inner.fd, FlockOperation::Unlock);
    }
}

impl WipObject for LinuxEntry {
    fn name(&self) -> &str {
        &self.name
    }

    fn description(&self) -> Option<&str> {
        Some(entry_description(self.kind))
    }

    fn validator(&self) -> Option<&Validator> {
        self.validator.as_ref()
    }
}

impl FileRead for LinuxEntry {
    fn read(&self, offset: Option<u64>, length: Option<u64>) -> wip::OperationResult<wip::Bytes> {
        check_context(&self.context).map_err(resource_host)?;
        if !self.supports_file_read() {
            return Err(internal_not_committed());
        }
        let offset = offset.unwrap_or(0);
        let stat = fstat(self.fd()).map_err(|_| internal_not_committed())?;
        let size = u64::try_from(stat.st_size).map_err(|_| internal_not_committed())?;
        let available = size.saturating_sub(offset);
        let length = match length {
            Some(length) if length <= self.limits.max_file_read_bytes as u64 => {
                length.min(available)
            }
            Some(_) => return Err(resource_not_committed()),
            None if available <= self.limits.max_file_read_bytes as u64 => available,
            None => return Err(resource_not_committed()),
        };
        let length = usize::try_from(length).map_err(|_| resource_not_committed())?;
        let mut bytes = vec![0_u8; length];
        let mut read = 0_usize;
        while read < length {
            check_context(&self.context).map_err(resource_host)?;
            let count = pread(self.fd(), &mut bytes[read..], offset + read as u64)
                .map_err(|_| internal_not_committed())?;
            if count == 0 {
                break;
            }
            read += count;
        }
        bytes.truncate(read);
        Ok(wip::Bytes(bytes))
    }
}

impl SymlinkInspect for LinuxEntry {
    fn read_link(&self) -> wip::OperationResult<wip::Bytes> {
        check_context(&self.context).map_err(resource_host)?;
        if !self.supports_symlink_inspect() {
            return Err(internal_not_committed());
        }
        let target = readlinkat(self.fd(), "", Vec::new()).map_err(|_| internal_not_committed())?;
        Ok(wip::Bytes(target.as_bytes().to_vec()))
    }
}

impl FileWrite for LinuxEntry {
    fn replace(&self, contents: wip::Bytes) -> wip::OperationResult<()> {
        check_context(&self.context).map_err(resource_host)?;
        if !self.supports_file_write() {
            return Err(internal_not_committed());
        }
        if contents.0.len() > self.limits.max_file_write_bytes {
            return Err(resource_not_committed());
        }
        let current = fstat(self.fd()).map_err(|_| internal_not_committed())?;
        if self.initial_validator.as_deref() != Some(validator_from_stat(&current).as_slice()) {
            return Err(HostError::internal(
                "filesystem object changed before write",
                EffectGuarantee::NotCommitted,
            ));
        }
        if self.injection == FailureInjection::BeforeWrite {
            return Err(internal_not_committed());
        }
        ftruncate(self.fd(), 0).map_err(|_| internal_not_committed())?;
        if self.injection == FailureInjection::AfterTruncate {
            return Err(HostError::internal(
                "injected failure after write began",
                EffectGuarantee::Unknown,
            ));
        }
        let mut written = 0_usize;
        while written < contents.0.len() {
            if context_expired(&self.context) {
                return Err(HostError::resource_limit(
                    "write interrupted after execution began",
                    EffectGuarantee::Unknown,
                ));
            }
            let count =
                pwrite(self.fd(), &contents.0[written..], written as u64).map_err(|_| {
                    HostError::internal("filesystem write failed", EffectGuarantee::Unknown)
                })?;
            if count == 0 {
                return Err(HostError::internal(
                    "filesystem write made no progress",
                    EffectGuarantee::Unknown,
                ));
            }
            written += count;
        }
        rustix::fs::fdatasync(self.fd()).map_err(|_| {
            HostError::internal(
                "filesystem synchronization failed",
                EffectGuarantee::Unknown,
            )
        })?;
        Ok(())
    }
}

fn open_root(path: &Path) -> Result<OwnedFd, std::io::Error> {
    let start = if path.is_absolute() { "/" } else { "." };
    let mut current = openat(
        CWD,
        start,
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
        FsMode::empty(),
    )
    .map_err(std::io::Error::from)?;
    for component in path.components() {
        let name = match component {
            Component::RootDir | Component::CurDir => continue,
            Component::Normal(name) => name,
            Component::ParentDir | Component::Prefix(_) => {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidInput,
                    "filesystem root must not contain parent or platform-prefix components",
                ));
            }
        };
        let probe = openat(
            &current,
            name,
            OFlags::PATH | OFlags::NOFOLLOW | OFlags::CLOEXEC,
            FsMode::empty(),
        )
        .map_err(std::io::Error::from)?;
        let stat = fstat(&probe).map_err(std::io::Error::from)?;
        if entry_kind(stat.st_mode) != EntryKind::Directory {
            return Err(std::io::Error::new(
                std::io::ErrorKind::NotADirectory,
                "filesystem root contains a symlink or non-directory component",
            ));
        }
        current = reopen(&probe, OFlags::RDONLY | OFlags::DIRECTORY)
            .map_err(|_| std::io::Error::other("filesystem root could not be pinned"))?;
    }
    Ok(current)
}

fn lock_with_context<'a>(
    inner: &'a EntryInner,
    context: &ExecutionContext,
    maximum_wait: std::time::Duration,
) -> Result<EntryLock<'a>, ProjectionError> {
    let now = std::time::Instant::now();
    let fallback_deadline = now.checked_add(maximum_wait).unwrap_or(now);
    let deadline = context.deadline().map_or(fallback_deadline, |configured| {
        configured.min(fallback_deadline)
    });
    loop {
        check_context(context)?;
        if std::time::Instant::now() >= deadline {
            return Err(ProjectionError::Resource);
        }
        // flock treats clones' shared open-file description as a single owner.
        let guard = match inner.execution.try_lock() {
            Ok(guard) => guard,
            Err(TryLockError::Poisoned(error)) => error.into_inner(),
            Err(TryLockError::WouldBlock) => {
                std::thread::sleep(std::time::Duration::from_millis(1));
                continue;
            }
        };
        match flock(&inner.fd, FlockOperation::NonBlockingLockExclusive) {
            Ok(()) => {
                return Ok(EntryLock {
                    inner,
                    _guard: guard,
                });
            }
            Err(rustix::io::Errno::WOULDBLOCK) => {
                drop(guard);
                std::thread::sleep(std::time::Duration::from_millis(1));
            }
            Err(_) => return Err(ProjectionError::Internal),
        }
    }
}

fn reopen(fd: &OwnedFd, flags: OFlags) -> Result<OwnedFd, ProjectionError> {
    let path = format!("/proc/self/fd/{}", fd.as_raw_fd());
    openat(CWD, path, flags | OFlags::CLOEXEC, FsMode::empty()).map_err(classify_lookup)
}

fn validator_from_stat(stat: &rustix::fs::Stat) -> Vec<u8> {
    let mut bytes = Vec::with_capacity(56);
    bytes.extend_from_slice(&stat.st_dev.to_be_bytes());
    bytes.extend_from_slice(&stat.st_ino.to_be_bytes());
    bytes.extend_from_slice(&u64::from(stat.st_mode).to_be_bytes());
    bytes.extend_from_slice(&stat.st_size.to_be_bytes());
    bytes.extend_from_slice(&stat.st_mtime.to_be_bytes());
    bytes.extend_from_slice(&stat.st_mtime_nsec.to_be_bytes());
    bytes.extend_from_slice(&stat.st_ctime.to_be_bytes());
    bytes.extend_from_slice(&stat.st_ctime_nsec.to_be_bytes());
    bytes
}

fn classify_lookup(error: rustix::io::Errno) -> ProjectionError {
    use rustix::io::Errno;
    if matches!(
        error,
        Errno::NOENT
            | Errno::NOTDIR
            | Errno::LOOP
            | Errno::ACCESS
            | Errno::PERM
            | Errno::INVAL
            | Errno::NAMETOOLONG
    ) {
        ProjectionError::NotFound
    } else {
        ProjectionError::Internal
    }
}

fn context_expired(context: &ExecutionContext) -> bool {
    context.cancellation().is_cancelled()
        || context
            .deadline()
            .is_some_and(|deadline| std::time::Instant::now() >= deadline)
}

fn check_context(context: &ExecutionContext) -> Result<(), ProjectionError> {
    if context_expired(context) || context.resource_budget() == Some(0) {
        Err(ProjectionError::Resource)
    } else {
        Ok(())
    }
}

fn resource_host(_: ProjectionError) -> HostError {
    resource_not_committed()
}

fn resource_not_committed() -> HostError {
    HostError::resource_limit(
        "filesystem operation resource limit exceeded",
        EffectGuarantee::NotCommitted,
    )
}

fn internal_not_committed() -> HostError {
    HostError::internal("filesystem operation failed", EffectGuarantee::NotCommitted)
}

fn cstr_dot() -> &'static CStr {
    c"."
}

fn cstr_dot_dot() -> &'static CStr {
    c".."
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;
    use std::fs;

    use tempfile::tempdir;
    use wip_host::{
        CallErrorKind, CallRequest, ExecutionContext, InterfaceTarget, ObjectProvider,
        OperationAttempt, ProviderResult, Target, WipObject,
    };
    use wip_protocol::{ObserveRequest, ProtocolErrorCode, Value};

    use super::{FailureInjection, FlockOperation, LinuxBackend, LinuxEntry};
    use crate::engine::{DefaultPolicy, FilesystemBackend, Projection};
    use crate::interfaces::{FileRead, FileWrite};
    use crate::{AccessMode, DefaultComposition, ResolvedEntry, ResourceLimits};

    type DefaultProjection = Projection<LinuxBackend, DefaultComposition, DefaultPolicy>;

    fn projection(root: &std::path::Path, access: AccessMode) -> DefaultProjection {
        crate::default_projection(root, access, ResourceLimits::default()).expect("open projection")
    }

    fn call(
        projection: &DefaultProjection,
        path: &str,
        validator: Option<Vec<u8>>,
        interface: wip_protocol::InterfaceReference,
        operation: &str,
        arguments: BTreeMap<String, Value>,
    ) -> Result<wip_protocol::CallOperationResponse, wip_host::CallError> {
        projection.call_operation(CallRequest {
            target: Target {
                path: path.into(),
                validator,
            },
            interface: InterfaceTarget {
                scope_ref: None,
                reference: interface,
                validator: None,
            },
            operation: operation.into(),
            arguments,
        })
    }

    fn resolved(
        backend: &LinuxBackend,
        context: &ExecutionContext,
        path: &str,
    ) -> ResolvedEntry<LinuxEntry> {
        match backend.resolve(context, path) {
            ProviderResult::Found(entry) => entry,
            ProviderResult::NotFound => panic!("entry should resolve"),
            ProviderResult::Failure(error) => panic!("resolution failed: {error:?}"),
        }
    }

    #[test]
    fn execution_refreshes_both_validators_under_lock_and_returns_post_write_validator() {
        let temporary = tempdir().expect("temporary root");
        let path = temporary.path().join("file");
        fs::write(&path, b"old").expect("write file");
        let backend = LinuxBackend::open(
            temporary.path(),
            AccessMode::ReadWrite,
            ResourceLimits::default(),
        )
        .expect("open backend");
        let context = ExecutionContext::new();
        let entry = resolved(&backend, &context, "/file");
        let original = entry.target().initial_validator.clone();
        let pinned = entry.target().clone();
        // Simulate a cooperating writer committing between resolution and execution.
        fs::write(&path, b"changed before lock acquisition").expect("external mutation");
        let lock_probe = fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(&path)
            .expect("open independent lock probe");
        let mut callback_validator = None;
        let attempt = backend.execute(&context, entry, |resolved| {
            let entry = resolved.target();
            assert!(std::sync::Arc::ptr_eq(&entry.inner, &pinned.inner));
            assert_ne!(entry.initial_validator, original);
            assert_eq!(
                entry
                    .validator()
                    .map(|validator| validator.as_bytes().to_vec()),
                entry.initial_validator,
            );
            callback_validator = entry.initial_validator.clone();
            assert_eq!(
                rustix::fs::flock(&lock_probe, FlockOperation::NonBlockingLockExclusive),
                Err(rustix::io::Errno::WOULDBLOCK),
            );
            entry
                .replace(wip_interface::Bytes(b"new".to_vec()))
                .expect("write uses refreshed initial validator");
            OperationAttempt::Started(Ok(wip_host::CallResult {
                result: Value::Bytes(Vec::new()),
                validator: None,
            }))
        });
        let ProviderResult::Found(OperationAttempt::Started(Ok(result))) = attempt else {
            panic!("execution should succeed: {attempt:?}");
        };
        assert_ne!(result.validator, callback_validator);
        assert_eq!(
            result.validator,
            resolved(&backend, &context, "/file")
                .target()
                .initial_validator,
        );
        assert_eq!(fs::read(&path).expect("read written file"), b"new");
        // The retained clone must not keep the operation's flock held.
        rustix::fs::flock(&lock_probe, FlockOperation::NonBlockingLockExclusive)
            .expect("execution released lock");
        rustix::fs::flock(&lock_probe, FlockOperation::Unlock).expect("unlock probe");
    }

    #[test]
    fn execution_uses_the_pinned_entry_after_path_replacement() {
        let temporary = tempdir().expect("temporary root");
        let path = temporary.path().join("file");
        fs::write(&path, b"old").expect("write file");
        let backend = LinuxBackend::open(
            temporary.path(),
            AccessMode::ReadWrite,
            ResourceLimits::default(),
        )
        .expect("open backend");
        let context = ExecutionContext::new();
        let entry = resolved(&backend, &context, "/file");
        let pinned = entry.target().clone();
        fs::remove_file(&path).expect("unlink pinned entry");
        fs::write(&path, b"replacement").expect("replace path");
        let attempt = backend.execute(&context, entry, |resolved| {
            let entry = resolved.target();
            assert!(std::sync::Arc::ptr_eq(&entry.inner, &pinned.inner));
            entry
                .replace(wip_interface::Bytes(b"pinned write".to_vec()))
                .expect("write pinned inode");
            OperationAttempt::Started(Ok(wip_host::CallResult {
                result: Value::Bytes(Vec::new()),
                validator: None,
            }))
        });
        assert!(matches!(
            attempt,
            ProviderResult::Found(OperationAttempt::Started(Ok(_)))
        ));
        assert_eq!(fs::read(path).expect("read replacement"), b"replacement");
        assert_eq!(
            pinned.read(None, None).expect("read pinned inode").0,
            b"pinned write"
        );
    }

    #[test]
    fn detached_or_replaced_ancestors_reject_before_callback_without_effects() {
        // Exercise both the immediate parent and a higher ancestor, with a
        // visible directory replacement, no replacement, and a symlink replacement.
        for ancestor in ["a", "a/b"] {
            for replacement in ["directory", "missing", "symlink"] {
                let temporary = tempdir().expect("temporary root");
                fs::create_dir_all(temporary.path().join("a/b")).expect("create lineage");
                fs::write(temporary.path().join("a/b/file"), b"original").expect("write original");
                let backend = LinuxBackend::open(
                    temporary.path(),
                    AccessMode::ReadWrite,
                    ResourceLimits::default(),
                )
                .expect("open backend");
                let context = ExecutionContext::new();
                let entry = resolved(&backend, &context, "/a/b/file");
                let pinned = entry.target().clone();
                let detached = temporary.path().join("detached");
                let ancestor_path = temporary.path().join(ancestor);
                fs::rename(&ancestor_path, &detached).expect("detach ancestor");
                if replacement == "directory" {
                    fs::create_dir(&ancestor_path).expect("replace directory");
                } else if replacement == "symlink" {
                    std::os::unix::fs::symlink(&detached, &ancestor_path)
                        .expect("replace ancestor with symlink to original");
                }
                let visible_leaf =
                    ancestor_path.join(if ancestor == "a" { "b/file" } else { "file" });
                if replacement == "directory" {
                    fs::create_dir_all(visible_leaf.parent().expect("leaf parent"))
                        .expect("create replacement lineage");
                    fs::write(&visible_leaf, b"replacement").expect("write replacement");
                }
                let mut called = false;
                let attempt = backend.execute(&context, entry, |entry| {
                    called = true;
                    entry
                        .target()
                        .replace(wip_interface::Bytes(b"bad".to_vec()))
                        .expect("would write pinned inode");
                    OperationAttempt::reject(CallErrorKind::ValidatorMismatch)
                });
                assert!(
                    matches!(attempt, ProviderResult::NotFound),
                    "{ancestor}: {replacement}"
                );
                assert!(!called);
                assert_eq!(
                    pinned.read(None, None).expect("read original inode").0,
                    b"original"
                );
                if replacement != "missing" {
                    assert_eq!(
                        fs::read(&visible_leaf).expect("read visible leaf"),
                        if replacement == "directory" {
                            b"replacement".as_slice()
                        } else {
                            b"original".as_slice()
                        }
                    );
                }
                // Rejection releases every guard, even while the target is retained.
                let root_probe = fs::File::open(temporary.path()).expect("open root probe");
                rustix::fs::flock(&root_probe, FlockOperation::NonBlockingLockExclusive)
                    .expect("rejection released root guard");
            }
        }
    }

    #[test]
    fn execution_retains_and_refreshes_ancestors_under_ordered_directory_guards() {
        let temporary = tempdir().expect("temporary root");
        fs::create_dir_all(temporary.path().join("a/b")).expect("create lineage");
        fs::write(temporary.path().join("a/b/file"), b"original").expect("write file");
        let backend = LinuxBackend::open(
            temporary.path(),
            AccessMode::ReadOnly,
            ResourceLimits::default(),
        )
        .expect("open backend");
        let context = ExecutionContext::new();
        let entry = resolved(&backend, &context, "/a/b/file");
        assert_eq!(
            entry
                .ancestors()
                .iter()
                .map(WipObject::name)
                .collect::<Vec<_>>(),
            ["", "a", "b"]
        );
        let pinned_ancestors = entry.ancestors().to_vec();
        let (mut ancestors, target) = entry.into_parts();
        // Stale snapshots must not escape the execution boundary.
        for ancestor in &mut ancestors {
            ancestor.validator = Some(wip_host::Validator::new(vec![1]));
            ancestor.initial_validator = Some(vec![1]);
        }
        let probes: Vec<_> = ["", "a", "a/b", "a/b/file"]
            .into_iter()
            .map(|path| fs::File::open(temporary.path().join(path)).expect("open lock probe"))
            .collect();
        let attempt = backend.execute(&context, ResolvedEntry::new(ancestors, target), |entry| {
            for (ancestor, pinned) in entry.ancestors().iter().zip(&pinned_ancestors) {
                assert!(std::sync::Arc::ptr_eq(&ancestor.inner, &pinned.inner));
                assert!(ancestor.validator().is_none());
                assert!(ancestor.initial_validator.is_none());
            }
            for probe in &probes {
                assert_eq!(
                    rustix::fs::flock(probe, FlockOperation::NonBlockingLockExclusive),
                    Err(rustix::io::Errno::WOULDBLOCK)
                );
            }
            OperationAttempt::reject(CallErrorKind::ValidatorMismatch)
        });
        assert!(matches!(attempt, ProviderResult::Found(_)));
        for probe in &probes {
            rustix::fs::flock(probe, FlockOperation::NonBlockingLockExclusive)
                .expect("guard released after callback");
            rustix::fs::flock(probe, FlockOperation::Unlock).expect("unlock probe");
        }
    }

    #[test]
    fn configured_root_stays_pinned_after_textual_rename_and_replacement() {
        let temporary = tempdir().expect("temporary parent");
        let configured = temporary.path().join("root");
        fs::create_dir_all(configured.join("dir")).expect("create original root");
        fs::write(configured.join("dir/file"), b"original").expect("write original file");
        let backend = LinuxBackend::open(
            &configured,
            AccessMode::ReadWrite,
            ResourceLimits::default(),
        )
        .expect("open backend");
        let context = ExecutionContext::new();
        let root = resolved(&backend, &context, "/");
        assert!(root.ancestors().is_empty());
        let file = resolved(&backend, &context, "/dir/file");
        let moved = temporary.path().join("moved");
        fs::rename(&configured, &moved).expect("rename configured root");
        fs::create_dir_all(configured.join("dir")).expect("replace textual root");
        fs::write(configured.join("dir/file"), b"replacement").expect("write replacement file");
        for entry in [
            root,
            resolved(&backend, &context, "/"),
            resolved(&backend, &context, "/dir"),
        ] {
            let attempt = backend.execute(&context, entry, |entry| {
                assert_eq!(entry.target().kind, crate::EntryKind::Directory);
                OperationAttempt::reject(CallErrorKind::ValidatorMismatch)
            });
            assert!(
                matches!(attempt, ProviderResult::Found(_)),
                "root/directory locks once"
            );
        }
        let attempt = backend.execute(&context, file, |entry| {
            entry
                .target()
                .replace(wip_interface::Bytes(b"pinned write".to_vec()))
                .expect("write beneath original root");
            OperationAttempt::Started(Ok(wip_host::CallResult {
                result: Value::Bytes(Vec::new()),
                validator: None,
            }))
        });
        assert!(matches!(
            attempt,
            ProviderResult::Found(OperationAttempt::Started(Ok(_)))
        ));
        assert_eq!(
            fs::read(moved.join("dir/file")).expect("read original root"),
            b"pinned write"
        );
        assert_eq!(
            fs::read(configured.join("dir/file")).expect("read textual replacement"),
            b"replacement"
        );
        assert_eq!(
            resolved(&backend, &context, "/dir/file")
                .target()
                .read(None, None)
                .expect("resolution still uses pinned root")
                .0,
            b"pinned write"
        );
    }

    #[test]
    fn deep_resolution_counts_all_retained_ancestors_against_handle_budget() {
        let temporary = tempdir().expect("temporary root");
        fs::create_dir_all(temporary.path().join("a/b")).expect("create lineage");
        fs::write(temporary.path().join("a/b/file"), b"original").expect("write file");
        let context = ExecutionContext::new();
        for maximum in [3, 4] {
            let backend = LinuxBackend::open(
                temporary.path(),
                AccessMode::ReadOnly,
                ResourceLimits {
                    max_open_handles: maximum,
                    ..ResourceLimits::default()
                },
            )
            .expect("open backend");
            match backend.resolve(&context, "/a/b/file") {
                ProviderResult::Failure(error) if maximum == 3 => {
                    assert_eq!(error.kind(), wip_host::HostErrorKind::ResourceLimitExceeded);
                    assert_eq!(
                        error.effect_guarantee(),
                        wip_host::EffectGuarantee::NotCommitted
                    );
                }
                ProviderResult::Found(entry) if maximum == 4 => {
                    assert_eq!(entry.ancestors().len(), 3);
                    assert_eq!(backend.handles.current.load(super::Ordering::Acquire), 4);
                    let cloned =
                        ResolvedEntry::new(entry.ancestors().to_vec(), entry.target().clone());
                    drop(entry);
                    assert_eq!(backend.handles.current.load(super::Ordering::Acquire), 4);
                    drop(cloned);
                }
                _ => panic!("unexpected resolution outcome for budget {maximum}"),
            }
            assert_eq!(backend.handles.current.load(super::Ordering::Acquire), 0);
            let root = resolved(&backend, &context, "/");
            assert_eq!(backend.handles.current.load(super::Ordering::Acquire), 1);
            drop(root);
            assert_eq!(backend.handles.current.load(super::Ordering::Acquire), 0);
        }
    }

    #[test]
    fn rejected_execution_releases_lock_even_when_callback_retains_entry() {
        let temporary = tempdir().expect("temporary root");
        let path = temporary.path().join("file");
        fs::write(&path, b"old").expect("write file");
        let backend = LinuxBackend::open(
            temporary.path(),
            AccessMode::ReadOnly,
            ResourceLimits::default(),
        )
        .expect("open backend");
        let context = ExecutionContext::new();
        let entry = resolved(&backend, &context, "/file");
        let mut retained = None;
        let attempt = backend.execute(&context, entry, |entry| {
            retained = Some(entry);
            OperationAttempt::reject(CallErrorKind::ValidatorMismatch)
        });
        assert!(matches!(
            attempt,
            ProviderResult::Found(OperationAttempt::NotStarted(_))
        ));
        let lock_probe = fs::File::open(path).expect("open independent lock probe");
        rustix::fs::flock(&lock_probe, FlockOperation::NonBlockingLockExclusive)
            .expect("rejected execution released lock");
        rustix::fs::flock(&lock_probe, FlockOperation::Unlock).expect("unlock probe");
        assert!(retained.is_some());
    }

    #[test]
    fn cloned_entries_cannot_bypass_execution_lock_and_wait_is_bounded() {
        let temporary = tempdir().expect("temporary root");
        fs::write(temporary.path().join("file"), b"old").expect("write file");
        let backend = LinuxBackend::open(
            temporary.path(),
            AccessMode::ReadOnly,
            ResourceLimits {
                request_timeout: std::time::Duration::from_millis(20),
                ..ResourceLimits::default()
            },
        )
        .expect("open backend");
        let context = ExecutionContext::new();
        let entry = resolved(&backend, &context, "/file");
        let cloned = ResolvedEntry::new(entry.ancestors().to_vec(), entry.target().clone());
        let attempt = backend.execute(&context, entry, |entry| {
            let started = std::time::Instant::now();
            let nested = backend.execute(&context, entry, |_| {
                panic!("shared open-file description must not bypass execution lock");
            });
            let ProviderResult::Failure(error) = nested else {
                panic!("nested clone execution must time out: {nested:?}");
            };
            assert_eq!(error.kind(), wip_host::HostErrorKind::ResourceLimitExceeded);
            assert_eq!(
                error.effect_guarantee(),
                wip_host::EffectGuarantee::NotCommitted
            );
            assert!(started.elapsed() < std::time::Duration::from_secs(1));
            OperationAttempt::reject(CallErrorKind::ValidatorMismatch)
        });
        assert!(matches!(attempt, ProviderResult::Found(_)));
        assert!(matches!(
            backend.execute(&context, cloned, |_| {
                OperationAttempt::reject(CallErrorKind::ValidatorMismatch)
            }),
            ProviderResult::Found(_)
        ));
    }

    #[test]
    fn children_budget_fails_completely_and_drops_temporary_handles() {
        let temporary = tempdir().expect("temporary root");
        fs::write(temporary.path().join("a"), b"a").expect("write file");
        fs::write(temporary.path().join("b"), b"b").expect("write file");
        let backend = LinuxBackend::open(
            temporary.path(),
            AccessMode::ReadOnly,
            ResourceLimits {
                max_open_handles: 4,
                ..ResourceLimits::default()
            },
        )
        .expect("open backend");
        let context = ExecutionContext::new();
        let root = resolved(&backend, &context, "/");
        assert!(backend.children(&context, root.target(), 1).is_err());
        assert_eq!(backend.handles.current.load(super::Ordering::Acquire), 1);
        let children = backend
            .children(&context, root.target(), 2)
            .expect("complete children");
        assert_eq!(children.len(), 2);
        let cloned = children[0].clone();
        assert_eq!(backend.handles.current.load(super::Ordering::Acquire), 3);
        drop(children);
        assert_eq!(backend.handles.current.load(super::Ordering::Acquire), 2);
        drop(cloned);
        assert_eq!(backend.handles.current.load(super::Ordering::Acquire), 1);
    }

    #[test]
    fn direct_capabilities_enforce_access_limits_and_execution_context() {
        let temporary = tempdir().expect("temporary root");
        let path = temporary.path().join("file");
        fs::write(&path, b"old").expect("write file");
        let backend = LinuxBackend::open(
            temporary.path(),
            AccessMode::ReadOnly,
            ResourceLimits {
                max_file_read_bytes: 2,
                ..ResourceLimits::default()
            },
        )
        .expect("open backend");
        let context = ExecutionContext::new();
        let entry = resolved(&backend, &context, "/file");
        let target = entry.target();
        assert_eq!(target.access_mode(), AccessMode::ReadOnly);
        assert!(target.supports_file_read());
        assert!(!target.supports_file_write());
        assert!(!target.supports_symlink_inspect());
        assert!(target.read(None, None).is_err());
        assert_eq!(target.read(None, Some(2)).expect("bounded read").0, b"ol");
        assert!(target.replace(wip_interface::Bytes(Vec::new())).is_err());
        assert_eq!(fs::read(&path).expect("read unchanged file"), b"old");
        let cancellation = wip_host::CancellationToken::new();
        cancellation.cancel();
        let attempt = backend.execute(
            &ExecutionContext::new().with_cancellation(cancellation),
            entry,
            |_| panic!("cancelled execution must not invoke callback"),
        );
        assert!(matches!(attempt, ProviderResult::Failure(_)));
    }

    #[test]
    fn projects_entry_kinds_without_following_symlinks() {
        let temporary = tempdir().expect("temporary root");
        fs::write(temporary.path().join("file.txt"), b"hello").expect("write file");
        fs::create_dir(temporary.path().join("dir")).expect("create directory");
        #[cfg(unix)]
        std::os::unix::fs::symlink("/etc/passwd", temporary.path().join("link"))
            .expect("create symlink");
        let projection = projection(temporary.path(), AccessMode::ReadOnly);

        let root = projection.observe_object("/").expect("observe root");
        assert_eq!(root.name, "");
        assert_eq!(root.description.as_deref(), Some("Filesystem directory"));
        assert!(root.interfaces.is_empty());
        assert!(root.validator.is_none());

        let file = projection
            .observe_object("/file.txt")
            .expect("observe file");
        assert_eq!(file.interfaces.len(), 1);
        assert!(file.validator.is_some());
        assert!(file.r#ref.is_none());

        let directory = projection
            .observe_object("/dir")
            .expect("observe directory");
        assert!(directory.interfaces.is_empty());

        #[cfg(unix)]
        {
            let link = projection
                .observe_object("/link")
                .expect("observe link itself");
            assert_eq!(
                link.description.as_deref(),
                Some("Symbolic link (not followed)")
            );
            assert_eq!(link.interfaces.len(), 1);
            assert_eq!(
                projection
                    .observe_object("/link/shadow")
                    .expect_err("must not follow")
                    .kind(),
                CallErrorKind::NotFound
            );
            let target = call(
                &projection,
                "/link",
                link.validator,
                link.interfaces[0].clone(),
                "read_link",
                BTreeMap::new(),
            )
            .expect("inspect symlink");
            assert_eq!(target.result, Value::Bytes(b"/etc/passwd".to_vec()));
        }

        assert_eq!(
            projection
                .observe_object("/missing")
                .expect_err("missing")
                .kind(),
            CallErrorKind::NotFound
        );
    }

    #[test]
    fn reads_are_bounded_and_observation_is_complete() {
        let temporary = tempdir().expect("temporary root");
        fs::write(temporary.path().join("a"), b"abcdef").expect("write file");
        fs::write(temporary.path().join("b"), b"b").expect("write file");
        let projection = projection(temporary.path(), AccessMode::ReadOnly);
        let file = projection.observe_object("/a").expect("observe file");
        let result = call(
            &projection,
            "/a",
            file.validator.clone(),
            file.interfaces[0].clone(),
            "read",
            BTreeMap::from([
                ("offset".into(), Value::Integer(2)),
                ("length".into(), Value::Integer(3)),
            ]),
        )
        .expect("read range");
        assert_eq!(result.result, Value::Bytes(b"cde".to_vec()));

        let complete = projection
            .observe(
                &ExecutionContext::new(),
                &ObserveRequest {
                    path: "/".into(),
                    depth: 1,
                },
            )
            .expect("complete observation");
        let children = complete.children.as_ref().expect("root children observed");
        assert_eq!(children.len(), 2);
        assert!(children.iter().all(|child| child.children.is_none()));
    }

    #[test]
    fn writes_require_current_validator_and_return_new_validator() {
        let temporary = tempdir().expect("temporary root");
        let path = temporary.path().join("file");
        fs::write(&path, b"old").expect("write file");
        let projection = projection(temporary.path(), AccessMode::ReadWrite);
        let observed = projection.observe_object("/file").expect("observe file");
        assert_eq!(observed.interfaces.len(), 2);
        let write_interface = observed.interfaces[1].clone();

        let missing = call(
            &projection,
            "/file",
            None,
            write_interface.clone(),
            "replace",
            BTreeMap::from([("contents".into(), Value::Bytes(b"new".to_vec()))]),
        )
        .expect_err("write validator required");
        assert_eq!(missing.kind(), CallErrorKind::ValidatorRequired);

        let stale_validator = observed.validator.clone();
        fs::remove_file(&path).expect("remove old entry");
        fs::write(&path, b"replacement").expect("replace path");
        let stale = call(
            &projection,
            "/file",
            stale_validator,
            write_interface,
            "replace",
            BTreeMap::from([("contents".into(), Value::Bytes(b"bad".to_vec()))]),
        )
        .expect_err("replacement identity must mismatch");
        assert_eq!(stale.kind(), CallErrorKind::ValidatorMismatch);
        assert_eq!(fs::read(&path).expect("read replacement"), b"replacement");

        let current = projection
            .observe_object("/file")
            .expect("observe replacement");
        let result = call(
            &projection,
            "/file",
            current.validator.clone(),
            current.interfaces[1].clone(),
            "replace",
            BTreeMap::from([("contents".into(), Value::Bytes(b"committed".to_vec()))]),
        )
        .expect("replace file");
        assert_ne!(result.validator, current.validator);
        assert_eq!(fs::read(path).expect("read committed file"), b"committed");
    }

    #[test]
    fn post_dispatch_failures_preserve_effect_guarantees() {
        let temporary = tempdir().expect("temporary root");
        let path = temporary.path().join("file");
        fs::write(&path, b"old").expect("write file");
        let projection = projection(temporary.path(), AccessMode::ReadWrite);
        let observed = projection.observe_object("/file").expect("observe file");
        let request = || {
            call(
                &projection,
                "/file",
                observed.validator.clone(),
                observed.interfaces[1].clone(),
                "replace",
                BTreeMap::from([("contents".into(), Value::Bytes(b"new".to_vec()))]),
            )
        };

        projection
            .backend()
            .set_injection(FailureInjection::BeforeWrite);
        assert_eq!(
            request().expect_err("pre-write failure").kind(),
            CallErrorKind::Internal
        );
        assert_eq!(fs::read(&path).expect("old file remains"), b"old");

        projection
            .backend()
            .set_injection(FailureInjection::AfterTruncate);
        assert_eq!(
            request().expect_err("post-truncate failure").kind(),
            CallErrorKind::OperationOutcomeUnknown
        );
        assert_eq!(fs::read(path).expect("truncated file"), b"");
    }

    #[test]
    fn large_directory_fails_without_partial_observation() {
        let temporary = tempdir().expect("temporary root");
        for index in 0..2_000 {
            fs::write(temporary.path().join(format!("entry-{index}")), b"x")
                .expect("write directory entry");
        }
        let projection = projection(temporary.path(), AccessMode::ReadOnly);
        let error = projection
            .observe(
                &ExecutionContext::new(),
                &ObserveRequest {
                    path: "/".into(),
                    depth: 1,
                },
            )
            .expect_err("complete observation exceeds node budget");
        assert_eq!(error.code, ProtocolErrorCode::ResourceLimitExceeded);
    }

    #[cfg(unix)]
    #[test]
    fn non_unicode_names_are_hidden_and_special_files_are_explicit() {
        use std::ffi::OsString;
        use std::os::unix::ffi::OsStringExt;
        use std::os::unix::net::UnixListener;

        let temporary = tempdir().expect("temporary root");
        let invalid_name = OsString::from_vec(vec![b'n', b'o', b'n', 0xff]);
        fs::write(temporary.path().join(invalid_name), b"hidden").expect("write non-Unicode file");
        let socket_path = temporary.path().join("socket");
        let _socket = UnixListener::bind(&socket_path).expect("bind filesystem socket");
        let projection = projection(temporary.path(), AccessMode::ReadOnly);

        let tree = projection
            .observe(
                &ExecutionContext::new(),
                &ObserveRequest {
                    path: "/".into(),
                    depth: 1,
                },
            )
            .expect("list root");
        let children = tree.children.as_ref().expect("root children observed");
        assert_eq!(children.len(), 1);
        assert_eq!(children[0].object.name, "socket");
        assert_eq!(
            children[0].object.description.as_deref(),
            Some("Unsupported filesystem entry")
        );
        assert!(children[0].object.interfaces.is_empty());
    }

    #[test]
    fn cancellation_is_not_clean_tree_exhaustion() {
        let temporary = tempdir().expect("temporary root");
        fs::write(temporary.path().join("entry"), b"x").expect("write file");
        let projection = projection(temporary.path(), AccessMode::ReadOnly);
        let cancellation = wip_host::CancellationToken::new();
        cancellation.cancel();
        let error = projection
            .observe(
                &ExecutionContext::new().with_cancellation(cancellation),
                &ObserveRequest {
                    path: "/".into(),
                    depth: 1,
                },
            )
            .expect_err("cancelled traversal");
        assert_eq!(error.code, ProtocolErrorCode::ResourceLimitExceeded);
    }

    #[test]
    fn filesystem_invalid_names_are_not_published() {
        let temporary = tempdir().expect("temporary root");
        let projection = projection(temporary.path(), AccessMode::ReadOnly);
        let error = projection
            .observe_object("/bad\0name")
            .expect_err("NUL cannot name a backend entry");
        assert_eq!(error.kind(), CallErrorKind::NotFound);
    }

    #[cfg(unix)]
    #[test]
    fn configured_root_rejects_symlink_components() {
        let temporary = tempdir().expect("temporary parent");
        let actual = temporary.path().join("actual");
        fs::create_dir(&actual).expect("create actual root");
        let linked = temporary.path().join("linked");
        std::os::unix::fs::symlink(&actual, &linked).expect("create root symlink");
        assert!(
            LinuxBackend::open(&linked, AccessMode::ReadOnly, ResourceLimits::default()).is_err()
        );
    }

    #[test]
    fn file_lock_wait_observes_the_operation_deadline() {
        let temporary = tempdir().expect("temporary root");
        let path = temporary.path().join("file");
        fs::write(&path, b"locked").expect("write file");
        let locked = fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(&path)
            .expect("open lock owner");
        rustix::fs::flock(&locked, FlockOperation::LockExclusive).expect("hold external lock");

        let limits = ResourceLimits {
            request_timeout: std::time::Duration::from_millis(20),
            ..ResourceLimits::default()
        };
        let projection = crate::default_projection(temporary.path(), AccessMode::ReadOnly, limits)
            .expect("open projection");
        let observed = projection.observe_object("/file").expect("observe file");
        let started = std::time::Instant::now();
        let error = call(
            &projection,
            "/file",
            observed.validator,
            observed.interfaces[0].clone(),
            "read",
            BTreeMap::from([("length".into(), Value::Integer(1))]),
        )
        .expect_err("lock wait must time out");
        assert_eq!(error.kind(), CallErrorKind::ResourceLimitExceeded);
        assert!(started.elapsed() < std::time::Duration::from_secs(1));
        rustix::fs::flock(&locked, FlockOperation::Unlock).expect("release lock");
    }
}