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 car_winenv::PlainCanonicalize;
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,
}

/// How often a model in continuous use re-stamps its [`UsageStamp`].
pub const USAGE_STAMP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(10 * 60);

const USAGE_TRACKING_MARKER: &str = "tracking-since.json";
const USAGE_COVERAGE: &str = "coverage.json";

/// Longest stretch without a beat — awake time, not wall time — that still
/// counts as one continuous run of a stamping daemon. The daemon beats every
/// 15 minutes; four missed beats is a stopped or stalled daemon.
pub const COVERAGE_GAP_SECS: u64 = 60 * 60;

/// The current contiguous run during which a usage-stamping daemon was up.
/// Silence only counts inside it: idle needs the whole window observed, so
/// only the run in progress matters, never earlier ones.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
struct UsageCoverage {
    run_start: u64,
    last_beat: u64,
}

/// The last time a local model was used, written when its activity lease is
/// taken — the one point every local load passes, whatever the modality.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct UsageStamp {
    pub model_id: String,
    pub last_used: u64,
}

#[derive(Debug, Serialize, Deserialize)]
struct UsageTracking {
    since: u64,
}

fn unix_now() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

#[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,
    /// When each local model was last used by any CAR on this machine
    /// ([`UsageStamp`]). Beside the leases, not under one daemon's state root,
    /// because the weights it judges are machine-shared too.
    usage_root: PathBuf,
    /// Last stamp this process wrote per model: a lease is taken per request,
    /// and a durable write per request is not worth a minute of resolution.
    usage_throttle: std::sync::Arc<Mutex<std::collections::HashMap<String, std::time::Instant>>>,
    /// This process's last coverage beat: (wall clock, monotonic instant).
    /// The instant does not advance while the machine sleeps, which is what
    /// tells sleep (no one could use a model) from a stalled daemon.
    last_beat: std::sync::Arc<Mutex<Option<(u64, std::time::Instant)>>>,
    /// 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: receipts_dir(&state_root),
            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"),
            usage_root: shared_coordination_root.join("usage"),
            usage_throttle: Default::default(),
            last_beat: Default::default(),
            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)?;
        read_receipt_at(&self.receipt_path(model_id), model_id)
    }

    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);
        self.note_used(model_id);
        Ok(ModelLease { _file: file })
    }

    /// Record that `model_id` is being used now. At most one durable write
    /// per model per [`USAGE_STAMP_INTERVAL`] from this process, and never a
    /// reason to refuse the load: a failure is logged and the lease stands.
    /// Stamped before admission, so an attempt admission then refuses still
    /// counts as use — the safe direction for a signal that gates deletion.
    fn note_used(&self, model_id: &str) {
        let now = std::time::Instant::now();
        {
            let mut written = self
                .usage_throttle
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            if written
                .get(model_id)
                .is_some_and(|at| now.duration_since(*at) < USAGE_STAMP_INTERVAL)
            {
                return;
            }
            written.insert(model_id.to_string(), now);
        }
        let stamp = UsageStamp {
            model_id: model_id.to_string(),
            last_used: unix_now(),
        };
        if let Err(error) =
            write_private_json(&self.usage_root.join(hashed_filename(model_id)), &stamp)
        {
            tracing::debug!(model = model_id, %error, "could not record model use");
        }
    }

    /// Mark when usage tracking began on this machine, once. Until a model
    /// has gone unstamped for longer than an idle threshold *since this
    /// moment*, its silence is no evidence: the day stamping ships, every
    /// model is unstamped.
    ///
    /// The daemon calls this at startup, and nothing else may: the process
    /// that serves the models is the one whose leases stamp them. A newer CLI
    /// or test starting the clock while an older daemon serves unstamped
    /// would make every model look idle a month later.
    pub fn start_usage_tracking(&self) {
        let marker = self.usage_root.join(USAGE_TRACKING_MARKER);
        if std::fs::symlink_metadata(&marker).is_ok() {
            return;
        }
        if let Err(error) = write_private_json(&marker, &UsageTracking { since: unix_now() }) {
            tracing::debug!(%error, "could not record when model usage tracking began");
        }
    }

    /// Backdate the tracking marker, as if this machine had been tracking
    /// usage since `since`.
    #[cfg(test)]
    pub(crate) fn set_usage_tracking_since_for_test(&self, since: u64) {
        write_private_json(
            &self.usage_root.join(USAGE_TRACKING_MARKER),
            &UsageTracking { since },
        )
        .unwrap();
        write_private_json(
            &self.usage_root.join(USAGE_COVERAGE),
            &UsageCoverage {
                run_start: since,
                last_beat: unix_now(),
            },
        )
        .unwrap();
    }

    /// Record that a usage-stamping daemon is up now. The run continues from
    /// the previous beat when that beat was this process's own and less than
    /// [`COVERAGE_GAP_SECS`] of awake time ago (a sleeping machine pauses the
    /// clock rather than breaking the run), or when another stamping process
    /// beat within that window of wall time (a restart). Otherwise a run
    /// starts now. Daemon only, like [`Self::start_usage_tracking`].
    pub fn beat(&self) {
        let now = unix_now();
        let instant = std::time::Instant::now();
        let path = self.usage_root.join(USAGE_COVERAGE);
        let previous = read_private_state(&path)
            .ok()
            .and_then(|bytes| serde_json::from_slice::<UsageCoverage>(&bytes).ok());
        let mut mine = self
            .last_beat
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let run_start = match (previous, *mine) {
            (Some(prev), Some((wall, at))) if prev.last_beat == wall => {
                if instant.duration_since(at).as_secs() <= COVERAGE_GAP_SECS {
                    prev.run_start
                } else {
                    now
                }
            }
            (Some(prev), _) if now.saturating_sub(prev.last_beat) <= COVERAGE_GAP_SECS => {
                prev.run_start
            }
            _ => now,
        };
        let coverage = UsageCoverage {
            run_start,
            last_beat: now,
        };
        if let Err(error) = write_private_json(&path, &coverage) {
            tracing::debug!(%error, "could not record model usage coverage");
        }
        *mine = Some((now, instant));
    }

    /// Since when a usage-stamping daemon has been continuously up, if one is.
    pub fn usage_covered_since(&self) -> Option<u64> {
        read_private_state(&self.usage_root.join(USAGE_COVERAGE))
            .ok()
            .and_then(|bytes| serde_json::from_slice::<UsageCoverage>(&bytes).ok())
            .map(|coverage| coverage.run_start)
    }

    /// When usage tracking began on this machine, if it has.
    pub fn usage_tracking_since(&self) -> Option<u64> {
        read_private_state(&self.usage_root.join(USAGE_TRACKING_MARKER))
            .ok()
            .and_then(|bytes| serde_json::from_slice::<UsageTracking>(&bytes).ok())
            .map(|marker| marker.since)
    }

    /// When `model_id` was last used, if it ever was since tracking began.
    pub fn last_used(&self, model_id: &str) -> Option<u64> {
        let path = self.usage_root.join(hashed_filename(model_id));
        read_private_state(&path)
            .ok()
            .and_then(|bytes| serde_json::from_slice::<UsageStamp>(&bytes).ok())
            .filter(|stamp| stamp.model_id == model_id)
            .map(|stamp| stamp.last_used)
    }

    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.plain_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
                .plain_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)?;
        // Both the HF home and the hub itself: `HF_HUB_CACHE` can put the hub
        // anywhere, and that is where the shared bytes live.
        for hf_root in [crate::hf_cache::hf_home(), crate::hf_cache::hub_dir()] {
            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
                .plain_canonicalize()
                .map_err(|source| ModelManagementError::Io {
                    path: receipt.managed_path.clone(),
                    source,
                })?;
            // A receipt written before canonical paths were plain holds the
            // Windows verbatim spelling; compare like with like.
            if !receipt
                .shared_cache_references
                .iter()
                .any(|path| car_winenv::plain_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)?;
            let recorded: Vec<PathBuf> = receipt
                .shared_cache_references
                .iter()
                .map(|path| car_winenv::plain_path(path))
                .collect();
            if current_references != recorded {
                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,
}

/// The exclusive side of a model's activity lease, held by a removal while
/// it deletes: no request anywhere on the machine holds a lease on the model
/// while this lives.
pub(crate) struct IdleActivity {
    _file: OwnedFileLock,
}

impl ModelMutationGuard {
    /// Take the model's activity lease exclusively, as a receipt-backed
    /// removal does for the whole of its deletion; `ModelInUse` while any
    /// process holds a lease. Under this guard a new lease cannot start (a
    /// lease enters the shared side of the mutation lock first), so once this
    /// succeeds the model stays idle until it is dropped.
    pub(crate) fn hold_idle_activity(&self) -> Result<IdleActivity, ModelManagementError> {
        Ok(IdleActivity {
            _file: self.store.lock_idle_activity(&self.model_id)?,
        })
    }

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

/// The receipt file name [`read_receipt`] reads for `model_id`, for tests that
/// stand a receipt up without a store. Its one caller is a Unix-only test.
#[cfg(all(test, unix))]
pub(crate) fn receipt_file_name_for_test(model_id: &str) -> String {
    hashed_filename(model_id)
}

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.plain_canonicalize()
        .map_err(|source| ModelManagementError::Io {
            path: path.to_path_buf(),
            source,
        })
}

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

/// Read a model's install receipt under `state_root` without constructing a
/// [`ModelManagementStore`] — whose constructor resumes interrupted removals,
/// which deletes. For read-only paths such as `car doctor`.
pub fn read_receipt(
    state_root: &Path,
    model_id: &str,
) -> Result<Option<InstallReceipt>, ModelManagementError> {
    validate_model_id(model_id)?;
    let state_root = crate::resource_policy::normalized_state_root_key(state_root);
    read_receipt_at(
        &receipts_dir(&state_root).join(hashed_filename(model_id)),
        model_id,
    )
}

/// Where install receipts live under an already-normalized state root — the
/// one spelling the store and [`read_receipt`] both use.
fn receipts_dir(state_root: &Path) -> PathBuf {
    state_root.join("model-management").join("receipts")
}

fn read_receipt_at(
    path: &Path,
    model_id: &str,
) -> Result<Option<InstallReceipt>, ModelManagementError> {
    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: path.to_path_buf(),
                source,
            })
        }
    };
    let receipt: InstallReceipt =
        serde_json::from_slice(&bytes).map_err(|error| ModelManagementError::InvalidState {
            path: path.to_path_buf(),
            message: error.to_string(),
        })?;
    ensure_identity(path, model_id, &receipt.model_id)?;
    Ok(Some(receipt))
}

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

/// A hub directory being retired: journaled before it is detached into its
/// quarantine, removed once the quarantine is gone, so a crash between the
/// two leaves a record the next daemon start finishes from.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct DetachedTreeJournal {
    source: PathBuf,
    quarantine: PathBuf,
    identity: ArtifactIdentity,
}

/// A failed [`ModelManagementStore::detach_and_remove_tree`]: whether the
/// directory had already been renamed into its quarantine when it failed.
#[derive(Debug)]
pub(crate) struct DetachError {
    pub(crate) detached: bool,
    pub(crate) error: ModelManagementError,
}

/// A retirement's intent to clear the tombstone its model removal writes:
/// retiring leaves a row fetchable, not disabled. Journaled before the
/// removal so a crash between removal and clear is finished at next start.
#[derive(Debug, Serialize, Deserialize)]
struct RetiringIntent {
    model_id: String,
}

/// Quarantine names a retirement creates — the only names resume will delete.
pub(crate) const RETIRE_QUARANTINE_PREFIX: &str = ".car-retire-";

impl ModelManagementStore {
    fn detached_tree_journals(&self) -> PathBuf {
        self.management_state_dir.join("hub-removals")
    }

    fn retiring_intent_path(&self, model_id: &str) -> PathBuf {
        self.management_state_dir
            .join("retiring")
            .join(hashed_filename(model_id))
    }

    /// Journal that `model_id`'s removal is a retirement, before it starts.
    pub(crate) fn note_retiring(&self, model_id: &str) -> Result<(), ModelManagementError> {
        validate_model_id(model_id)?;
        write_private_json(
            &self.retiring_intent_path(model_id),
            &RetiringIntent {
                model_id: model_id.to_string(),
            },
        )
    }

    /// The retirement finished: its tombstone is cleared.
    pub(crate) fn retired(&self, model_id: &str) {
        let _ = std::fs::remove_file(self.retiring_intent_path(model_id));
    }

    /// Finish retirements whose model removal completed but whose tombstone
    /// was never cleared. An intent whose removal has not completed — the
    /// receipt is still there — is kept while a removal journal can still
    /// finish it, and dropped once nothing can.
    pub fn resume_retiring(&self) -> Vec<ModelManagementError> {
        let mut errors = Vec::new();
        let Ok(read) = std::fs::read_dir(self.management_state_dir.join("retiring")) else {
            return errors;
        };
        for entry in read.filter_map(Result::ok) {
            let path = entry.path();
            let Some(intent) = read_private_state(&path)
                .ok()
                .and_then(|bytes| serde_json::from_slice::<RetiringIntent>(&bytes).ok())
                .filter(|intent| {
                    path.file_name() == Some(hashed_filename(&intent.model_id).as_ref())
                })
            else {
                tracing::warn!(path = %path.display(), "unreadable retirement intent left in place");
                continue;
            };
            let id = &intent.model_id;
            let receipt_present = self.receipt_path(id).exists();
            if receipt_present {
                if !self.removal_journal_path(id).exists() {
                    // The removal never ran; the row is still installed.
                    self.retired(id);
                }
                continue;
            }
            match self.clear_tombstone(id) {
                Ok(()) => self.retired(id),
                Err(error) => errors.push(error),
            }
        }
        errors
    }

    /// Detach `source` — a hub repo directory CAR is retiring — into
    /// `quarantine` (a sibling, so the rename stays on one filesystem) and
    /// delete it through the same descriptor-bound walk a model removal uses:
    /// children opened by descriptor before they are trusted, symlinks
    /// unlinked and never followed, no filesystem crossing.
    ///
    /// The directory's identity is captured first, `verify` (the caller's
    /// containment check) runs next, and the identity is re-read immediately
    /// before the journal and rename, so what is renamed is what was checked.
    /// A failure after the rename says so ([`DetachError::detached`]): the
    /// directory is gone from its path and the next start finishes it.
    /// Refused outright where directory removal is unsupported, before
    /// anything is written or renamed.
    pub(crate) fn detach_and_remove_tree(
        &self,
        source: &Path,
        quarantine: &Path,
        label: &str,
        verify: impl FnOnce() -> Result<(), String>,
    ) -> Result<(), DetachError> {
        let before = |error| DetachError {
            detached: false,
            error,
        };
        if !directory_removal_supported() {
            return Err(before(ModelManagementError::UnsafeManagedPath {
                model_id: label.to_string(),
                path: source.to_path_buf(),
                reason: "directory removal is not supported on this platform".into(),
            }));
        }
        let identity = artifact_identity(source, label).map_err(before)?;
        if identity.kind != ManagedArtifactKind::Directory {
            return Err(before(ModelManagementError::UnsafeManagedPath {
                model_id: label.to_string(),
                path: source.to_path_buf(),
                reason: "a retired repo must be a directory".into(),
            }));
        }
        if quarantine.parent() != source.parent() {
            return Err(before(ModelManagementError::UnsafeManagedPath {
                model_id: label.to_string(),
                path: quarantine.to_path_buf(),
                reason: "the quarantine must sit beside the directory it detaches".into(),
            }));
        }
        verify().map_err(|reason| {
            before(ModelManagementError::UnsafeManagedPath {
                model_id: label.to_string(),
                path: source.to_path_buf(),
                reason,
            })
        })?;
        let again = artifact_identity(source, label).map_err(before)?;
        if again.device != identity.device || again.inode != identity.inode {
            return Err(before(ModelManagementError::UnsafeManagedPath {
                model_id: label.to_string(),
                path: source.to_path_buf(),
                reason: "the repo directory changed while it was being checked".into(),
            }));
        }
        let journal = DetachedTreeJournal {
            source: source.to_path_buf(),
            quarantine: quarantine.to_path_buf(),
            identity,
        };
        let journal_path = self
            .detached_tree_journals()
            .join(hashed_filename(&quarantine.to_string_lossy()));
        if let Some(parent) = journal_path.parent() {
            std::fs::create_dir_all(parent)
                .map_err(|source| ModelManagementError::Io {
                    path: parent.to_path_buf(),
                    source,
                })
                .map_err(before)?;
        }
        write_private_json(&journal_path, &journal).map_err(before)?;
        if let Err(io) = atomic_rename_noreplace(source, quarantine) {
            let _ = std::fs::remove_file(&journal_path);
            return Err(before(ModelManagementError::Io {
                path: source.to_path_buf(),
                source: io,
            }));
        }
        self.finish_detached_tree(&journal, &journal_path, label)
            .map_err(|error| DetachError {
                detached: true,
                error,
            })
    }

    #[cfg(any(target_os = "macos", target_os = "linux"))]
    fn finish_detached_tree(
        &self,
        journal: &DetachedTreeJournal,
        journal_path: &Path,
        label: &str,
    ) -> Result<(), ModelManagementError> {
        self.remove_quarantined_directory(&journal.quarantine, &journal.identity, label)?;
        let _ = std::fs::remove_file(journal_path);
        Ok(())
    }

    #[cfg(not(any(target_os = "macos", target_os = "linux")))]
    fn finish_detached_tree(
        &self,
        journal: &DetachedTreeJournal,
        _journal_path: &Path,
        label: &str,
    ) -> Result<(), ModelManagementError> {
        Err(ModelManagementError::UnsafeManagedPath {
            model_id: label.to_string(),
            path: journal.quarantine.clone(),
            reason: "directory removal is not supported on this platform".into(),
        })
    }

    /// Finish retirements a crash interrupted: a journaled quarantine still
    /// on disk is deleted (identity-checked); a journal whose rename never
    /// happened is dropped. Daemon start only.
    pub fn resume_detached_trees(&self) -> Vec<ModelManagementError> {
        let mut errors = Vec::new();
        let Ok(read) = std::fs::read_dir(self.detached_tree_journals()) else {
            return errors;
        };
        for entry in read.filter_map(Result::ok) {
            let path = entry.path();
            let journal: DetachedTreeJournal = match read_private_state(&path)
                .ok()
                .and_then(|bytes| serde_json::from_slice(&bytes).ok())
            {
                Some(journal) => journal,
                None => {
                    tracing::warn!(path = %path.display(), "unreadable retirement journal left in place");
                    continue;
                }
            };
            // Only ever a name a retirement creates; the identity check in the
            // walk is what binds it to the directory that was journaled. (The
            // parent comparison is a consistency check on the journal, not a
            // guard: both paths come from it.)
            let named = journal
                .quarantine
                .file_name()
                .and_then(|name| name.to_str())
                .is_some_and(|name| name.starts_with(RETIRE_QUARANTINE_PREFIX));
            if !named || journal.quarantine.parent() != journal.source.parent() {
                tracing::warn!(path = %path.display(), "retirement journal names a foreign quarantine; left in place");
                continue;
            }
            match std::fs::symlink_metadata(&journal.quarantine) {
                // The rename never happened, or the walk already finished.
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                    let _ = std::fs::remove_file(&path);
                    continue;
                }
                Err(source) => {
                    errors.push(ModelManagementError::Io {
                        path: journal.quarantine.clone(),
                        source,
                    });
                    continue;
                }
                Ok(_) => {}
            }
            if let Err(error) = self.finish_detached_tree(&journal, &path, "retired repo") {
                errors.push(error);
            }
        }
        errors
    }
}

#[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 directory the quarantine sits in —
        // the validated models root for a model removal, the hub for a retired
        // repo — and only if it is still the captured object and now empty.
        // (Looking it up under the models root alone found nothing for a hub
        // quarantine and reported success, leaving the directory behind.)
        let CapturedDirectory { file: root, .. } = directories.pop().expect("root");
        drop(root);
        let container = quarantine
            .parent()
            .ok_or_else(|| unsafe_path("quarantine path has no parent"))?;
        let models =
            object_bound::open_directory(container).map_err(|source| ModelManagementError::Io {
                path: container.to_path_buf(),
                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",
            )),
            // The walk held this root open moments ago; its absence now means
            // something moved it, never that the removal completed.
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                Err(unsafe_path(QUARANTINE_VANISHED_REASON))
            }
            Err(source) => Err(io_error(source)),
        }
    }
}

/// What [`unlink_checked`] may remove.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum UnlinkKind {
    /// A regular file only.
    File,
    /// A symlink only.
    Link,
    /// Either.
    FileOrLink,
}

/// Unlink `path`'s last component relative to a descriptor for its parent
/// directory — the parent opened once, refusing a symlink at its own leaf —
/// and only if the entry is still of `kind`. A path-based `remove_file` would
/// resolve the whole path again after the caller checked it.
#[cfg(any(target_os = "macos", target_os = "linux"))]
pub(crate) fn unlink_checked(path: &Path, kind: UnlinkKind) -> std::io::Result<()> {
    use object_bound::EntryKind;
    let parent = path.parent().ok_or_else(|| {
        std::io::Error::new(std::io::ErrorKind::InvalidInput, "path has no parent")
    })?;
    let name = path.file_name().ok_or_else(|| {
        std::io::Error::new(std::io::ErrorKind::InvalidInput, "path has no file name")
    })?;
    let dir = object_bound::open_directory(parent)?;
    let name = object_bound::c_name(Path::new(name))?;
    let (actual, _) = object_bound::stat_entry(&dir, &name)?;
    let allowed = matches!(
        (kind, actual),
        (
            UnlinkKind::File | UnlinkKind::FileOrLink,
            EntryKind::RegularFile
        ) | (
            UnlinkKind::Link | UnlinkKind::FileOrLink,
            EntryKind::Symlink
        )
    );
    if !allowed {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            format!("{} is no longer a {kind:?} entry", path.display()),
        ));
    }
    object_bound::unlink_entry(&dir, &name)
}

/// [`unlink_checked`] where descriptor-relative removal is unavailable: the
/// kind is re-checked, then the path is removed.
#[cfg(not(any(target_os = "macos", target_os = "linux")))]
pub(crate) fn unlink_checked(path: &Path, kind: UnlinkKind) -> std::io::Result<()> {
    let meta = std::fs::symlink_metadata(path)?;
    let allowed = match kind {
        UnlinkKind::File => meta.is_file(),
        UnlinkKind::Link => meta.file_type().is_symlink(),
        UnlinkKind::FileOrLink => meta.is_file() || meta.file_type().is_symlink(),
    };
    if !allowed {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            format!("{} is no longer a {kind:?} entry", path.display()),
        ));
    }
    // Windows removes a directory symlink or junction with `RemoveDirectory`,
    // which takes the link and never what it points at; `DeleteFile` refuses
    // one outright.
    #[cfg(windows)]
    {
        use std::os::windows::fs::FileTypeExt;
        if meta.file_type().is_symlink_dir() {
            return std::fs::remove_dir(path);
        }
    }
    std::fs::remove_file(path)
}

// 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 {
    /// A retirement a crash interrupted after detaching its repo is finished
    /// at the next start; one whose rename never happened is just dropped.
    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn interrupted_hub_retirements_resume() {
        let root = tempfile::tempdir().unwrap();
        let store =
            ModelManagementStore::new(root.path().join("state"), root.path().join("models"));
        let hub = root.path().join("hub");
        let repo = hub.join("models--a--b");
        std::fs::create_dir_all(repo.join("blobs")).unwrap();
        std::fs::write(repo.join("blobs/x"), b"bytes").unwrap();
        let quarantine = hub.join(".car-retire-1");
        let journal = DetachedTreeJournal {
            source: repo.clone(),
            quarantine: quarantine.clone(),
            identity: artifact_identity(&repo, "t").unwrap(),
        };
        let journals = store.detached_tree_journals();
        std::fs::create_dir_all(&journals).unwrap();
        write_private_json(&journals.join("j1"), &journal).unwrap();
        std::fs::rename(&repo, &quarantine).unwrap();
        // A second journal whose rename never happened.
        let never = DetachedTreeJournal {
            source: hub.join("models--c--d"),
            quarantine: hub.join(".car-retire-2"),
            identity: journal.identity.clone(),
        };
        write_private_json(&journals.join("j2"), &never).unwrap();

        assert!(store.resume_detached_trees().is_empty());
        assert!(!quarantine.exists());
        assert!(std::fs::read_dir(&journals).unwrap().next().is_none());

        // A journal aimed at something that is not a retirement quarantine
        // is left alone, and so is its target.
        let foreign = hub.join("models--e--f");
        std::fs::create_dir_all(&foreign).unwrap();
        std::fs::write(foreign.join("keep"), b"k").unwrap();
        let aimed = DetachedTreeJournal {
            source: hub.join("models--g--h"),
            quarantine: foreign.clone(),
            identity: artifact_identity(&foreign, "t").unwrap(),
        };
        write_private_json(&journals.join("j3"), &aimed).unwrap();
        // A real quarantine whose contents are no longer what was journaled.
        let swapped = hub.join(".car-retire-3");
        std::fs::create_dir_all(&swapped).unwrap();
        std::fs::write(swapped.join("other"), b"o").unwrap();
        // The identity of a directory that still exists: the first journal's
        // was deleted above, and Linux reuses a freed inode, so `swapped` can
        // be created with that very identity and match it.
        let stale = DetachedTreeJournal {
            source: hub.join("models--i--j"),
            quarantine: swapped.clone(),
            identity: aimed.identity.clone(),
        };
        write_private_json(&journals.join("j4"), &stale).unwrap();

        let errors = store.resume_detached_trees();
        assert_eq!(errors.len(), 1, "{errors:?}");
        assert!(foreign.join("keep").exists());
        assert!(swapped.join("other").exists());
        assert!(journals.join("j3").exists() && journals.join("j4").exists());
    }

    /// Tracking starts when the daemon starts it, not at the first use — so
    /// a machine that never loads a model still has a start to measure from —
    /// and a later start never moves it.
    #[test]
    fn usage_tracking_starts_with_the_store_and_stays_put() {
        let root = tempfile::tempdir().unwrap();
        let store =
            ModelManagementStore::new(root.path().join("state"), root.path().join("models"));
        assert_eq!(
            store.usage_tracking_since(),
            None,
            "construction starts nothing"
        );
        store.start_usage_tracking();
        let since = store
            .usage_tracking_since()
            .expect("marker written when started");
        let marker = store.usage_root.join(USAGE_TRACKING_MARKER);
        write_private_json(&marker, &UsageTracking { since: 7 }).unwrap();
        let again =
            ModelManagementStore::new(root.path().join("state"), root.path().join("models"));
        again.start_usage_tracking();
        assert_eq!(again.usage_tracking_since(), Some(7));
        assert!(since > 7);
        assert_eq!(store.last_used("org/model"), None, "never used is not used");
    }

    /// A lease stamps the model's use once per interval, not once per request.
    #[test]
    fn a_lease_stamps_use_at_most_once_per_interval() {
        let root = tempfile::tempdir().unwrap();
        let store =
            ModelManagementStore::new(root.path().join("state"), root.path().join("models"));
        drop(store.acquire_lease("org/model").unwrap());
        let stamped = store.last_used("org/model").expect("first lease stamps");
        assert!(stamped > 0);
        std::fs::remove_file(store.usage_root.join(hashed_filename("org/model"))).unwrap();
        drop(store.acquire_lease("org/model").unwrap());
        assert_eq!(
            store.last_used("org/model"),
            None,
            "throttled: no second write"
        );
        // Another model is its own stamp.
        drop(store.acquire_lease("org/other").unwrap());
        assert!(store.last_used("org/other").is_some());
        // A stamp naming a different model under this file is not this model's.
        write_private_json(
            &store.usage_root.join(hashed_filename("org/model")),
            &UsageStamp {
                model_id: "org/forged".into(),
                last_used: 1,
            },
        )
        .unwrap();
        assert_eq!(store.last_used("org/model"), None);
    }

    /// A beat continues the run across a restart within the gap and across
    /// sleep within one process, and a long silence starts a new run.
    #[test]
    fn coverage_runs_continue_across_restarts_and_break_on_long_gaps() {
        let root = tempfile::tempdir().unwrap();
        let store =
            ModelManagementStore::new(root.path().join("state"), root.path().join("models"));
        assert_eq!(store.usage_covered_since(), None);
        store.beat();
        let first = store.usage_covered_since().unwrap();
        store.beat();
        assert_eq!(store.usage_covered_since(), Some(first), "same run");
        // A restart: a new process (fresh store) beating soon after.
        let restarted =
            ModelManagementStore::new(root.path().join("state"), root.path().join("models"));
        restarted.beat();
        assert_eq!(restarted.usage_covered_since(), Some(first));
        // The last beat was long ago and not ours: a new run.
        write_private_json(
            &restarted.usage_root.join(USAGE_COVERAGE),
            &UsageCoverage {
                run_start: 5,
                last_beat: 10,
            },
        )
        .unwrap();
        let later =
            ModelManagementStore::new(root.path().join("state"), root.path().join("models"));
        later.beat();
        assert!(later.usage_covered_since().unwrap() > 10);
    }

    #[test]
    fn an_interrupted_retirement_clears_its_tombstone_at_resume() {
        let root = tempfile::tempdir().unwrap();
        let store =
            ModelManagementStore::new(root.path().join("state"), root.path().join("models"));
        let id = "mlx/some:4bit";
        store.note_retiring(id).unwrap();
        write_private_json(
            &store.tombstone_path(id),
            &RemovalTombstone {
                model_id: id.into(),
                removal_generation: 1,
                artifact_kind: None,
                preserved_shared_cache_references: Vec::new(),
            },
        )
        .unwrap();
        assert!(!store.car_enabled(id).unwrap());
        assert!(store.tombstone_path(id).exists());
        assert!(store.resume_retiring().is_empty());
        assert!(!store.tombstone_path(id).exists());
        assert!(!store.retiring_intent_path(id).exists());
    }

    use super::*;

    // Invoked only by the parent process fixture below, never by a model backend.
    #[test]
    #[ignore = "re-executed by identity_initialization_serializes_processes_before_json_is_written"]
    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",
                        "--include-ignored",
                    ])
                    .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.plain_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.plain_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").plain_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));
    }

    /// Reading receipts for the revoked-copies report constructs no store, so
    /// a read-only path (`car doctor`, the portfolio) never finishes an
    /// interrupted removal. A store constructed afterwards does — the control
    /// that the quarantine below is one a store would resume.
    #[cfg(any(target_os = "macos", target_os = "linux"))]
    #[test]
    fn the_revoked_copies_report_never_resumes_a_removal() {
        let (root, models, store) = supported_platform_store();
        let state = root.path().join("state");
        let managed = models.join("Managed");
        std::fs::create_dir_all(&managed).unwrap();
        std::fs::write(managed.join("weights"), 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: receipt.clone(),
            quarantine_path: quarantine.clone(),
        };
        write_private_json(&store.removal_journal_path("org/model"), &journal).unwrap();
        std::fs::rename(&managed, &quarantine).unwrap();
        drop(store);
        crate::catalog::save_revoked_names(
            &crate::catalog::revoked_names_path(&state),
            &std::collections::BTreeMap::from([("Managed".into(), "org/model".into())]),
        )
        .unwrap();

        assert_eq!(read_receipt(&state, "org/model").unwrap(), Some(receipt));
        let _ = crate::retire::revoked_copies(&state, &models);
        assert!(quarantine.join("weights").exists(), "nothing resumed");

        let _store = ModelManagementStore::new(state, models.clone());
        assert!(
            std::fs::symlink_metadata(&quarantine).is_err(),
            "control: a store resumes it"
        );
    }

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