car-inference 0.56.1

//! What retiring a local model would delete from the Hugging Face hub, and why
//! — planned without touching anything.
//!
//! Removing a model from CAR used to unlink CAR's projection and leave the
//! weights in the shared hub cache, freeing nothing. The concierge retires
//! dead weights on the user's behalf, so it needs to know exactly which bytes
//! are CAR's to delete. The answer is a [`RetirePlan`]; executing one is a
//! separate, journaled step.
//!
//! **Ownership matches what CAR fetches.** [`hf_references`] is the single
//! account of that, derived from a row's [`ModelSource`]:
//!
//! - an MLX / managed vLLM-MLX row fetches its whole `hf_repo` — CAR owns that
//!   repo directory;
//! - a GGUF row fetches one quantization (`hf_filename`) from a repo that holds
//!   many, and one `tokenizer.json` from its base-model repo (`Qwen/Qwen3-8B`
//!   holds full weights a user's own `transformers` may be using); the Flux row
//!   fetches one tokenizer from the full FLUX checkpoint. For these CAR owns
//!   **only the files it fetched**: their snapshot entries, and a blob only when
//!   no other entry in the repo still points at it.
//!
//! A repo is CAR's at all only when **CAR's own builtin catalog** names it (the
//! user's decision: catalog repos are CAR's) or CAR **recorded fetching** into
//! it ([`record_download`], written only for files hf-hub did not already hold).
//! A user-registered row pointing at a repo does not make it CAR's, and
//! nothing else in the hub — other tools' caches, text encoders a runtime
//! fetched on its own — is ever touched. A reference another **installed** row
//! also holds is kept. File-level ownership is never widened to the repo.
//!
//! **Paths are built only from checked inputs**: an absolute hub, a repo id
//! that passes [`crate::hf_cache::checked_repo_dir_in`], a repo directory that
//! is not a symlink and whose canonical parent is the canonical hub, and blobs
//! that resolve inside that repo's own `blobs/` (neither `blobs/` nor
//! `snapshots/` may be a symlink). Every emitted path is in canonical form,
//! nothing is derived from `refs/main`, and an I/O error refuses the plan
//! rather than shrinking the evidence it reasons from.

use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::path::{Path, PathBuf};

use serde::Serialize;

use crate::schema::{ModelSchema, ModelSource};

/// One thing a row fetches from the hub.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct HfReference {
    pub repo: String,
    /// `None`: the whole repo. `Some`: only these paths within a snapshot.
    pub files: Option<BTreeSet<String>>,
}

/// What `schema` fetches from the Hugging Face hub — the one account the
/// downloader's provenance and the retire planner's ownership both use.
pub fn hf_references(schema: &ModelSchema) -> Vec<HfReference> {
    let file = |repo: &str, path: &str| HfReference {
        repo: repo.to_string(),
        files: Some(BTreeSet::from([path.to_string()])),
    };
    match &schema.source {
        ModelSource::Local {
            hf_repo,
            hf_filename,
            tokenizer_repo,
        } => vec![
            file(hf_repo, hf_filename),
            file(tokenizer_repo, "tokenizer.json"),
        ],
        ModelSource::Mlx { hf_repo, .. } | ModelSource::ManagedVllmMlx { hf_repo, .. } => {
            let mut refs = vec![HfReference {
                repo: hf_repo.clone(),
                files: None,
            }];
            if crate::registry::uses_flux_auxiliary(&schema.name, hf_repo) {
                let (_, _, base, path) = crate::registry::FLUX_AUXILIARY;
                refs.push(file(base, path));
            }
            if crate::registry::uses_ltx_text_encoder(schema) {
                refs.push(HfReference {
                    repo: crate::registry::LTX_TEXT_ENCODER.to_string(),
                    files: None,
                });
            }
            refs
        }
        _ => Vec::new(),
    }
}

/// The repo a row *is* — fetched whole as its own weights — as opposed to an
/// auxiliary it also loads (the Flux tokenizer, the LTX text encoder). Rows
/// that share a primary repo are twins and retire together; sharing an
/// auxiliary makes nobody a twin, it only keeps the auxiliary while either
/// row needs it.
pub fn primary_whole_repo(schema: &ModelSchema) -> Option<&str> {
    match &schema.source {
        ModelSource::Mlx { hf_repo, .. } | ModelSource::ManagedVllmMlx { hf_repo, .. } => {
            Some(hf_repo.as_str())
        }
        _ => None,
    }
}

/// File under the state root's `model-management/` recording what CAR fetched.
pub const PROVENANCE_FILE: &str = "hf-provenance.jsonl";

#[derive(Debug, Serialize, serde::Deserialize)]
struct ProvenanceRecord {
    at: u64,
    model_id: String,
    repo: String,
    /// The snapshot (commit) the files landed in, when known.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    revision: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    files: Option<BTreeSet<String>>,
    hub: PathBuf,
}

fn provenance_path(state_root: &Path) -> PathBuf {
    state_root.join("model-management").join(PROVENANCE_FILE)
}

/// One file a download fetched: `(repo, file, snapshot revision)`.
pub type Fetched = (String, String, Option<String>);

tokio::task_local! {
    static FETCHED: std::cell::RefCell<Vec<Fetched>>;
}

/// The snapshot revision a cache path lies in: the component after
/// `snapshots/` in `…/models--x--y/snapshots/<rev>/<file>`.
pub(crate) fn snapshot_revision(path: &Path) -> Option<String> {
    // Anchored after the repo directory, so a hub that itself lives under a
    // directory named `snapshots` does not read the wrong component.
    let mut parts = path.components().map(|c| c.as_os_str().to_string_lossy());
    parts.by_ref().find(|c| c.starts_with("models--"))?;
    (parts.next()? == "snapshots").then_some(())?;
    parts.next().map(|c| c.into_owned())
}

/// Note that the download in progress fetched `file` from `repo` — called at
/// the one point a byte actually arrives. Outside [`collecting_fetches`] it is
/// a no-op.
pub(crate) fn note_fetched(repo: &str, file: &str, revision: Option<String>) {
    // A fetch whose snapshot is unknown is not recorded: "any revision" would
    // claim more than CAR fetched.
    if revision.is_none() {
        return;
    }
    let _ = FETCHED.try_with(|f| {
        f.borrow_mut()
            .push((repo.to_string(), file.to_string(), revision))
    });
}

/// Run `fut`, returning what it fetched alongside its output.
pub(crate) async fn collecting_fetches<F: std::future::Future>(
    fut: F,
) -> (F::Output, Vec<Fetched>) {
    FETCHED
        .scope(std::cell::RefCell::new(Vec::new()), async move {
            let out = fut.await;
            let fetched = FETCHED.with(|f| std::mem::take(&mut *f.borrow_mut()));
            (out, fetched)
        })
        .await
}

/// Append what a download fetched: one record per repo, naming the files.
/// Best effort — a record that cannot be written costs only the ownership of
/// a future renamed row, and must never fail the download.
pub fn record_download(state_root: &Path, hub: &Path, model_id: &str, fetched: &[Fetched]) {
    use std::io::Write;
    if fetched.is_empty() {
        return;
    }
    let hub = hub.canonicalize().unwrap_or_else(|_| hub.to_path_buf());
    let at = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let mut by_repo: BTreeMap<(&str, Option<&str>), BTreeSet<String>> = BTreeMap::new();
    for (repo, file, revision) in fetched {
        by_repo
            .entry((repo, revision.as_deref()))
            .or_default()
            .insert(file.clone());
    }
    let path = provenance_path(state_root);
    let mut body = String::new();
    for ((repo, revision), files) in by_repo {
        let record = ProvenanceRecord {
            at,
            model_id: model_id.to_string(),
            repo: repo.to_string(),
            revision: revision.map(str::to_string),
            files: Some(files),
            hub: hub.clone(),
        };
        if let Ok(line) = serde_json::to_string(&record) {
            body.push_str(&line);
            body.push('\n');
        }
    }
    let write = || -> std::io::Result<()> {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let mut f = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&path)?;
        f.write_all(body.as_bytes())
    };
    if let Err(error) = write() {
        tracing::warn!(%error, path = %path.display(), "could not record download provenance");
    }
}

/// A file CAR owns in a repo: its path within a snapshot, and the snapshot it
/// was fetched into (`None`: any snapshot, as the catalog names it).
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct OwnedFile {
    pub path: String,
    pub revision: Option<String>,
}

/// What of one repo is CAR's.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Owned {
    /// The catalog fetches the whole repo.
    Whole,
    /// Only these files.
    Files(BTreeSet<OwnedFile>),
}

/// What is CAR's in the hub, per repo.
pub type Ownership = BTreeMap<String, Owned>;

/// The files recorded downloads fetched into `hub`, each bound to the
/// snapshot it landed in.
pub fn recorded_files(state_root: &Path, hub: &Path) -> Ownership {
    let mut out: Ownership = BTreeMap::new();
    let Ok(text) = std::fs::read_to_string(provenance_path(state_root)) else {
        return out;
    };
    let hub = hub.canonicalize().unwrap_or_else(|_| hub.to_path_buf());
    for record in text
        .lines()
        .filter_map(|line| serde_json::from_str::<ProvenanceRecord>(line).ok())
        .filter(|r| r.hub == hub)
    {
        let files = record
            .files
            .unwrap_or_default()
            .into_iter()
            .map(|path| OwnedFile {
                path,
                revision: record.revision.clone(),
            });
        match out
            .entry(record.repo)
            .or_insert(Owned::Files(BTreeSet::new()))
        {
            Owned::Files(set) => set.extend(files),
            Owned::Whole => {}
        }
    }
    out
}

/// What CAR's compiled catalog fetches, at the granularity it fetches it:
/// a repo it takes whole is [`Owned::Whole`], a repo it takes one file from
/// (a GGUF row's base-model tokenizer) is only that file. The user's rule,
/// "catalog repos are CAR's", is about what the catalog fetches; it never
/// reaches a user's own full checkout of a repo the catalog takes one file
/// from.
///
/// Builtin rows only. A signed catalog arrives from the network between
/// releases, and a row it adds may name a repo the user downloaded
/// themselves before the row existed; naming is not fetching. What CAR
/// fetched for a signed row is CAR's through [`recorded_files`] instead.
pub fn catalog_ownership() -> Ownership {
    let mut out: Ownership = BTreeMap::new();
    for r in crate::registry::builtin_catalog()
        .iter()
        .flat_map(hf_references)
    {
        let entry = out.entry(r.repo).or_insert(Owned::Files(BTreeSet::new()));
        match (entry, r.files) {
            (slot, None) => *slot = Owned::Whole,
            (Owned::Files(set), Some(files)) => {
                set.extend(files.into_iter().map(|path| OwnedFile {
                    path,
                    revision: None,
                }))
            }
            (Owned::Whole, Some(_)) => {}
        }
    }
    out
}

/// Catalog ownership merged with recorded downloads.
pub fn ownership(catalog: Ownership, recorded: Ownership) -> Ownership {
    let mut out = recorded;
    for (repo, owned) in catalog {
        match (
            out.entry(repo).or_insert(Owned::Files(BTreeSet::new())),
            owned,
        ) {
            (slot, Owned::Whole) => *slot = Owned::Whole,
            (Owned::Files(set), Owned::Files(more)) => set.extend(more),
            (Owned::Whole, Owned::Files(_)) => {}
        }
    }
    out
}

/// A planned retirement. Empty `deletions` with no `refusals` means there is
/// nothing of CAR's left to reclaim.
#[derive(Debug, Clone, Default, Serialize)]
pub struct RetirePlan {
    pub model_id: String,
    /// Other rows retired with it: rows backed by a repo this retirement
    /// deletes whole, which could not keep their weights anyway (the two
    /// Gemma 12B rows share one repo and would otherwise pin each other).
    pub also_retires: Vec<String>,
    /// CAR's own projections of the model under its models directory.
    pub projections: Vec<Projection>,
    /// Directories under the models directory named for a retired row that
    /// hold real files without a receipt — the user's, never removed.
    pub user_content: Vec<PathBuf>,
    pub deletions: Vec<HubDeletion>,
    /// References another installed row also holds; left in place.
    pub kept: Vec<KeptReference>,
    /// Repos the model references that CAR does not own; never touched.
    pub unowned: Vec<String>,
    pub refusals: Vec<RetireRefusal>,
    /// Bytes the deletions would free.
    pub freed_bytes: u64,
    /// A fingerprint of what this plan would delete. Passing it back as
    /// `expect` makes execution refuse unless the plan made under the locks is
    /// the same — consent to a reviewed dry run, not to whatever is current.
    pub digest: String,
}

/// One projection under the models directory and, when an install receipt
/// backs it, the row whose receipt that is — decided when planning, so
/// execution never re-infers which removal to run.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct Projection {
    pub path: PathBuf,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub receipt_for: Option<String>,
}

impl RetirePlan {
    /// Compute [`RetirePlan::digest`] over everything execution would act on.
    pub fn seal(&mut self) {
        use sha2::{Digest, Sha256};
        // Canonical order first: targets come from a HashMap and entries from
        // `read_dir`, whose orders change between processes, and a digest a
        // user approved before a daemon restart must still match after it.
        self.also_retires.sort();
        self.projections.sort_by(|a, b| a.path.cmp(&b.path));
        for deletion in &mut self.deletions {
            if let HubDeletion::Files { entries, blobs, .. } = deletion {
                entries.sort();
                blobs.sort();
            }
        }
        self.deletions.sort_by(|a, b| a.repo().cmp(b.repo()));
        let material = serde_json::json!({
            "model_id": self.model_id,
            "also_retires": self.also_retires,
            "projections": self.projections,
            "deletions": self.deletions,
        });
        let mut hash = Sha256::new();
        hash.update(material.to_string().as_bytes());
        self.digest = format!("{:x}", hash.finalize());
    }
}

impl HubDeletion {
    fn repo(&self) -> &str {
        match self {
            HubDeletion::Repo { repo, .. } | HubDeletion::Files { repo, .. } => repo,
        }
    }
}

impl RetirePlan {
    pub fn is_refused(&self) -> bool {
        !self.refusals.is_empty()
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum HubDeletion {
    /// The whole `models--…` directory.
    Repo {
        repo: String,
        path: PathBuf,
        bytes: u64,
    },
    /// Snapshot entries CAR fetched, and the blobs only they referenced.
    Files {
        repo: String,
        entries: Vec<PathBuf>,
        blobs: Vec<PathBuf>,
        bytes: u64,
    },
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct KeptReference {
    pub repo: String,
    pub kept_for: String,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(tag = "reason", rename_all = "snake_case")]
pub enum RetireRefusal {
    /// The model is loaded, serving, leased, or being torn down.
    InUse,
    /// The model is a lane default; retiring it would leave a lane pointing at
    /// nothing.
    LaneDefault { use_case: String },
    /// The model is a default something else depends on — a configured
    /// generation/embedding/classification model, a local speech default, the
    /// local tool model CAR's setup guidance installs (or the MLX twin such a
    /// default runs as).
    Protected { why: String },
    /// A download of the model is in flight.
    Downloading,
    /// The plan made under the locks needs a lock that was not taken — the
    /// registry changed between planning and locking. Try again.
    StateChanged,
    /// A path failed a safety check; the plan deletes nothing.
    UnsafePath { path: PathBuf, why: String },
    /// The model downloads no weights (remote, OS-provided, runtime-only).
    NothingToRetire,
    /// The model downloads weights somewhere this planner does not manage yet
    /// (whisper.cpp keeps its own store outside the hub).
    Unsupported { why: String },
}

/// What a retirement did.
#[derive(Debug, Clone, Serialize)]
pub struct RetireOutcome {
    /// The plan made under the locks — the one that was executed — or, for a
    /// dry run or a refusal, the plan that would have been.
    pub plan: RetirePlan,
    /// Whether anything was deleted.
    pub executed: bool,
    /// Paths deleted, in order.
    pub removed: Vec<PathBuf>,
    /// What stopped the execution part-way; already-removed paths stay
    /// removed and a journaled hub directory is finished at the next start.
    pub errors: Vec<String>,
}

/// Whether two paths name the same directory entry: equal leaf names under
/// parents that both resolve, to the same directory. The leaves are not
/// resolved, so a dangling link compares by where it is, not where it led.
pub fn same_entry(a: &Path, b: &Path) -> bool {
    let parent = |p: &Path| p.parent().and_then(|d| d.canonicalize().ok());
    a.file_name().is_some()
        && a.file_name() == b.file_name()
        && matches!((parent(a), parent(b)), (Some(x), Some(y)) if x == y)
}

/// Re-check, immediately before deleting it, that `dir` is still a real repo
/// directory directly in `hub`. See [`check_repo_dir`].
pub(crate) fn verify_repo_dir(dir: &Path, hub: &Path) -> Result<(), String> {
    let hub = hub.canonicalize().map_err(|e| e.to_string())?;
    check_repo_dir(dir, &hub)
}

/// Re-check a snapshot entry before unlinking it: still a link or a regular
/// file, and its directory still inside the repo's `snapshots/`.
pub(crate) fn verify_entry(entry: &Path, repo_snapshots: &Path) -> Result<(), String> {
    let meta = std::fs::symlink_metadata(entry).map_err(|e| e.to_string())?;
    if !(meta.file_type().is_symlink() || meta.is_file()) {
        return Err("no longer a link or a file".into());
    }
    let parent = entry
        .parent()
        .ok_or("no parent")?
        .canonicalize()
        .map_err(|e| e.to_string())?;
    let snapshots = repo_snapshots.canonicalize().map_err(|e| e.to_string())?;
    if !parent.starts_with(&snapshots) {
        return Err("no longer inside the repo's snapshots".into());
    }
    Ok(())
}

/// Re-check a blob before unlinking it: still a regular file directly in the
/// repo's `blobs/`.
pub(crate) fn verify_blob(blob: &Path, repo_blobs: &Path) -> Result<(), String> {
    let meta = std::fs::symlink_metadata(blob).map_err(|e| e.to_string())?;
    if !meta.is_file() {
        return Err("no longer a regular file".into());
    }
    let parent = blob
        .parent()
        .ok_or("no parent")?
        .canonicalize()
        .map_err(|e| e.to_string())?;
    if parent != repo_blobs.canonicalize().map_err(|e| e.to_string())? {
        return Err("no longer in the repo's blobs".into());
    }
    Ok(())
}

/// Remove a links-only projection: re-check that it is still a direct child
/// of `models_dir`, then unlink every link (descriptor-relative) and remove
/// each directory once empty, deepest first. A real file found along the way
/// stops it — that tree is no longer CAR's to remove.
pub(crate) fn remove_links_only_tree(
    root: &Path,
    models_dir: &Path,
) -> Result<Vec<PathBuf>, String> {
    use crate::model_management::{unlink_checked, UnlinkKind};
    let models_dir = models_dir.canonicalize().map_err(|e| e.to_string())?;
    let parent = root
        .parent()
        .ok_or("projection has no parent")?
        .canonicalize()
        .map_err(|e| e.to_string())?;
    if parent != models_dir {
        return Err("the projection is no longer directly in the models directory".into());
    }
    let meta = std::fs::symlink_metadata(root).map_err(|e| e.to_string())?;
    if meta.file_type().is_symlink() {
        unlink_checked(root, UnlinkKind::Link).map_err(|e| e.to_string())?;
        return Ok(vec![root.to_path_buf()]);
    }
    if !links_only(root) {
        return Err("the projection now holds real files".into());
    }
    let mut dirs = vec![root.to_path_buf()];
    let mut links = Vec::new();
    let mut i = 0;
    while i < dirs.len() {
        for child in std::fs::read_dir(&dirs[i]).map_err(|e| e.to_string())? {
            let path = child.map_err(|e| e.to_string())?.path();
            let meta = std::fs::symlink_metadata(&path).map_err(|e| e.to_string())?;
            if meta.file_type().is_symlink() {
                links.push(path);
            } else if meta.is_dir() {
                dirs.push(path);
            } else {
                return Err(format!("{} is a real file", path.display()));
            }
        }
        i += 1;
    }
    let mut removed = Vec::new();
    for link in links {
        unlink_checked(&link, UnlinkKind::Link).map_err(|e| e.to_string())?;
        removed.push(link);
    }
    for dir in dirs.into_iter().rev() {
        std::fs::remove_dir(&dir).map_err(|e| e.to_string())?;
    }
    removed.push(root.to_path_buf());
    Ok(removed)
}

/// After file-level deletions, remove a repo left as an empty shell: no
/// snapshot entries, no blobs, nothing but empty directories and the `refs/`
/// pointers to the revisions just emptied. Anything else and it is left.
pub(crate) fn prune_empty_repo(dir: &Path) {
    let mut files = Vec::new();
    let mut dirs = vec![dir.to_path_buf()];
    let mut i = 0;
    while i < dirs.len() {
        let Ok(read) = std::fs::read_dir(&dirs[i]) else {
            return;
        };
        for child in read {
            let Ok(child) = child else { return };
            let path = child.path();
            let Ok(meta) = std::fs::symlink_metadata(&path) else {
                return;
            };
            if meta.is_dir() {
                dirs.push(path);
            } else if meta.is_file() && path.starts_with(dir.join("refs")) {
                files.push(path);
            } else {
                return;
            }
        }
        i += 1;
    }
    for file in files {
        if std::fs::remove_file(&file).is_err() {
            return;
        }
    }
    for d in dirs.into_iter().rev() {
        if std::fs::remove_dir(&d).is_err() {
            return;
        }
    }
}

/// Everything [`plan_hub_retirement`] reads, injected so tests use tempdirs.
pub struct RetireInputs<'a> {
    /// The rows retired together, the requested one first.
    pub targets: &'a [&'a ModelSchema],
    /// The full registry: catalog, user and project rows — read only to find
    /// what other installed rows still hold.
    pub registry: &'a [&'a ModelSchema],
    /// Whether a row's weights are installed and enabled in CAR.
    pub installed: &'a dyn Fn(&ModelSchema) -> bool,
    /// What is CAR's: see [`ownership`].
    pub owned: &'a Ownership,
    /// The hub cache directory; must be absolute.
    pub hub: &'a Path,
}

/// Plan the hub side of retiring `inputs.target`. Engine-level refusals (in
/// use, lane default, downloading) are added by the caller.
pub fn plan_hub_retirement(inputs: &RetireInputs<'_>) -> RetirePlan {
    let Some(target) = inputs.targets.first().copied() else {
        return RetirePlan::default();
    };
    let mut plan = RetirePlan {
        model_id: target.id.clone(),
        also_retires: inputs.targets[1..].iter().map(|t| t.id.clone()).collect(),
        ..Default::default()
    };
    let target_ids: HashSet<&str> = inputs.targets.iter().map(|t| t.id.as_str()).collect();
    let refs: Vec<HfReference> = inputs
        .targets
        .iter()
        .flat_map(|t| hf_references(t))
        .collect();
    if !target.downloads_weights() {
        plan.refusals.push(RetireRefusal::NothingToRetire);
        return plan;
    }
    if refs.is_empty() {
        plan.refusals.push(RetireRefusal::Unsupported {
            why: format!(
                "{} keeps its weights outside the Hugging Face hub",
                target.id
            ),
        });
        return plan;
    }
    let hub = match inputs.hub.is_absolute().then(|| inputs.hub.canonicalize()) {
        Some(Ok(hub)) => hub,
        Some(Err(error)) if error.kind() == std::io::ErrorKind::NotFound => return plan,
        _ => {
            plan.refusals.push(RetireRefusal::UnsafePath {
                path: inputs.hub.to_path_buf(),
                why: "the hub cache path is not an absolute, readable directory".into(),
            });
            return plan;
        }
    };

    // What every OTHER installed row holds, per repo.
    let mut held: BTreeMap<String, (Option<BTreeSet<String>>, String)> = BTreeMap::new();
    for row in inputs.registry.iter().copied() {
        if target_ids.contains(row.id.as_str()) || !(inputs.installed)(row) {
            continue;
        }
        for r in hf_references(row) {
            let entry = held
                .entry(r.repo)
                .or_insert((Some(BTreeSet::new()), row.id.clone()));
            merge_files(&mut entry.0, r.files);
        }
    }

    let mut mine: BTreeMap<String, Option<BTreeSet<String>>> = BTreeMap::new();
    for r in refs {
        merge_files(mine.entry(r.repo).or_insert(Some(BTreeSet::new())), r.files);
    }

    for (repo, claimed) in mine {
        let Some(owned) = inputs.owned.get(&repo) else {
            plan.unowned.push(repo);
            continue;
        };
        let files = claimed;
        let Some(dir) = crate::hf_cache::checked_repo_dir_in(&hub, &repo) else {
            plan.refusals.push(RetireRefusal::UnsafePath {
                path: hub.join(&repo),
                why: "the repo id is not a Hub repo id".into(),
            });
            continue;
        };
        match std::fs::symlink_metadata(&dir) {
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
            Err(error) => {
                plan.refusals.push(RetireRefusal::UnsafePath {
                    path: dir,
                    why: error.to_string(),
                });
                continue;
            }
            Ok(_) => {}
        }
        if let Err(why) = check_repo_dir(&dir, &hub) {
            plan.refusals
                .push(RetireRefusal::UnsafePath { path: dir, why });
            continue;
        }
        let kept_files = match held.get(&repo) {
            Some((None, holder)) => {
                plan.kept.push(KeptReference {
                    repo: repo.clone(),
                    kept_for: holder.clone(),
                });
                continue;
            }
            Some((Some(held_files), holder)) => {
                let overlaps = files
                    .as_ref()
                    .is_none_or(|mine| !mine.is_disjoint(held_files));
                if overlaps {
                    plan.kept.push(KeptReference {
                        repo: repo.clone(),
                        kept_for: holder.clone(),
                    });
                }
                held_files.clone()
            }
            None => BTreeSet::new(),
        };
        match plan_repo(&dir, &repo, files.as_ref(), owned, &kept_files) {
            Ok(Some(deletion)) => {
                plan.freed_bytes += match &deletion {
                    HubDeletion::Repo { bytes, .. } | HubDeletion::Files { bytes, .. } => *bytes,
                };
                plan.deletions.push(deletion);
            }
            Ok(None) => {}
            Err(why) => plan
                .refusals
                .push(RetireRefusal::UnsafePath { path: dir, why }),
        }
    }
    plan
}

fn merge_files(slot: &mut Option<BTreeSet<String>>, more: Option<BTreeSet<String>>) {
    match (slot.as_mut(), more) {
        (Some(files), Some(more)) => files.extend(more),
        (_, None) => *slot = None,
        (None, Some(_)) => {}
    }
}

/// The repo directory must be a real directory directly in the hub, under
/// exactly the name its id gives, with real `blobs/` and `snapshots/`.
fn check_repo_dir(dir: &Path, hub: &Path) -> Result<(), String> {
    let meta = std::fs::symlink_metadata(dir).map_err(|e| e.to_string())?;
    if meta.file_type().is_symlink() {
        return Err("the repo directory is a symlink".into());
    }
    if !meta.is_dir() {
        return Err("the repo path is not a directory".into());
    }
    let name = dir.file_name().ok_or("the repo path has no name")?;
    if dir.canonicalize().map_err(|e| e.to_string())? != hub.join(name) {
        return Err("the repo directory is not directly inside the hub".into());
    }
    for sub in ["blobs", "snapshots"] {
        match std::fs::symlink_metadata(dir.join(sub)) {
            Ok(m) if m.file_type().is_symlink() => {
                return Err(format!("the repo's {sub}/ is a symlink"));
            }
            Ok(m) if !m.is_dir() => return Err(format!("the repo's {sub}/ is not a directory")),
            Ok(_) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => return Err(e.to_string()),
        }
    }
    Ok(())
}

/// One snapshot entry: its path, and the blob it resolves to when it is a
/// symlink to a regular file in this repo's `blobs/` (or itself, when it is a
/// regular file).
struct Entry {
    snapshot_path: PathBuf,
    /// The snapshot (revision) directory it lies in.
    snapshot: String,
    relative: String,
    bytes_at: PathBuf,
}

fn plan_repo(
    dir: &Path,
    repo: &str,
    files: Option<&BTreeSet<String>>,
    owned: &Owned,
    kept: &BTreeSet<String>,
) -> Result<Option<HubDeletion>, String> {
    let entries = snapshot_entries(dir)?;
    // A repo the catalog fetches whole, claimed whole, nothing kept: the
    // directory is CAR's, refs and locks included — unless a transfer is still
    // writing into it.
    if *owned == Owned::Whole && files.is_none() && kept.is_empty() {
        if transfer_in_progress(dir)? {
            return Err("the repo has a download in progress".into());
        }
        let bytes = tree_bytes(dir);
        return Ok(
            (bytes > 0 || !entries.is_empty()).then(|| HubDeletion::Repo {
                repo: repo.to_string(),
                path: dir.to_path_buf(),
                bytes,
            }),
        );
    }
    // Otherwise only CAR's entries go, never the repo: an empty or partial
    // snapshot set is not evidence that the rest is CAR's.
    let claimed_and_free =
        |e: &Entry| !kept.contains(&e.relative) && files.is_none_or(|f| f.contains(&e.relative));
    let doomed: Box<dyn Fn(&Entry) -> bool> = match owned {
        Owned::Whole => Box::new(claimed_and_free),
        Owned::Files(of) => {
            let owns = |e: &Entry| {
                of.contains(&OwnedFile {
                    path: e.relative.clone(),
                    revision: None,
                }) || of.contains(&OwnedFile {
                    path: e.relative.clone(),
                    revision: Some(e.snapshot.clone()),
                })
            };
            // File-level ownership deletes a file only from a snapshot whose
            // every entry is CAR's to delete: in a snapshot another tool also
            // uses (a user's full checkout of a base repo, their own quant
            // beside CAR's), CAR's file stays, because hf tools reuse an
            // existing pointer instead of fetching their own copy.
            let clean: HashSet<&str> = {
                let mut dirty: HashSet<&str> = HashSet::new();
                for e in &entries {
                    if !(owns(e) && claimed_and_free(e)) {
                        dirty.insert(e.snapshot.as_str());
                    }
                }
                entries
                    .iter()
                    .map(|e| e.snapshot.as_str())
                    .filter(|s| !dirty.contains(s))
                    .collect()
            };
            let clean: HashSet<String> = clean.into_iter().map(str::to_string).collect();
            Box::new(move |e: &Entry| clean.contains(&e.snapshot) && owns(e) && claimed_and_free(e))
        }
    };
    let surviving: HashSet<&Path> = entries
        .iter()
        .filter(|e| !doomed(e))
        .map(|e| e.bytes_at.as_path())
        .collect();
    let mut gone_entries = Vec::new();
    let mut blobs = BTreeSet::new();
    for e in entries.iter().filter(|e| doomed(e)) {
        gone_entries.push(e.snapshot_path.clone());
        if !surviving.contains(e.bytes_at.as_path()) && e.bytes_at != e.snapshot_path {
            blobs.insert(e.bytes_at.clone());
        }
    }
    if gone_entries.is_empty() {
        return Ok(None);
    }
    let bytes = unique_file_bytes(
        blobs
            .iter()
            .chain(gone_entries.iter())
            .map(PathBuf::as_path),
    );
    Ok(Some(HubDeletion::Files {
        repo: repo.to_string(),
        entries: gone_entries,
        blobs: blobs.into_iter().collect(),
        bytes,
    }))
}

/// How long a partial blob may sit untouched before, with no lock held on
/// it, it counts as abandoned rather than a transfer in progress.
pub const ABANDONED_PARTIAL_SECS: u64 = 24 * 60 * 60;

/// Delete abandoned partial downloads from repos CAR fetches whole: a blob named as a
/// partial (`.incomplete` from Python `huggingface_hub`, `.sync.part` from
/// hf-hub), a regular file, no writer holding its lock in either library's
/// place for it, and untouched for [`ABANDONED_PARTIAL_SECS`]. Each is
/// re-checked just before it is unlinked, relative to a descriptor for its
/// directory. Repos CAR does not own are never read. Returns what went, with
/// its size, and what could not.
pub fn discard_abandoned_partials(
    hub: &Path,
    owned: &Ownership,
) -> (Vec<(PathBuf, u64)>, Vec<String>) {
    use crate::model_management::{unlink_checked, UnlinkKind};
    let now = std::time::SystemTime::now();
    let mut discarded = Vec::new();
    let mut errors = Vec::new();
    for (repo, owns) in owned {
        let Some(dir) = crate::hf_cache::checked_repo_dir_in(hub, repo) else {
            continue;
        };
        // Only a repo CAR fetches whole: in a repo it took files from, a
        // partial is not one of CAR's (partials are never recorded), and may
        // be someone's paused download — their resume point.
        if *owns != Owned::Whole {
            continue;
        }
        let Ok(read) = std::fs::read_dir(dir.join("blobs")) else {
            continue;
        };
        let dir_name = dir
            .file_name()
            .map(|n| n.to_os_string())
            .unwrap_or_default();
        for entry in read.filter_map(Result::ok) {
            let name = entry.file_name().to_string_lossy().into_owned();
            let Some(blob) = name
                .strip_suffix(".incomplete")
                .or_else(|| name.strip_suffix(".sync.part"))
            else {
                continue;
            };
            let path = entry.path();
            let abandoned = || {
                let held = [
                    dir.join("blobs").join(format!("{blob}.lock")),
                    hub.join(".locks")
                        .join(&dir_name)
                        .join(format!("{blob}.lock")),
                ]
                .iter()
                .any(|lock| lock.exists() && lock_is_held(lock));
                let meta = std::fs::symlink_metadata(&path).ok()?;
                let age = meta
                    .modified()
                    .ok()
                    .and_then(|modified| now.duration_since(modified).ok())?;
                (meta.is_file() && !held && age.as_secs() >= ABANDONED_PARTIAL_SECS)
                    .then_some(meta.len())
            };
            let Some(bytes) = abandoned() else {
                continue;
            };
            match unlink_checked(&path, UnlinkKind::File) {
                Ok(()) => discarded.push((path, bytes)),
                Err(error) => errors.push(format!("{}: {error}", path.display())),
            }
        }
    }
    (discarded, errors)
}

/// Whether a transfer is writing into the repo, from a CAR process this one's
/// locks do not cover or from another tool. Both hub writers hold an `flock`
/// while they write: hf-hub (CAR's own downloader) on `blobs/<blob>.lock`,
/// Python `huggingface_hub` on `<hub>/.locks/models--…/<blob>.lock`. A held
/// lock is a live transfer — including one whose process is suspended, as a
/// sleeping laptop's is. Failing that, a partial blob (`.incomplete`,
/// `.sync.part`) modified within [`ABANDONED_PARTIAL_SECS`] still counts as
/// live, and one whose age cannot be read does too. A partial with no lock
/// and older than that belongs to a download whose process is gone; it goes
/// with the repo, and a later resume of it starts over.
pub(crate) fn transfer_in_progress(dir: &Path) -> Result<bool, String> {
    let mut lock_dirs = vec![dir.join("blobs")];
    if let (Some(hub), Some(name)) = (dir.parent(), dir.file_name()) {
        lock_dirs.push(hub.join(".locks").join(name));
    }
    for lock_dir in &lock_dirs {
        let read = match std::fs::read_dir(lock_dir) {
            Ok(read) => read,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
            Err(e) => return Err(e.to_string()),
        };
        for entry in read {
            let path = entry.map_err(|e| e.to_string())?.path();
            if path.extension().is_some_and(|e| e == "lock") && lock_is_held(&path) {
                return Ok(true);
            }
        }
    }
    let now = std::time::SystemTime::now();
    match std::fs::read_dir(dir.join("blobs")) {
        Ok(read) => {
            for entry in read {
                let entry = entry.map_err(|e| e.to_string())?;
                let name = entry.file_name().to_string_lossy().into_owned();
                if !(name.ends_with(".incomplete") || name.ends_with(".sync.part")) {
                    continue;
                }
                let age = entry
                    .metadata()
                    .and_then(|m| m.modified())
                    .ok()
                    .and_then(|modified| now.duration_since(modified).ok());
                if age.is_none_or(|age| age.as_secs() < ABANDONED_PARTIAL_SECS) {
                    return Ok(true);
                }
            }
            Ok(false)
        }
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
        Err(e) => Err(e.to_string()),
    }
}

/// Whether another process holds an `flock` on `path`: a non-blocking
/// exclusive attempt that fails with "would block". A file that cannot be
/// opened to probe counts as held.
#[cfg(unix)]
fn lock_is_held(path: &Path) -> bool {
    use std::os::fd::AsRawFd;
    let Ok(file) = std::fs::File::open(path) else {
        return true;
    };
    // SAFETY: flock on a descriptor this function owns for its whole life.
    let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
    if rc == 0 {
        // SAFETY: as above; release the probe's own lock.
        unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_UN) };
        false
    } else {
        true
    }
}

#[cfg(not(unix))]
fn lock_is_held(_path: &Path) -> bool {
    // No probe here; the partial-age rule still applies.
    false
}

/// Bytes of the regular files among `paths`, each inode counted once.
fn unique_file_bytes<'a>(paths: impl Iterator<Item = &'a Path>) -> u64 {
    let mut seen = HashSet::new();
    let mut total = 0;
    for path in paths {
        let Ok(meta) = std::fs::symlink_metadata(path) else {
            continue;
        };
        if !meta.is_file() {
            continue;
        }
        #[cfg(unix)]
        let key = {
            use std::os::unix::fs::MetadataExt;
            (meta.dev(), meta.ino())
        };
        #[cfg(not(unix))]
        let key = path.to_path_buf();
        if seen.insert(key) {
            total += meta.len();
        }
    }
    total
}

/// Every file entry under `snapshots/*/`, with what it resolves to. Only a
/// missing `snapshots/` reads as empty; any other I/O error is an error.
fn snapshot_entries(dir: &Path) -> Result<Vec<Entry>, String> {
    let io = |e: std::io::Error| e.to_string();
    let blobs = match dir.join("blobs").canonicalize() {
        Ok(blobs) if blobs == dir.join("blobs") => Some(blobs),
        Ok(_) => return Err("the repo's blobs/ resolves elsewhere".into()),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
        Err(e) => return Err(io(e)),
    };
    let snapshots = dir.join("snapshots");
    let read = match std::fs::read_dir(&snapshots) {
        Ok(read) => read,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
        Err(e) => return Err(io(e)),
    };
    let mut out = Vec::new();
    for snapshot in read {
        let root = snapshot.map_err(io)?.path();
        if !std::fs::symlink_metadata(&root).map_err(io)?.is_dir() {
            continue;
        }
        let mut stack = vec![root.clone()];
        while let Some(at) = stack.pop() {
            for child in std::fs::read_dir(&at).map_err(io)? {
                let path = child.map_err(io)?.path();
                let meta = std::fs::symlink_metadata(&path).map_err(io)?;
                if meta.is_dir() {
                    stack.push(path);
                    continue;
                }
                let relative = path
                    .strip_prefix(&root)
                    .map_err(|e| e.to_string())?
                    .to_string_lossy()
                    .replace('\\', "/");
                let bytes_at = if meta.file_type().is_symlink() {
                    match (path.canonicalize(), &blobs) {
                        (Ok(target), Some(blobs))
                            if target.parent() == Some(blobs.as_path())
                                && std::fs::symlink_metadata(&target)
                                    .is_ok_and(|m| m.is_file()) =>
                        {
                            target
                        }
                        // Dangling, pointing outside this repo's blobs, or at
                        // anything but a regular file: only the link itself is
                        // CAR's to remove.
                        _ => path.clone(),
                    }
                } else {
                    path.clone()
                };
                out.push(Entry {
                    snapshot_path: path,
                    snapshot: root
                        .file_name()
                        .map(|n| n.to_string_lossy().into_owned())
                        .unwrap_or_default(),
                    relative,
                    bytes_at,
                });
            }
        }
    }
    Ok(out)
}

/// Whether `path` is a link, or a directory tree holding nothing but links and
/// directories — the shape of a projection CAR creates lazily. Unreadable
/// counts as no.
pub fn links_only(path: &Path) -> bool {
    let mut stack = vec![path.to_path_buf()];
    while let Some(at) = stack.pop() {
        let Ok(meta) = std::fs::symlink_metadata(&at) else {
            return false;
        };
        if meta.file_type().is_symlink() {
            continue;
        }
        if !meta.is_dir() {
            return false;
        }
        let Ok(read) = std::fs::read_dir(&at) else {
            return false;
        };
        for child in read {
            let Ok(child) = child else {
                return false;
            };
            stack.push(child.path());
        }
    }
    true
}

/// Whose a hub repo that no registry row references is.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum OrphanOwner {
    /// Every entry in every snapshot is one CAR recorded fetching (or CAR
    /// owns the repo whole), and nothing in its catalog wants it now. A repo
    /// CAR took one file from is not this: the rest of it is someone else's
    /// checkout.
    Car,
    /// Nothing says CAR fetched it: another tool's, or the user's. Reported,
    /// never touched.
    User,
}

/// A `models--…` repo in the hub that no registry row references.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct HubOrphan {
    pub repo: String,
    /// Not published: the repo id names it, and a canonical cache path is
    /// diagnostic detail the wire does not need.
    #[serde(skip)]
    pub path: PathBuf,
    pub bytes: u64,
    pub owner: OrphanOwner,
}

/// Whether `name` is exactly one ordinary path component — the only shape a
/// model name may take when it is joined onto the models directory.
pub(crate) fn single_component(name: &str) -> bool {
    let mut components = Path::new(name).components();
    matches!(
        (components.next(), components.next()),
        (Some(std::path::Component::Normal(_)), None)
    )
}

/// A directory under the models directory, named for a model the signed
/// catalog revoked, that still holds real files. A revocation removes CAR's
/// links and never real bytes — so a compromised catalog key cannot order data
/// wiped — and discovery skips the name, so nothing else would mention it.
/// The record says a revoked model had this directory, not that these bytes
/// are that model's: a user's own model put there later is listed too.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, serde::Deserialize)]
pub struct RevokedCopy {
    /// The revoked model the directory was named for.
    pub model_id: String,
    pub path: PathBuf,
    /// Real-file bytes; links are not followed.
    pub bytes: u64,
}

/// Directories still holding real files that are named for revoked models:
/// each installed revoked row's `models_dir/<name>`, and the path its install
/// receipt records when that is a copy rather than a link. Reads only. Reported
/// and never deleted, by maintenance or anything else: removing them is the
/// user's decision.
pub fn revoked_copies(state_root: &Path, models_dir: &Path) -> Vec<RevokedCopy> {
    let names = crate::catalog::load_revoked_names(&crate::catalog::revoked_names_path(state_root));
    if names.is_empty() {
        return Vec::new();
    }
    let mut seen = HashSet::new();
    let mut out = Vec::new();
    for (name, id) in names {
        // The record is unsigned state; a name that is not one plain
        // component would point the walk outside the models directory.
        if !single_component(&name) {
            continue;
        }
        let receipt =
            // Never a `ModelManagementStore`: constructing one resumes
            // interrupted removals, and this path only reads.
            crate::model_management::read_receipt(state_root, &id).ok().flatten().filter(|r| {
                r.artifact_kind != crate::model_management::ManagedArtifactKind::Symlink
            });
        for path in std::iter::once(models_dir.join(&name)).chain(receipt.map(|r| r.managed_path)) {
            if std::fs::symlink_metadata(&path).is_err()
                || links_only(&path)
                || !seen.insert(path.clone())
            {
                continue;
            }
            out.push(RevokedCopy {
                model_id: id.clone(),
                bytes: tree_bytes(&path),
                path,
            });
        }
    }
    out
}

/// The first entry under `models_dir` — a link, or a link inside a directory
/// tree — that resolves into `repo_dir` (both canonical), if any. Bounded to
/// four levels, which covers every projection shape CAR creates.
pub(crate) fn path_resolving_into(models_dir: &Path, repo_dir: &Path) -> Option<PathBuf> {
    let mut stack = vec![(models_dir.to_path_buf(), 0usize)];
    while let Some((dir, depth)) = stack.pop() {
        let Ok(read) = std::fs::read_dir(&dir) else {
            continue;
        };
        for entry in read.filter_map(Result::ok) {
            let path = entry.path();
            let Ok(meta) = std::fs::symlink_metadata(&path) else {
                continue;
            };
            if meta.file_type().is_symlink() {
                if path
                    .canonicalize()
                    .is_ok_and(|target| target.starts_with(repo_dir))
                {
                    return Some(path);
                }
            } else if meta.is_dir() && depth < 4 {
                stack.push((path, depth + 1));
            }
        }
    }
    None
}

/// Whether CAR owns everything in the repo at `dir`: the whole repo, or a
/// recorded file (in its revision) for every entry of every snapshot. An
/// empty or unreadable repo is not CAR's.
fn wholly_cars(dir: &Path, owned: Option<&Owned>) -> bool {
    match owned {
        Some(Owned::Whole) => true,
        Some(Owned::Files(of)) => snapshot_entries(dir).is_ok_and(|entries| {
            !entries.is_empty()
                && entries.iter().all(|e| {
                    of.contains(&OwnedFile {
                        path: e.relative.clone(),
                        revision: None,
                    }) || of.contains(&OwnedFile {
                        path: e.relative.clone(),
                        revision: Some(e.snapshot.clone()),
                    })
                })
        }),
        None => false,
    }
}

/// The model repos in `hub` that no row in `referenced` names. A Hub repo id
/// cannot contain `--`, so `models--owner--name` decodes unambiguously; a
/// directory that does not decode to a valid id is skipped, not guessed at.
pub fn unreferenced_repos(
    hub: &Path,
    referenced: &std::collections::HashSet<String>,
    owned: &Ownership,
) -> Vec<HubOrphan> {
    let Ok(read) = std::fs::read_dir(hub) else {
        return Vec::new();
    };
    let mut out = Vec::new();
    for entry in read.filter_map(Result::ok) {
        let name = entry.file_name().to_string_lossy().into_owned();
        let Some(encoded) = name.strip_prefix("models--") else {
            continue;
        };
        let repo = match encoded.split_once("--") {
            Some((owner, rest)) if !rest.contains("--") => format!("{owner}/{rest}"),
            Some(_) => continue,
            None => encoded.to_string(),
        };
        if referenced.contains(&repo)
            || crate::hf_cache::checked_repo_dir_in(hub, &repo).as_deref()
                != Some(entry.path().as_path())
            || !entry.file_type().is_ok_and(|t| t.is_dir())
        {
            continue;
        }
        out.push(HubOrphan {
            owner: if wholly_cars(&entry.path(), owned.get(&repo)) {
                OrphanOwner::Car
            } else {
                OrphanOwner::User
            },
            bytes: tree_bytes(&entry.path()),
            path: entry.path(),
            repo,
        });
    }
    out.sort_by(|a, b| a.repo.cmp(&b.repo));
    out
}

/// Bytes of regular files under `dir` (or `dir` itself), symlinks not followed,
/// each inode once. Unreadable subtrees are skipped, so this can understate.
pub(crate) fn tree_bytes(dir: &Path) -> u64 {
    // Removing a link frees nothing, whatever it points at — and `read_dir`
    // would follow it into bytes that are not this path's.
    match std::fs::symlink_metadata(dir) {
        Ok(meta) if meta.file_type().is_symlink() => return 0,
        Ok(meta) if meta.is_file() => return meta.len(),
        Ok(_) => {}
        Err(_) => return 0,
    }
    let mut files = Vec::new();
    let mut stack = vec![dir.to_path_buf()];
    while let Some(at) = stack.pop() {
        let Ok(read) = std::fs::read_dir(&at) else {
            continue;
        };
        for child in read.filter_map(Result::ok) {
            let Ok(meta) = std::fs::symlink_metadata(child.path()) else {
                continue;
            };
            if meta.is_dir() {
                stack.push(child.path());
            } else if meta.is_file() {
                files.push(child.path());
            }
        }
    }
    unique_file_bytes(files.iter().map(PathBuf::as_path))
}

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

    /// Video rows reference the text encoder the LTX runtime loads, so it is
    /// neither an orphan nor kept when the video model retires.
    #[test]
    fn a_video_row_references_its_text_encoder() {
        let video = crate::registry::builtin_catalog()
            .into_iter()
            .find(crate::registry::uses_ltx_text_encoder)
            .expect("a catalog LTX row");
        assert!(hf_references(&video).contains(&HfReference {
            repo: crate::registry::LTX_TEXT_ENCODER.to_string(),
            files: None,
        }));
    }

    #[test]
    fn unreferenced_repos_are_split_by_whose_they_are() {
        let hub = tempfile::tempdir().unwrap();
        for dir in [
            "models--org--wanted",
            "models--org--dropped",
            "models--someone--theirs",
            "models--bad--na--me",
            "datasets--org--data",
        ] {
            std::fs::create_dir_all(hub.path().join(dir).join("blobs")).unwrap();
        }
        std::fs::write(
            hub.path().join("models--org--dropped/blobs/b"),
            vec![0u8; 9],
        )
        .unwrap();
        let referenced = std::collections::HashSet::from(["org/wanted".to_string()]);
        // A snapshot whose every entry CAR recorded is CAR's; one recorded
        // file beside a file CAR did not fetch is not.
        let snap = hub.path().join("models--org--fetched/snapshots/r1");
        std::fs::create_dir_all(&snap).unwrap();
        std::fs::write(snap.join("model.safetensors"), b"w").unwrap();
        std::fs::write(snap.join("config.json"), b"c").unwrap();
        let mixed = hub.path().join("models--org--mixed/snapshots/r1");
        std::fs::create_dir_all(&mixed).unwrap();
        std::fs::write(mixed.join("tokenizer.json"), b"t").unwrap();
        std::fs::write(mixed.join("weights.bin"), b"user").unwrap();
        let file = |path: &str, rev: Option<&str>| OwnedFile {
            path: path.into(),
            revision: rev.map(str::to_owned),
        };
        let owned = Ownership::from([
            ("org/dropped".to_string(), Owned::Whole),
            ("someone/theirs".to_string(), Owned::Files(BTreeSet::new())),
            (
                "org/fetched".to_string(),
                Owned::Files(BTreeSet::from([
                    file("model.safetensors", Some("r1")),
                    file("config.json", None),
                ])),
            ),
            (
                "org/mixed".to_string(),
                Owned::Files(BTreeSet::from([file("tokenizer.json", None)])),
            ),
        ]);
        let orphans = unreferenced_repos(hub.path(), &referenced, &owned);
        let summary: Vec<(&str, OrphanOwner, u64)> = orphans
            .iter()
            .map(|o| (o.repo.as_str(), o.owner, o.bytes))
            .collect();
        assert_eq!(
            summary,
            vec![
                ("org/dropped", OrphanOwner::Car, 9),
                ("org/fetched", OrphanOwner::Car, 2),
                ("org/mixed", OrphanOwner::User, 5),
                ("someone/theirs", OrphanOwner::User, 0)
            ]
        );
    }

    /// A digest approved before a restart matches after it: discovery order
    /// (HashMap targets, `read_dir` entries) never reaches the hash.
    #[test]
    fn the_digest_ignores_discovery_order() {
        let plan = |twins: [&str; 2], entries: [&str; 2]| {
            let mut plan = RetirePlan {
                model_id: "a".into(),
                also_retires: twins.iter().map(|t| t.to_string()).collect(),
                deletions: vec![HubDeletion::Files {
                    repo: "o/r".into(),
                    entries: entries.iter().map(PathBuf::from).collect(),
                    blobs: Vec::new(),
                    bytes: 1,
                }],
                ..RetirePlan::default()
            };
            plan.seal();
            plan.digest
        };
        assert_eq!(plan(["b", "c"], ["x", "y"]), plan(["c", "b"], ["y", "x"]));
        assert_ne!(plan(["b", "c"], ["x", "y"]), plan(["b", "d"], ["x", "y"]));
    }
    use crate::schema::{CostModel, ModelCapability, PerformanceEnvelope, TrustTier};

    fn row(id: &str, name: &str, source: ModelSource) -> ModelSchema {
        ModelSchema {
            id: id.into(),
            name: name.into(),
            provider: "test".into(),
            family: "qwen3".into(),
            version: String::new(),
            capabilities: vec![ModelCapability::Generate],
            context_length: 4096,
            max_output_tokens: None,
            param_count: "4B".into(),
            quantization: None,
            performance: PerformanceEnvelope::default(),
            cost: CostModel::default(),
            source,
            tags: vec![],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: TrustTier::Curated,
            deprecated: false,
            available: true,
            weights_ready: true,
        }
    }

    fn mlx(id: &str, repo: &str) -> ModelSchema {
        row(
            id,
            id,
            ModelSource::Mlx {
                hf_repo: repo.into(),
                hf_weight_file: None,
            },
        )
    }

    #[cfg(unix)]
    fn gguf(id: &str, repo: &str, file: &str, tokenizer: &str) -> ModelSchema {
        row(
            id,
            id,
            ModelSource::Local {
                hf_repo: repo.into(),
                hf_filename: file.into(),
                tokenizer_repo: tokenizer.into(),
            },
        )
    }

    /// Lay out a repo the way hf-hub does: blobs, and snapshot entries that
    /// are symlinks to them.
    #[cfg(unix)]
    fn repo(hub: &Path, id: &str, files: &[(&str, &str, usize)]) -> PathBuf {
        let dir = crate::hf_cache::repo_dir_in(hub, id);
        let snap = dir.join("snapshots/abc");
        std::fs::create_dir_all(dir.join("blobs")).unwrap();
        std::fs::create_dir_all(dir.join("refs")).unwrap();
        std::fs::write(dir.join("refs/main"), "abc").unwrap();
        for (path, blob, size) in files {
            let blob_path = dir.join("blobs").join(blob);
            if !blob_path.exists() {
                std::fs::write(&blob_path, vec![0u8; *size]).unwrap();
            }
            let entry = snap.join(path);
            std::fs::create_dir_all(entry.parent().unwrap()).unwrap();
            #[cfg(unix)]
            std::os::unix::fs::symlink(&blob_path, &entry).unwrap();
        }
        dir
    }

    /// Ownership as the catalog would give it for these rows: a repo fetched
    /// whole is whole, a repo fetched file by file is just those files.
    fn as_catalog(rows: &[&ModelSchema]) -> Ownership {
        let mut out = Ownership::new();
        for r in rows.iter().flat_map(|row| hf_references(row)) {
            let entry = out.entry(r.repo).or_insert(Owned::Files(BTreeSet::new()));
            match (entry, r.files) {
                (slot, None) => *slot = Owned::Whole,
                (Owned::Files(set), Some(files)) => {
                    set.extend(files.into_iter().map(|path| OwnedFile {
                        path,
                        revision: None,
                    }))
                }
                (Owned::Whole, Some(_)) => {}
            }
        }
        out
    }

    /// Plan with the target owned as a catalog row is.
    fn plan(
        target: &ModelSchema,
        registry: &[&ModelSchema],
        installed: &[&str],
        hub: &Path,
    ) -> RetirePlan {
        plan_owned(target, registry, installed, hub, &as_catalog(&[target]))
    }

    fn plan_owned(
        target: &ModelSchema,
        registry: &[&ModelSchema],
        installed: &[&str],
        hub: &Path,
        owned: &Ownership,
    ) -> RetirePlan {
        let installed: HashSet<String> = installed.iter().map(|s| s.to_string()).collect();
        let is_installed = |m: &ModelSchema| installed.contains(&m.id);
        plan_hub_retirement(&RetireInputs {
            targets: &[target],
            registry,
            installed: &is_installed,
            owned,
            hub,
        })
    }

    /// Add a snapshot named `rev` to an existing repo layout.
    #[cfg(unix)]
    fn snapshot(dir: &Path, rev: &str, files: &[(&str, &str, usize)]) {
        for (path, blob, size) in files {
            let blob_path = dir.join("blobs").join(blob);
            if !blob_path.exists() {
                std::fs::write(&blob_path, vec![0u8; *size]).unwrap();
            }
            let entry = dir.join("snapshots").join(rev).join(path);
            std::fs::create_dir_all(entry.parent().unwrap()).unwrap();
            std::os::unix::fs::symlink(&blob_path, &entry).unwrap();
        }
    }

    fn hub() -> (tempfile::TempDir, PathBuf) {
        let tmp = tempfile::tempdir().unwrap();
        let hub = tmp.path().canonicalize().unwrap().join("hub");
        std::fs::create_dir_all(&hub).unwrap();
        (tmp, hub)
    }

    #[cfg(unix)]
    #[test]
    fn an_mlx_row_owns_its_whole_repo() {
        let (_tmp, hub) = hub();
        let dir = repo(
            &hub,
            "mlx-community/Qwen3-30B-A3B-4bit",
            &[("config.json", "c", 10), ("model.safetensors", "w", 1000)],
        );
        let target = mlx("mlx/qwen3-30b-a3b:4bit", "mlx-community/Qwen3-30B-A3B-4bit");
        let p = plan(&target, &[&target], &[], &hub);
        assert!(!p.is_refused(), "{p:?}");
        assert_eq!(
            p.deletions,
            vec![HubDeletion::Repo {
                repo: "mlx-community/Qwen3-30B-A3B-4bit".into(),
                path: dir,
                bytes: 1013,
            }]
        );
        assert_eq!(p.freed_bytes, 1013);
    }

    /// A GGUF row takes one quant and one tokenizer, and owns only those.
    /// CAR's tokenizer stays in a snapshot the user's own full checkout of the
    /// base model also uses; where the snapshot is CAR's alone, it goes.
    #[cfg(unix)]
    #[test]
    fn a_gguf_row_owns_only_the_files_it_fetched() {
        let (_tmp, hub) = hub();
        let quant = repo(
            &hub,
            "Qwen/Qwen3-8B-GGUF",
            &[("Qwen3-8B-Q4_K_M.gguf", "q4", 500)],
        );
        repo(
            &hub,
            "Qwen/Qwen3-8B",
            &[
                ("tokenizer.json", "tok", 7),
                ("model-00001.safetensors", "big", 9000),
                ("config.json", "cfg", 3),
            ],
        );
        let target = gguf(
            "qwen/qwen3-8b:q4_k_m",
            "Qwen/Qwen3-8B-GGUF",
            "Qwen3-8B-Q4_K_M.gguf",
            "Qwen/Qwen3-8B",
        );
        let p = plan(&target, &[&target], &[], &hub);
        assert!(!p.is_refused(), "{p:?}");
        assert_eq!(
            p.deletions,
            vec![HubDeletion::Files {
                repo: "Qwen/Qwen3-8B-GGUF".into(),
                entries: vec![quant.join("snapshots/abc/Qwen3-8B-Q4_K_M.gguf")],
                blobs: vec![quant.join("blobs/q4")],
                bytes: 500,
            }],
            "the user's base checkout keeps the shared tokenizer"
        );

        // A base repo holding only CAR's tokenizer gives it up.
        let (_tmp2, hub2) = super::tests::hub();
        repo(&hub2, "Qwen/Qwen3-8B", &[("tokenizer.json", "tok", 7)]);
        let p = plan(&target, &[&target], &[], &hub2);
        assert!(
            matches!(&p.deletions[..],
            [HubDeletion::Files { repo, bytes: 7, .. }] if repo == "Qwen/Qwen3-8B"),
            "{p:?}"
        );
    }

    /// Two rows backed by one repo (`mlx/gemma-4-12b-it:4bit` and
    /// `vllm-mlx/gemma-4-12b-it`): retiring one keeps the repo for the other.
    #[cfg(unix)]
    #[test]
    fn a_repo_another_installed_row_uses_is_kept() {
        let (_tmp, hub) = hub();
        repo(
            &hub,
            "mlx-community/gemma-4-12B-it-4bit",
            &[("w", "w", 100)],
        );
        let a = mlx(
            "mlx/gemma-4-12b-it:4bit",
            "mlx-community/gemma-4-12B-it-4bit",
        );
        let b = mlx(
            "vllm-mlx/gemma-4-12b-it",
            "mlx-community/gemma-4-12B-it-4bit",
        );
        let p = plan(&a, &[&a, &b], &["vllm-mlx/gemma-4-12b-it"], &hub);
        assert!(p.deletions.is_empty(), "{p:?}");
        assert_eq!(p.kept[0].kept_for, "vllm-mlx/gemma-4-12b-it");
        // Not installed, it holds nothing.
        let p = plan(&a, &[&a, &b], &[], &hub);
        assert_eq!(p.deletions.len(), 1);
    }

    /// A blob two entries share survives while one of them stays.
    #[cfg(unix)]
    #[test]
    fn a_shared_blob_survives_while_any_entry_keeps_it() {
        let (_tmp, hub) = hub();
        let dir = repo(&hub, "Qwen/Qwen3-8B", &[("tokenizer.json", "same", 7)]);
        // The same blob, reached from a snapshot CAR does not own.
        snapshot(&dir, "def", &[("other.json", "same", 7)]);
        let target = gguf("g", "Qwen/Qwen3-8B-GGUF", "q.gguf", "Qwen/Qwen3-8B");
        let p = plan(&target, &[&target], &[], &hub);
        let [HubDeletion::Files {
            entries,
            blobs,
            bytes,
            ..
        }] = &p.deletions[..]
        else {
            panic!("{p:?}");
        };
        assert_eq!(entries, &vec![dir.join("snapshots/abc/tokenizer.json")]);
        assert!(blobs.is_empty(), "another snapshot still points at it");
        assert_eq!(*bytes, 0);
    }

    /// A repo neither the catalog nor a recorded download names is never CAR's
    /// — even when the target itself, a user row, points at it.
    #[cfg(unix)]
    #[test]
    fn an_unowned_repo_is_never_touched_even_by_its_own_row() {
        let (_tmp, hub) = hub();
        repo(&hub, "google/flan-t5-large", &[("w", "w", 100)]);
        let target = mlx("mlx/x", "google/flan-t5-large");
        let p = plan_owned(&target, &[&target], &[], &hub, &Ownership::new());
        assert!(p.deletions.is_empty());
        assert_eq!(p.unowned, vec!["google/flan-t5-large".to_string()]);
    }

    /// Rows retired together do not keep each other's repo.
    #[cfg(unix)]
    #[test]
    fn twins_retired_together_free_their_shared_repo() {
        let (_tmp, hub) = hub();
        repo(
            &hub,
            "mlx-community/gemma-4-12B-it-4bit",
            &[("w", "w", 100)],
        );
        let a = mlx(
            "mlx/gemma-4-12b-it:4bit",
            "mlx-community/gemma-4-12B-it-4bit",
        );
        let b = mlx(
            "vllm-mlx/gemma-4-12b-it",
            "mlx-community/gemma-4-12B-it-4bit",
        );
        let owned: Ownership = [(
            "mlx-community/gemma-4-12B-it-4bit".to_string(),
            Owned::Whole,
        )]
        .into();
        let installed = |_: &ModelSchema| true;
        let p = plan_hub_retirement(&RetireInputs {
            targets: &[&a, &b],
            registry: &[&a, &b],
            installed: &installed,
            owned: &owned,
            hub: &hub,
        });
        assert!(p.kept.is_empty(), "{p:?}");
        assert_eq!(p.also_retires, vec!["vllm-mlx/gemma-4-12b-it".to_string()]);
        assert!(matches!(
            &p.deletions[..],
            [HubDeletion::Repo { bytes: 103, .. }]
        ));
    }

    /// A repo only a recorded download makes CAR's yields only the recorded
    /// files, in the snapshot they were fetched into — even to a row that
    /// claims the whole repo.
    #[cfg(unix)]
    #[test]
    fn a_recorded_only_repo_is_owned_file_by_file_and_revision() {
        let (_tmp, hub) = hub();
        let dir = repo(&hub, "someone/model", &[("model.safetensors", "w", 100)]);
        // The user later fetched another revision with the same file name.
        snapshot(&dir, "def", &[("model.safetensors", "w2", 400)]);
        let target = mlx("user/model", "someone/model");
        let recorded: Ownership = [(
            "someone/model".to_string(),
            Owned::Files(BTreeSet::from([OwnedFile {
                path: "model.safetensors".into(),
                revision: Some("abc".into()),
            }])),
        )]
        .into();
        let p = plan_owned(
            &target,
            &[&target],
            &[],
            &hub,
            &ownership(Ownership::new(), recorded),
        );
        let [HubDeletion::Files { entries, bytes, .. }] = &p.deletions[..] else {
            panic!("{p:?}");
        };
        assert_eq!(entries, &vec![dir.join("snapshots/abc/model.safetensors")]);
        assert_eq!(*bytes, 100, "revision def is not CAR's");
        // The catalog's word is the whole repo.
        let catalog: Ownership = [("someone/model".to_string(), Owned::Whole)].into();
        let p = plan_owned(
            &target,
            &[&target],
            &[],
            &hub,
            &ownership(catalog, Ownership::new()),
        );
        assert!(matches!(&p.deletions[..], [HubDeletion::Repo { .. }]));
    }

    /// A user row claiming the whole of a repo the catalog takes one file
    /// from never deletes the repo — or the user's checkout in it.
    #[cfg(unix)]
    #[test]
    fn a_whole_claim_on_a_file_level_catalog_repo_is_never_a_repo_delete() {
        let (_tmp, hub) = hub();
        repo(
            &hub,
            "Qwen/Qwen3-8B",
            &[
                ("tokenizer.json", "tok", 7),
                ("model-00001.safetensors", "big", 9000),
            ],
        );
        let catalog_gguf = gguf(
            "qwen/qwen3-8b:q4_k_m",
            "Qwen/Qwen3-8B-GGUF",
            "q.gguf",
            "Qwen/Qwen3-8B",
        );
        let user = mlx("user/qwen3-8b", "Qwen/Qwen3-8B");
        let p = plan_owned(
            &user,
            &[&user, &catalog_gguf],
            &[],
            &hub,
            &as_catalog(&[&catalog_gguf]),
        );
        assert!(p.deletions.is_empty(), "{p:?}");
    }

    /// CAR's own downloader writes `.sync.part`; another CAR process's
    /// download in flight stops a whole-repo delete.
    #[cfg(unix)]
    #[test]
    fn a_whole_repo_with_an_hf_hub_partial_blob_is_refused() {
        let (_tmp, hub) = hub();
        let dir = repo(&hub, "mlx-community/Qwen3-4B-4bit", &[("w", "w", 10)]);
        let partial = dir.join("blobs/z.sync.part");
        std::fs::write(&partial, b"partial").unwrap();
        let target = mlx("mlx/q", "mlx-community/Qwen3-4B-4bit");
        let p = plan(&target, &[&target], &[], &hub);
        assert!(p.deletions.is_empty());
        assert!(
            matches!(&p.refusals[..], [RetireRefusal::UnsafePath { .. }]),
            "{p:?}"
        );

        // A partial untouched for over a day is abandoned, not in progress.
        let stale = std::time::SystemTime::now()
            - std::time::Duration::from_secs(ABANDONED_PARTIAL_SECS + 60);
        std::fs::File::options()
            .write(true)
            .open(&partial)
            .unwrap()
            .set_modified(stale)
            .unwrap();
        let p = plan(&target, &[&target], &[], &hub);
        assert!(
            matches!(&p.deletions[..], [HubDeletion::Repo { .. }]),
            "{p:?}"
        );
    }

    #[cfg(unix)]
    #[test]
    fn links_only_tells_a_projection_from_user_content() {
        let tmp = tempfile::tempdir().unwrap();
        let proj = tmp.path().join("proj");
        std::fs::create_dir_all(proj.join("sub")).unwrap();
        std::os::unix::fs::symlink("/nowhere", proj.join("a")).unwrap();
        std::os::unix::fs::symlink("/nowhere", proj.join("sub/b")).unwrap();
        assert!(links_only(&proj));
        std::fs::write(proj.join("sub/real.bin"), b"user bytes").unwrap();
        assert!(!links_only(&proj));
        let link = tmp.path().join("link");
        std::os::unix::fs::symlink(&proj, &link).unwrap();
        assert!(
            links_only(&link),
            "a link is CAR's to unlink, whatever it points at"
        );
    }

    #[cfg(unix)]
    #[test]
    fn symlinked_blobs_or_snapshots_refuse_the_plan() {
        for sub in ["blobs", "snapshots"] {
            let (tmp, hub) = hub();
            let dir = repo(&hub, "Qwen/Qwen3-8B", &[("tokenizer.json", "tok", 7)]);
            let elsewhere = tmp.path().join("elsewhere");
            std::fs::rename(dir.join(sub), &elsewhere).unwrap();
            std::os::unix::fs::symlink(&elsewhere, dir.join(sub)).unwrap();
            let target = gguf("g", "Qwen/Qwen3-8B-GGUF", "q.gguf", "Qwen/Qwen3-8B");
            let p = plan(&target, &[&target], &[], &hub);
            assert!(
                p.refusals
                    .iter()
                    .any(|r| matches!(r, RetireRefusal::UnsafePath { .. })),
                "{sub}: {p:?}"
            );
            assert!(p.deletions.is_empty(), "{sub}");
        }
    }

    /// A file-level owner of a repo with no snapshot entries — another tool's
    /// download mid-flight — deletes nothing.
    #[cfg(unix)]
    #[test]
    fn an_empty_or_partial_repo_is_not_widened_to_a_whole_repo_delete() {
        let (_tmp, hub) = hub();
        let dir = crate::hf_cache::repo_dir_in(&hub, "Qwen/Qwen3-8B");
        std::fs::create_dir_all(dir.join("blobs")).unwrap();
        std::fs::write(dir.join("blobs/x.incomplete"), vec![0u8; 900]).unwrap();
        let target = gguf("g", "Qwen/Qwen3-8B-GGUF", "q.gguf", "Qwen/Qwen3-8B");
        let p = plan(&target, &[&target], &[], &hub);
        assert!(p.deletions.is_empty(), "{p:?}");
        assert_eq!(p.freed_bytes, 0);
    }

    /// A whole-repo owner still waits for a transfer writing into the repo.
    #[cfg(unix)]
    #[test]
    fn a_whole_repo_with_an_incomplete_blob_is_refused() {
        let (_tmp, hub) = hub();
        let dir = repo(&hub, "mlx-community/Qwen3-4B-4bit", &[("w", "w", 10)]);
        std::fs::write(dir.join("blobs/y.incomplete"), b"partial").unwrap();
        let target = mlx("mlx/q", "mlx-community/Qwen3-4B-4bit");
        let p = plan(&target, &[&target], &[], &hub);
        assert!(p.deletions.is_empty());
        assert!(
            matches!(&p.refusals[..], [RetireRefusal::UnsafePath { .. }]),
            "{p:?}"
        );
    }

    /// An unreadable subtree refuses; it never shrinks the evidence.
    #[cfg(unix)]
    #[test]
    fn an_unreadable_snapshot_refuses_the_plan() {
        use std::os::unix::fs::PermissionsExt;
        let (_tmp, hub) = hub();
        let dir = repo(
            &hub,
            "Qwen/Qwen3-8B",
            &[("tokenizer.json", "tok", 7), ("sub/x", "x", 3)],
        );
        let locked = dir.join("snapshots/abc/sub");
        std::fs::set_permissions(&locked, std::fs::Permissions::from_mode(0o000)).unwrap();
        let target = gguf("g", "Qwen/Qwen3-8B-GGUF", "q.gguf", "Qwen/Qwen3-8B");
        let p = plan(&target, &[&target], &[], &hub);
        std::fs::set_permissions(&locked, std::fs::Permissions::from_mode(0o755)).unwrap();
        assert!(p.deletions.is_empty(), "{p:?}");
        assert!(p
            .refusals
            .iter()
            .any(|r| matches!(r, RetireRefusal::UnsafePath { .. })));
    }

    /// A blob hardlinked into the snapshot is counted once.
    #[cfg(unix)]
    #[test]
    fn hardlinked_bytes_are_counted_once() {
        let (_tmp, hub) = hub();
        let dir = crate::hf_cache::repo_dir_in(&hub, "mlx-community/H");
        std::fs::create_dir_all(dir.join("snapshots/s")).unwrap();
        std::fs::create_dir_all(dir.join("blobs")).unwrap();
        std::fs::write(dir.join("blobs/b"), vec![0u8; 100]).unwrap();
        std::fs::hard_link(dir.join("blobs/b"), dir.join("snapshots/s/w")).unwrap();
        let target = mlx("mlx/h", "mlx-community/H");
        let p = plan(&target, &[&target], &[], &hub);
        assert_eq!(p.freed_bytes, 100, "{p:?}");
    }

    #[cfg(unix)]
    #[test]
    fn unsafe_paths_refuse_the_plan() {
        let (tmp, hub) = hub();
        // A repo directory that is a symlink elsewhere.
        let elsewhere = tmp.path().join("elsewhere");
        std::fs::create_dir_all(&elsewhere).unwrap();
        std::os::unix::fs::symlink(&elsewhere, hub.join("models--a--b")).unwrap();
        let target = mlx("mlx/a", "a/b");
        let p = plan(&target, &[&target], &[], &hub);
        assert!(
            matches!(&p.refusals[..], [RetireRefusal::UnsafePath { .. }]),
            "{p:?}"
        );
        assert!(p.deletions.is_empty());
        // A relative hub.
        let p = plan(&target, &[&target], &[], Path::new("rel/hub"));
        assert!(matches!(
            &p.refusals[..],
            [RetireRefusal::UnsafePath { .. }]
        ));
        // A repo id outside the Hub grammar.
        let bad = mlx("mlx/bad", "../../etc");
        let p = plan(&bad, &[&bad], &[], &hub);
        assert!(
            matches!(&p.refusals[..], [RetireRefusal::UnsafePath { .. }]),
            "{p:?}"
        );
    }

    /// A snapshot entry that points outside the repo's blobs is only a link:
    /// the link may go, never its target.
    #[cfg(unix)]
    #[test]
    fn a_link_out_of_the_repo_never_frees_its_target() {
        let (tmp, hub) = hub();
        let foreign = tmp.path().join("foreign.bin");
        std::fs::write(&foreign, vec![0u8; 50]).unwrap();
        let dir = crate::hf_cache::repo_dir_in(&hub, "Qwen/Qwen3-8B");
        std::fs::create_dir_all(dir.join("snapshots/abc")).unwrap();
        std::fs::create_dir_all(dir.join("blobs")).unwrap();
        std::os::unix::fs::symlink(&foreign, dir.join("snapshots/abc/tokenizer.json")).unwrap();
        let target = gguf("g", "Qwen/Qwen3-8B-GGUF", "q.gguf", "Qwen/Qwen3-8B");
        let p = plan(&target, &[&target], &[], &hub);
        let HubDeletion::Files { entries, blobs, .. } = &p.deletions[0] else {
            panic!("{p:?}");
        };
        assert_eq!(entries, &vec![dir.join("snapshots/abc/tokenizer.json")]);
        assert!(blobs.is_empty());
    }

    #[test]
    fn a_remote_row_has_nothing_to_retire() {
        let (_tmp, hub) = hub();
        let target = crate::openrouter::curated_schemas()
            .into_iter()
            .next()
            .expect("a remote row");
        let p = plan(&target, &[&target], &[], &hub);
        assert_eq!(p.refusals, vec![RetireRefusal::NothingToRetire]);
    }

    #[test]
    fn a_cache_path_names_its_snapshot_revision() {
        assert_eq!(
            snapshot_revision(Path::new("/h/models--a--b/snapshots/0f1e/sub/tok.json")),
            Some("0f1e".into())
        );
        assert_eq!(
            snapshot_revision(Path::new("/h/models--a--b/blobs/x")),
            None
        );
        assert_eq!(
            snapshot_revision(Path::new(
                "/Volumes/snapshots/hub/models--a--b/snapshots/r9/f"
            )),
            Some("r9".into())
        );
    }

    /// Only what is noted inside the collector is recorded, grouped per repo.
    #[tokio::test]
    async fn only_noted_fetches_are_recorded() {
        note_fetched("outside/scope", "x", Some("r0".into()));
        let ((), unknown) = collecting_fetches(async { note_fetched("a/b", "x", None) }).await;
        assert!(unknown.is_empty(), "an unknown revision is not recorded");
        let ((), fetched) = collecting_fetches(async {
            note_fetched("a/b", "model.safetensors", Some("r1".into()));
            note_fetched("a/b", "config.json", Some("r1".into()));
        })
        .await;
        assert_eq!(fetched.len(), 2);
        let root = tempfile::tempdir().unwrap();
        record_download(root.path(), Path::new("/hub"), "m", &fetched);
        let text = std::fs::read_to_string(provenance_path(root.path())).unwrap();
        assert_eq!(text.lines().count(), 1, "{text}");
        assert!(text.contains("model.safetensors") && text.contains("config.json"));
        assert!(!text.contains("outside/scope"));
        record_download(root.path(), Path::new("/hub"), "m", &[]);
        assert_eq!(
            std::fs::read_to_string(provenance_path(root.path())).unwrap(),
            text
        );
    }

    #[test]
    fn the_flux_row_takes_one_file_from_its_base_checkpoint() {
        let flux = mlx("mlx/flux", crate::registry::FLUX_AUXILIARY.0);
        let refs = hf_references(&flux);
        assert_eq!(refs.len(), 2);
        assert_eq!(refs[1].repo, crate::registry::FLUX_AUXILIARY.2);
        assert_eq!(
            refs[1].files,
            Some(BTreeSet::from([crate::registry::FLUX_AUXILIARY
                .3
                .to_string()]))
        );
    }
}