car_inference/

model_management.rs

//! Crash-safe, identity-bound state for CAR-managed local model artifacts.
//!
//! Raw model IDs never enter filenames. Every record is stored under the
//! SHA-256 of the exact UTF-8 ID and repeats that ID inside a
//! `deny_unknown_fields` payload so a collision or corrupt file fails closed.

use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Mutex, OnceLock};

use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

const MAX_MODEL_ID_BYTES: usize = 4 * 1024;
const MAX_STATE_BYTES: u64 = 64 * 1024;
/// Refusal reason for a quarantine that disappeared between CAR's own
/// no-replace rename and the delete — on a first attempt or a resume, at the
/// path-based guard or at the narrower window before the binding descriptor
/// opens. Every one of those paths returns before the receipt and journal
/// unlinks, so the removal journal is still on disk and running the removal
/// again resumes it. Shared by both raise sites so the wording cannot drift.
const QUARANTINE_VANISHED_REASON: &str = "quarantine vanished before deletion; the removal journal is retained, so retrying the removal resumes and completes it";

#[derive(Debug, thiserror::Error)]
pub enum ModelManagementError {
    #[error("model-management I/O failed at {path}: {source}")]
    Io {
        path: PathBuf,
        #[source]
        source: std::io::Error,
    },
    #[error("invalid model-management state at {path}: {message}")]
    InvalidState { path: PathBuf, message: String },
    #[error("model-management identity mismatch at {path}: expected {expected}, found {actual}")]
    IdentityMismatch {
        path: PathBuf,
        expected: String,
        actual: String,
    },
    #[error("no CAR install receipt exists for {model_id}")]
    MissingReceipt { model_id: String },
    #[error("unsafe CAR-managed path for {model_id}: {path} ({reason})")]
    UnsafeManagedPath {
        model_id: String,
        path: PathBuf,
        reason: String,
    },
    #[error("local model {model_id} is in use")]
    ModelInUse { model_id: String },
    #[error("local model-management operation is already active for {model_id}")]
    MutationInProgress { model_id: String },
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ManagedArtifactKind {
    Symlink,
    Directory,
    File,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct InstallReceipt {
    pub model_id: String,
    pub managed_path: PathBuf,
    pub artifact_kind: ManagedArtifactKind,
    pub source_model_id: String,
    pub source_revision: Option<String>,
    pub creation_generation: u64,
    pub shared_cache_references: Vec<PathBuf>,
    pub adopted: bool,
}

#[derive(Debug, Clone)]
pub struct ModelManagementStore {
    models_dir: PathBuf,
    management_state_dir: PathBuf,
    receipts_root: PathBuf,
    tombstones_root: PathBuf,
    installs_root: PathBuf,
    removals_root: PathBuf,
    leases_root: PathBuf,
    mutation_locks_root: PathBuf,
    /// Test-only seam that runs between the phases of a directory removal so
    /// tests can race it the way a same-UID process would. Compiled out of
    /// release builds and of every target without directory removal.
    #[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
    removal_hook: Option<RemovalTestHook>,
    #[cfg(test)]
    identity_init_hook: Option<fn(&Path)>,
}

/// Whether this build can delete a CAR-managed directory through a
/// descriptor-bound walk (`ModelManagementStore::remove_quarantined_directory`).
/// Everywhere else recursive pathname deletion stays refused: directory
/// receipts remain usable evidence but are never removable.
pub(crate) const fn directory_removal_supported() -> bool {
    cfg!(any(target_os = "macos", target_os = "linux"))
}

#[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum RemovalPhase {
    /// The managed leaf has been renamed into its quarantine; nothing deleted yet.
    AfterRename,
    /// The quarantine passed the path-based identity check; the descriptor
    /// that binds every later operation has not been opened yet.
    BeforeRootOpen,
    /// Every entry of the quarantined directory has been captured by descriptor;
    /// nothing deleted yet.
    AfterCapture,
}

#[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
#[derive(Clone)]
pub(crate) struct RemovalTestHook(std::sync::Arc<dyn Fn(RemovalPhase, &Path) + Send + Sync>);

#[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
impl std::fmt::Debug for RemovalTestHook {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("RemovalTestHook")
    }
}

impl ModelManagementStore {
    pub fn new(state_root: PathBuf, models_dir: PathBuf) -> Self {
        let state_root = crate::resource_policy::normalized_state_root_key(&state_root);
        let management_state_dir = state_root.join("model-management");
        // `models_dir` is machine-shared even when two daemons use different
        // CAR_HOME roots. Resolve aliases before deriving lock paths so those
        // daemons coordinate on one immutable set of lock sentinels.
        let models_dir = crate::resource_policy::normalized_state_root_key(&models_dir);
        let shared_coordination_root = models_dir
            .parent()
            .unwrap_or(&models_dir)
            .join(".car-model-management");
        let store = Self {
            models_dir,
            receipts_root: management_state_dir.join("receipts"),
            tombstones_root: management_state_dir.join("tombstones"),
            installs_root: management_state_dir.join("installs"),
            removals_root: management_state_dir.join("removals"),
            leases_root: shared_coordination_root.join("activity"),
            mutation_locks_root: shared_coordination_root.join("mutation"),
            management_state_dir,
            #[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
            removal_hook: None,
            #[cfg(test)]
            identity_init_hook: None,
        };
        // A crash can leave a CAR-owned quarantine after the managed leaf has
        // already been atomically detached. Resume only those journaled,
        // identity-matching transactions whose cross-process locks are idle.
        store.resume_pending_quarantines();
        store
    }

    pub fn models_dir(&self) -> &Path {
        &self.models_dir
    }

    #[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
    pub(crate) fn with_removal_hook(
        mut self,
        hook: impl Fn(RemovalPhase, &Path) + Send + Sync + 'static,
    ) -> Self {
        self.removal_hook = Some(RemovalTestHook(std::sync::Arc::new(hook)));
        self
    }

    pub fn management_state_dir(&self) -> &Path {
        &self.management_state_dir
    }

    pub fn receipts_root(&self) -> &Path {
        &self.receipts_root
    }

    pub fn receipt_path(&self, model_id: &str) -> PathBuf {
        self.receipts_root.join(hashed_filename(model_id))
    }

    pub fn tombstones_root(&self) -> &Path {
        &self.tombstones_root
    }

    pub fn tombstone_path(&self, model_id: &str) -> PathBuf {
        self.tombstones_root.join(hashed_filename(model_id))
    }

    fn removal_journal_path(&self, model_id: &str) -> PathBuf {
        self.removals_root.join(hashed_filename(model_id))
    }

    fn install_journal_path(&self, model_id: &str) -> PathBuf {
        self.installs_root.join(hashed_filename(model_id))
    }

    pub fn leases_root(&self) -> &Path {
        &self.leases_root
    }

    pub fn lease_path(&self, model_id: &str) -> PathBuf {
        self.leases_root.join(hashed_filename(model_id))
    }

    pub fn mutation_lock_path(&self, model_id: &str) -> PathBuf {
        self.mutation_locks_root.join(hashed_filename(model_id))
    }

    pub fn load_receipt(
        &self,
        model_id: &str,
    ) -> Result<Option<InstallReceipt>, ModelManagementError> {
        validate_model_id(model_id)?;
        let path = self.receipt_path(model_id);
        let bytes = match read_private_state(&path) {
            Ok(bytes) => bytes,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(source) => return Err(ModelManagementError::Io { path, source }),
        };
        let receipt: InstallReceipt =
            serde_json::from_slice(&bytes).map_err(|error| ModelManagementError::InvalidState {
                path: path.clone(),
                message: error.to_string(),
            })?;
        ensure_identity(&path, model_id, &receipt.model_id)?;
        Ok(Some(receipt))
    }

    pub(crate) fn write_receipt(
        &self,
        receipt: &InstallReceipt,
    ) -> Result<(), ModelManagementError> {
        validate_model_id(&receipt.model_id)?;
        if let Some(existing) = self.load_receipt(&receipt.model_id)? {
            ensure_identity(
                &self.receipt_path(&receipt.model_id),
                &receipt.model_id,
                &existing.model_id,
            )?;
        }
        write_private_json(&self.receipt_path(&receipt.model_id), receipt)
    }

    /// Return whether CAR has a receipt whose current artifact still matches
    /// the ownership evidence recorded at install/adoption time.
    pub fn can_remove(&self, model_id: &str) -> Result<bool, ModelManagementError> {
        let Some(receipt) = self.load_receipt(model_id)? else {
            return Ok(false);
        };
        self.validate_managed_artifact(&receipt)?;
        // Recursive pathname deletion cannot be made object-bound portably.
        // Directory receipts are removable only where CAR has a
        // descriptor-bound walk (`directory_removal_supported`); elsewhere they
        // remain usable evidence but are never removable.
        Ok(
            receipt.artifact_kind != ManagedArtifactKind::Directory
                || directory_removal_supported(),
        )
    }

    pub fn acquire_lease(&self, model_id: &str) -> Result<ModelLease, ModelManagementError> {
        validate_model_id(model_id)?;
        // A load first enters the shared side of the immutable mutation gate,
        // then takes its durable activity lease. Removal takes the exclusive
        // side before it starts draining, closing the check/delete TOCTOU.
        let mutation = self.open_identity_lock(&self.mutation_lock_path(model_id), model_id)?;
        mutation
            .lock_shared()
            .map_err(|source| ModelManagementError::Io {
                path: self.mutation_lock_path(model_id),
                source,
            })?;
        let mut mutation = OwnedFileLock::new(mutation);
        validate_lock_identity(&mut mutation, &self.mutation_lock_path(model_id), model_id)?;
        let file = self.open_identity_lock(&self.lease_path(model_id), model_id)?;
        file.lock_shared()
            .map_err(|source| ModelManagementError::Io {
                path: self.lease_path(model_id),
                source,
            })?;
        let mut file = OwnedFileLock::new(file);
        validate_lock_identity(&mut file, &self.lease_path(model_id), model_id)?;
        drop(mutation);
        Ok(ModelLease { _file: file })
    }

    pub(crate) fn begin_mutation(
        &self,
        model_id: &str,
    ) -> Result<ModelMutationGuard, ModelManagementError> {
        validate_model_id(model_id)?;
        let mut file = self.open_identity_lock(&self.mutation_lock_path(model_id), model_id)?;
        validate_lock_identity(&mut file, &self.mutation_lock_path(model_id), model_id)?;
        match file.try_lock() {
            Ok(()) => Ok(ModelMutationGuard {
                store: self.clone(),
                model_id: model_id.to_string(),
                _file: OwnedFileLock::new(file),
            }),
            Err(std::fs::TryLockError::WouldBlock) => {
                Err(ModelManagementError::MutationInProgress {
                    model_id: model_id.to_string(),
                })
            }
            Err(std::fs::TryLockError::Error(source)) => Err(ModelManagementError::Io {
                path: self.mutation_lock_path(model_id),
                source,
            }),
        }
    }

    pub fn model_in_use(&self, model_id: &str) -> Result<bool, ModelManagementError> {
        validate_model_id(model_id)?;
        let mut file = self.open_identity_lock(&self.lease_path(model_id), model_id)?;
        validate_lock_identity(&mut file, &self.lease_path(model_id), model_id)?;
        match file.try_lock() {
            Ok(()) => {
                let _lock = OwnedFileLock::new(file);
                Ok(false)
            }
            Err(std::fs::TryLockError::WouldBlock) => Ok(true),
            Err(std::fs::TryLockError::Error(source)) => Err(ModelManagementError::Io {
                path: self.lease_path(model_id),
                source,
            }),
        }
    }

    pub fn car_enabled(&self, model_id: &str) -> Result<bool, ModelManagementError> {
        Ok(self.load_tombstone(model_id)?.is_none())
    }

    fn load_tombstone(
        &self,
        model_id: &str,
    ) -> Result<Option<RemovalTombstone>, ModelManagementError> {
        validate_model_id(model_id)?;
        let path = self.tombstone_path(model_id);
        let bytes = match read_private_state(&path) {
            Ok(bytes) => bytes,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(source) => return Err(ModelManagementError::Io { path, source }),
        };
        let tombstone: RemovalTombstone =
            serde_json::from_slice(&bytes).map_err(|error| ModelManagementError::InvalidState {
                path: path.clone(),
                message: error.to_string(),
            })?;
        ensure_identity(&path, model_id, &tombstone.model_id)?;
        Ok(Some(tombstone))
    }

    pub(crate) fn clear_tombstone(&self, model_id: &str) -> Result<(), ModelManagementError> {
        validate_model_id(model_id)?;
        let _ = self.car_enabled(model_id)?;
        let path = self.tombstone_path(model_id);
        match std::fs::remove_file(&path) {
            Ok(()) => sync_directory(path.parent().expect("tombstone path has parent"))
                .map_err(|source| ModelManagementError::Io { path, source }),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(source) => Err(ModelManagementError::Io { path, source }),
        }
    }

    /// Bind a usable artifact under the CAR models root to a durable receipt.
    /// The artifact path is derived by CAR; callers cannot pass arbitrary
    /// deletion targets through the public API.
    pub(crate) fn record_managed_artifact(
        &self,
        model_id: &str,
        source_model_id: &str,
        source_revision: Option<String>,
        creation_generation: u64,
        adopted: bool,
        managed_path: PathBuf,
    ) -> Result<InstallReceipt, ModelManagementError> {
        validate_model_id(model_id)?;
        let artifact_kind = artifact_kind(&managed_path, model_id)?;
        let shared_cache_references = collect_shared_references(&managed_path, model_id)?;
        let receipt = InstallReceipt {
            model_id: model_id.to_string(),
            managed_path,
            artifact_kind,
            source_model_id: source_model_id.to_string(),
            source_revision,
            creation_generation,
            shared_cache_references,
            adopted,
        };
        self.validate_managed_artifact(&receipt)?;
        self.write_receipt(&receipt)?;
        self.clear_tombstone(model_id)?;
        Ok(receipt)
    }

    pub(crate) fn begin_install_intent(
        &self,
        receipt: InstallReceipt,
        staging_path: Option<&Path>,
    ) -> Result<(), ModelManagementError> {
        validate_model_id(&receipt.model_id)?;
        validate_direct_child(&self.models_dir, &receipt.managed_path, &receipt.model_id)?;
        let staging_path = staging_path.map(Path::to_path_buf);
        let staging_identity = staging_path
            .as_deref()
            .map(|path| artifact_identity(path, &receipt.model_id))
            .transpose()?;
        if let Some(path) = staging_path.as_deref() {
            validate_direct_child(&self.models_dir, path, &receipt.model_id)?;
        }
        let intent = InstallJournal {
            model_id: receipt.model_id.clone(),
            receipt,
            staging_path,
            staging_identity,
        };
        let path = self.install_journal_path(&intent.model_id);
        if let Some(existing) = self.load_install_intent(&intent.model_id)? {
            if existing == intent {
                return Ok(());
            }
            return Err(ModelManagementError::InvalidState {
                path,
                message: "a different CAR install intent already exists for this model".into(),
            });
        }
        if std::fs::symlink_metadata(&intent.receipt.managed_path).is_ok() {
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: intent.model_id,
                path: intent.receipt.managed_path,
                reason: "pre-existing unreceipted leaf cannot become CAR-owned through an install intent"
                    .into(),
            });
        }
        write_private_json(&path, &intent)
    }

    pub(crate) fn install_receipt_for_publication(
        &self,
        model_id: &str,
        source_model_id: &str,
        creation_generation: u64,
        adopted: bool,
        managed_path: PathBuf,
        artifact_kind: ManagedArtifactKind,
        artifact_source: &Path,
    ) -> Result<InstallReceipt, ModelManagementError> {
        validate_model_id(model_id)?;
        let shared_cache_references = match artifact_kind {
            ManagedArtifactKind::Symlink => {
                vec![artifact_source
                    .canonicalize()
                    .map_err(|source| ModelManagementError::Io {
                        path: artifact_source.to_path_buf(),
                        source,
                    })?]
            }
            ManagedArtifactKind::Directory => collect_shared_references(artifact_source, model_id)?,
            ManagedArtifactKind::File => Vec::new(),
        };
        Ok(InstallReceipt {
            model_id: model_id.to_string(),
            managed_path,
            artifact_kind,
            source_model_id: source_model_id.to_string(),
            source_revision: None,
            creation_generation,
            shared_cache_references,
            adopted,
        })
    }

    pub(crate) fn resume_install_intent(
        &self,
        model_id: &str,
    ) -> Result<Option<InstallReceipt>, ModelManagementError> {
        let Some(intent) = self.load_install_intent(model_id)? else {
            return Ok(None);
        };
        if std::fs::symlink_metadata(&intent.receipt.managed_path).is_err() {
            if let (Some(staging), Some(identity)) = (
                intent.staging_path.as_deref(),
                intent.staging_identity.as_ref(),
            ) {
                if std::fs::symlink_metadata(staging).is_ok() {
                    ensure_artifact_identity(staging, model_id, identity)?;
                    atomic_rename_noreplace(staging, &intent.receipt.managed_path).map_err(
                        |source| ModelManagementError::Io {
                            path: intent.receipt.managed_path.clone(),
                            source,
                        },
                    )?;
                    sync_directory(&self.models_dir).map_err(|source| {
                        ModelManagementError::Io {
                            path: self.models_dir.clone(),
                            source,
                        }
                    })?;
                } else {
                    self.clear_install_intent(model_id)?;
                    return Ok(None);
                }
            } else {
                // No CAR-created leaf was published. Clearing only the intent
                // is safe and lets the caller create a fresh projection.
                self.clear_install_intent(model_id)?;
                return Ok(None);
            }
        }
        if let Some(identity) = intent.staging_identity.as_ref() {
            ensure_artifact_identity(&intent.receipt.managed_path, model_id, identity)?;
        }
        self.validate_managed_artifact(&intent.receipt)?;
        self.write_receipt(&intent.receipt)?;
        self.clear_install_intent(model_id)?;
        self.clear_tombstone(model_id)?;
        Ok(Some(intent.receipt))
    }

    fn load_install_intent(
        &self,
        model_id: &str,
    ) -> Result<Option<InstallJournal>, ModelManagementError> {
        let path = self.install_journal_path(model_id);
        let bytes = match read_private_state(&path) {
            Ok(bytes) => bytes,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(source) => return Err(ModelManagementError::Io { path, source }),
        };
        let intent: InstallJournal =
            serde_json::from_slice(&bytes).map_err(|error| ModelManagementError::InvalidState {
                path: path.clone(),
                message: error.to_string(),
            })?;
        ensure_identity(&path, model_id, &intent.model_id)?;
        ensure_identity(&path, model_id, &intent.receipt.model_id)?;
        validate_direct_child(&self.models_dir, &intent.receipt.managed_path, model_id)?;
        if let Some(staging) = intent.staging_path.as_deref() {
            validate_direct_child(&self.models_dir, staging, model_id)?;
            if !staging
                .file_name()
                .and_then(|name| name.to_str())
                .is_some_and(|name| name.starts_with(".car-install-"))
            {
                return Err(ModelManagementError::UnsafeManagedPath {
                    model_id: model_id.to_string(),
                    path: staging.to_path_buf(),
                    reason: "install journal staging path lacks CAR staging identity".into(),
                });
            }
        }
        Ok(Some(intent))
    }

    fn clear_install_intent(&self, model_id: &str) -> Result<(), ModelManagementError> {
        let path = self.install_journal_path(model_id);
        match std::fs::remove_file(&path) {
            Ok(()) => sync_directory(&self.installs_root)
                .map_err(|source| ModelManagementError::Io { path, source }),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(source) => Err(ModelManagementError::Io { path, source }),
        }
    }

    /// Create or reuse a CAR-owned symlink projection for an artifact that
    /// physically lives in a shared cache. Shared bytes remain outside CAR's
    /// ownership and are never removed with the projection.
    pub(crate) fn materialize_managed_projection(
        &self,
        managed_leaf: &str,
        source: &Path,
    ) -> Result<PathBuf, ModelManagementError> {
        validate_managed_leaf(managed_leaf, &self.models_dir)?;
        create_private_dir(&self.models_dir)?;
        let models_root = canonical_directory(&self.models_dir)?;
        let source_parent =
            source
                .parent()
                .ok_or_else(|| ModelManagementError::UnsafeManagedPath {
                    model_id: managed_leaf.to_string(),
                    path: source.to_path_buf(),
                    reason: "source has no parent".into(),
                })?;
        if canonical_directory(source_parent)? == models_root
            && source == self.models_dir.join(managed_leaf)
        {
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: managed_leaf.to_string(),
                path: source.to_path_buf(),
                reason: "pre-existing unreceipted managed projection cannot become CAR-owned"
                    .into(),
            });
        }
        let target = source
            .canonicalize()
            .map_err(|source_error| ModelManagementError::Io {
                path: source.to_path_buf(),
                source: source_error,
            })?;
        let managed = self.models_dir.join(managed_leaf);
        if std::fs::symlink_metadata(&managed).is_ok() {
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: managed_leaf.to_string(),
                path: managed,
                reason: "pre-existing managed projection has no CAR creation receipt".into(),
            });
        }
        static NEXT_PROJECTION: AtomicU64 = AtomicU64::new(1);
        let tmp = self.models_dir.join(format!(
            ".managed-projection.{}.{}",
            std::process::id(),
            NEXT_PROJECTION.fetch_add(1, Ordering::Relaxed)
        ));
        create_symlink(&target, &tmp)?;
        let result = atomic_rename_noreplace(&tmp, &managed).map_err(|source| {
            ModelManagementError::UnsafeManagedPath {
                model_id: managed_leaf.to_string(),
                path: managed.clone(),
                reason: format!("managed projection publication raced or failed: {source}"),
            }
        });
        if result.is_err() {
            let _ = std::fs::remove_file(&tmp);
        }
        result?;
        sync_directory(&self.models_dir).map_err(|source| ModelManagementError::Io {
            path: self.models_dir.clone(),
            source,
        })?;
        Ok(managed)
    }

    pub(crate) fn create_install_staging(
        &self,
        model_id: &str,
    ) -> Result<PathBuf, ModelManagementError> {
        validate_model_id(model_id)?;
        create_private_dir(&self.models_dir)?;
        let digest = hex::encode(Sha256::digest(model_id.as_bytes()));
        for _ in 0..16 {
            let candidate = self.models_dir.join(format!(
                ".car-install-{}-{:032x}",
                &digest[..16],
                rand::random::<u128>()
            ));
            match std::fs::create_dir(&candidate) {
                Ok(()) => {
                    harden_private_directory(&candidate)?;
                    return Ok(candidate);
                }
                Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue,
                Err(source) => {
                    return Err(ModelManagementError::Io {
                        path: candidate,
                        source,
                    });
                }
            }
        }
        Err(ModelManagementError::InvalidState {
            path: self.models_dir.clone(),
            message: "could not allocate a collision-resistant install staging directory".into(),
        })
    }

    pub(crate) fn publish_install_staging(
        &self,
        model_id: &str,
        staging: &Path,
        managed_leaf: &str,
    ) -> Result<PathBuf, ModelManagementError> {
        validate_model_id(model_id)?;
        validate_direct_child(&self.models_dir, staging, model_id)?;
        if !staging
            .file_name()
            .and_then(|name| name.to_str())
            .is_some_and(|name| name.starts_with(".car-install-"))
        {
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: model_id.to_string(),
                path: staging.to_path_buf(),
                reason: "install staging path lacks CAR staging identity".into(),
            });
        }
        validate_managed_leaf(managed_leaf, &self.models_dir)?;
        let managed = self.models_dir.join(managed_leaf);
        atomic_rename_noreplace(staging, &managed).map_err(|source| {
            ModelManagementError::UnsafeManagedPath {
                model_id: model_id.to_string(),
                path: managed.clone(),
                reason: format!("could not atomically publish CAR-owned install: {source}"),
            }
        })?;
        sync_directory(&self.models_dir).map_err(|source| ModelManagementError::Io {
            path: self.models_dir.clone(),
            source,
        })?;
        Ok(managed)
    }

    pub(crate) fn discard_install_staging(&self, staging: &Path) {
        // Deliberately fail closed. A same-UID attacker can swap a verified
        // directory before a pathname-recursive delete. Leave failed staging
        // journal/evidence for future object-bound cleanup instead of risking
        // deletion of shared model bytes.
        let _ = staging;
    }

    /// Project a hand-installed or shared artifact into a distinct CAR-owned
    /// leaf. The original path is never promoted into CAR ownership merely by
    /// adoption, even when it already sits beneath `models_dir`.
    pub(crate) fn materialize_adopted_projection(
        &self,
        model_id: &str,
        source: &Path,
    ) -> Result<PathBuf, ModelManagementError> {
        validate_model_id(model_id)?;
        let managed = self.adopted_projection_path(model_id)?;
        let leaf = managed
            .file_name()
            .and_then(|name| name.to_str())
            .expect("adopted projection is an UTF-8 direct child");
        self.materialize_managed_projection(leaf, source)
    }

    pub(crate) fn adopted_projection_path(
        &self,
        model_id: &str,
    ) -> Result<PathBuf, ModelManagementError> {
        validate_model_id(model_id)?;
        let digest = hex::encode(Sha256::digest(model_id.as_bytes()));
        Ok(self.models_dir.join(format!(".car-adopted-{digest}")))
    }

    fn remove_with_mutation(
        &self,
        model_id: &str,
        removal_generation: u64,
    ) -> Result<RemoveFromCarResult, ModelManagementError> {
        let _activity = self.lock_idle_activity(model_id)?;

        if let Some(journal) = self.load_removal_journal(model_id)? {
            write_private_json(
                &self.tombstone_path(model_id),
                &RemovalTombstone {
                    model_id: model_id.to_string(),
                    removal_generation: journal.removal_generation,
                    artifact_kind: Some(journal.receipt.artifact_kind),
                    preserved_shared_cache_references: journal
                        .receipt
                        .shared_cache_references
                        .clone(),
                },
            )?;
            return self.resume_removal_journal(journal);
        }

        let Some(receipt) = self.load_receipt(model_id)? else {
            if let Some(tombstone) = self.load_tombstone(model_id)? {
                if let Some(artifact_kind) = tombstone.artifact_kind {
                    return Ok(RemoveFromCarResult {
                        model_id: model_id.to_string(),
                        artifact_kind,
                        preserved_shared_cache_references: tombstone
                            .preserved_shared_cache_references,
                    });
                }
            }
            return Err(ModelManagementError::MissingReceipt {
                model_id: model_id.to_string(),
            });
        };
        self.validate_managed_artifact(&receipt)?;
        if receipt.artifact_kind == ManagedArtifactKind::Directory && !directory_removal_supported()
        {
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: model_id.to_string(),
                path: receipt.managed_path,
                reason: "recursive directory removal is unsupported without object-bound traversal"
                    .into(),
            });
        }
        let artifact_identity = artifact_identity(&receipt.managed_path, model_id)?;
        let quarantine = self.allocate_quarantine(model_id)?;

        write_private_json(
            &self.tombstone_path(model_id),
            &RemovalTombstone {
                model_id: model_id.to_string(),
                removal_generation,
                artifact_kind: Some(receipt.artifact_kind),
                preserved_shared_cache_references: receipt.shared_cache_references.clone(),
            },
        )?;
        let journal = RemovalJournal {
            model_id: model_id.to_string(),
            removal_generation,
            receipt,
            quarantine_path: quarantine,
            artifact_identity,
        };
        write_private_json(&self.removal_journal_path(model_id), &journal)?;
        self.resume_removal_journal(journal)
    }

    fn lock_idle_activity(&self, model_id: &str) -> Result<OwnedFileLock, ModelManagementError> {
        let mut activity = self.open_identity_lock(&self.lease_path(model_id), model_id)?;
        validate_lock_identity(&mut activity, &self.lease_path(model_id), model_id)?;
        match activity.try_lock() {
            Ok(()) => Ok(OwnedFileLock::new(activity)),
            Err(std::fs::TryLockError::WouldBlock) => Err(ModelManagementError::ModelInUse {
                model_id: model_id.to_string(),
            }),
            Err(std::fs::TryLockError::Error(source)) => Err(ModelManagementError::Io {
                path: self.lease_path(model_id),
                source,
            }),
        }
    }

    fn load_removal_journal(
        &self,
        model_id: &str,
    ) -> Result<Option<RemovalJournal>, ModelManagementError> {
        let path = self.removal_journal_path(model_id);
        let bytes = match read_private_state(&path) {
            Ok(bytes) => bytes,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(source) => return Err(ModelManagementError::Io { path, source }),
        };
        let journal: RemovalJournal =
            serde_json::from_slice(&bytes).map_err(|error| ModelManagementError::InvalidState {
                path: path.clone(),
                message: error.to_string(),
            })?;
        ensure_identity(&path, model_id, &journal.model_id)?;
        ensure_identity(&path, model_id, &journal.receipt.model_id)?;
        validate_direct_child(&self.models_dir, &journal.receipt.managed_path, model_id)?;
        if journal.receipt.artifact_kind == ManagedArtifactKind::Directory
            && !directory_removal_supported()
        {
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: model_id.to_string(),
                path: journal.receipt.managed_path,
                reason: "recursive directory removal journal is unsupported".into(),
            });
        }
        validate_direct_child(&self.models_dir, &journal.quarantine_path, model_id)?;
        if !journal
            .quarantine_path
            .file_name()
            .and_then(|name| name.to_str())
            .is_some_and(|name| name.starts_with(".car-remove-"))
        {
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: model_id.to_string(),
                path: journal.quarantine_path,
                reason: "removal journal has an invalid quarantine path".into(),
            });
        }
        Ok(Some(journal))
    }

    fn allocate_quarantine(&self, model_id: &str) -> Result<PathBuf, ModelManagementError> {
        let digest = hex::encode(Sha256::digest(model_id.as_bytes()));
        for _ in 0..16 {
            let candidate = self.models_dir.join(format!(
                ".car-remove-{}-{:032x}",
                &digest[..16],
                rand::random::<u128>()
            ));
            if std::fs::symlink_metadata(&candidate)
                .is_err_and(|error| error.kind() == std::io::ErrorKind::NotFound)
            {
                return Ok(candidate);
            }
        }
        Err(ModelManagementError::InvalidState {
            path: self.models_dir.clone(),
            message: "could not allocate a collision-resistant removal quarantine".into(),
        })
    }

    fn resume_removal_journal(
        &self,
        journal: RemovalJournal,
    ) -> Result<RemoveFromCarResult, ModelManagementError> {
        let model_id = &journal.model_id;
        if journal.receipt.artifact_kind == ManagedArtifactKind::Directory
            && !directory_removal_supported()
        {
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: model_id.clone(),
                path: journal.receipt.managed_path,
                reason: "recursive directory removal is unsupported without object-bound traversal"
                    .into(),
            });
        }
        let original_exists = std::fs::symlink_metadata(&journal.receipt.managed_path).is_ok();
        let quarantine_exists = std::fs::symlink_metadata(&journal.quarantine_path).is_ok();
        let mut renamed_now = false;
        if original_exists && quarantine_exists {
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: model_id.clone(),
                path: journal.receipt.managed_path.clone(),
                reason: "both managed artifact and removal quarantine exist".into(),
            });
        }
        if original_exists {
            self.validate_managed_artifact(&journal.receipt)?;
            ensure_artifact_identity(
                &journal.receipt.managed_path,
                model_id,
                &journal.artifact_identity,
            )?;
            // Revalidate the parent and leaf identity immediately before the
            // same-root no-replace rename. Unsupported platforms fail closed.
            validate_direct_child(&self.models_dir, &journal.receipt.managed_path, model_id)?;
            ensure_artifact_identity(
                &journal.receipt.managed_path,
                model_id,
                &journal.artifact_identity,
            )?;
            atomic_rename_noreplace(&journal.receipt.managed_path, &journal.quarantine_path)
                .map_err(|source| ModelManagementError::Io {
                    path: journal.receipt.managed_path.clone(),
                    source,
                })?;
            sync_directory(&self.models_dir).map_err(|source| ModelManagementError::Io {
                path: self.models_dir.clone(),
                source,
            })?;
            renamed_now = true;
            #[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
            if let Some(hook) = &self.removal_hook {
                (hook.0)(RemovalPhase::AfterRename, &journal.quarantine_path);
            }
        }

        if std::fs::symlink_metadata(&journal.quarantine_path).is_err() && renamed_now {
            // We detached the leaf moments ago and it is already gone from its
            // quarantine: something else moved it. Never report that as a
            // completed removal; the journal stays so a later resume can look
            // again.
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: model_id.clone(),
                path: journal.quarantine_path,
                reason: QUARANTINE_VANISHED_REASON.into(),
            });
        }

        if std::fs::symlink_metadata(&journal.quarantine_path).is_ok() {
            // A resumed directory removal may already be partially deleted, so
            // the quarantine is matched on kind, device, and inode only.
            ensure_removal_identity(
                &journal.quarantine_path,
                model_id,
                &journal.artifact_identity,
            )?;
            match journal.receipt.artifact_kind {
                #[cfg(any(target_os = "macos", target_os = "linux"))]
                ManagedArtifactKind::Directory => {
                    self.remove_quarantined_directory(
                        &journal.quarantine_path,
                        &journal.artifact_identity,
                        model_id,
                    )?;
                }
                #[cfg(not(any(target_os = "macos", target_os = "linux")))]
                ManagedArtifactKind::Directory => {
                    return Err(ModelManagementError::UnsafeManagedPath {
                        model_id: model_id.clone(),
                        path: journal.quarantine_path,
                        reason:
                            "recursive directory removal is unsupported without object-bound traversal"
                                .into(),
                    });
                }
                ManagedArtifactKind::Symlink | ManagedArtifactKind::File => {
                    std::fs::remove_file(&journal.quarantine_path).map_err(|source| {
                        ModelManagementError::Io {
                            path: journal.quarantine_path.clone(),
                            source,
                        }
                    })?;
                }
            }
            sync_directory(&self.models_dir).map_err(|source| ModelManagementError::Io {
                path: self.models_dir.clone(),
                source,
            })?;
        }

        let receipt_path = self.receipt_path(model_id);
        match std::fs::remove_file(&receipt_path) {
            Ok(()) => {
                sync_directory(&self.receipts_root).map_err(|source| ModelManagementError::Io {
                    path: self.receipts_root.clone(),
                    source,
                })?
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(source) => {
                return Err(ModelManagementError::Io {
                    path: receipt_path,
                    source,
                });
            }
        }
        let journal_path = self.removal_journal_path(model_id);
        match std::fs::remove_file(&journal_path) {
            Ok(()) => {
                sync_directory(&self.removals_root).map_err(|source| ModelManagementError::Io {
                    path: self.removals_root.clone(),
                    source,
                })?
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(source) => {
                return Err(ModelManagementError::Io {
                    path: journal_path,
                    source,
                });
            }
        }
        Ok(RemoveFromCarResult {
            model_id: model_id.clone(),
            artifact_kind: journal.receipt.artifact_kind,
            preserved_shared_cache_references: journal.receipt.shared_cache_references,
        })
    }

    fn resume_pending_quarantines(&self) {
        let Ok(entries) = std::fs::read_dir(&self.removals_root) else {
            return;
        };
        for entry in entries.flatten() {
            if !is_hashed_state_filename(&entry.file_name()) {
                continue;
            }
            let Ok(bytes) = read_private_state(&entry.path()) else {
                continue;
            };
            let Ok(candidate) = serde_json::from_slice::<RemovalJournal>(&bytes) else {
                continue;
            };
            if entry.path() != self.removal_journal_path(&candidate.model_id) {
                continue;
            }
            let Ok(guard) = self.begin_mutation(&candidate.model_id) else {
                continue;
            };
            let Ok(Some(journal)) = self.load_removal_journal(&candidate.model_id) else {
                continue;
            };
            if std::fs::symlink_metadata(&journal.receipt.managed_path).is_ok() {
                continue;
            }
            let _ = guard.resume_existing(journal);
        }
    }

    fn open_identity_lock(
        &self,
        path: &Path,
        model_id: &str,
    ) -> Result<File, ModelManagementError> {
        let parent = path.parent().expect("lock path has parent");
        create_private_dir(parent)?;
        // A new identity inode is visible before it can be locked and filled.
        // Serialize that window with a permanent, non-JSON gate. The gate itself
        // needs no initialization payload, so whichever opener locks it first
        // can safely initialize an identity. Never rename or unlink either inode.
        let initialization_path = parent.join(".initialization.lock");
        let initialization = match open_private_new(&initialization_path) {
            Ok(file) => file,
            Err(ModelManagementError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::AlreadyExists =>
            {
                open_existing_identity_lock(&initialization_path)?
            }
            Err(error) => return Err(error),
        };
        initialization
            .lock()
            .map_err(|source| ModelManagementError::Io {
                path: initialization_path,
                source,
            })?;
        let _initialization = OwnedFileLock::new(initialization);
        // This guard drops on return, before callers wait for any existing
        // activity/mutation lock; existing lease holders never need this gate.
        match open_private_new(path) {
            Ok(file) => {
                #[cfg(test)]
                if let Some(hook) = self.identity_init_hook {
                    hook(path);
                }
                file.lock().map_err(|source| ModelManagementError::Io {
                    path: path.to_path_buf(),
                    source,
                })?;
                let mut file = OwnedFileLock::new(file);
                let bytes = serde_json::to_vec(&IdentityRecord {
                    model_id: model_id.to_string(),
                })
                .map_err(|error| ModelManagementError::InvalidState {
                    path: path.to_path_buf(),
                    message: error.to_string(),
                })?;
                file.write_all(&bytes)
                    .and_then(|_| file.sync_all())
                    .map_err(|source| ModelManagementError::Io {
                        path: path.to_path_buf(),
                        source,
                    })?;
                return file
                    .into_unlocked_file()
                    .map_err(|source| ModelManagementError::Io {
                        path: path.to_path_buf(),
                        source,
                    });
            }
            Err(ModelManagementError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::AlreadyExists => {}
            Err(error) => return Err(error),
        }
        open_existing_identity_lock(path)
    }

    fn validate_managed_artifact(
        &self,
        receipt: &InstallReceipt,
    ) -> Result<(), ModelManagementError> {
        let models_root = validate_models_root(&self.models_dir, &receipt.model_id)?;
        if let Some(hf_root) = shared_hugging_face_root() {
            let hf_root = crate::resource_policy::normalized_state_root_key(&hf_root);
            if models_root.starts_with(&hf_root) || hf_root.starts_with(&models_root) {
                return Err(ModelManagementError::UnsafeManagedPath {
                    model_id: receipt.model_id.clone(),
                    path: receipt.managed_path.clone(),
                    reason: "managed models root overlaps the shared Hugging Face cache".into(),
                });
            }
        }
        let parent = receipt.managed_path.parent().ok_or_else(|| {
            ModelManagementError::UnsafeManagedPath {
                model_id: receipt.model_id.clone(),
                path: receipt.managed_path.clone(),
                reason: "managed path has no parent".into(),
            }
        })?;
        let canonical_parent = canonical_directory(parent)?;
        if canonical_parent != models_root || receipt.managed_path == models_root {
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: receipt.model_id.clone(),
                path: receipt.managed_path.clone(),
                reason: "artifact is not a direct child of the managed models root".into(),
            });
        }
        let actual = artifact_kind(&receipt.managed_path, &receipt.model_id)?;
        if actual != receipt.artifact_kind {
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: receipt.model_id.clone(),
                path: receipt.managed_path.clone(),
                reason: format!(
                    "receipt says {:?}, filesystem is {:?}",
                    receipt.artifact_kind, actual
                ),
            });
        }
        if actual == ManagedArtifactKind::Symlink {
            let target =
                receipt
                    .managed_path
                    .canonicalize()
                    .map_err(|source| ModelManagementError::Io {
                        path: receipt.managed_path.clone(),
                        source,
                    })?;
            if !receipt
                .shared_cache_references
                .iter()
                .any(|path| path == &target)
            {
                return Err(ModelManagementError::UnsafeManagedPath {
                    model_id: receipt.model_id.clone(),
                    path: receipt.managed_path.clone(),
                    reason: "symlink target does not match a preserved shared-cache reference"
                        .into(),
                });
            }
        } else {
            let current_references =
                collect_shared_references(&receipt.managed_path, &receipt.model_id)?;
            if current_references != receipt.shared_cache_references {
                return Err(ModelManagementError::UnsafeManagedPath {
                    model_id: receipt.model_id.clone(),
                    path: receipt.managed_path.clone(),
                    reason: "shared-cache references no longer match the install receipt".into(),
                });
            }
        }
        Ok(())
    }
}

// Closing a descriptor alone does not release flock when a concurrently
// spawned child inherited the open-file description. End ownership explicitly
// at the Rust guard boundary, including temporary initialization/activity locks.
struct OwnedFileLock(Option<File>);

impl OwnedFileLock {
    fn new(file: File) -> Self {
        Self(Some(file))
    }

    fn into_unlocked_file(mut self) -> std::io::Result<File> {
        self.unlock()?;
        Ok(self.0.take().expect("owned lock has a file"))
    }
}

impl std::ops::Deref for OwnedFileLock {
    type Target = File;

    fn deref(&self) -> &File {
        self.0.as_ref().expect("owned lock has a file")
    }
}

impl std::ops::DerefMut for OwnedFileLock {
    fn deref_mut(&mut self) -> &mut File {
        self.0.as_mut().expect("owned lock has a file")
    }
}

impl Drop for OwnedFileLock {
    fn drop(&mut self) {
        if let Some(file) = &self.0 {
            let _ = file.unlock();
        }
    }
}

pub(crate) struct ModelMutationGuard {
    store: ModelManagementStore,
    model_id: String,
    _file: OwnedFileLock,
}

impl ModelMutationGuard {
    pub(crate) fn remove(
        self,
        removal_generation: u64,
    ) -> Result<RemoveFromCarResult, ModelManagementError> {
        self.store
            .remove_with_mutation(&self.model_id, removal_generation)
    }

    fn resume_existing(
        self,
        journal: RemovalJournal,
    ) -> Result<RemoveFromCarResult, ModelManagementError> {
        if journal.model_id != self.model_id {
            return Err(ModelManagementError::IdentityMismatch {
                path: self.store.removal_journal_path(&self.model_id),
                expected: self.model_id,
                actual: journal.model_id,
            });
        }
        let _activity = self.store.lock_idle_activity(&self.model_id)?;
        write_private_json(
            &self.store.tombstone_path(&self.model_id),
            &RemovalTombstone {
                model_id: self.model_id.clone(),
                removal_generation: journal.removal_generation,
                artifact_kind: Some(journal.receipt.artifact_kind),
                preserved_shared_cache_references: journal.receipt.shared_cache_references.clone(),
            },
        )?;
        self.store.resume_removal_journal(journal)
    }
}

pub struct ModelLease {
    _file: OwnedFileLock,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RemoveFromCarResult {
    pub model_id: String,
    pub artifact_kind: ManagedArtifactKind,
    pub preserved_shared_cache_references: Vec<PathBuf>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct IdentityRecord {
    model_id: String,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct RemovalTombstone {
    model_id: String,
    removal_generation: u64,
    #[serde(default)]
    artifact_kind: Option<ManagedArtifactKind>,
    #[serde(default)]
    preserved_shared_cache_references: Vec<PathBuf>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct ArtifactIdentity {
    kind: ManagedArtifactKind,
    #[serde(default)]
    device: u64,
    #[serde(default)]
    inode: u64,
    #[serde(default)]
    file_len: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct RemovalJournal {
    model_id: String,
    removal_generation: u64,
    receipt: InstallReceipt,
    quarantine_path: PathBuf,
    artifact_identity: ArtifactIdentity,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct InstallJournal {
    model_id: String,
    receipt: InstallReceipt,
    #[serde(default)]
    staging_path: Option<PathBuf>,
    #[serde(default)]
    staging_identity: Option<ArtifactIdentity>,
}

fn hashed_filename(model_id: &str) -> String {
    let digest = Sha256::digest(model_id.as_bytes());
    format!("{}.json", hex::encode(digest))
}

fn is_hashed_state_filename(name: &std::ffi::OsStr) -> bool {
    let Some(name) = name.to_str() else {
        return false;
    };
    let Some(hex) = name.strip_suffix(".json") else {
        return false;
    };
    hex.len() == 64
        && hex
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
}

fn validate_model_id(model_id: &str) -> Result<(), ModelManagementError> {
    if !model_id.is_empty() && model_id.len() <= MAX_MODEL_ID_BYTES {
        return Ok(());
    }
    Err(ModelManagementError::InvalidState {
        path: PathBuf::new(),
        message: "model id must contain 1..=4096 UTF-8 bytes".into(),
    })
}

fn validate_managed_leaf(
    managed_leaf: &str,
    models_dir: &Path,
) -> Result<(), ModelManagementError> {
    if Path::new(managed_leaf).components().count() == 1
        && matches!(
            Path::new(managed_leaf).components().next(),
            Some(std::path::Component::Normal(_))
        )
    {
        return Ok(());
    }
    Err(ModelManagementError::UnsafeManagedPath {
        model_id: managed_leaf.to_string(),
        path: models_dir.join(managed_leaf),
        reason: "managed model name must be one ordinary path component".into(),
    })
}

fn validate_direct_child(
    root: &Path,
    candidate: &Path,
    model_id: &str,
) -> Result<(), ModelManagementError> {
    let canonical_root = validate_models_root(root, model_id)?;
    let parent = candidate
        .parent()
        .ok_or_else(|| ModelManagementError::UnsafeManagedPath {
            model_id: model_id.to_string(),
            path: candidate.to_path_buf(),
            reason: "managed path has no parent".into(),
        })?;
    if canonical_directory(parent)? != canonical_root {
        return Err(ModelManagementError::UnsafeManagedPath {
            model_id: model_id.to_string(),
            path: candidate.to_path_buf(),
            reason: "managed path is not a direct child of the models root".into(),
        });
    }
    Ok(())
}

fn validate_models_root(root: &Path, model_id: &str) -> Result<PathBuf, ModelManagementError> {
    let metadata = std::fs::symlink_metadata(root).map_err(|source| ModelManagementError::Io {
        path: root.to_path_buf(),
        source,
    })?;
    if metadata.file_type().is_symlink() {
        return Err(ModelManagementError::UnsafeManagedPath {
            model_id: model_id.to_string(),
            path: root.to_path_buf(),
            reason: "managed models root must not be a symlink".into(),
        });
    }
    #[cfg(windows)]
    {
        use std::os::windows::fs::MetadataExt;
        const FILE_ATTRIBUTE_REPARSE_POINT: u32 = 0x400;
        if metadata.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT != 0 {
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: model_id.to_string(),
                path: root.to_path_buf(),
                reason: "managed models root must not be a reparse point".into(),
            });
        }
    }
    if !metadata.is_dir() {
        return Err(ModelManagementError::UnsafeManagedPath {
            model_id: model_id.to_string(),
            path: root.to_path_buf(),
            reason: "managed models root is not a directory".into(),
        });
    }
    canonical_directory(root)
}

fn ensure_identity(path: &Path, expected: &str, actual: &str) -> Result<(), ModelManagementError> {
    if expected == actual {
        return Ok(());
    }
    Err(ModelManagementError::IdentityMismatch {
        path: path.to_path_buf(),
        expected: expected.to_string(),
        actual: actual.to_string(),
    })
}

fn validate_lock_identity(
    file: &mut File,
    path: &Path,
    model_id: &str,
) -> Result<(), ModelManagementError> {
    file.rewind().map_err(|source| ModelManagementError::Io {
        path: path.to_path_buf(),
        source,
    })?;
    let mut bytes = Vec::new();
    file.take(16 * 1024)
        .read_to_end(&mut bytes)
        .map_err(|source| ModelManagementError::Io {
            path: path.to_path_buf(),
            source,
        })?;
    let record: IdentityRecord =
        serde_json::from_slice(&bytes).map_err(|error| ModelManagementError::InvalidState {
            path: path.to_path_buf(),
            message: error.to_string(),
        })?;
    ensure_identity(path, model_id, &record.model_id)
}

fn canonical_directory(path: &Path) -> Result<PathBuf, ModelManagementError> {
    path.canonicalize()
        .map_err(|source| ModelManagementError::Io {
            path: path.to_path_buf(),
            source,
        })
}

fn shared_hugging_face_root() -> Option<PathBuf> {
    std::env::var_os("HF_HOME").map(PathBuf::from).or_else(|| {
        std::env::var_os("HOME")
            .or_else(|| std::env::var_os("USERPROFILE"))
            .map(|home| PathBuf::from(home).join(".cache/huggingface"))
    })
}

fn artifact_identity(
    path: &Path,
    model_id: &str,
) -> Result<ArtifactIdentity, ModelManagementError> {
    let metadata = std::fs::symlink_metadata(path).map_err(|source| ModelManagementError::Io {
        path: path.to_path_buf(),
        source,
    })?;
    reject_windows_reparse(&metadata, path, model_id)?;
    let kind = if metadata.file_type().is_symlink() {
        ManagedArtifactKind::Symlink
    } else if metadata.is_dir() {
        ManagedArtifactKind::Directory
    } else if metadata.is_file() {
        ManagedArtifactKind::File
    } else {
        return Err(ModelManagementError::UnsafeManagedPath {
            model_id: model_id.to_string(),
            path: path.to_path_buf(),
            reason: "unsupported artifact type".into(),
        });
    };
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt;
        Ok(ArtifactIdentity {
            kind,
            device: metadata.dev(),
            inode: metadata.ino(),
            file_len: metadata.len(),
        })
    }
    #[cfg(not(unix))]
    {
        Ok(ArtifactIdentity {
            kind,
            device: 0,
            inode: 0,
            file_len: metadata.len(),
        })
    }
}

fn ensure_artifact_identity(
    path: &Path,
    model_id: &str,
    expected: &ArtifactIdentity,
) -> Result<(), ModelManagementError> {
    let actual = artifact_identity(path, model_id)?;
    if &actual == expected {
        return Ok(());
    }
    Err(ModelManagementError::UnsafeManagedPath {
        model_id: model_id.to_string(),
        path: path.to_path_buf(),
        reason: "managed artifact identity changed during removal".into(),
    })
}

impl ArtifactIdentity {
    /// Removal-phase match. A directory's length changes as its entries are
    /// deleted, so a resumed directory removal compares kind, device, and inode
    /// only; files and symlinks keep the exact comparison. Install-intent
    /// recovery keeps using exact equality.
    fn matches_for_removal(&self, actual: &ArtifactIdentity) -> bool {
        self.kind == actual.kind
            && self.device == actual.device
            && self.inode == actual.inode
            && (self.kind == ManagedArtifactKind::Directory || self.file_len == actual.file_len)
    }
}

fn ensure_removal_identity(
    path: &Path,
    model_id: &str,
    expected: &ArtifactIdentity,
) -> Result<(), ModelManagementError> {
    let actual = artifact_identity(path, model_id)?;
    if expected.matches_for_removal(&actual) {
        return Ok(());
    }
    Err(ModelManagementError::UnsafeManagedPath {
        model_id: model_id.to_string(),
        path: path.to_path_buf(),
        reason: "managed artifact identity changed during removal".into(),
    })
}

fn artifact_kind(path: &Path, model_id: &str) -> Result<ManagedArtifactKind, ModelManagementError> {
    let metadata = std::fs::symlink_metadata(path).map_err(|source| ModelManagementError::Io {
        path: path.to_path_buf(),
        source,
    })?;
    reject_windows_reparse(&metadata, path, model_id)?;
    if metadata.file_type().is_symlink() {
        Ok(ManagedArtifactKind::Symlink)
    } else if metadata.is_dir() {
        Ok(ManagedArtifactKind::Directory)
    } else if metadata.is_file() {
        Ok(ManagedArtifactKind::File)
    } else {
        Err(ModelManagementError::UnsafeManagedPath {
            model_id: model_id.to_string(),
            path: path.to_path_buf(),
            reason: "unsupported artifact type".into(),
        })
    }
}

fn collect_shared_references(
    path: &Path,
    model_id: &str,
) -> Result<Vec<PathBuf>, ModelManagementError> {
    let metadata = std::fs::symlink_metadata(path).map_err(|source| ModelManagementError::Io {
        path: path.to_path_buf(),
        source,
    })?;
    if metadata.file_type().is_symlink() {
        return path
            .canonicalize()
            .map(|target| vec![target])
            .map_err(|source| ModelManagementError::Io {
                path: path.to_path_buf(),
                source,
            });
    }
    if !metadata.is_dir() {
        return Ok(Vec::new());
    }
    let mut references = Vec::new();
    for entry in std::fs::read_dir(path).map_err(|source| ModelManagementError::Io {
        path: path.to_path_buf(),
        source,
    })? {
        let entry = entry.map_err(|source| ModelManagementError::Io {
            path: path.to_path_buf(),
            source,
        })?;
        let entry_path = entry.path();
        let entry_metadata =
            std::fs::symlink_metadata(&entry_path).map_err(|source| ModelManagementError::Io {
                path: entry_path.clone(),
                source,
            })?;
        if entry_metadata.file_type().is_symlink() {
            let target = entry_path
                .canonicalize()
                .map_err(|source| ModelManagementError::Io {
                    path: entry_path.clone(),
                    source,
                })?;
            if std::fs::metadata(&target)
                .map_err(|source| ModelManagementError::Io {
                    path: target.clone(),
                    source,
                })?
                .is_dir()
            {
                return Err(ModelManagementError::UnsafeManagedPath {
                    model_id: model_id.to_string(),
                    path: entry_path,
                    reason: "managed directory contains a directory symlink".into(),
                });
            }
            references.push(target);
        } else if entry_metadata.is_dir() {
            references.extend(collect_shared_references(&entry_path, model_id)?);
        }
    }
    references.sort();
    references.dedup();
    Ok(references)
}

#[cfg(unix)]
fn create_symlink(target: &Path, link: &Path) -> Result<(), ModelManagementError> {
    std::os::unix::fs::symlink(target, link).map_err(|source| ModelManagementError::Io {
        path: link.to_path_buf(),
        source,
    })
}

#[cfg(windows)]
fn create_symlink(target: &Path, link: &Path) -> Result<(), ModelManagementError> {
    let result = if target.is_dir() {
        std::os::windows::fs::symlink_dir(target, link)
    } else {
        std::os::windows::fs::symlink_file(target, link)
    };
    result.map_err(|source| ModelManagementError::Io {
        path: link.to_path_buf(),
        source,
    })
}

fn create_private_dir(path: &Path) -> Result<(), ModelManagementError> {
    std::fs::create_dir_all(path).map_err(|source| ModelManagementError::Io {
        path: path.to_path_buf(),
        source,
    })?;
    harden_private_directory(path)?;
    Ok(())
}

fn harden_private_directory(path: &Path) -> Result<(), ModelManagementError> {
    let metadata = std::fs::symlink_metadata(path).map_err(|source| ModelManagementError::Io {
        path: path.to_path_buf(),
        source,
    })?;
    if metadata.file_type().is_symlink() {
        return Err(ModelManagementError::UnsafeManagedPath {
            model_id: "model-management-state".into(),
            path: path.to_path_buf(),
            reason: "private model-management directory must not be a symlink".into(),
        });
    }
    reject_windows_reparse(&metadata, path, "model-management-state")?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700)).map_err(
            |source| ModelManagementError::Io {
                path: path.to_path_buf(),
                source,
            },
        )?;
    }
    Ok(())
}

fn reject_windows_reparse(
    metadata: &std::fs::Metadata,
    path: &Path,
    model_id: &str,
) -> Result<(), ModelManagementError> {
    #[cfg(windows)]
    {
        use std::os::windows::fs::MetadataExt;
        const FILE_ATTRIBUTE_REPARSE_POINT: u32 = 0x400;
        if metadata.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT != 0 {
            return Err(ModelManagementError::UnsafeManagedPath {
                model_id: model_id.to_string(),
                path: path.to_path_buf(),
                reason: "model-management path is a Windows reparse point".into(),
            });
        }
    }
    #[cfg(not(windows))]
    let _ = (metadata, path, model_id);
    Ok(())
}

fn open_private_new(path: &Path) -> Result<File, ModelManagementError> {
    let mut options = OpenOptions::new();
    options.read(true).write(true).create_new(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.mode(0o600);
    }
    options
        .open(path)
        .map_err(|source| ModelManagementError::Io {
            path: path.to_path_buf(),
            source,
        })
}

fn open_existing_identity_lock(path: &Path) -> Result<File, ModelManagementError> {
    let metadata = std::fs::symlink_metadata(path).map_err(|source| ModelManagementError::Io {
        path: path.to_path_buf(),
        source,
    })?;
    if metadata.file_type().is_symlink() {
        return Err(ModelManagementError::InvalidState {
            path: path.to_path_buf(),
            message: "model-management lock sentinel must not be a symlink".into(),
        });
    }
    reject_windows_reparse(&metadata, path, "model-management-lock")?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt;
        if metadata.nlink() != 1 {
            return Err(ModelManagementError::InvalidState {
                path: path.to_path_buf(),
                message: "model-management lock sentinel must not be hard-linked".into(),
            });
        }
    }
    let mut options = OpenOptions::new();
    options.read(true).write(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.custom_flags(libc::O_NOFOLLOW);
    }
    #[cfg(windows)]
    {
        use std::os::windows::fs::OpenOptionsExt;
        const FILE_FLAG_OPEN_REPARSE_POINT: u32 = 0x0020_0000;
        options.custom_flags(FILE_FLAG_OPEN_REPARSE_POINT);
    }
    let file = options
        .open(path)
        .map_err(|source| ModelManagementError::Io {
            path: path.to_path_buf(),
            source,
        })?;
    let opened_metadata = file.metadata().map_err(|source| ModelManagementError::Io {
        path: path.to_path_buf(),
        source,
    })?;
    if !opened_metadata.is_file() {
        return Err(ModelManagementError::InvalidState {
            path: path.to_path_buf(),
            message: "model-management lock sentinel is not a regular file".into(),
        });
    }
    reject_windows_reparse(&opened_metadata, path, "model-management-lock")?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt;
        if opened_metadata.nlink() != 1 {
            return Err(ModelManagementError::InvalidState {
                path: path.to_path_buf(),
                message: "opened model-management lock sentinel is hard-linked".into(),
            });
        }
    }
    Ok(file)
}

fn read_private_state(path: &Path) -> std::io::Result<Vec<u8>> {
    let mut options = OpenOptions::new();
    options.read(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.custom_flags(libc::O_NOFOLLOW);
    }
    let file = options.open(path)?;
    let metadata = file.metadata()?;
    if !metadata.is_file() || metadata.len() > MAX_STATE_BYTES {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "model-management state must be a bounded regular file",
        ));
    }
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt;
        if metadata.nlink() != 1 {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "model-management state must not be hard-linked",
            ));
        }
    }
    let mut bytes = Vec::with_capacity(metadata.len() as usize);
    file.take(MAX_STATE_BYTES + 1).read_to_end(&mut bytes)?;
    if bytes.len() as u64 > MAX_STATE_BYTES {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "model-management state exceeds the size limit",
        ));
    }
    Ok(bytes)
}

fn write_private_json<T: Serialize>(path: &Path, value: &T) -> Result<(), ModelManagementError> {
    let _guard = state_mutation_lock()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let parent = path
        .parent()
        .expect("model-management record path has a parent");
    create_private_dir(parent)?;
    let tmp = (0..16)
        .map(|_| {
            parent.join(format!(
                ".{}.{}-{:032x}.tmp",
                std::process::id(),
                path.file_name()
                    .and_then(|name| name.to_str())
                    .unwrap_or("state"),
                rand::random::<u128>()
            ))
        })
        .find(|candidate| {
            std::fs::symlink_metadata(candidate)
                .is_err_and(|error| error.kind() == std::io::ErrorKind::NotFound)
        })
        .ok_or_else(|| ModelManagementError::InvalidState {
            path: parent.to_path_buf(),
            message: "could not allocate collision-resistant state staging file".into(),
        })?;
    let bytes =
        serde_json::to_vec_pretty(value).map_err(|error| ModelManagementError::InvalidState {
            path: path.to_path_buf(),
            message: error.to_string(),
        })?;
    let result = (|| {
        let mut file = open_private_new(&tmp)?;
        file.write_all(&bytes)
            .and_then(|_| file.sync_all())
            .map_err(|source| ModelManagementError::Io {
                path: tmp.clone(),
                source,
            })?;
        atomic_replace(&tmp, path).map_err(|source| ModelManagementError::Io {
            path: path.to_path_buf(),
            source,
        })?;
        car_secrets::harden_owner_only(path);
        sync_directory(parent).map_err(|source| ModelManagementError::Io {
            path: parent.to_path_buf(),
            source,
        })
    })();
    if result.is_err() {
        let _ = std::fs::remove_file(&tmp);
    }
    result
}

fn state_mutation_lock() -> &'static Mutex<()> {
    static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
    LOCK.get_or_init(|| Mutex::new(()))
}

#[cfg(not(windows))]
fn atomic_replace(source: &Path, destination: &Path) -> std::io::Result<()> {
    std::fs::rename(source, destination)
}

#[cfg(target_os = "macos")]
fn atomic_rename_noreplace(source: &Path, destination: &Path) -> std::io::Result<()> {
    use std::ffi::CString;
    use std::os::unix::ffi::OsStrExt;

    let source = CString::new(source.as_os_str().as_bytes())
        .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidInput, error))?;
    let destination = CString::new(destination.as_os_str().as_bytes())
        .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidInput, error))?;
    // SAFETY: both C strings are NUL-terminated and live for the call.
    let result =
        unsafe { libc::renamex_np(source.as_ptr(), destination.as_ptr(), libc::RENAME_EXCL) };
    if result == 0 {
        Ok(())
    } else {
        Err(std::io::Error::last_os_error())
    }
}

#[cfg(target_os = "linux")]
fn atomic_rename_noreplace(source: &Path, destination: &Path) -> std::io::Result<()> {
    use std::ffi::CString;
    use std::os::unix::ffi::OsStrExt;

    let source = CString::new(source.as_os_str().as_bytes())
        .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidInput, error))?;
    let destination = CString::new(destination.as_os_str().as_bytes())
        .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidInput, error))?;
    // SAFETY: both C strings are NUL-terminated and live for the call.
    let result = unsafe {
        libc::renameat2(
            libc::AT_FDCWD,
            source.as_ptr(),
            libc::AT_FDCWD,
            destination.as_ptr(),
            libc::RENAME_NOREPLACE,
        )
    };
    if result == 0 {
        Ok(())
    } else {
        Err(std::io::Error::last_os_error())
    }
}

#[cfg(windows)]
fn atomic_rename_noreplace(source: &Path, destination: &Path) -> std::io::Result<()> {
    use std::os::windows::ffi::OsStrExt;

    const MOVEFILE_WRITE_THROUGH: u32 = 0x8;
    #[link(name = "kernel32")]
    unsafe extern "system" {
        fn MoveFileExW(existing: *const u16, new: *const u16, flags: u32) -> i32;
    }
    let source = source
        .as_os_str()
        .encode_wide()
        .chain(Some(0))
        .collect::<Vec<_>>();
    let destination = destination
        .as_os_str()
        .encode_wide()
        .chain(Some(0))
        .collect::<Vec<_>>();
    // No REPLACE_EXISTING flag: a pre-existing destination fails closed.
    let moved = unsafe {
        MoveFileExW(
            source.as_ptr(),
            destination.as_ptr(),
            MOVEFILE_WRITE_THROUGH,
        )
    };
    if moved == 0 {
        Err(std::io::Error::last_os_error())
    } else {
        Ok(())
    }
}

#[cfg(not(any(target_os = "macos", target_os = "linux", windows)))]
fn atomic_rename_noreplace(_source: &Path, _destination: &Path) -> std::io::Result<()> {
    Err(std::io::Error::new(
        std::io::ErrorKind::Unsupported,
        "atomic no-replace publication is unavailable on this platform",
    ))
}

#[cfg(windows)]
fn atomic_replace(source: &Path, destination: &Path) -> std::io::Result<()> {
    use std::os::windows::ffi::OsStrExt;

    const MOVEFILE_REPLACE_EXISTING: u32 = 0x1;
    const MOVEFILE_WRITE_THROUGH: u32 = 0x8;
    #[link(name = "kernel32")]
    unsafe extern "system" {
        fn MoveFileExW(existing: *const u16, new: *const u16, flags: u32) -> i32;
    }
    let source = source
        .as_os_str()
        .encode_wide()
        .chain(Some(0))
        .collect::<Vec<_>>();
    let destination = destination
        .as_os_str()
        .encode_wide()
        .chain(Some(0))
        .collect::<Vec<_>>();
    // SAFETY: both pointers remain valid NUL-terminated UTF-16 buffers for
    // the duration of the call.
    let replaced = unsafe {
        MoveFileExW(
            source.as_ptr(),
            destination.as_ptr(),
            MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH,
        )
    };
    if replaced == 0 {
        Err(std::io::Error::last_os_error())
    } else {
        Ok(())
    }
}

#[cfg(unix)]
fn sync_directory(path: &Path) -> std::io::Result<()> {
    File::open(path)?.sync_all()
}

/// Object-bound deletion of a quarantined managed directory (macOS/Linux).
///
/// Everything after the first `open` is relative to descriptors CAR itself
/// opened: children are opened before they are trusted, symlinks are unlinked
/// and never followed, a change of device (and on Linux any mount crossing)
/// refuses the walk, entries are captured first and each is rechecked against
/// its captured identity immediately before its unlink, and a directory is
/// removed only when it is empty and still the object that was captured. POSIX
/// has no unlink-by-descriptor, so the one residual is the recheck → `unlinkat`
/// window on a captured name, the same window `std::fs::remove_dir_all` and
/// `rm -rf` accept; a name that appears after capture is never opened or
/// unlinked and makes its parent's `rmdir` fail closed instead.
#[cfg(any(target_os = "macos", target_os = "linux"))]
mod object_bound {
    use std::ffi::{CStr, CString};
    use std::fs::File;
    use std::os::fd::{AsRawFd, FromRawFd};
    use std::os::unix::ffi::OsStrExt;
    use std::os::unix::fs::MetadataExt;
    use std::path::Path;

    pub(super) const MAX_REMOVAL_DEPTH: usize = 32;

    const DIRECTORY_FLAGS: libc::c_int =
        libc::O_RDONLY | libc::O_DIRECTORY | libc::O_CLOEXEC | libc::O_NOFOLLOW;

    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
    pub(super) struct Identity {
        pub(super) device: u64,
        pub(super) inode: u64,
    }

    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
    pub(super) enum EntryKind {
        Directory,
        RegularFile,
        Symlink,
        Other,
    }

    pub(super) fn c_name(path: &Path) -> std::io::Result<CString> {
        CString::new(path.as_os_str().as_bytes())
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidInput, error))
    }

    /// Open a directory by path once, refusing to follow a symlink at the leaf.
    pub(super) fn open_directory(path: &Path) -> std::io::Result<File> {
        let name = c_name(path)?;
        // SAFETY: `name` is NUL-terminated and lives for the call.
        let raw = unsafe { libc::open(name.as_ptr(), DIRECTORY_FLAGS) };
        if raw < 0 {
            return Err(std::io::Error::last_os_error());
        }
        // SAFETY: `raw` is a newly-owned descriptor returned by `open`.
        Ok(unsafe { File::from_raw_fd(raw) })
    }

    /// Open `name` relative to `parent` as a directory. Fails with `ENOTDIR`
    /// for a regular file and `ELOOP` for a symlink, which the walker treats
    /// as "not a directory".
    #[cfg(target_os = "macos")]
    pub(super) fn open_child_directory(parent: &File, name: &CStr) -> std::io::Result<File> {
        // SAFETY: the parent descriptor is open and `name` is NUL-terminated.
        let raw = unsafe { libc::openat(parent.as_raw_fd(), name.as_ptr(), DIRECTORY_FLAGS) };
        if raw < 0 {
            return Err(std::io::Error::last_os_error());
        }
        // SAFETY: `raw` is a newly-owned descriptor returned by `openat`.
        Ok(unsafe { File::from_raw_fd(raw) })
    }

    /// Linux: `openat2` with `RESOLVE_BENEATH | RESOLVE_NO_SYMLINKS |
    /// RESOLVE_NO_XDEV` so the open can neither escape the parent, follow a
    /// symlink, nor cross a mount (bind mounts included, which share `st_dev`).
    #[cfg(target_os = "linux")]
    pub(super) fn open_child_directory(parent: &File, name: &CStr) -> std::io::Result<File> {
        // SAFETY: `open_how` is plain data; zero is a valid initial state.
        let mut how: libc::open_how = unsafe { std::mem::zeroed() };
        how.flags = DIRECTORY_FLAGS as u64;
        how.resolve = libc::RESOLVE_BENEATH | libc::RESOLVE_NO_SYMLINKS | libc::RESOLVE_NO_XDEV;
        // SAFETY: the parent descriptor is open, `name` is NUL-terminated, and
        // `how` is a valid `open_how` of the size passed.
        let raw = unsafe {
            libc::syscall(
                libc::SYS_openat2,
                parent.as_raw_fd(),
                name.as_ptr(),
                &how as *const libc::open_how,
                std::mem::size_of::<libc::open_how>(),
            )
        };
        if raw < 0 {
            return Err(std::io::Error::last_os_error());
        }
        // SAFETY: `raw` is a newly-owned descriptor returned by `openat2`.
        Ok(unsafe { File::from_raw_fd(raw as libc::c_int) })
    }

    /// A child that reports a different device than the root is on another
    /// filesystem; the walk never descends across one. Linux additionally
    /// refuses bind mounts through `RESOLVE_NO_XDEV`, which share `st_dev`.
    pub(super) fn same_filesystem(root: &Identity, child: &Identity) -> bool {
        root.device == child.device
    }

    pub(super) fn is_not_a_directory(error: &std::io::Error) -> bool {
        matches!(
            error.raw_os_error(),
            Some(libc::ENOTDIR) | Some(libc::ELOOP)
        )
    }

    pub(super) fn identity_of(file: &File) -> std::io::Result<Identity> {
        let metadata = file.metadata()?;
        Ok(Identity {
            device: metadata.dev(),
            inode: metadata.ino(),
        })
    }

    /// `fstatat(AT_SYMLINK_NOFOLLOW)` of `name` relative to `parent`.
    // `st_dev` / `st_ino` widths differ per target (macOS `dev_t` is `i32`,
    // Linux `dev_t` is `u64`), so the casts are needed on some targets and
    // redundant on others.
    #[allow(clippy::unnecessary_cast)]
    pub(super) fn stat_entry(parent: &File, name: &CStr) -> std::io::Result<(EntryKind, Identity)> {
        // SAFETY: `stat` is plain data; zero is a valid initial state.
        let mut stat: libc::stat = unsafe { std::mem::zeroed() };
        // SAFETY: the parent descriptor is open, `name` is NUL-terminated, and
        // `stat` is a valid out-pointer.
        let result = unsafe {
            libc::fstatat(
                parent.as_raw_fd(),
                name.as_ptr(),
                &mut stat,
                libc::AT_SYMLINK_NOFOLLOW,
            )
        };
        if result != 0 {
            return Err(std::io::Error::last_os_error());
        }
        let kind = match stat.st_mode & libc::S_IFMT {
            libc::S_IFDIR => EntryKind::Directory,
            libc::S_IFREG => EntryKind::RegularFile,
            libc::S_IFLNK => EntryKind::Symlink,
            _ => EntryKind::Other,
        };
        Ok((
            kind,
            Identity {
                device: stat.st_dev as u64,
                inode: stat.st_ino as u64,
            },
        ))
    }

    /// Enumerate `directory` through a duplicate descriptor so the retained
    /// one keeps its position; names are sorted for deterministic order.
    pub(super) fn list_names(directory: &File) -> std::io::Result<Vec<CString>> {
        let dot = c".";
        // SAFETY: the directory descriptor is open and `dot` is NUL-terminated.
        let raw = unsafe { libc::openat(directory.as_raw_fd(), dot.as_ptr(), DIRECTORY_FLAGS) };
        if raw < 0 {
            return Err(std::io::Error::last_os_error());
        }
        // SAFETY: `raw` is a newly-owned descriptor; `fdopendir` takes ownership.
        let stream = unsafe { libc::fdopendir(raw) };
        if stream.is_null() {
            let error = std::io::Error::last_os_error();
            // SAFETY: `fdopendir` failed, so the descriptor is still ours to close.
            unsafe {
                libc::close(raw);
            }
            return Err(error);
        }
        let mut names = Vec::new();
        let enumeration = loop {
            set_errno(0);
            // SAFETY: `stream` is a valid open directory stream.
            let entry = unsafe { libc::readdir(stream) };
            if entry.is_null() {
                let errno = errno();
                break if errno == 0 {
                    Ok(())
                } else {
                    Err(std::io::Error::from_raw_os_error(errno))
                };
            }
            // SAFETY: `entry` points at a valid dirent whose name is NUL-terminated.
            let bytes = unsafe { CStr::from_ptr((*entry).d_name.as_ptr()) }.to_bytes();
            if bytes != b"." && bytes != b".." {
                match CString::new(bytes) {
                    Ok(name) => names.push(name),
                    Err(error) => {
                        break Err(std::io::Error::new(std::io::ErrorKind::InvalidData, error))
                    }
                }
            }
        };
        // SAFETY: `stream` is a valid open directory stream owned by us.
        let closed = unsafe { libc::closedir(stream) };
        enumeration?;
        if closed < 0 {
            return Err(std::io::Error::last_os_error());
        }
        names.sort();
        Ok(names)
    }

    /// `unlinkat(parent, name, 0)`; a missing entry counts as already removed.
    pub(super) fn unlink_entry(parent: &File, name: &CStr) -> std::io::Result<()> {
        // SAFETY: the parent descriptor is open and `name` is NUL-terminated.
        let result = unsafe { libc::unlinkat(parent.as_raw_fd(), name.as_ptr(), 0) };
        if result == 0 {
            return Ok(());
        }
        let error = std::io::Error::last_os_error();
        if error.kind() == std::io::ErrorKind::NotFound {
            return Ok(());
        }
        Err(error)
    }

    /// `unlinkat(parent, name, AT_REMOVEDIR)`: only ever removes an empty
    /// directory. A missing entry counts as already removed.
    pub(super) fn remove_empty_directory(parent: &File, name: &CStr) -> std::io::Result<()> {
        // SAFETY: the parent descriptor is open and `name` is NUL-terminated.
        let result =
            unsafe { libc::unlinkat(parent.as_raw_fd(), name.as_ptr(), libc::AT_REMOVEDIR) };
        if result == 0 {
            return Ok(());
        }
        let error = std::io::Error::last_os_error();
        if error.kind() == std::io::ErrorKind::NotFound {
            return Ok(());
        }
        Err(error)
    }

    pub(super) fn is_not_empty(error: &std::io::Error) -> bool {
        matches!(
            error.raw_os_error(),
            Some(libc::ENOTEMPTY) | Some(libc::EEXIST)
        )
    }

    #[cfg(target_os = "macos")]
    fn errno() -> i32 {
        // SAFETY: `__error` returns the thread-local errno pointer.
        unsafe { *libc::__error() }
    }

    #[cfg(target_os = "macos")]
    fn set_errno(value: i32) {
        // SAFETY: `__error` returns the thread-local errno pointer.
        unsafe {
            *libc::__error() = value;
        }
    }

    #[cfg(target_os = "linux")]
    fn errno() -> i32 {
        // SAFETY: `__errno_location` returns the thread-local errno pointer.
        unsafe { *libc::__errno_location() }
    }

    #[cfg(target_os = "linux")]
    fn set_errno(value: i32) {
        // SAFETY: `__errno_location` returns the thread-local errno pointer.
        unsafe {
            *libc::__errno_location() = value;
        }
    }
}

#[cfg(any(target_os = "macos", target_os = "linux"))]
impl ModelManagementStore {
    /// Delete the quarantined directory `quarantine` whose pre-rename identity
    /// is `expected`. See `object_bound` for the guarantees.
    fn remove_quarantined_directory(
        &self,
        quarantine: &Path,
        expected: &ArtifactIdentity,
        model_id: &str,
    ) -> Result<(), ModelManagementError> {
        use object_bound::{EntryKind, Identity, MAX_REMOVAL_DEPTH};
        use std::ffi::CString;
        use std::fs::File;

        struct CapturedDirectory {
            file: File,
            name: CString,
            parent: usize,
            depth: usize,
            identity: Identity,
        }
        struct CapturedEntry {
            parent: usize,
            name: CString,
            identity: Identity,
        }

        let unsafe_path = |reason: &str| ModelManagementError::UnsafeManagedPath {
            model_id: model_id.to_string(),
            path: quarantine.to_path_buf(),
            reason: reason.into(),
        };
        let io_error = |source: std::io::Error| ModelManagementError::Io {
            path: quarantine.to_path_buf(),
            source,
        };

        #[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
        if let Some(hook) = &self.removal_hook {
            (hook.0)(RemovalPhase::BeforeRootOpen, quarantine);
        }

        // Bind the root: open it once, then trust only the descriptor. The
        // caller observed the quarantine moments ago, so its absence here means
        // something moved it; that is never a completed removal.
        let root = match object_bound::open_directory(quarantine) {
            Ok(root) => root,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                return Err(unsafe_path(QUARANTINE_VANISHED_REASON));
            }
            Err(source) => return Err(io_error(source)),
        };
        let root_identity = object_bound::identity_of(&root).map_err(io_error)?;
        if root_identity.device != expected.device || root_identity.inode != expected.inode {
            return Err(unsafe_path(
                "quarantined directory identity changed before deletion",
            ));
        }

        // Capture phase: open every child by descriptor before trusting it.
        let mut directories = vec![CapturedDirectory {
            file: root,
            name: CString::default(),
            parent: 0,
            depth: 0,
            identity: root_identity,
        }];
        let mut entries: Vec<CapturedEntry> = Vec::new();
        let mut pending = vec![0usize];
        while let Some(index) = pending.pop() {
            let depth = directories[index].depth;
            if depth > MAX_REMOVAL_DEPTH {
                return Err(unsafe_path(
                    "managed directory nesting exceeds the removal depth cap",
                ));
            }
            let names = object_bound::list_names(&directories[index].file).map_err(io_error)?;
            for name in names {
                match object_bound::open_child_directory(&directories[index].file, &name) {
                    Ok(child) => {
                        let identity = object_bound::identity_of(&child).map_err(io_error)?;
                        if !object_bound::same_filesystem(&root_identity, &identity) {
                            return Err(unsafe_path(
                                "managed directory crosses a filesystem boundary",
                            ));
                        }
                        directories.push(CapturedDirectory {
                            file: child,
                            name,
                            parent: index,
                            depth: depth + 1,
                            identity,
                        });
                        pending.push(directories.len() - 1);
                    }
                    Err(error) if object_bound::is_not_a_directory(&error) => {
                        let (kind, identity) =
                            object_bound::stat_entry(&directories[index].file, &name)
                                .map_err(io_error)?;
                        match kind {
                            EntryKind::RegularFile | EntryKind::Symlink => {
                                entries.push(CapturedEntry {
                                    parent: index,
                                    name,
                                    identity,
                                });
                            }
                            EntryKind::Directory | EntryKind::Other => {
                                return Err(unsafe_path(
                                    "managed directory contains an entry CAR cannot classify",
                                ));
                            }
                        }
                    }
                    Err(error) if error.raw_os_error() == Some(libc::EXDEV) => {
                        return Err(unsafe_path(
                            "managed directory crosses a filesystem boundary",
                        ));
                    }
                    Err(error)
                        if matches!(
                            error.raw_os_error(),
                            Some(libc::ENOSYS) | Some(libc::EINVAL) | Some(libc::E2BIG)
                        ) =>
                    {
                        return Err(unsafe_path(
                            "descriptor-bound directory open is unavailable on this kernel",
                        ));
                    }
                    Err(source) => return Err(io_error(source)),
                }
            }
        }

        #[cfg(all(test, any(target_os = "macos", target_os = "linux")))]
        if let Some(hook) = &self.removal_hook {
            (hook.0)(RemovalPhase::AfterCapture, quarantine);
        }

        // Delete phase: only captured entries, each rechecked against its
        // captured identity immediately before the unlink.
        for entry in &entries {
            let parent = &directories[entry.parent].file;
            match object_bound::stat_entry(parent, &entry.name) {
                Ok((_, identity)) if identity == entry.identity => {
                    object_bound::unlink_entry(parent, &entry.name).map_err(io_error)?;
                }
                Ok(_) => {
                    return Err(unsafe_path(
                        "managed directory entry identity changed during deletion",
                    ));
                }
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                Err(source) => return Err(io_error(source)),
            }
        }

        // Directories deepest first: every child index is greater than its
        // parent's, so popping from the end empties children before parents.
        while directories.len() > 1 {
            let CapturedDirectory {
                file,
                name,
                parent,
                identity,
                ..
            } = directories.pop().expect("non-empty");
            drop(file);
            let parent = &directories[parent].file;
            match object_bound::stat_entry(parent, &name) {
                Ok((EntryKind::Directory, actual)) if actual == identity => {
                    object_bound::remove_empty_directory(parent, &name).map_err(|error| {
                        if object_bound::is_not_empty(&error) {
                            unsafe_path("managed directory contains entries that were not captured")
                        } else {
                            io_error(error)
                        }
                    })?;
                }
                Ok(_) => {
                    return Err(unsafe_path(
                        "managed directory entry identity changed during deletion",
                    ));
                }
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                Err(source) => return Err(io_error(source)),
            }
        }

        // The root itself: relative to the validated models root, and only
        // if it is still the captured object and now empty.
        let CapturedDirectory { file: root, .. } = directories.pop().expect("root");
        drop(root);
        let models = object_bound::open_directory(&self.models_dir).map_err(|source| {
            ModelManagementError::Io {
                path: self.models_dir.clone(),
                source,
            }
        })?;
        let leaf = quarantine
            .file_name()
            .map(Path::new)
            .ok_or_else(|| unsafe_path("quarantine path has no file name"))?;
        let leaf = object_bound::c_name(leaf).map_err(io_error)?;
        match object_bound::stat_entry(&models, &leaf) {
            Ok((EntryKind::Directory, actual)) if actual == root_identity => {
                object_bound::remove_empty_directory(&models, &leaf).map_err(|error| {
                    if object_bound::is_not_empty(&error) {
                        unsafe_path("managed directory contains entries that were not captured")
                    } else {
                        io_error(error)
                    }
                })
            }
            Ok(_) => Err(unsafe_path(
                "quarantined directory identity changed during deletion",
            )),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(source) => Err(io_error(source)),
        }
    }
}

// Keep both cfg halves of `sync_directory` here, ABOVE `mod tests`. The
// non-unix half used to sit at the end of the file, where it was invisible on
// Unix (it compiles out) but tripped `clippy::items_after_test_module` on the
// Windows leg of `-D warnings`.
#[cfg(not(unix))]
fn sync_directory(_path: &Path) -> std::io::Result<()> {
    Ok(())
}

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

    // Invoked only by the parent process fixture below, never by a model backend.
    #[test]
    fn identity_initialization_child() {
        use std::io::{BufRead, Write};
        let Some(root) = std::env::var_os("CAR_IDENTITY_TEST_ROOT") else {
            return;
        };
        let root = PathBuf::from(root);
        let mut store = ModelManagementStore::new(root.join("state"), root.join("models"));
        if std::env::var("CAR_IDENTITY_TEST_ROLE").as_deref() == Ok("creator") {
            store.identity_init_hook = Some(|path| {
                let phase = std::env::var("CAR_IDENTITY_TEST_PHASE").unwrap();
                if path.parent().unwrap().file_name().unwrap() == phase.as_str() {
                    println!("IDENTITY_CREATED");
                    std::io::stdout().flush().unwrap();
                    let mut release = String::new();
                    std::io::stdin().lock().read_line(&mut release).unwrap();
                    assert_eq!(release.trim(), "release");
                }
            });
        } else {
            println!("READER_STARTED");
            std::io::stdout().flush().unwrap();
        }
        let _lease = store.acquire_lease("fixture/model").unwrap();
    }

    #[test]
    fn identity_initialization_serializes_processes_before_json_is_written() {
        use std::io::{BufRead, BufReader, Write};
        use std::process::{Command, Stdio};
        fn wait_marker(reader: &mut impl BufRead, marker: &str) {
            let mut line = String::new();
            loop {
                line.clear();
                assert_ne!(
                    reader.read_line(&mut line).unwrap(),
                    0,
                    "child exited before {marker}"
                );
                if line.contains(marker) {
                    return;
                }
            }
        }
        for phase in ["mutation", "activity"] {
            let root = tempfile::tempdir().unwrap();
            let spawn = |role: &str| {
                Command::new(std::env::current_exe().unwrap())
                    .args([
                        "--exact",
                        "model_management::tests::identity_initialization_child",
                        "--nocapture",
                    ])
                    .env("CAR_IDENTITY_TEST_ROOT", root.path())
                    .env("CAR_IDENTITY_TEST_ROLE", role)
                    .env("CAR_IDENTITY_TEST_PHASE", phase)
                    .stdin(Stdio::piped())
                    .stdout(Stdio::piped())
                    .stderr(Stdio::inherit())
                    .spawn()
                    .unwrap()
            };
            let mut creator = spawn("creator");
            let mut creator_out = BufReader::new(creator.stdout.take().unwrap());
            wait_marker(&mut creator_out, "IDENTITY_CREATED");
            let store =
                ModelManagementStore::new(root.path().join("state"), root.path().join("models"));
            let path = if phase == "mutation" {
                store.mutation_lock_path("fixture/model")
            } else {
                store.lease_path("fixture/model")
            };
            assert!(
                std::fs::read(&path).unwrap().is_empty(),
                "probe must hit the original partial-publication window"
            );
            let gate =
                open_existing_identity_lock(&path.parent().unwrap().join(".initialization.lock"))
                    .unwrap();
            assert!(
                gate.try_lock().is_err(),
                "creator must hold the initialization gate before publishing an empty identity"
            );
            let mut contender = spawn("reader");
            let mut contender_out = BufReader::new(contender.stdout.take().unwrap());
            wait_marker(&mut contender_out, "READER_STARTED");
            assert!(contender.try_wait().unwrap().is_none());
            writeln!(creator.stdin.take().unwrap(), "release").unwrap();
            assert!(creator.wait().unwrap().success());
            assert!(contender.wait().unwrap().success());
            let before = std::fs::metadata(&path).unwrap();
            drop(store.acquire_lease("fixture/model").unwrap());
            #[cfg(unix)]
            {
                use std::os::unix::fs::MetadataExt;
                assert_eq!(before.ino(), std::fs::metadata(&path).unwrap().ino());
            }
            #[cfg(not(unix))]
            let _ = before;
        }
    }

    #[test]
    fn identity_initialization_preserves_persistent_corruption_refusal() {
        for phase in ["mutation", "activity"] {
            for corrupt in [b"".as_slice(), b"not json".as_slice()] {
                let root = tempfile::tempdir().unwrap();
                let store = ModelManagementStore::new(
                    root.path().join("state"),
                    root.path().join("models"),
                );
                drop(store.acquire_lease("fixture/model").unwrap());
                let path = if phase == "mutation" {
                    store.mutation_lock_path("fixture/model")
                } else {
                    store.lease_path("fixture/model")
                };
                std::fs::write(&path, corrupt).unwrap();
                assert!(matches!(
                    store.acquire_lease("fixture/model"),
                    Err(ModelManagementError::InvalidState { .. })
                ));
                assert_eq!(std::fs::read(&path).unwrap(), corrupt);
            }
        }
    }

    #[cfg(unix)]
    #[test]
    fn identity_initialization_rejects_linked_gates_and_identity_files() {
        for phase in ["mutation", "activity"] {
            for gate in [false, true] {
                for symlink in [false, true] {
                    let root = tempfile::tempdir().unwrap();
                    let store = ModelManagementStore::new(
                        root.path().join("state"),
                        root.path().join("models"),
                    );
                    let identity = if phase == "mutation" {
                        store.mutation_lock_path("fixture/model")
                    } else {
                        store.lease_path("fixture/model")
                    };
                    create_private_dir(identity.parent().unwrap()).unwrap();
                    let path = if gate {
                        identity.parent().unwrap().join(".initialization.lock")
                    } else {
                        identity
                    };
                    let outside = root.path().join("outside");
                    std::fs::write(&outside, b"untouched").unwrap();
                    if symlink {
                        std::os::unix::fs::symlink(&outside, &path).unwrap();
                    } else {
                        std::fs::hard_link(&outside, &path).unwrap();
                    }
                    assert!(matches!(
                        store.acquire_lease("fixture/model"),
                        Err(ModelManagementError::InvalidState { .. })
                    ));
                    assert_eq!(std::fs::read(&outside).unwrap(), b"untouched");
                }
            }
        }
    }

    fn directory_receipt(model_id: &str, managed_path: PathBuf) -> InstallReceipt {
        InstallReceipt {
            model_id: model_id.into(),
            managed_path,
            artifact_kind: ManagedArtifactKind::Directory,
            source_model_id: model_id.into(),
            source_revision: None,
            creation_generation: 1,
            shared_cache_references: vec![],
            adopted: false,
        }
    }

    #[test]
    fn raw_receipt_identity_is_private_and_collision_fails_closed() {
        let root = tempfile::tempdir().unwrap();
        let models = root.path().join("models");
        std::fs::create_dir_all(&models).unwrap();
        let store = ModelManagementStore::new(root.path().join("state"), models.clone());
        let receipt = directory_receipt("org/model", models.join("Model"));
        store.write_receipt(&receipt).unwrap();
        assert_eq!(store.load_receipt("org/model").unwrap(), Some(receipt));

        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            assert_eq!(
                std::fs::metadata(store.receipt_path("org/model"))
                    .unwrap()
                    .permissions()
                    .mode()
                    & 0o077,
                0
            );
        }

        let collision = directory_receipt("different/model", models.join("Other"));
        std::fs::write(
            store.receipt_path("org/model"),
            serde_json::to_vec(&collision).unwrap(),
        )
        .unwrap();
        assert!(matches!(
            store.load_receipt("org/model"),
            Err(ModelManagementError::IdentityMismatch { .. })
        ));
    }

    #[test]
    fn journaled_removal_rejects_leaf_identity_swap() {
        let root = tempfile::tempdir().unwrap();
        let models = root.path().join("models");
        std::fs::create_dir_all(&models).unwrap();
        let store = ModelManagementStore::new(root.path().join("state"), models.clone());
        let managed = models.join("Managed");
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("weights"), b"original").unwrap();
        let receipt = directory_receipt("org/model", managed.clone());
        store.write_receipt(&receipt).unwrap();
        let journal = RemovalJournal {
            model_id: "org/model".into(),
            removal_generation: 2,
            artifact_identity: artifact_identity(&managed, "org/model").unwrap(),
            receipt,
            quarantine_path: store.allocate_quarantine("org/model").unwrap(),
        };
        write_private_json(&store.removal_journal_path("org/model"), &journal).unwrap();
        let displaced = models.join("displaced-original");
        std::fs::rename(&managed, &displaced).unwrap();
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("sentinel"), b"replacement").unwrap();

        assert!(matches!(
            store.begin_mutation("org/model").unwrap().remove(2),
            Err(ModelManagementError::UnsafeManagedPath { .. })
        ));
        assert_eq!(
            std::fs::read(managed.join("sentinel")).unwrap(),
            b"replacement"
        );
    }

    #[test]
    fn journaled_removal_rejects_models_parent_swap() {
        let root = tempfile::tempdir().unwrap();
        let models = root.path().join("models");
        std::fs::create_dir_all(&models).unwrap();
        let store = ModelManagementStore::new(root.path().join("state"), models.clone());
        let managed = models.join("Managed");
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("weights"), b"original").unwrap();
        let receipt = directory_receipt("org/model", managed.clone());
        store.write_receipt(&receipt).unwrap();
        let journal = RemovalJournal {
            model_id: "org/model".into(),
            removal_generation: 2,
            artifact_identity: artifact_identity(&managed, "org/model").unwrap(),
            receipt,
            quarantine_path: store.allocate_quarantine("org/model").unwrap(),
        };
        write_private_json(&store.removal_journal_path("org/model"), &journal).unwrap();
        std::fs::rename(&models, root.path().join("models-original")).unwrap();
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("sentinel"), b"replacement-parent").unwrap();

        assert!(matches!(
            store.begin_mutation("org/model").unwrap().remove(2),
            Err(ModelManagementError::UnsafeManagedPath { .. })
        ));
        assert_eq!(
            std::fs::read(managed.join("sentinel")).unwrap(),
            b"replacement-parent"
        );
    }

    #[cfg(not(any(target_os = "macos", target_os = "linux")))]
    #[test]
    fn directory_receipts_fail_closed_without_recursive_cleanup() {
        let root = tempfile::tempdir().unwrap();
        let models = root.path().join("models");
        std::fs::create_dir_all(&models).unwrap();
        let store = ModelManagementStore::new(root.path().join("state"), models.clone());
        let managed = models.join("Managed");
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("shared-sentinel"), b"preserve").unwrap();
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        assert!(!store.can_remove("org/model").unwrap());
        let error = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap_err();
        assert!(matches!(
            error,
            ModelManagementError::UnsafeManagedPath { .. }
        ));
        assert_eq!(
            std::fs::read(managed.join("shared-sentinel")).unwrap(),
            b"preserve"
        );
    }

    #[test]
    fn failed_directory_staging_is_left_for_object_bound_cleanup() {
        let root = tempfile::tempdir().unwrap();
        let models = root.path().join("models");
        std::fs::create_dir_all(&models).unwrap();
        let store = ModelManagementStore::new(root.path().join("state"), models);
        let staging = store.create_install_staging("org/model").unwrap();
        std::fs::write(staging.join("shared-sentinel"), b"preserve").unwrap();

        store.discard_install_staging(&staging);

        assert_eq!(
            std::fs::read(staging.join("shared-sentinel")).unwrap(),
            b"preserve"
        );
    }

    #[cfg(unix)]
    #[test]
    fn completed_symlink_removal_retry_is_idempotent() {
        use std::os::unix::fs::symlink;

        let root = tempfile::tempdir().unwrap();
        let models = root.path().join("models");
        let shared = root.path().join("shared");
        std::fs::create_dir_all(&models).unwrap();
        std::fs::create_dir_all(&shared).unwrap();
        std::fs::write(shared.join("sentinel"), b"preserve").unwrap();
        let managed = models.join("Managed");
        symlink(&shared, &managed).unwrap();
        let store = ModelManagementStore::new(root.path().join("state"), models);
        let receipt = InstallReceipt {
            model_id: "org/model".into(),
            managed_path: managed,
            artifact_kind: ManagedArtifactKind::Symlink,
            source_model_id: "org/model".into(),
            source_revision: None,
            creation_generation: 1,
            shared_cache_references: vec![shared.canonicalize().unwrap()],
            adopted: false,
        };
        store.write_receipt(&receipt).unwrap();

        let first = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap();
        let retry = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(3)
            .unwrap();
        assert_eq!(retry, first);
        assert!(shared.join("sentinel").exists());
    }

    #[test]
    fn pulled_directory_publish_before_receipt_resumes_from_install_intent() {
        let root = tempfile::tempdir().unwrap();
        let models = root.path().join("models");
        std::fs::create_dir_all(&models).unwrap();
        let store = ModelManagementStore::new(root.path().join("state"), models.clone());
        let staging = store.create_install_staging("org/model").unwrap();
        std::fs::write(staging.join("weights"), b"owned").unwrap();
        let receipt = store
            .install_receipt_for_publication(
                "org/model",
                "org/model",
                1,
                false,
                models.join("Managed"),
                ManagedArtifactKind::Directory,
                &staging,
            )
            .unwrap();
        store.begin_install_intent(receipt, Some(&staging)).unwrap();
        store
            .publish_install_staging("org/model", &staging, "Managed")
            .unwrap();
        assert!(store.load_receipt("org/model").unwrap().is_none());

        let resumed = store.resume_install_intent("org/model").unwrap().unwrap();
        assert_eq!(resumed.managed_path, models.join("Managed"));
        assert!(store.load_receipt("org/model").unwrap().is_some());
        #[cfg(any(target_os = "macos", target_os = "linux"))]
        assert!(store.can_remove("org/model").unwrap());
        #[cfg(not(any(target_os = "macos", target_os = "linux")))]
        assert!(!store.can_remove("org/model").unwrap());
    }

    #[cfg(unix)]
    #[test]
    fn adopted_projection_publish_before_receipt_resumes_from_install_intent() {
        let root = tempfile::tempdir().unwrap();
        let models = root.path().join("models");
        let source = root.path().join("hand-installed");
        std::fs::create_dir_all(&models).unwrap();
        std::fs::create_dir_all(&source).unwrap();
        std::fs::write(source.join("sentinel"), b"preserve").unwrap();
        let store = ModelManagementStore::new(root.path().join("state"), models);
        let managed = store.adopted_projection_path("org/model").unwrap();
        let receipt = store
            .install_receipt_for_publication(
                "org/model",
                "org/model",
                1,
                true,
                managed.clone(),
                ManagedArtifactKind::Symlink,
                &source,
            )
            .unwrap();
        store.begin_install_intent(receipt, None).unwrap();
        store
            .materialize_adopted_projection("org/model", &source)
            .unwrap();
        assert!(store.load_receipt("org/model").unwrap().is_none());

        let resumed = store.resume_install_intent("org/model").unwrap().unwrap();
        assert_eq!(resumed.managed_path, managed);
        assert!(store.can_remove("org/model").unwrap());
        assert!(source.join("sentinel").exists());
    }

    #[cfg(unix)]
    #[test]
    fn install_intent_never_claims_a_preexisting_unreceipted_leaf() {
        use std::os::unix::fs::symlink;

        let root = tempfile::tempdir().unwrap();
        let models = root.path().join("models");
        let source = root.path().join("shared");
        std::fs::create_dir_all(&models).unwrap();
        std::fs::create_dir_all(&source).unwrap();
        let managed = models.join("Managed");
        symlink(&source, &managed).unwrap();
        let store = ModelManagementStore::new(root.path().join("state"), models);
        let receipt = store
            .install_receipt_for_publication(
                "org/model",
                "org/model",
                1,
                false,
                managed.clone(),
                ManagedArtifactKind::Symlink,
                &source,
            )
            .unwrap();

        let error = store.begin_install_intent(receipt, None).unwrap_err();
        assert!(matches!(
            error,
            ModelManagementError::UnsafeManagedPath { .. }
        ));
        assert!(managed.exists());
        assert!(store.load_receipt("org/model").unwrap().is_none());
    }

    #[test]
    fn forged_install_journal_paths_fail_before_filesystem_mutation() {
        let root = tempfile::tempdir().unwrap();
        let models = root.path().join("models");
        std::fs::create_dir_all(&models).unwrap();
        let outside = root.path().join("outside");
        std::fs::create_dir_all(&outside).unwrap();
        std::fs::write(outside.join("sentinel"), b"preserve").unwrap();
        let store = ModelManagementStore::new(root.path().join("state"), models.clone());

        let forged_managed = InstallJournal {
            model_id: "org/model".into(),
            receipt: directory_receipt("org/model", outside.clone()),
            staging_path: None,
            staging_identity: None,
        };
        write_private_json(&store.install_journal_path("org/model"), &forged_managed).unwrap();
        assert!(matches!(
            store.resume_install_intent("org/model"),
            Err(ModelManagementError::UnsafeManagedPath { .. })
        ));
        assert!(outside.join("sentinel").exists());

        let forged_staging = InstallJournal {
            model_id: "org/other".into(),
            receipt: directory_receipt("org/other", models.join("Other")),
            staging_path: Some(outside.clone()),
            staging_identity: Some(artifact_identity(&outside, "org/other").unwrap()),
        };
        write_private_json(&store.install_journal_path("org/other"), &forged_staging).unwrap();
        assert!(matches!(
            store.resume_install_intent("org/other"),
            Err(ModelManagementError::UnsafeManagedPath { .. })
        ));
        assert!(outside.join("sentinel").exists());
    }

    #[cfg(unix)]
    #[test]
    fn startup_ignores_noncanonical_removal_journal_without_touching_valid_model() {
        use std::os::unix::fs::symlink;

        let root = tempfile::tempdir().unwrap();
        let state = root.path().join("state");
        let models = root.path().join("models");
        let shared = root.path().join("shared");
        let outside = root.path().join("outside");
        std::fs::create_dir_all(&models).unwrap();
        std::fs::create_dir_all(&shared).unwrap();
        std::fs::create_dir_all(&outside).unwrap();
        std::fs::write(shared.join("sentinel"), b"shared").unwrap();
        std::fs::write(outside.join("sentinel"), b"outside").unwrap();
        let managed = models.join("Managed");
        symlink(&shared, &managed).unwrap();
        let store = ModelManagementStore::new(state.clone(), models.clone());
        let legitimate = InstallReceipt {
            model_id: "org/model".into(),
            managed_path: managed.clone(),
            artifact_kind: ManagedArtifactKind::Symlink,
            source_model_id: "org/model".into(),
            source_revision: None,
            creation_generation: 1,
            shared_cache_references: vec![shared.canonicalize().unwrap()],
            adopted: false,
        };
        store.write_receipt(&legitimate).unwrap();
        create_private_dir(&store.removals_root).unwrap();
        let forged = RemovalJournal {
            model_id: "org/model".into(),
            removal_generation: 99,
            receipt: InstallReceipt {
                managed_path: outside.join("missing-leaf"),
                ..legitimate
            },
            quarantine_path: models.join(".car-remove-forged"),
            artifact_identity: ArtifactIdentity {
                kind: ManagedArtifactKind::Symlink,
                device: 1,
                inode: 1,
                file_len: 0,
            },
        };
        write_private_json(&store.removals_root.join("junk.json"), &forged).unwrap();
        drop(store);

        let recovered = ModelManagementStore::new(state, models);
        assert!(std::fs::symlink_metadata(&managed).is_ok());
        assert!(recovered.car_enabled("org/model").unwrap());
        assert_eq!(std::fs::read(shared.join("sentinel")).unwrap(), b"shared");
        assert_eq!(std::fs::read(outside.join("sentinel")).unwrap(), b"outside");
    }

    // try_clone retains the same open-file description as an inherited fork
    // descriptor, without forking a multithreaded test process.
    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn mutation_guard_releases_lock_even_with_an_inherited_descriptor() {
        let (_root, _models, store) = supported_platform_store();
        let guard = store.begin_mutation("org/model").unwrap();
        let inherited = guard._file.try_clone().unwrap();
        assert!(matches!(
            store.begin_mutation("org/model"),
            Err(ModelManagementError::MutationInProgress { .. })
        ));
        drop(guard);
        let replacement = store.begin_mutation("org/model").unwrap();
        // Releasing the old duplicate must not unlock the replacement owner.
        drop(inherited);
        assert!(matches!(
            store.begin_mutation("org/model"),
            Err(ModelManagementError::MutationInProgress { .. })
        ));
        drop(replacement);
        assert!(store.begin_mutation("org/model").is_ok());
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn lease_and_idle_guards_release_despite_inherited_descriptors() {
        let (_root, _models, store) = supported_platform_store();
        let lease = store.acquire_lease("org/model").unwrap();
        let inherited = lease._file.try_clone().unwrap();
        assert!(store.model_in_use("org/model").unwrap());
        drop(lease);
        assert!(!store.model_in_use("org/model").unwrap());
        let idle = store.lock_idle_activity("org/model").unwrap();
        let inherited_idle = idle.try_clone().unwrap();
        assert!(store.model_in_use("org/model").unwrap());
        drop(idle);
        assert!(!store.model_in_use("org/model").unwrap());
        drop((inherited, inherited_idle));
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn temporary_owner_releases_initialization_gate_despite_inherited_descriptor() {
        let (_root, _models, store) = supported_platform_store();
        let identity = store
            .open_identity_lock(&store.mutation_lock_path("org/model"), "org/model")
            .unwrap();
        let path = store
            .mutation_lock_path("org/model")
            .parent()
            .unwrap()
            .join(".initialization.lock");
        let gate = open_existing_identity_lock(&path).unwrap();
        gate.lock().unwrap();
        let gate = OwnedFileLock::new(gate);
        let inherited = gate.try_clone().unwrap();
        let contender = open_existing_identity_lock(&path).unwrap();
        assert!(contender.try_lock().is_err());
        drop(gate);
        contender.try_lock().unwrap();
        contender.unlock().unwrap();
        drop((identity, inherited));
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    fn supported_platform_store() -> (tempfile::TempDir, PathBuf, ModelManagementStore) {
        let root = tempfile::tempdir().unwrap();
        let models = root.path().join("models");
        std::fs::create_dir_all(&models).unwrap();
        let store = ModelManagementStore::new(root.path().join("state"), models.clone());
        (root, models, store)
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    fn no_quarantine_left(models: &Path) -> bool {
        std::fs::read_dir(models).unwrap().all(|entry| {
            !entry
                .unwrap()
                .file_name()
                .to_string_lossy()
                .starts_with(".car-remove-")
        })
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    const MAX_REMOVAL_DEPTH_FOR_TESTS: usize = super::object_bound::MAX_REMOVAL_DEPTH;

    /// Unwraps a fail-closed refusal to its reason. Every race test funnels
    /// through here so no single site can quietly weaken into accepting a
    /// different error variant.
    #[cfg(any(target_os = "macos", target_os = "linux"))]
    fn unsafe_managed_reason(error: &ModelManagementError) -> &str {
        match error {
            ModelManagementError::UnsafeManagedPath { reason, .. } => reason,
            other => panic!("expected UnsafeManagedPath, got {other:?}"),
        }
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn directory_receipt_is_removable_and_leaves_a_tombstone() {
        let (_root, models, store) = supported_platform_store();
        let managed = models.join("Managed");
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("weights"), b"owned").unwrap();
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        assert!(store.can_remove("org/model").unwrap());
        let result = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap();

        assert_eq!(result.artifact_kind, ManagedArtifactKind::Directory);
        assert!(std::fs::symlink_metadata(&managed).is_err());
        assert!(store.load_receipt("org/model").unwrap().is_none());
        let tombstone = store.load_tombstone("org/model").unwrap().unwrap();
        assert_eq!(
            tombstone.artifact_kind,
            Some(ManagedArtifactKind::Directory)
        );
        assert_eq!(tombstone.removal_generation, 2);
        assert!(no_quarantine_left(&models));
        assert!(store.load_removal_journal("org/model").unwrap().is_none());
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn directory_removal_preserves_shared_cache_symlink_targets() {
        use std::os::unix::fs::symlink;

        let (root, models, store) = supported_platform_store();
        let shared = root.path().join("shared-cache");
        std::fs::create_dir_all(&shared).unwrap();
        std::fs::write(shared.join("blob"), b"shared-bytes").unwrap();
        let managed = models.join("Managed");
        let nested = managed.join("a").join("b").join("c");
        std::fs::create_dir_all(&nested).unwrap();
        std::fs::write(managed.join("config.json"), b"{}").unwrap();
        std::fs::write(nested.join("weights"), b"owned").unwrap();
        symlink(shared.join("blob"), managed.join("blob.safetensors")).unwrap();
        let blob = shared.join("blob").canonicalize().unwrap();
        let mut receipt = directory_receipt("org/model", managed.clone());
        receipt.shared_cache_references = vec![blob.clone()];
        store.write_receipt(&receipt).unwrap();

        let result = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap();

        assert!(std::fs::symlink_metadata(&managed).is_err());
        assert_eq!(std::fs::read(&blob).unwrap(), b"shared-bytes");
        assert_eq!(std::fs::read_dir(&shared).unwrap().count(), 1);
        assert_eq!(result.preserved_shared_cache_references, vec![blob]);
        assert!(no_quarantine_left(&models));
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn directory_removal_resumes_after_a_partial_quarantine_delete() {
        let (_root, models, store) = supported_platform_store();
        let managed = models.join("Managed");
        std::fs::create_dir_all(managed.join("sub")).unwrap();
        std::fs::write(managed.join("weights"), b"owned").unwrap();
        std::fs::write(managed.join("sub").join("more"), b"owned").unwrap();
        let receipt = directory_receipt("org/model", managed.clone());
        store.write_receipt(&receipt).unwrap();
        let quarantine = store.allocate_quarantine("org/model").unwrap();
        let journal = RemovalJournal {
            model_id: "org/model".into(),
            removal_generation: 2,
            artifact_identity: artifact_identity(&managed, "org/model").unwrap(),
            receipt,
            quarantine_path: quarantine.clone(),
        };
        write_private_json(&store.removal_journal_path("org/model"), &journal).unwrap();
        // A crash after the rename and after some entries were already deleted.
        std::fs::rename(&managed, &quarantine).unwrap();
        std::fs::remove_file(quarantine.join("weights")).unwrap();

        let resumed = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap();
        assert_eq!(resumed.artifact_kind, ManagedArtifactKind::Directory);
        assert!(std::fs::symlink_metadata(&quarantine).is_err());
        assert!(no_quarantine_left(&models));

        // A third request replays the durable tombstone result. The generation
        // it reports must stay the one that actually performed the removal (2),
        // not the generation this caller asked for (3), and the preserved
        // reference list must replay empty as the receipt recorded it.
        let replayed = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(3)
            .unwrap();
        assert_eq!(replayed.artifact_kind, ManagedArtifactKind::Directory);
        let tombstone = store.load_tombstone("org/model").unwrap().unwrap();
        assert_eq!(tombstone.removal_generation, 2);
        assert!(replayed.preserved_shared_cache_references.is_empty());
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn directory_removal_refuses_trees_deeper_than_the_cap() {
        let (_root, models, store) = supported_platform_store();
        let managed = models.join("Managed");
        let mut deep = managed.clone();
        for level in 0..(MAX_REMOVAL_DEPTH_FOR_TESTS + 1) {
            deep = deep.join(format!("level{level}"));
        }
        std::fs::create_dir_all(&deep).unwrap();
        std::fs::write(deep.join("sentinel"), b"deep").unwrap();
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        let error = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap_err();

        let reason = unsafe_managed_reason(&error);
        assert!(reason.contains("depth"), "unexpected reason: {reason}");
        // The leaf was already detached into its quarantine; nothing in it was deleted.
        let journal = store.load_removal_journal("org/model").unwrap().unwrap();
        let quarantined_leaf = deep
            .strip_prefix(&managed)
            .map(|rest| journal.quarantine_path.join(rest))
            .unwrap();
        assert_eq!(
            std::fs::read(quarantined_leaf.join("sentinel")).unwrap(),
            b"deep"
        );
    }

    // Deliberately ungated: `matches_for_removal` is pure struct logic that
    // compiles on every target, so narrowing this to macOS/Linux like its
    // siblings would silently drop Windows coverage.
    #[test]
    fn removal_identity_ignores_length_for_directories_only() {
        let directory = ArtifactIdentity {
            kind: ManagedArtifactKind::Directory,
            device: 7,
            inode: 42,
            file_len: 96,
        };
        let shrunk = ArtifactIdentity {
            file_len: 32,
            ..directory.clone()
        };
        assert!(directory.matches_for_removal(&shrunk));
        assert!(!directory.matches_for_removal(&ArtifactIdentity {
            inode: 43,
            ..directory.clone()
        }));
        assert!(!directory.matches_for_removal(&ArtifactIdentity {
            device: 8,
            ..directory.clone()
        }));

        let file = ArtifactIdentity {
            kind: ManagedArtifactKind::File,
            device: 7,
            inode: 42,
            file_len: 96,
        };
        assert!(file.matches_for_removal(&file));
        assert!(!file.matches_for_removal(&ArtifactIdentity {
            file_len: 95,
            ..file.clone()
        }));
        assert!(!file.matches_for_removal(&ArtifactIdentity {
            kind: ManagedArtifactKind::Symlink,
            ..file.clone()
        }));
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn same_filesystem_predicate_compares_devices_only() {
        use super::object_bound::{same_filesystem, Identity};
        let root = Identity {
            device: 5,
            inode: 1,
        };
        assert!(same_filesystem(
            &root,
            &Identity {
                device: 5,
                inode: 999
            }
        ));
        assert!(!same_filesystem(
            &root,
            &Identity {
                device: 6,
                inode: 1
            }
        ));
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn directory_removal_refuses_entries_added_after_capture() {
        let (_root, models, store) = supported_platform_store();
        let managed = models.join("Managed");
        std::fs::create_dir_all(managed.join("sub")).unwrap();
        std::fs::write(managed.join("weights"), b"owned").unwrap();
        std::fs::write(managed.join("sub").join("more"), b"owned").unwrap();
        let store = store.with_removal_hook(|phase, quarantine| {
            if phase == RemovalPhase::AfterCapture {
                std::fs::write(quarantine.join("sub").join("late"), b"not captured").unwrap();
            }
        });
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        let error = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap_err();

        let reason = unsafe_managed_reason(&error);
        assert!(
            reason.contains("not captured"),
            "unexpected reason: {reason}"
        );
        let journal = store.load_removal_journal("org/model").unwrap().unwrap();
        let late = journal.quarantine_path.join("sub").join("late");
        assert_eq!(std::fs::read(&late).unwrap(), b"not captured");

        // Once the foreign entry is gone the journaled removal completes.
        std::fs::remove_file(&late).unwrap();
        let store = store.with_removal_hook(|_, _| {});
        let resumed = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap();
        assert_eq!(resumed.artifact_kind, ManagedArtifactKind::Directory);
        assert!(no_quarantine_left(&models));
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn directory_removal_refuses_root_swap_after_rename() {
        let (_root, models, store) = supported_platform_store();
        let managed = models.join("Managed");
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("weights"), b"owned").unwrap();
        let displaced = models.join("displaced");
        let store = store.with_removal_hook(move |phase, quarantine| {
            if phase == RemovalPhase::AfterRename {
                std::fs::rename(quarantine, &displaced).unwrap();
                std::fs::create_dir_all(quarantine).unwrap();
                std::fs::write(quarantine.join("sentinel"), b"replacement").unwrap();
            }
        });
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        let error = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap_err();

        let reason = unsafe_managed_reason(&error);
        assert!(
            reason.contains("identity changed"),
            "unexpected reason: {reason}"
        );
        let journal = store.load_removal_journal("org/model").unwrap().unwrap();
        assert_eq!(
            std::fs::read(journal.quarantine_path.join("sentinel")).unwrap(),
            b"replacement"
        );
        assert_eq!(
            std::fs::read(models.join("displaced").join("weights")).unwrap(),
            b"owned"
        );
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn directory_symlink_planted_after_rename_is_unlinked_not_followed() {
        use std::os::unix::fs::symlink;

        let (root, models, store) = supported_platform_store();
        let outside = root.path().join("outside");
        std::fs::create_dir_all(&outside).unwrap();
        std::fs::write(outside.join("sentinel"), b"outside").unwrap();
        let managed = models.join("Managed");
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("weights"), b"owned").unwrap();
        let target = outside.clone();
        let store = store.with_removal_hook(move |phase, quarantine| {
            if phase == RemovalPhase::AfterRename {
                symlink(&target, quarantine.join("escape")).unwrap();
            }
        });
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        let result = store.begin_mutation("org/model").unwrap().remove(2);

        assert_eq!(
            std::fs::read(outside.join("sentinel")).map_err(|e| e.to_string()),
            Ok(b"outside".to_vec()),
            "outside sentinel was deleted; removal result: {result:?}"
        );
        assert!(result.is_ok(), "{result:?}");
        assert!(std::fs::symlink_metadata(&managed).is_err());
        assert!(no_quarantine_left(&models));
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn directory_removal_refuses_root_swap_between_path_check_and_descriptor_open() {
        let (_root, models, store) = supported_platform_store();
        let managed = models.join("Managed");
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("weights"), b"owned").unwrap();
        let displaced = models.join("displaced");
        let store = store.with_removal_hook(move |phase, quarantine| {
            if phase == RemovalPhase::BeforeRootOpen {
                std::fs::rename(quarantine, &displaced).unwrap();
                std::fs::create_dir_all(quarantine).unwrap();
                std::fs::write(quarantine.join("sentinel"), b"replacement").unwrap();
            }
        });
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        let error = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap_err();

        let reason = unsafe_managed_reason(&error);
        assert_eq!(
            reason,
            "quarantined directory identity changed before deletion"
        );
        let journal = store.load_removal_journal("org/model").unwrap().unwrap();
        assert_eq!(
            std::fs::read(journal.quarantine_path.join("sentinel")).unwrap(),
            b"replacement"
        );
        assert_eq!(
            std::fs::read(models.join("displaced").join("weights")).unwrap(),
            b"owned"
        );
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn directory_removal_refuses_when_the_quarantine_vanishes_after_our_own_rename() {
        let (_root, models, store) = supported_platform_store();
        let managed = models.join("Managed");
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("weights"), b"owned").unwrap();
        let managed_for_hook = managed.clone();
        let store = store.with_removal_hook(move |phase, quarantine| {
            if phase == RemovalPhase::AfterRename {
                // A racer moves the quarantine straight back to its managed name.
                std::fs::rename(quarantine, &managed_for_hook).unwrap();
            }
        });
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        let error = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap_err();

        let reason = unsafe_managed_reason(&error);
        assert!(reason.contains("vanished"), "unexpected reason: {reason}");
        assert_eq!(std::fs::read(managed.join("weights")).unwrap(), b"owned");
        assert!(store.load_receipt("org/model").unwrap().is_some());
        assert!(store.load_removal_journal("org/model").unwrap().is_some());
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn first_attempt_vanish_refusal_names_the_retry_that_completes_the_removal() {
        let (_root, models, store) = supported_platform_store();
        let managed = models.join("Managed");
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("weights"), b"owned").unwrap();
        let managed_for_hook = managed.clone();
        let sabotaged = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let store = store.with_removal_hook(move |phase, quarantine| {
            // Sabotage the first attempt only: a racer moves the quarantine back
            // to its managed name right after CAR's own rename. The retry below
            // then runs against an untouched store.
            if phase == RemovalPhase::AfterRename
                && !sabotaged.swap(true, std::sync::atomic::Ordering::SeqCst)
            {
                std::fs::rename(quarantine, &managed_for_hook).unwrap();
            }
        });
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        let error = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap_err();

        // Asserted verbatim: the reason is the operator's only signal that
        // retrying is the remedy rather than corruption, so dropping the
        // journal-retention half of the sentence must fail this test.
        assert_eq!(
            unsafe_managed_reason(&error),
            "quarantine vanished before deletion; the removal journal is retained, so retrying the removal resumes and completes it"
        );
        assert!(store.load_removal_journal("org/model").unwrap().is_some());

        // And the retry the reason promises does finish the removal.
        let retried = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap();

        assert_eq!(retried.artifact_kind, ManagedArtifactKind::Directory);
        assert!(std::fs::symlink_metadata(&managed).is_err());
        assert!(no_quarantine_left(&models));
        assert!(store.load_removal_journal("org/model").unwrap().is_none());
        assert!(store.load_receipt("org/model").unwrap().is_none());
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn directory_removal_refuses_a_captured_file_replaced_before_unlink() {
        let (root, models, store) = supported_platform_store();
        let outside = root.path().join("outside");
        std::fs::create_dir_all(&outside).unwrap();
        let managed = models.join("Managed");
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("weights"), b"owned").unwrap();
        let parked = outside.join("parked");
        let store = store.with_removal_hook(move |phase, quarantine| {
            if phase == RemovalPhase::AfterCapture {
                std::fs::rename(quarantine.join("weights"), &parked).unwrap();
                std::fs::write(quarantine.join("weights"), b"victim").unwrap();
            }
        });
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        let error = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap_err();

        let reason = unsafe_managed_reason(&error);
        assert!(
            reason.contains("identity changed"),
            "unexpected reason: {reason}"
        );
        let journal = store.load_removal_journal("org/model").unwrap().unwrap();
        assert_eq!(
            std::fs::read(journal.quarantine_path.join("weights")).unwrap(),
            b"victim"
        );
        assert_eq!(std::fs::read(outside.join("parked")).unwrap(), b"owned");
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn directory_removal_refuses_a_captured_directory_replaced_before_rmdir() {
        let (root, models, store) = supported_platform_store();
        let outside = root.path().join("outside");
        std::fs::create_dir_all(&outside).unwrap();
        let managed = models.join("Managed");
        std::fs::create_dir_all(managed.join("sub")).unwrap();
        std::fs::write(managed.join("sub").join("more"), b"owned").unwrap();
        let parked = outside.join("parked-sub");
        let parked_for_hook = parked.clone();
        let store = store.with_removal_hook(move |phase, quarantine| {
            if phase == RemovalPhase::AfterCapture {
                std::fs::rename(quarantine.join("sub"), &parked_for_hook).unwrap();
                std::fs::create_dir_all(quarantine.join("sub")).unwrap();
                std::fs::write(quarantine.join("sub").join("victim"), b"victim").unwrap();
            }
        });
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        let error = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap_err();

        let reason = unsafe_managed_reason(&error);
        assert!(
            reason.contains("identity changed"),
            "unexpected reason: {reason}"
        );
        let journal = store.load_removal_journal("org/model").unwrap().unwrap();
        assert_eq!(
            std::fs::read(journal.quarantine_path.join("sub").join("victim")).unwrap(),
            b"victim"
        );
        // The captured subtree moved elsewhere is still CAR's object: its
        // captured entry was unlinked there, the directory itself was not.
        assert!(std::fs::symlink_metadata(&parked).is_ok());
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn directory_removal_accepts_a_tree_at_the_depth_cap() {
        let (_root, models, store) = supported_platform_store();
        let managed = models.join("Managed");
        let mut deep = managed.clone();
        for level in 0..MAX_REMOVAL_DEPTH_FOR_TESTS {
            deep = deep.join(format!("level{level}"));
        }
        std::fs::create_dir_all(&deep).unwrap();
        std::fs::write(deep.join("sentinel"), b"deep").unwrap();
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        let result = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap();

        assert_eq!(result.artifact_kind, ManagedArtifactKind::Directory);
        assert!(std::fs::symlink_metadata(&managed).is_err());
        assert!(no_quarantine_left(&models));
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn directory_removal_refuses_when_the_quarantine_vanishes_before_the_descriptor_open() {
        let (_root, models, store) = supported_platform_store();
        let managed = models.join("Managed");
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("weights"), b"owned").unwrap();
        let managed_for_hook = managed.clone();
        let store = store.with_removal_hook(move |phase, quarantine| {
            if phase == RemovalPhase::BeforeRootOpen {
                // The path-based checks passed; a racer now moves the quarantine
                // back to its managed name before the binding descriptor opens.
                std::fs::rename(quarantine, &managed_for_hook).unwrap();
            }
        });
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        let error = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap_err();

        // Spelled out rather than compared against QUARANTINE_VANISHED_REASON:
        // this window is user-reachable on a first attempt too, and the receipt
        // and journal assertions below are exactly the retention the sentence
        // promises, so the promise is pinned here as well as at the guard.
        assert_eq!(
            unsafe_managed_reason(&error),
            "quarantine vanished before deletion; the removal journal is retained, so retrying the removal resumes and completes it"
        );
        assert_eq!(std::fs::read(managed.join("weights")).unwrap(), b"owned");
        assert!(store.load_receipt("org/model").unwrap().is_some());
        assert!(store.load_removal_journal("org/model").unwrap().is_some());
    }

    /// Pins — does not endorse — what the retry does when the racer parked the
    /// quarantine somewhere other than the managed name. The resume finds
    /// neither the artifact nor the quarantine, so it cannot re-detach or
    /// delete anything; it unlinks the receipt and journal and reports success
    /// while the bytes survive at the parked path. Recorded so a deliberate
    /// change to that behaviour has to change this test.
    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn retry_after_the_quarantine_was_parked_elsewhere_settles_without_deleting_it() {
        let (root, models, store) = supported_platform_store();
        let outside = root.path().join("outside");
        std::fs::create_dir_all(&outside).unwrap();
        let managed = models.join("Managed");
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("weights"), b"owned").unwrap();
        let parked = outside.join("parked");
        let parked_for_hook = parked.clone();
        let store = store.with_removal_hook(move |phase, quarantine| {
            if phase == RemovalPhase::AfterRename {
                // Not back to the managed name: out of the models root entirely.
                std::fs::rename(quarantine, &parked_for_hook).unwrap();
            }
        });
        store
            .write_receipt(&directory_receipt("org/model", managed.clone()))
            .unwrap();

        let error = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap_err();

        assert_eq!(
            unsafe_managed_reason(&error),
            "quarantine vanished before deletion; the removal journal is retained, so retrying the removal resumes and completes it"
        );
        assert!(store.load_removal_journal("org/model").unwrap().is_some());

        let retried = store
            .begin_mutation("org/model")
            .unwrap()
            .remove(2)
            .unwrap();

        assert_eq!(retried.artifact_kind, ManagedArtifactKind::Directory);
        assert!(store.load_removal_journal("org/model").unwrap().is_none());
        assert!(store.load_receipt("org/model").unwrap().is_none());
        assert!(no_quarantine_left(&models));
        // The parked copy is untouched: CAR never followed it out of the root.
        assert_eq!(std::fs::read(parked.join("weights")).unwrap(), b"owned");
    }
}