//! Walk one or more scan roots, group split-GGUF shards, parse each
//! launchable file's header on a bounded pool, and stream results to
//! the caller over an `mpsc` channel (origin: R1, R5, R9).
//!
//! The walk uses the `ignore` crate so `.gitignore` rules and the
//! caller's exclude globs are honoured for free. CPU-bound parsing
//! runs on `tokio::task::spawn_blocking` so the scan tasks don't
//! starve the runtime when the user has hundreds of GGUFs on disk.

use std::path::{Path, PathBuf};
use std::sync::Arc;

use futures::stream::{self, StreamExt};
use ignore::WalkBuilder;
use tokio::sync::mpsc;

use crate::discovery::metadata_cache::{self, CachedParse, MetadataCache};
use crate::discovery::split_gguf::{group, DiscoveredEntry};
use crate::discovery::{DiscoveredModel, ModelSource};
use crate::gguf::{read_path, summarise_metadata, GgufError, HeaderReadOptions, ModelMetadata};

/// One root to scan plus how to label files found beneath it.
#[derive(Debug, Clone)]
pub struct ScanRoot {
  pub path: PathBuf,
  pub source: ModelSource,
}

/// Options for [`scan`]. `excludes` are appended to the gitignore-
/// derived ignores; absolute or relative-to-root globs both work.
#[derive(Debug, Clone, Default)]
pub struct ScanOptions {
  pub excludes: Vec<String>,
  /// Capacity for the streaming channel. The TUI is usually faster
  /// than disk, but a tiny capacity makes back-pressure visible in
  /// tests; production defaults to a comfortable buffer.
  pub channel_capacity: Option<usize>,
  /// Optional per-file parse cache. When `Some`, unchanged
  /// `(canonical path, mtime, size)` triples reuse the cached
  /// parse instead of re-reading + re-parsing the header. The
  /// daemon's discovery task wires a shared cache so successive
  /// watcher-driven re-scans don't re-parse the whole tree.
  pub metadata_cache: Option<MetadataCache>,
}

impl ScanOptions {
  pub fn channel_capacity(&self) -> usize {
    self.channel_capacity.unwrap_or(64)
  }
}

/// Begin a scan across `roots`. Returns the receiver immediately; the
/// scan runs in the background and closes the channel when every root
/// has been walked.
///
/// Errors per-file (unreadable directories, parse failures) are
/// surfaced via `DiscoveredModel.parse_error` rather than aborting the
/// whole scan — a single bad model file should not blind the user to
/// the rest of their library (origin: R9 "scan continues with other
/// roots").
pub fn scan(roots: Vec<ScanRoot>, opts: ScanOptions) -> mpsc::Receiver<DiscoveredModel> {
  let (tx, rx) = mpsc::channel(opts.channel_capacity());
  let excludes = Arc::new(opts.excludes);
  let cache = opts.metadata_cache;
  tokio::spawn(async move {
    for root in roots {
      walk_root(root, Arc::clone(&excludes), cache.clone(), tx.clone()).await;
    }
    // dropping `tx` here closes the receiver
  });
  rx
}

async fn walk_root(
  root: ScanRoot,
  excludes: Arc<Vec<String>>,
  cache: Option<MetadataCache>,
  tx: mpsc::Sender<DiscoveredModel>,
) {
  let path = root.path.clone();
  let source = root.source;
  let excludes_for_walk = Arc::clone(&excludes);
  let paths = tokio::task::spawn_blocking(move || collect_gguf_paths(&path, &excludes_for_walk))
    .await
    .unwrap_or_else(|join_err| {
      log::warn!(
        "scan walker task for {} panicked: {join_err}",
        root.path.display()
      );
      Vec::new()
    });

  // Parse files in parallel on the blocking pool. The per-file
  // `build_discovered_model` already pushes its CPU-bound work onto
  // `spawn_blocking`; `buffer_unordered` just lets several of those
  // happen at once instead of strict one-at-a-time await. On a cold
  // HF cache with hundreds of GGUFs this cuts first-scan latency from
  // serial-disk-bound to parallel-disk-bound. We deliberately do NOT
  // use rayon — the work is mostly waiting on disk, and we want the
  // tokio scheduler to interleave with the rest of the daemon.
  let entries: Vec<_> = group(paths);
  let cache_ref = cache.clone();
  let mut stream = stream::iter(entries.into_iter().map(|entry| {
    let cache_ref = cache_ref.clone();
    async move { build_discovered_model(entry, source, cache_ref.as_ref()).await }
  }))
  .buffer_unordered(parallel_parse_limit());
  while let Some(model) = stream.next().await {
    if tx.send(model).await.is_err() {
      return;
    }
  }
}

/// Is this `.gguf` file a multimodal projector companion (e.g.,
/// `mmproj-model-f16.gguf`)? Projector files are not independently
/// launchable — they are tensors that pair with a parent chat model
/// for vision/audio input. The user policy is to hide them from the
/// Models list unless they could be launched on their own, which they
/// cannot. The filename prefix is the upstream convention (used by
/// `llama.cpp`'s `convert_hf_to_gguf.py` and every published
/// HuggingFace repo that ships a projector). Filtering on the name
/// avoids paying the cost of a header re-read.
fn is_projector_companion(path: &Path) -> bool {
  path
    .file_name()
    .and_then(|n| n.to_str())
    .map(|n| n.starts_with("mmproj-") || n.starts_with("mmproj_"))
    .unwrap_or(false)
}

/// Concurrency cap for [`walk_root`]'s per-file parse. Default to
/// `num_cpus()`-flavoured but capped — too many parallel
/// `spawn_blocking` calls land everything on the blocking pool
/// regardless. Empirically 8 saturates a single NVMe.
fn parallel_parse_limit() -> usize {
  std::thread::available_parallelism()
    .map(|n| n.get().clamp(2, 8))
    .unwrap_or(4)
}

/// Synchronous file-system walk. Returns every `.gguf` file under
/// `root` honouring gitignore semantics and the caller's exclude
/// globs. Unreadable subdirectories are logged and skipped rather
/// than aborting the walk.
fn collect_gguf_paths(root: &Path, excludes: &[String]) -> Vec<PathBuf> {
  if !root.exists() {
    log::warn!("scan root does not exist: {}", root.display());
    return Vec::new();
  }
  let mut builder = WalkBuilder::new(root);
  builder
    .standard_filters(true)
    .require_git(false)
    // Follow symlinks so users who alias a GGUF into a scan root
    // (e.g., `ln -s /big-disk/model.gguf ~/models/`) still see the
    // model. The `ignore` walker detects cycles, so following links
    // doesn't expose us to symlink loops. A hostile symlink pointing
    // at a non-GGUF file is bounded by (a) the `.gguf` extension
    // gate below and (b) the GGUF parser's BadMagic short-circuit
    // and 4 MiB header cap — opening such a file reads at most a
    // few KB and surfaces as a parse error.
    .follow_links(true)
    .hidden(false);
  if !excludes.is_empty() {
    let mut overrides = ignore::overrides::OverrideBuilder::new(root);
    for pat in excludes {
      // `ignore`'s override globs treat a leading `!` as include-back,
      // so prefix every user exclude with `!` to mean "exclude this".
      // A plain `*.tmp` glob would otherwise be interpreted as
      // "include only files matching this".
      if let Err(e) = overrides.add(&format!("!{pat}")) {
        log::warn!("invalid scan exclude glob {pat:?}: {e}");
      }
    }
    match overrides.build() {
      Ok(o) => {
        builder.overrides(o);
      }
      Err(e) => log::warn!("scan exclude globs failed to compile: {e}"),
    }
  }

  let mut out = Vec::new();
  let mut seen: std::collections::BTreeSet<PathBuf> = std::collections::BTreeSet::new();
  for result in builder.build() {
    match result {
      Ok(entry) => {
        let p = entry.path();
        // Skip `.gguf.part` (mid-download) and only emit regular files
        // ending in `.gguf`. With `follow_links(true)` above, an entry
        // pointing at a symlink reports the *target*'s file type.
        if p.extension().and_then(|s| s.to_str()) == Some("gguf")
          && entry.file_type().map(|t| t.is_file()).unwrap_or(false)
          && !is_projector_companion(p)
        {
          // Canonicalise before dedup so a real file and a symlink to
          // it collapse to a single row. Falling back to the raw path
          // if canonicalisation fails (broken symlink, permission
          // denied) keeps the row visible — the user can investigate.
          let raw = p.to_path_buf();
          let canonical = crate::util::paths::canonicalize(p).unwrap_or_else(|_| raw.clone());
          if seen.insert(canonical.clone()) {
            // For most files we emit the canonical path so user-managed
            // aliases (e.g. `ln -s /big-disk/m.gguf ~/models/`) display
            // under their target name. The exception is HuggingFace's
            // hub layout: blobs are sha256-named files with no `.gguf`
            // extension, surfaced via `snapshots/<rev>/<name>.gguf`
            // symlinks. llama.cpp's split-GGUF loader parses the
            // filename for `-NNNNN-of-NNNNN.gguf` and rejects bare
            // sha256 names, so emitting the canonical path would make
            // every HF-cached multi-part model fail to launch with
            // `invalid split file name`. When canonicalisation strips
            // the `.gguf` extension we treat that as the HF-blob signal
            // and keep the symlink path. Single-file HF models still
            // load fine either way; the path swap only matters for the
            // split-aware loader.
            let emit = if canonical.extension().and_then(|s| s.to_str()) == Some("gguf") {
              canonical
            } else {
              raw
            };
            out.push(emit);
          }
        }
      }
      Err(e) => log::warn!("scan walker error under {}: {e}", root.display()),
    }
  }
  out
}

async fn build_discovered_model(
  entry: DiscoveredEntry,
  source: ModelSource,
  cache: Option<&MetadataCache>,
) -> DiscoveredModel {
  match entry {
    DiscoveredEntry::Single(path) => parse_into_model(path, source, Vec::new(), cache).await,
    DiscoveredEntry::Split(group) => {
      // Siblings exclude the launch file itself so the field's purpose
      // ("sibling shards") matches its content.
      let siblings = group
        .shards
        .into_iter()
        .filter(|p| *p != group.launch_path)
        .collect();
      parse_into_model(group.launch_path, source, siblings, cache).await
    }
  }
}

async fn parse_into_model(
  path: PathBuf,
  source: ModelSource,
  siblings: Vec<PathBuf>,
  cache: Option<&MetadataCache>,
) -> DiscoveredModel {
  let parent = path.parent().map(Path::to_path_buf).unwrap_or_default();
  let probe_path = path.clone();
  let (mtime, size) = tokio::task::spawn_blocking(move || metadata_cache::probe(&probe_path))
    .await
    .unwrap_or((None, 0));

  // Cache lookup first. A hit short-circuits the header read entirely.
  if let Some(c) = cache {
    if let Some(hit) = c.get(&path, mtime, size).await {
      let mut hit_metadata = hit.metadata;
      apply_split_total_weights(&mut hit_metadata, &path, &siblings).await;
      return DiscoveredModel {
        path,
        parent,
        source,
        metadata: hit_metadata,
        parse_error: hit.parse_error,
        split_siblings: siblings,
        display_label: None,
      };
    }
  }

  let path_for_parse = path.clone();
  let parsed: Result<_, GgufError> =
    tokio::task::spawn_blocking(move || read_path(&path_for_parse, HeaderReadOptions::default()))
      .await
      .unwrap_or_else(|join_err| {
        Err(GgufError::Io(std::io::Error::other(format!(
          "parser task panicked: {join_err}"
        ))))
      });
  let cached = match parsed {
    Ok(read) => CachedParse {
      metadata: Some(summarise_metadata(&read.header)),
      parse_error: None,
    },
    Err(e) => CachedParse {
      metadata: None,
      parse_error: Some(e.to_string()),
    },
  };
  if let Some(c) = cache {
    c.put(path.clone(), mtime, size, cached.clone()).await;
  }
  let mut metadata = cached.metadata;
  apply_split_total_weights(&mut metadata, &path, &siblings).await;
  DiscoveredModel {
    path,
    parent,
    source,
    metadata,
    parse_error: cached.parse_error,
    split_siblings: siblings,
    display_label: None,
  }
}

/// For split-GGUF entries, replace the shard-1-only `weights_bytes`
/// with an approximation of the total tensor footprint across every
/// shard. The per-shard `summarise_metadata` only sees shard 1's
/// header, so a 2-shard 80B model was reporting ~half its real size
/// — visible as a wrong SIZE column in `llamastash list`, an
/// undersized estimate from the recommender's VRAM-fit predicate, and
/// the same wrong number in `llamastash show`.
///
/// File size is a tight upper bound on tensor bytes (GGUF header +
/// per-tensor alignment padding is <1% on quant models), and reading
/// the file metadata is cheap, so we sum on-disk sizes instead of
/// reading each sibling's header. No-op when `siblings` is empty.
async fn apply_split_total_weights(
  metadata: &mut Option<ModelMetadata>,
  path: &Path,
  siblings: &[PathBuf],
) {
  if siblings.is_empty() {
    return;
  }
  let Some(meta) = metadata.as_mut() else {
    return;
  };
  let primary = path.to_path_buf();
  let sibling_paths: Vec<PathBuf> = siblings.to_vec();
  let total = tokio::task::spawn_blocking(move || {
    crate::discovery::shard_sizes::on_disk_total(&primary, &sibling_paths)
  })
  .await
  .unwrap_or(0);
  if total > 0 {
    meta.weights_bytes = Some(total);
  }
}

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

  use std::fs;

  use crate::gguf::test_fixtures::build_minimal_gguf;

  fn temp_dir(label: &str) -> PathBuf {
    crate::util::test_temp::unique_temp_dir(&format!("scanner-{label}"))
  }

  #[test]
  fn collect_gguf_paths_skips_part_files() {
    let dir = temp_dir("part");
    fs::write(dir.join("a.gguf"), build_minimal_gguf("llama")).unwrap();
    fs::write(dir.join("a.gguf.part"), b"in-progress").unwrap();
    let paths = collect_gguf_paths(&dir, &[]);
    assert_eq!(paths.len(), 1);
    assert!(paths[0].ends_with("a.gguf"));
    fs::remove_dir_all(&dir).ok();
  }

  #[test]
  fn collect_gguf_paths_drops_mmproj_projector_companions() {
    // Multimodal projector files (`mmproj-*.gguf`) ride along with a
    // parent chat model but are not launchable on their own; without
    // this filter they showed up in the TUI's Models list as
    // selectable rows that would fail at launch time.
    let dir = temp_dir("mmproj");
    fs::write(dir.join("model.gguf"), build_minimal_gguf("llama")).unwrap();
    fs::write(
      dir.join("mmproj-model-f16.gguf"),
      build_minimal_gguf("llama"),
    )
    .unwrap();
    fs::write(
      dir.join("mmproj_model_v2.gguf"),
      build_minimal_gguf("llama"),
    )
    .unwrap();
    let paths = collect_gguf_paths(&dir, &[]);
    let names: Vec<String> = paths
      .iter()
      .map(|p| p.file_name().unwrap().to_string_lossy().into_owned())
      .collect();
    assert_eq!(paths.len(), 1, "expected only model.gguf, got {names:?}");
    assert!(names[0].ends_with("model.gguf"));
    fs::remove_dir_all(&dir).ok();
  }

  #[test]
  fn collect_gguf_paths_honours_exclude_globs() {
    let dir = temp_dir("excl");
    fs::create_dir_all(dir.join("keep")).unwrap();
    fs::create_dir_all(dir.join("skip")).unwrap();
    fs::write(dir.join("keep/a.gguf"), build_minimal_gguf("llama")).unwrap();
    fs::write(dir.join("skip/b.gguf"), build_minimal_gguf("llama")).unwrap();
    let paths = collect_gguf_paths(&dir, &["skip/**".to_string()]);
    assert_eq!(paths.len(), 1);
    assert!(paths[0].to_string_lossy().contains("keep"));
    fs::remove_dir_all(&dir).ok();
  }

  #[test]
  fn nonexistent_root_returns_empty_without_panic() {
    let bogus = PathBuf::from("/nonexistent/scan-root-llamastash");
    assert!(collect_gguf_paths(&bogus, &[]).is_empty());
  }

  #[cfg(unix)]
  #[test]
  fn symlinked_gguf_is_canonicalised_and_deduped() {
    let dir = temp_dir("symlinks");
    fs::write(dir.join("real.gguf"), build_minimal_gguf("llama")).unwrap();
    // Alias the real file with a sibling symlink under the same root.
    let alias = dir.join("alias.gguf");
    std::os::unix::fs::symlink(dir.join("real.gguf"), &alias).unwrap();

    let paths = collect_gguf_paths(&dir, &[]);
    // One canonical row, not two — the symlink collapses onto the
    // real file via `canonicalize` + dedup.
    assert_eq!(
      paths.len(),
      1,
      "real + symlink should collapse to one canonical row, got {paths:?}"
    );
    // The emitted path is the canonical (target) path, not the alias.
    let canon_real = fs::canonicalize(dir.join("real.gguf")).unwrap();
    assert_eq!(paths[0], canon_real);
    fs::remove_dir_all(&dir).ok();
  }

  #[cfg(unix)]
  #[test]
  fn hf_cache_blob_symlink_keeps_symlink_path() {
    // Regression: in the HuggingFace hub layout, the canonical file
    // is a sha256-named blob (no `.gguf` extension) and the launch-
    // friendly path lives behind a snapshot symlink that preserves
    // the upstream name. llama.cpp's split loader requires the
    // `-NNNNN-of-NNNNN.gguf` naming convention, so emitting the
    // canonical blob path makes every multi-part HF model fail to
    // load with `invalid split file name`. The walker must therefore
    // keep the symlink path when the canonical target lacks a
    // `.gguf` extension. Layout mirrors `~/.cache/huggingface/hub`.
    let dir = temp_dir("hfcache");
    let blobs = dir.join("blobs");
    let snap = dir.join("snapshots/main");
    fs::create_dir_all(&blobs).unwrap();
    fs::create_dir_all(&snap).unwrap();
    let blob = blobs.join("403434e5c8454520");
    fs::write(&blob, build_minimal_gguf("llama")).unwrap();
    let named = snap.join("qwen2.5-32b-q4_k_m-00001-of-00005.gguf");
    std::os::unix::fs::symlink(&blob, &named).unwrap();

    let paths = collect_gguf_paths(&dir, &[]);
    assert_eq!(paths.len(), 1, "blob + symlink collapse to one row");
    let emitted = &paths[0];
    assert!(
      emitted.extension().and_then(|s| s.to_str()) == Some("gguf"),
      "emitted path must keep `.gguf` extension, got {emitted:?}"
    );
    assert_eq!(
      emitted.file_name().and_then(|s| s.to_str()),
      Some("qwen2.5-32b-q4_k_m-00001-of-00005.gguf"),
      "emitted path must be the snapshot symlink (split-aware name), \
       not the canonical blob"
    );
    fs::remove_dir_all(&dir).ok();
  }

  #[cfg(unix)]
  #[test]
  fn symlink_to_gguf_outside_root_is_followed_once() {
    // The target file lives outside `root`; a symlink under `root`
    // points at it. follow_links must surface the row.
    let outside = temp_dir("symlinks-outside-target");
    let target = outside.join("target.gguf");
    fs::write(&target, build_minimal_gguf("llama")).unwrap();
    let root = temp_dir("symlinks-outside-root");
    std::os::unix::fs::symlink(&target, root.join("aliased.gguf")).unwrap();

    let paths = collect_gguf_paths(&root, &[]);
    assert_eq!(paths.len(), 1, "symlink target outside root must surface");
    assert_eq!(paths[0], fs::canonicalize(&target).unwrap());
    fs::remove_dir_all(&outside).ok();
    fs::remove_dir_all(&root).ok();
  }

  #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
  async fn scan_streams_discovered_models_with_metadata() {
    let dir = temp_dir("stream");
    fs::write(dir.join("a.gguf"), build_minimal_gguf("llama")).unwrap();
    fs::write(dir.join("b.gguf"), build_minimal_gguf("qwen3")).unwrap();
    let roots = vec![ScanRoot {
      path: dir.clone(),
      source: ModelSource::UserPath,
    }];
    let mut rx = scan(roots, ScanOptions::default());
    let mut got = Vec::new();
    while let Some(m) = rx.recv().await {
      got.push(m);
    }
    assert_eq!(got.len(), 2);
    for m in &got {
      assert!(m.metadata.is_some(), "minimal gguf should parse");
      assert_eq!(m.source, ModelSource::UserPath);
      assert!(m.split_siblings.is_empty());
    }
    fs::remove_dir_all(&dir).ok();
  }

  #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
  async fn scan_surfaces_parse_failure_without_dropping_row() {
    let dir = temp_dir("badparse");
    fs::write(dir.join("bad.gguf"), b"this is not a GGUF").unwrap();
    let roots = vec![ScanRoot {
      path: dir.clone(),
      source: ModelSource::UserPath,
    }];
    let mut rx = scan(roots, ScanOptions::default());
    let m = rx.recv().await.expect("one model surfaced");
    assert!(rx.recv().await.is_none(), "only one file in dir");
    assert!(m.metadata.is_none(), "invalid file → no metadata");
    assert!(m.parse_error.is_some(), "diagnostic must accompany failure");
    fs::remove_dir_all(&dir).ok();
  }

  #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
  async fn scan_groups_split_shards_into_one_entry() {
    let dir = temp_dir("split");
    let bytes = build_minimal_gguf("llama");
    fs::write(dir.join("model-00001-of-00003.gguf"), &bytes).unwrap();
    fs::write(dir.join("model-00002-of-00003.gguf"), &bytes).unwrap();
    fs::write(dir.join("model-00003-of-00003.gguf"), &bytes).unwrap();
    let roots = vec![ScanRoot {
      path: dir.clone(),
      source: ModelSource::UserPath,
    }];
    let mut rx = scan(roots, ScanOptions::default());
    let m = rx.recv().await.expect("one grouped entry");
    assert!(
      rx.recv().await.is_none(),
      "shard set should collapse to one"
    );
    assert_eq!(m.split_siblings.len(), 2, "shard 1 plus 2 siblings");
    assert!(m
      .path
      .to_string_lossy()
      .ends_with("model-00001-of-00003.gguf"));
    fs::remove_dir_all(&dir).ok();
  }

  #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
  async fn split_shards_report_summed_weights_bytes() {
    // Regression: a multi-shard set used to report only shard 1's
    // header-derived weights_bytes, so `llamastash list`,
    // `show`, and the recommender's VRAM-fit predicate all saw
    // ~half the real size for a 2-shard 80B Q5_K_M model. The
    // scanner now sums every shard's on-disk size into
    // `metadata.weights_bytes` so the displayed/used value covers
    // the whole model.
    let dir = temp_dir("split-size");
    let shard_bytes = build_minimal_gguf("qwen3");
    let per_shard = shard_bytes.len() as u64;
    fs::write(dir.join("m-00001-of-00002.gguf"), &shard_bytes).unwrap();
    fs::write(dir.join("m-00002-of-00002.gguf"), &shard_bytes).unwrap();
    let roots = vec![ScanRoot {
      path: dir.clone(),
      source: ModelSource::UserPath,
    }];
    let mut rx = scan(roots, ScanOptions::default());
    let m = rx.recv().await.expect("one grouped entry");
    let weights = m
      .metadata
      .as_ref()
      .expect("metadata present")
      .weights_bytes
      .expect("split should set summed weights_bytes");
    assert_eq!(
      weights,
      per_shard * 2,
      "split weights_bytes must equal sum of every shard's on-disk size"
    );
    fs::remove_dir_all(&dir).ok();
  }

  #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
  async fn metadata_cache_reuses_parse_for_unchanged_file() {
    use crate::discovery::metadata_cache::MetadataCache;

    let dir = temp_dir("cache-hit");
    fs::write(dir.join("a.gguf"), build_minimal_gguf("llama")).unwrap();
    let cache = MetadataCache::new(8);
    let opts = ScanOptions {
      metadata_cache: Some(cache.clone()),
      ..ScanOptions::default()
    };
    let roots = vec![ScanRoot {
      path: dir.clone(),
      source: ModelSource::UserPath,
    }];

    async fn drain(mut rx: mpsc::Receiver<DiscoveredModel>) {
      while rx.recv().await.is_some() {}
    }

    // First scan: cache empty → one miss → one entry inserted.
    drain(scan(roots.clone(), opts.clone())).await;
    assert_eq!(cache.len().await, 1, "first scan populates cache");

    // Second scan: cache hit → still one entry, parse skipped.
    drain(scan(roots.clone(), opts.clone())).await;
    assert_eq!(cache.len().await, 1, "second scan does not duplicate");
    fs::remove_dir_all(&dir).ok();
  }

  #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
  async fn metadata_cache_invalidates_when_size_changes() {
    use crate::discovery::metadata_cache::MetadataCache;
    use crate::gguf::test_fixtures::FixtureBuilder;

    let dir = temp_dir("cache-invalid");
    let path = dir.join("a.gguf");
    // First write: minimal arch=llama header.
    fs::write(&path, build_minimal_gguf("llama")).unwrap();
    let cache = MetadataCache::new(8);
    let opts = ScanOptions {
      metadata_cache: Some(cache.clone()),
      ..ScanOptions::default()
    };
    let roots = vec![ScanRoot {
      path: dir.clone(),
      source: ModelSource::UserPath,
    }];

    // Prime the cache.
    let mut first_rx = scan(roots.clone(), opts.clone());
    let first = first_rx.recv().await.expect("one model");
    while first_rx.recv().await.is_some() {}
    let first_arch = first
      .metadata
      .as_ref()
      .and_then(|m| m.arch.clone())
      .unwrap();
    assert_eq!(first_arch, "llama");

    // Mutate the file: different arch, different total tensor count
    // → different on-disk size and (typically) different mtime.
    let updated_bytes = FixtureBuilder::new()
      .with_arch("phi3")
      .with_context_length(4096)
      .with_tensor("blk.0.attn_q.weight", &[10, 10], 1)
      .build();
    fs::write(&path, &updated_bytes).unwrap();

    // Force the on-disk mtime to advance even on filesystems whose
    // mtime resolution is coarse (some CI tmpfs sets mtime to whole
    // seconds and rewrites within the same second can otherwise look
    // unchanged). Size *also* changed above, which alone invalidates
    // the cache; this just makes the invalidation independent of fs
    // mtime granularity.
    let _ = std::process::Command::new("touch")
      .arg("-t")
      .arg("203601011200")
      .arg(&path)
      .status();

    let mut second_rx = scan(roots, opts);
    let second = second_rx.recv().await.expect("one model");
    while second_rx.recv().await.is_some() {}
    let second_arch = second
      .metadata
      .as_ref()
      .and_then(|m| m.arch.clone())
      .unwrap();
    assert_eq!(
      second_arch, "phi3",
      "size change must invalidate cache and re-parse"
    );
  }
}
