gonzalo_cli/

lib.rs

//! Command implementations for the gonzalo admin CLI.

use anyhow::Context;
use anyhow::Result;
use gonzalo_core::{
    BlobStore, Body, ContentHash, Identity, KeyPrefix, Manifest, Meta, PutResult, Record,
    RecordKey, RecordKind, Revision, Store,
};
use gonzalo_graph::{CodeGraph, EXTRACTION_VERSION, GraphStore, Language, build};
use gonzalo_graph_sqlite::{SqliteGraphStore, view_db_path};
use gonzalo_parse::{ParserPool, worker_extraction_version};
use gonzalo_store_fs::FsStore;
use gonzalo_ticket::IngestSummary;
use gonzalo_ticket_config::{Config, Connection, parse_category};
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

mod walk;
pub use walk::{IgnoredCounts, IndexFilter, SourceFiles, UnindexedCounts};

mod watch;
pub use watch::{WatchConfig, watch};

// ─── list ────────────────────────────────────────────────────────────────────

/// Return all record keys in the store, optionally filtered by namespace /
/// collection.
pub async fn list(
    root: &Path,
    namespace: Option<String>,
    collection: Option<String>,
) -> Result<Vec<RecordKey>> {
    let store = FsStore::new(root);
    let prefix = KeyPrefix {
        namespace,
        collection,
    };
    let keys = store.list(&prefix).await?;
    Ok(keys)
}

// ─── get ─────────────────────────────────────────────────────────────────────

/// Fetch a single record, or `None` if it does not exist.
pub async fn get(root: &Path, ns: &str, col: &str, id: &str) -> Result<Option<Record>> {
    let store = FsStore::new(root);
    let key = RecordKey::new(ns, col, id);
    Ok(store.get(&key).await?)
}

// ─── status ──────────────────────────────────────────────────────────────────

/// Count of records grouped by `"namespace/collection"`.
pub async fn status(root: &Path) -> Result<BTreeMap<String, usize>> {
    let keys = list(root, None, None).await?;
    let mut map: BTreeMap<String, usize> = BTreeMap::new();
    for k in keys {
        *map.entry(format!("{}/{}", k.namespace, k.collection))
            .or_insert(0) += 1;
    }
    Ok(map)
}

// ─── migrate ─────────────────────────────────────────────────────────────────

/// Summary returned by [`migrate`].
pub struct MigrateSummary {
    pub imported: usize,
    pub skipped: usize,
}

/// Recursively import every file under `src` as a record in the fs store at
/// `root`. Idempotent: if the key already exists, skip it.
pub async fn migrate(
    root: &Path,
    src: &Path,
    namespace: &str,
    collection: &str,
    kind: RecordKind,
) -> Result<MigrateSummary> {
    let store = FsStore::new(root);
    let mut imported = 0usize;
    let mut skipped = 0usize;

    // Collect all file paths recursively using std::fs (no walkdir dep).
    let files = collect_files(src)?;

    for abs_path in files {
        // Build relative path string with `/` as separator.
        let rel = abs_path
            .strip_prefix(src)
            .map_err(|e| anyhow::anyhow!("strip_prefix failed: {e}"))?;
        let rel_str = rel
            .components()
            .map(|c| c.as_os_str().to_string_lossy().into_owned())
            .collect::<Vec<_>>()
            .join("/");

        // Use the relative path verbatim as the record id. The store now
        // encodes arbitrary key characters reversibly and injectively, so two
        // distinct source files can never collide onto one record (the old
        // `segment()` collapse silently dropped one of `docs/api.md` and
        // `docs_api.md`). Ids stay human-readable (`docs/api.md`).
        let key = RecordKey::new(namespace, collection, rel_str);

        // Idempotency: skip if already present.
        if store.get(&key).await?.is_some() {
            skipped += 1;
            continue;
        }

        let file_bytes = std::fs::read(&abs_path)?;
        let body = Body::Inline(file_bytes);
        let record = Record {
            key,
            kind,
            revision: Revision::initial(body.bytes()),
            parent: None,
            body,
            meta: Meta {
                author: Identity::new(CLI_AUTHOR),
                origin_system: "migrate".into(),
                created: 0,
                updated: 0,
                labels: BTreeMap::new(),
            },
            links: vec![],
            ancestors: Vec::new(),
            deleted_at: None,
            deleted_blob: None,
            deleted_kind: None,
        };

        match store.put(record, None).await? {
            PutResult::Committed(_) => imported += 1,
            PutResult::Conflict(_) => skipped += 1,
        }
    }

    Ok(MigrateSummary { imported, skipped })
}

/// Walk `dir` recursively and return a sorted list of all file paths.
fn collect_files(dir: &Path) -> Result<Vec<std::path::PathBuf>> {
    let mut out = Vec::new();
    collect_files_inner(dir, &mut out)?;
    out.sort();
    Ok(out)
}

fn collect_files_inner(dir: &Path, out: &mut Vec<std::path::PathBuf>) -> Result<()> {
    // Name the directory: a bare "No such file or directory (os error 2)"
    // leaves the user guessing which path was wrong (#299).
    let entries = std::fs::read_dir(dir).with_context(|| format!("reading {}", dir.display()))?;
    for entry in entries {
        let entry = entry?;
        let ft = entry.file_type()?;
        if ft.is_dir() {
            collect_files_inner(&entry.path(), out)?;
        } else if ft.is_file() {
            out.push(entry.path());
        }
    }
    Ok(())
}

// ─── index ───────────────────────────────────────────────────────────────────

/// Summary returned by [`index`].
pub struct IndexSummary {
    /// Source files parsed into slices.
    pub files: usize,
    /// Paths newly added to the view.
    pub added: usize,
    /// Paths whose slice content changed.
    pub modified: usize,
    /// Paths removed from the view since the last index.
    pub deleted: usize,
    /// Files skipped because an isolated parse worker crashed or hung on them
    /// (only possible when parsing through the pool).
    pub skipped: usize,
    /// Which files those were. Counting them without naming them leaves no way
    /// to find the offender but bisecting by directory (#270).
    pub skipped_paths: Vec<String>,
    /// Paths excluded from the view by an [`IndexFilter`] rule — vendored or
    /// generated files, dependency/output directories, and gitignored trees
    /// (#209). Distinct from `skipped`, which is a parse failure.
    pub ignored: IgnoredCounts,
    /// Files gonzalo has no grammar for. Distinct from both of the above:
    /// recognized as a file, would have been indexed, but its extension names
    /// no language gonzalo parses (#259).
    pub unindexed: UnindexedCounts,
    /// Whether this run used the git-diff-driven incremental driver (only the
    /// changed set re-parsed) rather than the full tree walk.
    pub incremental: bool,
}

/// The `gonzalo-parse-worker` file name for this platform.
fn worker_file_name() -> &'static str {
    if cfg!(windows) {
        "gonzalo-parse-worker.exe"
    } else {
        "gonzalo-parse-worker"
    }
}

/// Where a parse worker was found.
///
/// `#[non_exhaustive]`: this set is already known to be growing — #212's own
/// proposal anticipates a `--parse-worker <path>` flag, which is a fourth
/// source. Sealing it now costs nothing; adding a variant later to a sealed
/// enum is not a breaking change, while adding one to an open enum is (#241).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum WorkerSource {
    /// The `GONZALO_PARSE_WORKER` environment override.
    Env,
    /// Beside the running executable (the installed / `cargo build` layout).
    Sibling,
    /// On `PATH` — the symlink and split-install layouts, where a sibling
    /// check cannot succeed but the worker is plainly available.
    Path,
}

impl WorkerSource {
    fn describe(self) -> &'static str {
        match self {
            Self::Env => "GONZALO_PARSE_WORKER",
            Self::Sibling => "beside the executable",
            Self::Path => "on PATH",
        }
    }
}

/// How an index run will parse.
///
/// The two modes produce identical graphs, which is exactly why the fallback
/// went unnoticed: the worker is a pure isolation wrapper, so its absence costs
/// nothing until a grammar `abort()`s and takes the whole run with it instead of
/// skipping one file (#212). Reporting the mode is what makes the guarantee
/// checkable at the moment it matters.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum ParseMode {
    /// Crash-isolated: parses run in a worker subprocess pool.
    Isolated { path: PathBuf, source: WorkerSource },
    /// In-process: no worker was found, so a grammar crash aborts the run.
    /// Carries every location that was searched, so the warning can say where
    /// to put one.
    InProcess { searched: Vec<PathBuf> },
}

impl ParseMode {
    /// The worker binary to parse through, or `None` to parse in-process.
    pub fn worker(&self) -> Option<&Path> {
        match self {
            Self::Isolated { path, .. } => Some(path),
            Self::InProcess { .. } => None,
        }
    }

    /// Whether parsing is crash-isolated.
    pub fn is_isolated(&self) -> bool {
        matches!(self, Self::Isolated { .. })
    }

    /// One line naming the active mode, for `gonzalo index` to print.
    pub fn summary(&self) -> String {
        match self {
            Self::Isolated { path, source } => {
                format!("isolated worker {} ({})", path.display(), source.describe())
            }
            Self::InProcess { .. } => "in-process (no worker found)".to_string(),
        }
    }

    /// The warning to emit when isolation is inactive, naming everywhere that
    /// was searched. `None` when parsing is isolated.
    pub fn warning(&self) -> Option<String> {
        let Self::InProcess { searched } = self else {
            return None;
        };
        let where_looked = if searched.is_empty() {
            "  (nowhere — no executable path and no PATH entries)".to_string()
        } else {
            searched
                .iter()
                .map(|p| format!("  {}", p.display()))
                .collect::<Vec<_>>()
                .join("\n")
        };
        Some(format!(
            "gonzalo index: warning — parsing IN-PROCESS, not crash-isolated. A tree-sitter \
             grammar that aborts will take down this run instead of skipping one file.\n\
             Searched for {}:\n{where_looked}\n\
             Install it (`cargo install gonzalo-parse`), put it on PATH or beside this \
             executable, or point GONZALO_PARSE_WORKER at it. Use --require-parse-worker to \
             make this a hard error.",
            worker_file_name()
        ))
    }
}

/// Locate the `gonzalo-parse-worker` binary, given the inputs that decide it.
///
/// Split from the environment so it can be tested: setting a process
/// environment variable is `unsafe` under edition 2024, which this workspace
/// forbids outright.
///
/// Order is override, then adjacency, then `PATH` — most specific first.
fn resolve_worker_from(
    env_override: Option<PathBuf>,
    exe: Option<PathBuf>,
    search_path: &[PathBuf],
) -> ParseMode {
    let name = worker_file_name();
    let mut searched = Vec::new();

    if let Some(path) = env_override {
        if path.is_file() {
            return ParseMode::Isolated {
                path,
                source: WorkerSource::Env,
            };
        }
        searched.push(path);
    }

    if let Some(exe) = exe {
        let sibling = exe.with_file_name(name);
        if sibling.is_file() {
            return ParseMode::Isolated {
                path: sibling,
                source: WorkerSource::Sibling,
            };
        }
        searched.push(sibling);
    }

    for dir in search_path {
        let candidate = dir.join(name);
        if candidate.is_file() {
            return ParseMode::Isolated {
                path: candidate,
                source: WorkerSource::Path,
            };
        }
        searched.push(candidate);
    }

    ParseMode::InProcess { searched }
}

/// Locate the `gonzalo-parse-worker` binary for crash-isolated parsing:
/// the `GONZALO_PARSE_WORKER` override, else a sibling of the current
/// executable, else `PATH`. Falls back to in-process parsing when none of those
/// has one — reporting where it looked (#212).
pub fn resolve_parse_worker() -> ParseMode {
    let search_path: Vec<PathBuf> = std::env::var_os("PATH")
        .map(|p| std::env::split_paths(&p).collect())
        .unwrap_or_default();
    resolve_worker_from(
        std::env::var_os("GONZALO_PARSE_WORKER").map(PathBuf::from),
        std::env::current_exe().ok(),
        &search_path,
    )
}

/// Parse one file's `content` as `language`, through the crash-isolated `pool`
/// if one is available (a crash/hang yields `None` — skip the file), or
/// in-process otherwise.
async fn parse_file(
    pool: Option<&ParserPool>,
    language: Language,
    content: &str,
    rel: &str,
) -> Option<CodeGraph> {
    match pool {
        Some(pool) => match pool.parse(language, content).await {
            Ok(graph) => Some(graph),
            Err(e) => {
                eprintln!("{}", skip_message(rel, &e.to_string()));
                None
            }
        },
        None => Some(build(language, content)),
    }
}

/// The line printed when a file is skipped, naming it.
///
/// A skipped file is silently absent from the view — `search` finds none of its
/// symbols, `callers` misses every call it makes — so the one line saying that
/// happened has to say *where*. Without the path the only way to find the
/// offender is bisecting by directory, which is what landing #266 cost (#270).
fn skip_message(rel: &str, error: &str) -> String {
    format!("gonzalo index: skipping {rel} — parse worker error: {error}")
}

/// Skipped paths for the index summary, bounded, with a count of what was left
/// out so a tree full of crashes does not become a wall of text.
pub fn named_skips(paths: &[String]) -> String {
    const MOST: usize = 5;
    if paths.is_empty() {
        return String::new();
    }
    let listed = paths
        .iter()
        .take(MOST)
        .map(String::as_str)
        .collect::<Vec<_>>()
        .join(", ");
    let more = paths.len().saturating_sub(MOST);
    if more > 0 {
        format!("{listed}, and {more} more")
    } else {
        listed
    }
}

/// Index the source files under `src` into the `(repo, view)` code-graph view:
/// parse each file into a content-addressed slice, then reconcile the view's
/// manifest to the tree (ADR 0012). Re-running updates the view.
///
/// When `src` is the root of a git repo and a base commit was recorded on a
/// prior run, the changed set is sourced directly from `git diff` (only the
/// added/modified files are re-parsed and deleted files dropped) — the
/// incremental driver of gonzalo#93. Otherwise the full tree is walked. Either
/// way the manifest is reconciled as a set, so a full walk always converges the
/// view even if an incremental run ever missed a change. Slices orphaned by
/// deletions are left for a separate GC pass (which must see all live views).
pub async fn index(root: &Path, src: &Path, repo: &str, view: &str) -> Result<IndexSummary> {
    index_with(root, src, repo, view, &IndexFilter::default()).await
}

/// [`index`], with control over which paths enter the view.
///
/// `filter` carries any `--include` overrides; the built-in vendored/generated
/// rules and `.gitignore` apply either way (#209).
pub async fn index_with(
    root: &Path,
    src: &Path,
    repo: &str,
    view: &str,
    filter: &IndexFilter,
) -> Result<IndexSummary> {
    let mode = resolve_parse_worker();
    index_with_worker(root, src, repo, view, filter, mode.worker()).await
}

/// The warning shown when the parse path's extraction version is not this
/// build's. `reported` is the worker's version, or `0` for "would not say".
///
/// Unknown and known-stale are **different states**, and saying so matters most
/// on the upgrade everyone performs at least once: every worker released before
/// `--extraction-version` (#228) answers nothing, while producing extraction
/// that may well be current. Telling that user to "install a matching worker to
/// get version N extraction" claims their data is wrong when it is not (#239).
fn worker_version_warning(reported: u32) -> String {
    if reported == 0 {
        format!(
            "gonzalo index: note — the parse worker did not report an extraction version, so \
             it predates `--extraction-version` (gonzalo 0.5.0 or earlier). Its output may \
             well match this build's version {EXTRACTION_VERSION}, but that cannot be \
             confirmed, so the view is credited `unknown` and rebuilt in full. Upgrading \
             gonzalo-parse-worker alongside gonzalo clears this and rebuilds once more."
        )
    } else {
        format!(
            "gonzalo index: WARNING — the parse worker reports extraction version {reported}, \
             but this build records {EXTRACTION_VERSION}. The two disagree about what a parse \
             records. Parsing anyway, rebuilding the view in full and crediting it to the \
             worker; install a matching gonzalo-parse-worker to get {EXTRACTION_VERSION} \
             extraction."
        )
    }
}

/// Print `msg` to stderr at most once per process.
///
/// `--watch` re-enters the index path on every debounced change, so a
/// per-index warning becomes a paragraph per file save (#239). The condition
/// cannot change mid-process anyway: the binaries on disk are what they are.
fn warn_once(msg: &str) {
    static ONCE: std::sync::Once = std::sync::Once::new();
    ONCE.call_once(|| eprintln!("{msg}"));
}

/// How long to wait for a worker to answer [`worker_extraction_version`].
/// Generous next to a process spawn; a worker that misses it reads as unknown
/// rather than wedging the run.
const WORKER_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);

/// [`index_with`], with the parse worker pinned rather than resolved.
///
/// `worker` is the `gonzalo-parse-worker` binary to parse through, or `None` to
/// parse in-process. [`index_with`] passes whatever [`resolve_parse_worker`]
/// finds; pinning it explicitly is what lets a test drive a known parse path.
pub async fn index_with_worker(
    root: &Path,
    src: &Path,
    repo: &str,
    view: &str,
    filter: &IndexFilter,
    worker: Option<&Path>,
) -> Result<IndexSummary> {
    let store = FsStore::new(root);

    // The extraction format the parse path will actually produce — which is not
    // necessarily this build's. The worker is a separately installed binary, so
    // ask it; `None` (a pre-#228 worker, a missing one, a hang) is unknown, and
    // unknown is recorded as 0 so it can never be mistaken for agreement (#228).
    // With no worker the CLI parses in-process, so its own constant is accurate.
    let parse_version = match worker {
        Some(bin) => worker_extraction_version(bin, WORKER_PROBE_TIMEOUT)
            .await
            .unwrap_or(0),
        None => EXTRACTION_VERSION,
    };
    if parse_version != EXTRACTION_VERSION {
        warn_once(&worker_version_warning(parse_version));
    }

    // Parse through a crash-isolated worker pool when a worker binary is
    // available (so a grammar crash on one file skips that file instead of
    // aborting the index); otherwise parse in-process.
    let pool = worker.map(|bin| {
        let workers = std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(4)
            .clamp(1, 8);
        ParserPool::new(bin, workers, std::time::Duration::from_secs(30))
    });

    // The persistent per-view graph, queried by the daemon without re-assembly.
    let db_path = view_db_path(&root.join("graphs"), repo, view);
    let mut graph = SqliteGraphStore::open(&db_path)
        .with_context(|| format!("opening graph db for {repo}/{view}"))?;

    // Load the view's current manifest (empty if new).
    let key = Manifest::key(repo, view);
    let existing = store.get(&key).await?;
    let current = match &existing {
        Some(rec) => Manifest::from_body(&rec.body)?,
        None => Manifest::new(),
    };

    // Choose the driver: incremental when `src` is a git repo root, a base was
    // recorded, and the diff against it is readable; otherwise a full walk.
    let base_path = db_path.with_extension("base");
    // A view built by a parser that recorded different things must be rebuilt
    // in full: the incremental driver carries unchanged slices forward, so
    // without this an existing view keeps pre-upgrade extraction forever.
    // Compared against the version the *parse path* produces, not this build's
    // — otherwise a stale worker's output is carried forward as current (#228).
    let version_path = db_path.with_extension("fmt");
    let recorded_version: Option<u32> = std::fs::read_to_string(&version_path)
        .ok()
        .and_then(|s| s.trim().parse().ok());
    let format_changed = recorded_version != Some(parse_version);

    let recorded_base = std::fs::read_to_string(&base_path)
        .ok()
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty())
        .filter(|_| !format_changed);
    let incremental_changed = recorded_base
        .as_deref()
        .and_then(|base| gonzalo_store_git::changed_paths(src, base).ok());

    // Stage every persistent-graph mutation instead of applying it inline. The
    // SqliteGraphStore is advanced only after the manifest that describes it
    // commits, so a concurrent-writer Conflict (below) leaves the graph
    // untouched rather than advanced ahead of a manifest that never landed (#153).
    let mut staging = GraphStaging::default();
    let DesiredView {
        entries: desired,
        files,
        skipped_paths,
        ignored,
        unindexed,
        incremental,
    } = match incremental_changed {
        Some(changed) => {
            build_desired_incremental(
                &store,
                &mut staging,
                pool.as_ref(),
                src,
                &current,
                &changed,
                filter,
            )
            .await?
        }
        None => build_desired_full(&store, &mut staging, pool.as_ref(), src, filter).await?,
    };

    // Reconcile against the current manifest and stage removed paths for the
    // persistent graph (applied only after the manifest commit).
    let recon = current.reconcile(&desired);
    for path in &recon.deleted {
        staging.removes.push(path.clone());
    }

    // Write the manifest record (create-or-update under OCC).
    let body = recon.manifest.to_body();
    let (revision, expected, parent) = match &existing {
        Some(rec) => (
            rec.revision.next(body.bytes()),
            Some(rec.revision.clone()),
            Some(rec.revision.clone()),
        ),
        None => (Revision::initial(body.bytes()), None, None),
    };
    let record = Record {
        key,
        kind: RecordKind::GraphManifest,
        revision,
        parent,
        body,
        meta: Meta {
            author: Identity::new("gonzalo-index"),
            origin_system: "gonzalo-index".into(),
            created: 0,
            updated: 0,
            labels: BTreeMap::new(),
        },
        links: Vec::new(),
        ancestors: Vec::new(),
        deleted_at: None,
        deleted_blob: None,
        deleted_kind: None,
    };
    match store.put(record, expected).await? {
        PutResult::Committed(_) => {}
        PutResult::Conflict(_) => {
            // The manifest moved under us: abandon the run WITHOUT touching the
            // persistent graph, so the SqliteGraphStore never advances ahead of a
            // committed manifest (#153). Orphaned slice blobs written above are
            // reclaimed by a later gc pass.
            anyhow::bail!("manifest for {repo}/{view} changed concurrently; retry the index")
        }
    }

    // Only added/modified paths need checking: an unchanged path was referenced
    // by both the old manifest and the new one, so no sweep ever saw it as
    // garbage.
    let newly_referenced: Vec<(String, ContentHash)> = recon
        .added
        .iter()
        .chain(recon.modified.iter())
        .filter_map(|p| recon.manifest.get(p).map(|h| (p.clone(), h.clone())))
        .collect();
    // An error here returns after the manifest committed but before
    // `staging.apply`, so the persistent graph stays behind the manifest (#153's
    // invariant holds) and the next run starts from a stale base; hence the
    // message tells the operator to re-index.
    let restored = ensure_slices_present(&store, &newly_referenced, &staging).await?;
    if restored > 0 {
        eprintln!("restored {restored} slice blob(s) swept during this index run");
    }

    // Manifest committed — now advance the persistent graph to match it.
    staging.apply(&mut graph);

    // Record the current HEAD as the base for the next run's incremental diff.
    // Only succeeds when `src` is a git repo root with at least one commit.
    if let Ok(sha) = gonzalo_store_git::head_commit(src) {
        if let Some(parent) = base_path.parent() {
            std::fs::create_dir_all(parent).ok();
        }
        std::fs::write(&base_path, sha).ok();
    }
    // Record the extraction format this view was built with, so the next run can
    // tell whether an incremental pass is still valid. This is the version the
    // parse path actually produced — never this build's constant when something
    // else did the parsing, or the view would claim an upgrade it never got and
    // no later re-index would revisit it (#228).
    if let Some(parent) = version_path.parent() {
        std::fs::create_dir_all(parent).ok();
        std::fs::write(&version_path, parse_version.to_string()).ok();
    }

    Ok(IndexSummary {
        files,
        added: recon.added.len(),
        modified: recon.modified.len(),
        deleted: recon.deleted.len(),
        skipped: skipped_paths.len(),
        skipped_paths,
        ignored,
        unindexed,
        incremental,
    })
}

/// A set of persistent-graph mutations collected during a [`index`] run but not
/// yet applied. Staging the writes lets [`index`] commit the view's manifest
/// first and only then advance the [`SqliteGraphStore`] — so a manifest Conflict
/// leaves the persistent graph untouched (#153).
#[derive(Default)]
struct GraphStaging {
    /// `(relative path, parsed slice)` to (re)insert; insert replaces the path's
    /// existing rows.
    inserts: Vec<(String, CodeGraph)>,
    /// Relative paths whose rows should be removed.
    removes: Vec<String>,
}

impl GraphStaging {
    /// Apply every staged mutation to the persistent graph. Called only after
    /// the manifest commit succeeds.
    fn apply(self, graph: &mut SqliteGraphStore) {
        for (rel, slice) in self.inserts {
            graph.insert(&rel, slice);
        }
        for rel in self.removes {
            graph.remove_path(&rel);
        }
    }
}

/// Full-walk desired set: parse every supported source file under `src` that
/// `filter` admits.
/// What one indexing driver produced: the manifest entries the view should
/// hold, plus the counts [`IndexSummary`] reports.
///
/// A named type rather than a tuple because the two drivers must agree on it
/// exactly, and a six-element tuple says nothing about which `usize` is which.
struct DesiredView {
    entries: BTreeMap<String, ContentHash>,
    /// Files parsed into the view this run.
    files: usize,
    /// Files a parse worker crashed or hung on, by path (#270).
    skipped_paths: Vec<String>,
    ignored: IgnoredCounts,
    unindexed: UnindexedCounts,
    /// Whether the git-diff-driven driver produced this, rather than a full walk.
    incremental: bool,
}

/// Confirm the slice blobs a just-committed manifest references still exist,
/// re-uploading any a sweep took.
///
/// A sweep can delete a blob between the upload and the manifest commit naming
/// it. GC's age filter covers a freshly uploaded slice, but not an **old** blob
/// this run newly references — content-addressed dedup makes that ordinary when
/// a file reverts to earlier content. Returns how many blobs it restored. See
/// ADR 0028.
async fn ensure_slices_present(
    store: &FsStore,
    wanted: &[(String, ContentHash)],
    staging: &GraphStaging,
) -> anyhow::Result<usize> {
    let mut restored = 0;
    for (path, hash) in wanted {
        if store.has_blob(hash).await? {
            continue;
        }
        let Some((_, slice)) = staging.inserts.iter().find(|(p, _)| p == path) else {
            anyhow::bail!(
                "slice blob {} for {path} is missing and this run did not parse it, \
                 so it cannot be restored; re-index the view",
                hash.0
            );
        };
        let put = store.put_blob(&slice.to_slice_bytes()).await?;
        anyhow::ensure!(
            &put == hash,
            "restored slice for {path} hashes to {} rather than the committed {}",
            put.0,
            hash.0
        );
        restored += 1;
    }
    Ok(restored)
}

async fn build_desired_full(
    store: &FsStore,
    staging: &mut GraphStaging,
    pool: Option<&ParserPool>,
    src: &Path,
    filter: &IndexFilter,
) -> Result<DesiredView> {
    let mut desired: BTreeMap<String, ContentHash> = BTreeMap::new();
    let mut skipped_paths: Vec<String> = Vec::new();
    let walk::SourceFiles {
        files: sources,
        ignored,
        unindexed,
    } = walk::source_files(src, filter)?;
    for (path, language) in sources {
        let content = std::fs::read_to_string(&path)
            .with_context(|| format!("reading {}", path.display()))?;
        let rel = path
            .strip_prefix(src)
            .unwrap_or(&path)
            .to_string_lossy()
            .replace('\\', "/");
        let Some(slice) = parse_file(pool, language, &content, &rel).await else {
            skipped_paths.push(rel.clone());
            continue;
        };
        let hash = store.put_blob(&slice.to_slice_bytes()).await?;
        staging.inserts.push((rel.clone(), slice)); // insert replaces this path's rows
        desired.insert(rel, hash);
    }
    let files = desired.len();
    Ok(DesiredView {
        entries: desired,
        files,
        skipped_paths,
        ignored,
        unindexed,
        incremental: false,
    })
}

/// Incremental desired set: start from the current manifest and apply only the
/// git-reported changes — re-parse added/modified source files, drop deleted
/// ones, and carry every unchanged path forward untouched. `files` counts the
/// files re-parsed this run.
async fn build_desired_incremental(
    store: &FsStore,
    staging: &mut GraphStaging,
    pool: Option<&ParserPool>,
    src: &Path,
    current: &Manifest,
    changed: &gonzalo_store_git::ChangedPaths,
    filter: &IndexFilter,
) -> Result<DesiredView> {
    let mut desired = current.entries.clone();
    let mut files = 0usize;
    let mut skipped_paths: Vec<String> = Vec::new();
    // Gitignored paths never reach here — `git2`'s diff omits them — so only the
    // path-only rules apply, and only files are ever counted.
    let mut ignored = IgnoredCounts::default();
    let mut unindexed = UnindexedCounts::default();

    // Re-apply the filter to paths carried forward from the previous run, not
    // just to changed ones. A view indexed under laxer rules keeps its vendored
    // bundles forever otherwise: `mermaid.min.js` never changes, so it never
    // appears in the diff, so an incremental run never reconsiders it — and once
    // a base commit is recorded there is no full walk to clean it up. Making the
    // carried-forward set self-healing is what lets an existing view benefit
    // from #209 rather than only newly created ones.
    let stale = walk::stale_entries(src, filter, desired.keys().map(String::as_str));
    for rel in stale {
        desired.remove(&rel);
        staging.removes.push(rel);
        ignored.files += 1;
    }

    for rel in changed.added.iter().chain(changed.modified.iter()) {
        if !filter.is_indexable(rel) {
            // A path that a previous, laxer walk admitted must also be dropped
            // from the view, not merely skipped, or the two drivers disagree.
            if desired.remove(rel).is_some() {
                staging.removes.push(rel.clone());
            }
            ignored.files += 1;
            continue;
        }
        let Some(language) = Path::new(rel)
            .extension()
            .and_then(|e| e.to_str())
            .and_then(Language::from_extension)
        else {
            // Indexable, but gonzalo has no grammar for it. The filter check
            // above already rejected anything deliberately excluded, so this
            // counts exactly the files a user would expect to have been
            // indexed and were not (#259).
            unindexed.record(Path::new(rel).extension().and_then(|e| e.to_str()));
            continue;
        };
        // A file git reports as changed but that we can no longer read (e.g.
        // it vanished between diff and read) is treated as a removal.
        let content = match std::fs::read_to_string(src.join(rel)) {
            Ok(c) => c,
            Err(_) => {
                if desired.remove(rel).is_some() {
                    staging.removes.push(rel.clone());
                }
                continue;
            }
        };
        let Some(slice) = parse_file(pool, language, &content, rel).await else {
            skipped_paths.push(rel.clone());
            continue;
        };
        let hash = store.put_blob(&slice.to_slice_bytes()).await?;
        staging.inserts.push((rel.clone(), slice));
        desired.insert(rel.clone(), hash);
        files += 1;
    }

    for rel in &changed.deleted {
        if desired.remove(rel).is_some() {
            staging.removes.push(rel.clone());
        }
    }

    Ok(DesiredView {
        entries: desired,
        files,
        skipped_paths,
        ignored,
        unindexed,
        incremental: true,
    })
}

// ─── gc ────────────────────────────────────────────────────────────────────

/// Summary returned by [`gc`].
pub struct GcSummary {
    /// Records scanned to build the mark set, tombstones included.
    pub scanned: usize,
    /// Unreferenced blobs deleted.
    pub freed: usize,
    /// Blobs kept because some record still references them.
    pub retained: usize,
    /// Unreferenced blobs kept because they are younger than the sweep policy's
    /// `min_age`. Without this an operator cannot tell "nothing to reclaim" from
    /// "reclaiming held back by the age rule".
    pub deferred: usize,
}

/// Sweep unreferenced blobs from the store at `root` (gonzalo#292).
///
/// Blobs are content-addressed and **shared** — identical content dedups across
/// views and records alike — so GC marks against every record in the store at
/// once. Four things keep a blob: a record whose body *is* the blob, a
/// tombstone pinning the blob its record used to hold, every slice a
/// `graph-manifest` names (ADR 0012), and every shard a `vector-manifest`
/// names (ADR 0027). Marking any one alone would delete the other three's
/// content, so this scans raw (tombstones included) and unions all four via
/// [`gonzalo_core::live_blob_hashes`].
///
/// Deleting a record therefore does not reclaim its bytes: the tombstone pins
/// them until `collect` removes it past the horizon. Reclaiming is `delete`,
/// then `collect`, then this.
pub async fn gc(root: &Path, min_age: Duration) -> Result<GcSummary> {
    let store = FsStore::new(root);

    // Every record, across all repos and collections (unset prefix = all), and
    // raw so tombstones — the things that pin deleted records' blobs — are seen.
    let keys = store.list_raw(&KeyPrefix::default()).await?;
    let mut records = Vec::with_capacity(keys.len());
    for key in &keys {
        if let Some(record) = store.get_raw(key).await? {
            records.push(record);
        }
    }

    let live = gonzalo_core::live_blob_hashes(&records)?;
    let policy = gonzalo_core::SweepPolicy {
        min_age,
        now: std::time::SystemTime::now(),
    };
    let report = gonzalo_core::sweep_blobs_with(&store, &live, policy).await?;
    Ok(GcSummary {
        scanned: records.len(),
        freed: report.freed.len(),
        retained: report.retained,
        deferred: report.deferred,
    })
}

/// The optional knobs of an indexing run (#331).
///
/// Collapsed into one struct so a new knob is a field rather than an argument
/// respelled at every fan-out layer and call site.
#[derive(Debug, Clone)]
pub struct IndexOptions {
    /// Sweep unreferenced blobs after a successful index.
    pub gc_after: bool,
    /// Blobs younger than this are never swept (ADR 0028).
    pub min_age: Duration,
    /// Which repo-relative paths are eligible for indexing (#209).
    pub filter: IndexFilter,
    /// Pin the parse worker rather than resolving one.
    pub worker: Option<PathBuf>,
}

impl Default for IndexOptions {
    /// Written out rather than derived: a derived `min_age` would be
    /// `Duration::ZERO`, so a caller who sets `gc_after` and leaves the rest
    /// alone would reinstate the age-unaware sweep ADR 0028 removed. The
    /// default is the same floor `gc` uses.
    fn default() -> Self {
        Self {
            gc_after: false,
            min_age: gonzalo_core::DEFAULT_MIN_AGE,
            filter: IndexFilter::default(),
            worker: None,
        }
    }
}

/// [`index`] the `(repo, view)` view, then — when `opts.gc_after` — sweep
/// unreferenced blobs. The opt-in post-index trigger of gonzalo#104: the sweep
/// runs only after a successful index and always goes through [`gc`], which
/// marks against *every* record in the store (never a per-view subset), so a
/// slice the just-indexed view dropped but another view still references is
/// preserved.
///
/// The optional knobs ride in [`IndexOptions`] rather than as positional
/// arguments, so adding one is a field rather than an edit at every layer
/// between here and the call sites (#331).
pub async fn index_with_options(
    root: &Path,
    src: &Path,
    repo: &str,
    view: &str,
    opts: &IndexOptions,
) -> Result<(IndexSummary, Option<GcSummary>)> {
    let resolved;
    let worker = match &opts.worker {
        Some(path) => Some(path.as_path()),
        None => {
            resolved = resolve_parse_worker();
            resolved.worker()
        }
    };
    let summary = index_with_worker(root, src, repo, view, &opts.filter, worker).await?;
    let swept = if opts.gc_after {
        Some(gc(root, opts.min_age).await?)
    } else {
        None
    };
    Ok((summary, swept))
}

/// [`index_with_options`] for the common case: default filter, resolved parse
/// worker, and the sweep governed by `gc_after` / `min_age`.
pub async fn index_with_gc(
    root: &Path,
    src: &Path,
    repo: &str,
    view: &str,
    gc_after: bool,
    min_age: Duration,
) -> Result<(IndexSummary, Option<GcSummary>)> {
    index_with_options(
        root,
        src,
        repo,
        view,
        &IndexOptions {
            gc_after,
            min_age,
            ..IndexOptions::default()
        },
    )
    .await
}

// ─── watch (debounce core) ──────────────────────────────────────────────────

/// Coalesces a burst of filesystem change notifications so a rapid series of
/// edits triggers a single re-index rather than one per event. The clock is
/// injected (`now`), so the logic is deterministic and unit-testable without
/// real sleeps — the seam the watcher loop drives with `Instant::now()`.
#[derive(Debug)]
pub struct Debouncer {
    window: Duration,
    /// When the most recent unhandled change was observed.
    last_event: Option<Instant>,
}

impl Debouncer {
    /// A debouncer that fires once the tree has been quiet for `window`.
    pub fn new(window: Duration) -> Self {
        Self {
            window,
            last_event: None,
        }
    }

    /// Record that a change was observed at `now`.
    pub fn on_event(&mut self, now: Instant) {
        self.last_event = Some(now);
    }

    /// Whether a change is waiting to be handled.
    pub fn is_pending(&self) -> bool {
        self.last_event.is_some()
    }

    /// Whether a re-index is due at `now`: a change is pending and the tree has
    /// been quiet for at least `window` since the last event. False when nothing
    /// is pending.
    pub fn is_due(&self, now: Instant) -> bool {
        matches!(self.last_event, Some(t) if now.duration_since(t) >= self.window)
    }

    /// Clear the pending change after a re-index has run.
    pub fn clear(&mut self) {
        self.last_event = None;
    }
}

// ─── delete / reset / collect (tombstones, gonzalo#203) ─────────────────────

/// Open the fs store at `root` with an explicit ancestor cap (spec §3.9).
/// A cap of 0 is rejected here, before any record is touched.
pub fn open_store(root: &Path, ancestor_cap: usize) -> Result<FsStore> {
    FsStore::new(root)
        .with_ancestor_cap(ancestor_cap)
        .with_context(|| format!("opening store at {}", root.display()))
}

/// Outcome of [`delete`].
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum DeleteOutcome {
    /// A tombstone was written, or the record was already absent or deleted.
    Deleted,
    /// `expected` did not match. Nothing was written. `current` is the live
    /// record's revision.
    Conflict { current: Revision },
}

/// The identity the CLI stamps on records it writes. Matches `migrate` above
/// (which uses this constant) and the `ticket sync --author` default. The CLI
/// opens only `FsStore`, which honours a named deleter; over a daemon, spec
/// §3.6 applies instead and a non-admin token is stamped as itself.
pub const CLI_AUTHOR: &str = "gonzalo-cli";

/// Delete one record by writing a tombstone attributed to [`CLI_AUTHOR`]
/// (spec §3.1–§3.2). Deleting an absent or already-deleted key is `Deleted`.
pub async fn delete(
    root: &Path,
    ancestor_cap: usize,
    namespace: &str,
    collection: &str,
    id: &str,
    expected: Option<Revision>,
) -> Result<DeleteOutcome> {
    let store = open_store(root, ancestor_cap)?;
    let key = RecordKey::new(namespace, collection, id);
    let author = Some(Identity::new(CLI_AUTHOR));
    Ok(match store.delete_as(&key, expected, author).await? {
        gonzalo_core::DeleteResult::Deleted => DeleteOutcome::Deleted,
        gonzalo_core::DeleteResult::Conflict(conflict) => DeleteOutcome::Conflict {
            current: conflict.current.revision,
        },
    })
}

/// Restore the record at `namespace/collection/id` from its tombstone.
///
/// The restore is attributed to [`CLI_AUTHOR`], the operator restoring it,
/// rather than whoever deleted the record.
pub async fn undelete(
    root: &Path,
    ancestor_cap: usize,
    namespace: &str,
    collection: &str,
    id: &str,
) -> Result<Revision> {
    let store = open_store(root, ancestor_cap)?;
    let key = RecordKey::new(namespace, collection, id);
    let author = Identity::new(CLI_AUTHOR);
    Ok(gonzalo_core::undelete(&store, &key, gonzalo_core::now_ms(), Some(&author)).await?)
}

/// Physically remove the tombstone at `namespace/collection/id`, discarding the
/// restore window it holds.
///
/// Refuses anything that is not a tombstone. [`gonzalo_core::plan_purge`] does
/// not check the record's kind: it removes whatever matches the revision, so
/// without this check an operator could physically delete a live record
/// leaving no tombstone, which a peer that had not synced since would
/// resurrect (ADR 0021). Core keeps its contract because `collect` depends on
/// it; the check belongs here, at the operator surface (ADR 0030).
///
/// Benign race: a concurrent `collect` that purges the same key first makes
/// the plan a no-op, which `Store::purge` reports as success, so this can
/// print `purged:` for a tombstone it did not itself remove. That is
/// `Store::purge`'s documented idempotency.
pub async fn purge(
    root: &Path,
    ancestor_cap: usize,
    namespace: &str,
    collection: &str,
    id: &str,
) -> Result<Revision> {
    let store = open_store(root, ancestor_cap)?;
    let key = RecordKey::new(namespace, collection, id);
    let Some(record) = store.get_raw(&key).await? else {
        anyhow::bail!("record not found: {key}");
    };
    if !record.is_tombstone() {
        anyhow::bail!(
            "{key} is live at revision {}; purge only discards an existing tombstone. \
             To remove the record use `gonzalo delete`: it keeps a restore window \
             for manifest kinds, and purging that tombstone would discard it.",
            record.revision.counter
        );
    }
    let revision = record.revision.clone();
    // Exits 1, not `EXIT_CONFLICT`: purge takes no `--expected`, so there is no
    // stale input for a caller to refresh. The revision moved between our read
    // and the purge (a concurrent `undelete`, or a create over a non-manifest
    // tombstone), and re-running re-reads it.
    if let gonzalo_core::DeleteResult::Conflict(_) = store.purge(&key, revision.clone()).await? {
        anyhow::bail!("{key} changed while purging; nothing was removed, retry");
    }
    Ok(revision)
}

/// Tombstone every live record in `namespace` (optionally one `collection`)
/// via [`gonzalo_core::reset_as`], attributing each tombstone to
/// [`CLI_AUTHOR`].
pub async fn reset(
    root: &Path,
    ancestor_cap: usize,
    namespace: String,
    collection: Option<String>,
) -> Result<gonzalo_core::ResetReport> {
    let store = open_store(root, ancestor_cap)?;
    let prefix = KeyPrefix {
        namespace: Some(namespace),
        collection,
    };
    Ok(gonzalo_core::reset_as(&store, &prefix, Some(Identity::new(CLI_AUTHOR))).await?)
}

/// Exit code for a write conflict (`delete` with a stale `--expected`, or
/// `reset` leaving any record live). Distinct from 1 (error, via anyhow) and
/// 2 (usage error, via clap), so automation can tell "retry" from "broken"
/// (spec §3.10).
pub const EXIT_CONFLICT: u8 = 3;

/// The process exit code for a finished `reset`: 0 with no conflicts,
/// [`EXIT_CONFLICT`] otherwise (re-run to finish, spec §3.10).
pub fn reset_exit_code(report: &gonzalo_core::ResetReport) -> u8 {
    if report.conflicts.is_empty() {
        0
    } else {
        EXIT_CONFLICT
    }
}

/// Purge tombstones at least `horizon` old via [`gonzalo_core::collect`].
/// `namespace == None` collects across the whole store.
pub async fn collect(
    root: &Path,
    ancestor_cap: usize,
    namespace: Option<String>,
    collection: Option<String>,
    horizon: Duration,
    now_ms: i64,
) -> Result<gonzalo_core::CollectReport> {
    let store = open_store(root, ancestor_cap)?;
    let prefix = KeyPrefix {
        namespace,
        collection,
    };
    Ok(gonzalo_core::collect(&store, &prefix, horizon, now_ms).await?)
}

/// A collection horizon as the operator typed it, plus its parsed value, so
/// command output can echo the horizon it used (spec §3.10, §8.2).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Horizon {
    pub raw: String,
    pub duration: Duration,
}

/// Parse a duration of one positive whole number followed by exactly one unit:
/// `d`, `h`, `m` or `s` (e.g. `30d`, `12h`). Hand-rolled on purpose. The
/// workspace has no `humantime`, and a horizon never needs compound or
/// fractional forms.
pub fn parse_duration(raw: &str) -> std::result::Result<Duration, String> {
    let s = raw.trim();
    let split = s
        .find(|c: char| !c.is_ascii_digit())
        .ok_or_else(|| format!("{raw:?}: missing unit (use d, h, m or s, e.g. 30d)"))?;
    let (digits, unit) = s.split_at(split);
    if digits.is_empty() {
        return Err(format!(
            "{raw:?}: expected a number before the unit, e.g. 30d"
        ));
    }
    let per_unit: u64 = match unit {
        "d" => 86_400,
        "h" => 3_600,
        "m" => 60,
        "s" => 1,
        _ => return Err(format!("{raw:?}: unknown unit {unit:?} (use d, h, m or s)")),
    };
    let n: u64 = digits
        .parse()
        .map_err(|_| format!("{raw:?}: duration is too large"))?;
    if n == 0 {
        return Err(format!("{raw:?}: horizon must be greater than zero"));
    }
    let secs = n
        .checked_mul(per_unit)
        .ok_or_else(|| format!("{raw:?}: duration is too large"))?;
    Ok(Duration::from_secs(secs))
}

/// clap value parser for `--older-than`.
pub fn parse_horizon(raw: &str) -> std::result::Result<Horizon, String> {
    Ok(Horizon {
        raw: raw.trim().to_string(),
        duration: parse_duration(raw)?,
    })
}

/// clap value parser for `--expected`: a revision as JSON, exactly as
/// `gonzalo get` prints it, e.g. `{"counter":1,"hash":"…"}`.
pub fn parse_revision(raw: &str) -> std::result::Result<Revision, String> {
    serde_json::from_str(raw).map_err(|e| {
        format!(r#"expected a revision as JSON, e.g. {{"counter":1,"hash":"…"}} ({e})"#)
    })
}

// ─── sync_stores ─────────────────────────────────────────────────────────────

/// Summary returned by [`sync_stores`].
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct SyncSummary {
    pub copied_to_a: usize,
    pub copied_to_b: usize,
    /// Keys where one store was behind the other's revision chain and was
    /// overwritten (both directions, tombstones included).
    pub fast_forwarded: usize,
    pub merged: usize,
    /// The keys sync could not reconcile, so the caller can print them rather
    /// than only a count (gonzalo#299).
    pub conflicts: Vec<RecordKey>,
    /// Keys still racing when sync gave up after its pass limit. Non-empty means
    /// the stores may still disagree on them even with no conflicts, so the run
    /// is not proof of convergence (gonzalo#290).
    pub unconverged: Vec<RecordKey>,
}

/// The process exit code for a finished `sync`: 0 when it converged with no
/// conflicts, [`EXIT_CONFLICT`] when it reports conflicts to resolve or gave up
/// before converging. Both are recoverable outcomes a script retries or acts on,
/// matching `delete`, `reset` and `collect` (#290).
pub fn sync_exit_code(summary: &SyncSummary) -> u8 {
    if summary.conflicts.is_empty() && summary.unconverged.is_empty() {
        0
    } else {
        EXIT_CONFLICT
    }
}

/// Sync two filesystem stores via [`gonzalo_core::sync`], at the default
/// ancestor cap.
pub async fn sync_stores(a: &Path, b: &Path) -> Result<SyncSummary> {
    sync_stores_with_cap(a, b, gonzalo_core::DEFAULT_ANCESTOR_CAP).await
}

/// [`sync_stores`] with an explicit ancestor cap for both stores (spec §3.9).
pub async fn sync_stores_with_cap(a: &Path, b: &Path, ancestor_cap: usize) -> Result<SyncSummary> {
    let store_a = open_store(a, ancestor_cap)?;
    let store_b = open_store(b, ancestor_cap)?;
    let report = gonzalo_core::sync(&store_a, &store_b).await?;
    Ok(SyncSummary {
        copied_to_a: report.copied_to_a.len(),
        copied_to_b: report.copied_to_b.len(),
        fast_forwarded: report.fast_forwarded_to_a.len() + report.fast_forwarded_to_b.len(),
        merged: report.merged.len(),
        conflicts: report.conflicts.into_iter().map(|c| c.key).collect(),
        unconverged: report.unconverged,
    })
}

// ─── ticket sync ───────────────────────────────────────────────────────────

/// Per-connection ingest result.
pub struct TicketSyncReport {
    pub connection: String,
    pub summary: IngestSummary,
}

/// Load the ticket config, build each connection's source, and ingest its
/// tickets into the fs store at `root`.
pub async fn ticket_sync(
    config_path: &Path,
    root: &Path,
    author: &str,
) -> Result<Vec<TicketSyncReport>> {
    let config = Config::load(config_path).context("loading ticket config")?;
    let store = FsStore::new(root);
    let mut reports = Vec::new();
    for (name, source) in config.sources().context("building ticket sources")? {
        // Scope each record key by connection name so the same issue on two
        // boards yields two distinct records rather than colliding (#159).
        let summary = gonzalo_ticket::ingest(source.as_ref(), &store, author, Some(&name))
            .await
            .with_context(|| format!("syncing connection {name}"))?;
        reports.push(TicketSyncReport {
            connection: name,
            summary,
        });
    }
    Ok(reports)
}

// ─── ticket move ─────────────────────────────────────────────────────────────

/// Move a board card to the column for `category`. Selects the connection named
/// `connection`, or the sole connection if there is exactly one.
pub async fn ticket_move(
    config_path: &Path,
    connection: Option<&str>,
    uid: &str,
    category: &str,
) -> Result<()> {
    let cat = parse_category(category)
        .ok_or_else(|| anyhow::anyhow!("unknown state category {category:?}"))?;
    let config = Config::load(config_path).context("loading ticket config")?;
    let conn = select_connection(&config.connections, connection)?;
    let source = gonzalo_ticket_config::build_source(conn).context("building ticket source")?;
    source
        .set_state(uid, cat)
        .await
        .with_context(|| format!("moving {uid} to {category}"))?;
    Ok(())
}

/// Pick the requested connection by name, or the only one if unambiguous.
fn select_connection<'a>(
    connections: &'a [Connection],
    name: Option<&str>,
) -> Result<&'a Connection> {
    match name {
        Some(n) => connections
            .iter()
            .find(|c| c.name == n)
            .ok_or_else(|| anyhow::anyhow!("no connection named {n:?}")),
        None => match connections {
            [one] => Ok(one),
            [] => Err(anyhow::anyhow!("no connections configured")),
            _ => Err(anyhow::anyhow!(
                "multiple connections configured; pass --connection <name>"
            )),
        },
    }
}

// ─── Tests ────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {

    /// `IndexOptions::default()` must not mean "sweep everything immediately".
    ///
    /// `gc_after` is off by default, but a caller who turns it on and leaves
    /// `min_age` alone would otherwise reinstate exactly the age-unaware sweep
    /// #325 removed — blobs deleted out from under a writer about to reference
    /// them. The default is the same floor `gc` uses.
    #[test]
    fn default_index_options_keep_the_gc_grace_period() {
        let opts = IndexOptions::default();
        assert!(!opts.gc_after, "a sweep is opt-in");
        assert_eq!(
            opts.min_age,
            gonzalo_core::DEFAULT_MIN_AGE,
            "a defaulted min_age of zero sweeps blobs a writer is about to reference"
        );
    }
    use super::*;
    use gonzalo_graph::GraphStore;
    use tempfile::TempDir;

    fn write_file(dir: &Path, name: &str, contents: &str) {
        std::fs::write(dir.join(name), contents).unwrap();
    }

    // The real `index`/`index_with`/`index_with_gc` resolve a parse worker from
    // the environment, and since #212 that includes `PATH` — so on any machine
    // with `gonzalo-parse-worker` installed these tests would silently measure
    // that binary instead of this build. (Concretely: a released worker predates
    // `--extraction-version`, so #228 reads it as unknown and every run becomes a
    // full walk, and the incremental assertions below fail for reasons that have
    // nothing to do with what they test.)
    //
    // These shadow the real entry points with parsing pinned in-process, so a
    // test asserts about the code under test and nothing ambient. Worker
    // resolution itself is covered directly, against `resolve_worker_from`.

    async fn index(root: &Path, src: &Path, repo: &str, view: &str) -> Result<IndexSummary> {
        index_with(root, src, repo, view, &IndexFilter::default()).await
    }

    async fn index_with(
        root: &Path,
        src: &Path,
        repo: &str,
        view: &str,
        filter: &IndexFilter,
    ) -> Result<IndexSummary> {
        index_with_worker(root, src, repo, view, filter, None).await
    }

    async fn index_with_gc(
        root: &Path,
        src: &Path,
        repo: &str,
        view: &str,
        gc_after: bool,
        min_age: Duration,
    ) -> Result<(IndexSummary, Option<GcSummary>)> {
        index_with_options(
            root,
            src,
            repo,
            view,
            &IndexOptions {
                gc_after,
                min_age,
                ..IndexOptions::default()
            },
        )
        .await
    }

    // ── migrate: basic import ────────────────────────────────────────────────

    #[tokio::test]
    async fn migrate_imports_two_files() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();

        write_file(src.path(), "alpha.md", "hello alpha");
        write_file(src.path(), "beta.md", "hello beta");

        let summary = migrate(
            root.path(),
            src.path(),
            "testns",
            "testcol",
            RecordKind::Topic,
        )
        .await
        .unwrap();

        assert_eq!(summary.imported, 2, "should have imported 2 files");
        assert_eq!(summary.skipped, 0, "nothing should be skipped yet");
    }

    // ── list: shows the right keys after migrate ─────────────────────────────

    #[tokio::test]
    async fn list_returns_migrated_keys() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();

        write_file(src.path(), "alpha.md", "hello alpha");
        write_file(src.path(), "beta.md", "hello beta");

        migrate(
            root.path(),
            src.path(),
            "testns",
            "testcol",
            RecordKind::Topic,
        )
        .await
        .unwrap();

        let keys = list(root.path(), None, None).await.unwrap();
        assert_eq!(keys.len(), 2);
    }

    // ── migrate: idempotent on second run ────────────────────────────────────

    #[tokio::test]
    async fn migrate_is_idempotent() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();

        write_file(src.path(), "alpha.md", "hello alpha");
        write_file(src.path(), "beta.md", "hello beta");

        migrate(
            root.path(),
            src.path(),
            "testns",
            "testcol",
            RecordKind::Topic,
        )
        .await
        .unwrap();

        let second = migrate(
            root.path(),
            src.path(),
            "testns",
            "testcol",
            RecordKind::Topic,
        )
        .await
        .unwrap();

        assert_eq!(
            second.skipped, 2,
            "second run should skip both already-imported files"
        );
        assert_eq!(second.imported, 0, "second run should import nothing new");
    }

    // ── get: round-trips body ────────────────────────────────────────────────

    #[tokio::test]
    async fn get_returns_migrated_record_body() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();

        write_file(src.path(), "alpha.md", "hello alpha");

        migrate(
            root.path(),
            src.path(),
            "testns",
            "testcol",
            RecordKind::Topic,
        )
        .await
        .unwrap();

        // The id is the source-relative path verbatim.
        let record = get(root.path(), "testns", "testcol", "alpha.md")
            .await
            .unwrap();

        assert!(record.is_some(), "record should be present");
        let body = record.unwrap().body;
        assert_eq!(body.bytes(), b"hello alpha");
    }

    // ── status: correct namespace/collection count ───────────────────────────

    #[tokio::test]
    async fn status_groups_by_ns_col() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();

        write_file(src.path(), "alpha.md", "hello alpha");
        write_file(src.path(), "beta.md", "hello beta");

        migrate(
            root.path(),
            src.path(),
            "testns",
            "testcol",
            RecordKind::Topic,
        )
        .await
        .unwrap();

        let map = status(root.path()).await.unwrap();
        assert_eq!(map.get("testns/testcol").copied(), Some(2));
    }

    // ── sync_stores: propagates records ─────────────────────────────────────

    #[tokio::test]
    async fn sync_stores_copies_to_b() {
        let store_a = TempDir::new().unwrap();
        let store_b = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();

        write_file(src.path(), "note.md", "synced content");

        // Import only into store A.
        migrate(
            store_a.path(),
            src.path(),
            "testns",
            "testcol",
            RecordKind::Topic,
        )
        .await
        .unwrap();

        let summary = sync_stores(store_a.path(), store_b.path()).await.unwrap();
        assert_eq!(summary.copied_to_b, 1);

        // Store B should now have the key.
        let keys = list(store_b.path(), None, None).await.unwrap();
        assert_eq!(keys.len(), 1);
    }

    #[tokio::test]
    async fn sync_stores_fast_forwards_a_peer_that_is_behind() {
        let store_a = TempDir::new().unwrap();
        let store_b = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "note.md", "synced content");
        migrate(
            store_a.path(),
            src.path(),
            "testns",
            "testcol",
            RecordKind::Topic,
        )
        .await
        .unwrap();
        let _ = sync_stores(store_a.path(), store_b.path()).await.unwrap();

        // A edits the record; B is now simply behind.
        let fs_a = FsStore::new(store_a.path());
        let key = fs_a.list(&KeyPrefix::default()).await.unwrap().remove(0);
        let cur = fs_a.get(&key).await.unwrap().unwrap();
        let mut next = cur.clone();
        next.body = Body::Inline(b"synced content\nmore\n".to_vec());
        next.revision = cur.revision.next(next.body.bytes());
        next.parent = Some(cur.revision.clone());
        next.ancestors = Vec::new();
        assert!(matches!(
            fs_a.put(next, Some(cur.revision)).await.unwrap(),
            PutResult::Committed(_)
        ));

        let summary = sync_stores(store_a.path(), store_b.path()).await.unwrap();
        assert_eq!(summary.fast_forwarded, 1);
        assert_eq!(summary.merged, 0);
        assert!(summary.conflicts.is_empty());
        assert!(summary.unconverged.is_empty());
    }

    // ── ticket_sync: empty config → no reports ───────────────────────────────

    #[tokio::test]
    async fn ticket_sync_with_no_connections_returns_no_reports() {
        let root = TempDir::new().unwrap();
        let cfg = TempDir::new().unwrap();
        let cfg_path = cfg.path().join("tickets.toml");
        std::fs::write(&cfg_path, "").unwrap(); // empty config = zero connections

        let reports = ticket_sync(&cfg_path, root.path(), "tester").await.unwrap();
        assert!(reports.is_empty());
    }

    // ── ticket_move: unknown category errors before any network call ─────────

    #[tokio::test]
    async fn ticket_move_unknown_category_errors() {
        let cfg = TempDir::new().unwrap();
        let cfg_path = cfg.path().join("tickets.toml");
        std::fs::write(
            &cfg_path,
            "[[connection]]\nname=\"b\"\nprovider=\"github-projects\"\norg=\"o\"\nproject=1\ntoken_env=\"X\"\n",
        )
        .unwrap();
        // "frozen" is not a valid category → error before any network call.
        let err = ticket_move(&cfg_path, None, "o/r#1", "frozen")
            .await
            .unwrap_err();
        assert!(err.to_string().contains("category"), "got {err}");
    }

    // ── ticket_move: ambiguous connection requires --connection ─────────────

    #[tokio::test]
    async fn ticket_move_requires_connection_when_many() {
        let cfg = TempDir::new().unwrap();
        let cfg_path = cfg.path().join("tickets.toml");
        std::fs::write(
            &cfg_path,
            "[[connection]]\nname=\"a\"\nprovider=\"github-projects\"\norg=\"o\"\nproject=1\ntoken_env=\"X\"\n\
             [[connection]]\nname=\"b\"\nprovider=\"github-projects\"\norg=\"o\"\nproject=2\ntoken_env=\"Y\"\n",
        )
        .unwrap();
        let err = ticket_move(&cfg_path, None, "o/r#1", "done")
            .await
            .unwrap_err();
        assert!(err.to_string().contains("connection"), "got {err}");
    }

    // ── index: build a queryable view from a source tree ─────────────────────

    // Review Focus 3: the ergonomic regression this design accepts must be a
    // clean refusal with guidance, not a panic and not a half-written
    // manifest. This is the graph-indexer half of #333's hazard: `index`
    // derives `expected` from a consumer `get`, which hides the tombstone.
    #[tokio::test]
    async fn index_over_a_deleted_manifest_is_refused_with_guidance() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");

        index(root.path(), src.path(), "r", "main").await.unwrap();

        // Delete the view's manifest, as `gonzalo delete` or `reset` would.
        let store = open_store(root.path(), gonzalo_core::DEFAULT_ANCESTOR_CAP).unwrap();
        let key = Manifest::key("r", "main");
        assert_eq!(
            store.delete(&key, None).await.unwrap(),
            gonzalo_core::DeleteResult::Deleted
        );
        let tomb = store.get_raw(&key).await.unwrap().expect("tombstone");
        let retained = tomb.body.clone();

        // Re-indexing must refuse rather than overwrite the restore window.
        let err = index(root.path(), src.path(), "r", "main")
            .await
            .err()
            .expect("indexing over a deleted manifest must be refused");
        let msg = format!("{err:#}");
        assert!(
            msg.contains("gonzalo undelete") && msg.contains("gonzalo purge"),
            "the refusal must tell the operator both ways out, got {msg}"
        );

        // The manifest record is untouched, so the window is intact (orphan slice
        // blobs are gc'd).
        let after = store
            .get_raw(&key)
            .await
            .unwrap()
            .expect("tombstone survives");
        assert!(after.is_tombstone());
        assert_eq!(after.body, retained, "the retained body was not touched");

        // Follow the refusal's own advice: purge the tombstone, then the
        // indexer is unblocked. Nothing else asserts this for a key reached
        // through `index`.
        purge(
            root.path(),
            gonzalo_core::DEFAULT_ANCESTOR_CAP,
            &key.namespace,
            &key.collection,
            &key.id,
        )
        .await
        .unwrap();
        index(root.path(), src.path(), "r", "main")
            .await
            .expect("indexing succeeds once the tombstone is purged");
        let live = store.get_raw(&key).await.unwrap().expect("manifest");
        assert!(!live.is_tombstone(), "the view is live again");
    }

    #[tokio::test]
    async fn index_builds_a_queryable_view() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn helper() {}");
        std::fs::create_dir(src.path().join("sub")).unwrap();
        write_file(src.path(), "sub/b.rs", "fn caller() { helper(); }");

        let summary = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert_eq!(summary.files, 2);
        assert_eq!(summary.added, 2);
        assert_eq!(summary.modified, 0);
        assert_eq!(summary.deleted, 0);

        // The indexed view assembles and answers real queries.
        let store = FsStore::new(root.path());
        let manifest = Manifest::from_body(
            &store
                .get(&Manifest::key("r", "main"))
                .await
                .unwrap()
                .unwrap()
                .body,
        )
        .unwrap();
        let graph = gonzalo_graph::assemble(&manifest, &store).await.unwrap();
        let defs = graph.definitions("helper");
        assert_eq!(defs.len(), 1);
        assert_eq!(defs[0].path, "a.rs");
        assert_eq!(graph.callers_of("helper"), vec!["caller".to_string()]);
        assert!(
            graph
                .symbols_in_file("sub/b.rs")
                .iter()
                .any(|s| s.name == "caller")
        );
    }

    #[tokio::test]
    async fn reindex_reports_modifications_and_deletions() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "keep.rs", "fn keep() {}");
        write_file(src.path(), "gone.rs", "fn gone() {}");
        index(root.path(), src.path(), "r", "v").await.unwrap();

        // Change one file, remove another.
        write_file(src.path(), "keep.rs", "fn keep() { extra(); }");
        std::fs::remove_file(src.path().join("gone.rs")).unwrap();

        let summary = index(root.path(), src.path(), "r", "v").await.unwrap();
        assert_eq!(summary.files, 1);
        assert_eq!(summary.added, 0);
        assert_eq!(summary.modified, 1);
        assert_eq!(summary.deleted, 1);
    }

    #[tokio::test]
    async fn index_skips_non_rust_and_build_dirs() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "real.rs", "fn real() {}");
        write_file(src.path(), "notes.txt", "not source");
        std::fs::create_dir(src.path().join("target")).unwrap();
        write_file(src.path(), "target/gen.rs", "fn generated() {}");

        let summary = index(root.path(), src.path(), "r", "v").await.unwrap();
        assert_eq!(summary.files, 1, "only real.rs is indexed");
        assert_eq!(summary.added, 1);
    }

    #[tokio::test]
    async fn index_handles_multiple_languages() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "lib.rs", "fn rust_fn() {}");
        write_file(src.path(), "app.py", "def py_fn():\n    pass\n");

        let summary = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert_eq!(summary.files, 2, "both the .rs and .py file are indexed");

        // Both languages' symbols are queryable in the assembled view.
        let store = FsStore::new(root.path());
        let manifest = Manifest::from_body(
            &store
                .get(&Manifest::key("r", "main"))
                .await
                .unwrap()
                .unwrap()
                .body,
        )
        .unwrap();
        let graph = gonzalo_graph::assemble(&manifest, &store).await.unwrap();
        assert_eq!(graph.definitions("rust_fn")[0].path, "lib.rs");
        assert_eq!(graph.definitions("py_fn")[0].path, "app.py");
    }

    #[tokio::test]
    async fn index_writes_a_queryable_sqlite_graph() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(
            src.path(),
            "lib.rs",
            "fn helper() {}\nfn main() { helper(); }",
        );
        index(root.path(), src.path(), "r", "main").await.unwrap();

        // The persistent per-view graph exists under <root>/graphs and answers
        // queries without re-assembly.
        let db = view_db_path(&root.path().join("graphs"), "r", "main");
        let g = SqliteGraphStore::open(&db).unwrap();
        assert_eq!(g.definitions("helper")[0].path, "lib.rs");
        assert_eq!(g.callers_of("helper"), vec!["main".to_string()]);
    }

    #[tokio::test]
    async fn reindex_removes_deleted_paths_from_the_sqlite_graph() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "keep.rs", "fn keep() {}");
        write_file(src.path(), "gone.rs", "fn gone() {}");
        index(root.path(), src.path(), "r", "v").await.unwrap();

        std::fs::remove_file(src.path().join("gone.rs")).unwrap();
        index(root.path(), src.path(), "r", "v").await.unwrap();

        let g =
            SqliteGraphStore::open(view_db_path(&root.path().join("graphs"), "r", "v")).unwrap();
        assert_eq!(g.definitions("keep").len(), 1);
        assert!(
            g.definitions("gone").is_empty(),
            "deleted file's symbols must be gone from the graph"
        );
    }

    // ── index: manifest commit precedes the SQLite graph write (gonzalo#153) ──

    #[tokio::test]
    async fn build_desired_stages_graph_writes_without_touching_the_store() {
        // The reordered control flow (#153): parsing/desired-set construction
        // must only *stage* persistent-graph mutations. The SqliteGraphStore is
        // advanced solely by GraphStaging::apply — which index() calls strictly
        // after the manifest commit — so a manifest Conflict leaves the graph
        // untouched instead of advanced ahead of an uncommitted manifest.
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");

        let store = FsStore::new(root.path());
        let db_path = view_db_path(&root.path().join("graphs"), "r", "main");
        let mut graph = SqliteGraphStore::open(&db_path).unwrap();

        let mut staging = GraphStaging::default();
        let DesiredView {
            entries: desired,
            files,
            incremental,
            ..
        } = build_desired_full(
            &store,
            &mut staging,
            None,
            src.path(),
            &IndexFilter::default(),
        )
        .await
        .unwrap();
        assert!(!incremental);
        assert_eq!(files, 1);
        assert_eq!(desired.len(), 1);
        assert_eq!(staging.inserts.len(), 1, "the write is staged, not applied");

        // Nothing has reached the persistent graph yet — this is the state after
        // a manifest Conflict would `bail!`.
        assert!(
            graph.definitions("a").is_empty(),
            "SqliteGraphStore must be untouched until the manifest commits"
        );

        // Applying the staged writes (index()'s post-commit step) advances it.
        staging.apply(&mut graph);
        assert_eq!(graph.definitions("a")[0].path, "a.rs");
    }

    // ── index: git-driven incremental sync (gonzalo#93) ──────────────────────

    /// Init a git repo at `dir` and commit every current file.
    fn git_init_commit(dir: &Path) {
        let repo = git2::Repository::init(dir).unwrap();
        let mut index = repo.index().unwrap();
        index
            .add_all(["*"].iter(), git2::IndexAddOption::DEFAULT, None)
            .unwrap();
        index.write().unwrap();
        let tree = repo.find_tree(index.write_tree().unwrap()).unwrap();
        let sig = git2::Signature::now("t", "t@localhost").unwrap();
        repo.commit(Some("HEAD"), &sig, &sig, "c", &tree, &[])
            .unwrap();
    }

    // ── the parse-version warning says the right thing (gonzalo#239) ─────────

    #[test]
    fn an_unknown_worker_version_does_not_claim_the_extraction_is_wrong() {
        // Every worker released before #228 answers nothing while very likely
        // producing current extraction. Telling that user to upgrade "to get
        // version N extraction" asserts their data is wrong. It is not — it is
        // unconfirmed, which is a different thing and the whole point here.
        let msg = worker_version_warning(0);
        assert!(msg.contains("did not report"), "{msg}");
        assert!(msg.contains("cannot be confirmed"), "{msg}");
        assert!(
            !msg.contains("WARNING"),
            "unconfirmed is a note, not a warning: {msg}"
        );
        assert!(
            !msg.contains(&format!("to get {EXTRACTION_VERSION} extraction")),
            "must not claim the recorded extraction is wrong: {msg}"
        );
    }

    #[test]
    fn a_known_mismatch_still_warns_plainly() {
        // A worker that reports a *different* version genuinely disagrees about
        // what a parse records, so the strong wording is accurate there.
        let msg = worker_version_warning(1);
        assert!(msg.contains("WARNING"), "{msg}");
        assert!(msg.contains("reports extraction version 1"), "{msg}");
        assert!(msg.contains(&EXTRACTION_VERSION.to_string()), "{msg}");
    }

    // ── index: locating the parse worker (gonzalo#212) ───────────────────────

    /// Create an empty file at `dir/name` and return its path.
    fn touch(dir: &Path, name: &str) -> PathBuf {
        let path = dir.join(name);
        std::fs::write(&path, "").unwrap();
        path
    }

    #[test]
    fn the_env_override_wins_over_everything_else() {
        let over = TempDir::new().unwrap();
        let beside = TempDir::new().unwrap();
        let on_path = TempDir::new().unwrap();
        let chosen = touch(over.path(), "my-worker");
        touch(beside.path(), worker_file_name());
        touch(on_path.path(), worker_file_name());

        let mode = resolve_worker_from(
            Some(chosen.clone()),
            Some(beside.path().join("gonzalo")),
            &[on_path.path().to_path_buf()],
        );
        assert_eq!(
            mode,
            ParseMode::Isolated {
                path: chosen,
                source: WorkerSource::Env
            }
        );
    }

    #[test]
    fn a_sibling_of_the_executable_is_preferred_over_path() {
        let beside = TempDir::new().unwrap();
        let on_path = TempDir::new().unwrap();
        let sibling = touch(beside.path(), worker_file_name());
        touch(on_path.path(), worker_file_name());

        let mode = resolve_worker_from(
            None,
            Some(beside.path().join("gonzalo")),
            &[on_path.path().to_path_buf()],
        );
        assert_eq!(
            mode,
            ParseMode::Isolated {
                path: sibling,
                source: WorkerSource::Sibling
            }
        );
    }

    #[test]
    fn the_worker_is_found_on_path_when_it_is_not_beside_the_executable() {
        // The layouts a sibling check cannot reach: a symlinked `gonzalo` on
        // PATH, a container that copies one binary, `cargo run` from a
        // workspace target dir. The worker is plainly available in all three.
        let beside = TempDir::new().unwrap();
        let empty = TempDir::new().unwrap();
        let on_path = TempDir::new().unwrap();
        let found = touch(on_path.path(), worker_file_name());

        let mode = resolve_worker_from(
            None,
            Some(beside.path().join("gonzalo")),
            &[empty.path().to_path_buf(), on_path.path().to_path_buf()],
        );
        assert_eq!(
            mode,
            ParseMode::Isolated {
                path: found,
                source: WorkerSource::Path
            }
        );
    }

    #[test]
    fn no_worker_anywhere_falls_back_and_names_every_place_it_looked() {
        let missing = TempDir::new().unwrap();
        let beside = TempDir::new().unwrap();
        let on_path = TempDir::new().unwrap();
        let override_path = missing.path().join("not-here");

        let mode = resolve_worker_from(
            Some(override_path.clone()),
            Some(beside.path().join("gonzalo")),
            &[on_path.path().to_path_buf()],
        );

        let ParseMode::InProcess { searched } = &mode else {
            panic!("expected the in-process fallback, got {mode:?}");
        };
        // A bad override is a place it looked, not a reason to stop looking.
        assert_eq!(
            searched,
            &[
                override_path,
                beside.path().join(worker_file_name()),
                on_path.path().join(worker_file_name()),
            ]
        );
        assert!(!mode.is_isolated());
        assert!(mode.worker().is_none());

        // The warning has to be actionable: it names the binary, every path
        // tried, and what a grammar crash now costs.
        let warning = mode.warning().expect("in-process must warn");
        assert!(warning.contains(worker_file_name()));
        assert!(warning.contains(&beside.path().join(worker_file_name()).display().to_string()));
        assert!(warning.contains("--require-parse-worker"));
        assert!(warning.contains("IN-PROCESS"));
    }

    #[test]
    fn an_isolated_mode_names_its_worker_and_never_warns() {
        let dir = TempDir::new().unwrap();
        let worker = touch(dir.path(), worker_file_name());
        let mode = resolve_worker_from(None, Some(dir.path().join("gonzalo")), &[]);

        assert_eq!(mode.worker(), Some(worker.as_path()));
        assert!(mode.warning().is_none(), "isolation is the quiet path");
        let summary = mode.summary();
        assert!(summary.contains("isolated"));
        assert!(summary.contains(&worker.display().to_string()));
        assert!(summary.contains("beside the executable"));
    }

    #[test]
    fn a_directory_named_like_the_worker_is_not_mistaken_for_it() {
        // `is_file`, not `exists` — a directory of that name on PATH would
        // otherwise be selected and every parse would fail to spawn.
        let on_path = TempDir::new().unwrap();
        std::fs::create_dir(on_path.path().join(worker_file_name())).unwrap();
        let mode = resolve_worker_from(None, None, &[on_path.path().to_path_buf()]);
        assert!(!mode.is_isolated());
    }

    // ── index: the parse path's extraction version (gonzalo#228) ─────────────

    /// An executable stand-in for `gonzalo-parse-worker` that reports
    /// `reports` as its extraction version and parses everything to an empty
    /// slice. Enough to drive an index run without a real grammar.
    fn worker_reporting(dir: &Path, name: &str, reports: Option<u32>) -> PathBuf {
        use std::os::unix::fs::PermissionsExt;
        let answer = match reports {
            // A pre-#228 worker ignores argv and blocks on stdin; the probe
            // closes stdin, so it exits silently having answered nothing.
            None => String::new(),
            Some(v) => format!("[ \"$1\" = --extraction-version ] && {{ echo {v}; exit 0; }}\n"),
        };
        let path = dir.join(name);
        std::fs::write(
            &path,
            format!(
                "#!/bin/sh\n{answer}while IFS= read -r _; do \
                 printf '%s\\n' '{{\"symbols\":[],\"references\":[]}}'; done\n"
            ),
        )
        .unwrap();
        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
        path
    }

    fn fmt_of(root: &Path) -> String {
        let p = view_db_path(&root.join("graphs"), "r", "main").with_extension("fmt");
        std::fs::read_to_string(p).unwrap()
    }

    #[tokio::test]
    async fn a_stale_worker_forces_a_full_walk_and_is_never_credited() {
        // The hole #228 records: EXTRACTION_VERSION is compiled into the CLI,
        // but the *worker* is what parses. A new CLI driving an old worker used
        // to emit pre-upgrade extraction and then stamp `.fmt` with the new
        // version — leaving the view wrong AND marked current, so no later
        // re-index would ever repair it.
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");
        git_init_commit(src.path());
        let bins = TempDir::new().unwrap();

        // A worker that agrees with this build behaves exactly as before.
        let current = worker_reporting(bins.path(), "current", Some(EXTRACTION_VERSION));
        let filter = IndexFilter::default();
        index_with_worker(
            root.path(),
            src.path(),
            "r",
            "main",
            &filter,
            Some(current.as_path()),
        )
        .await
        .unwrap();
        assert_eq!(fmt_of(root.path()), EXTRACTION_VERSION.to_string());
        assert!(
            index_with_worker(
                root.path(),
                src.path(),
                "r",
                "main",
                &filter,
                Some(current.as_path())
            )
            .await
            .unwrap()
            .incremental,
            "a worker at the current version still indexes incrementally"
        );

        // Swap in a worker one version behind — the exact upgrade-order slip.
        let stale = worker_reporting(bins.path(), "stale", Some(EXTRACTION_VERSION - 1));
        let summary = index_with_worker(
            root.path(),
            src.path(),
            "r",
            "main",
            &filter,
            Some(stale.as_path()),
        )
        .await
        .unwrap();
        assert!(
            !summary.incremental,
            "a worker that disagrees with the CLI must rebuild in full"
        );
        assert_eq!(
            fmt_of(root.path()),
            (EXTRACTION_VERSION - 1).to_string(),
            "and the view must be credited to what actually parsed, not to the CLI"
        );

        // Because the stamp is honest, fixing the worker repairs the view on the
        // next run instead of leaving it silently half-upgraded forever.
        let after = index_with_worker(
            root.path(),
            src.path(),
            "r",
            "main",
            &filter,
            Some(current.as_path()),
        )
        .await
        .unwrap();
        assert!(!after.incremental, "restoring the worker re-walks the view");
        assert_eq!(fmt_of(root.path()), EXTRACTION_VERSION.to_string());
    }

    #[tokio::test]
    async fn a_worker_that_cannot_report_its_version_is_recorded_as_unknown() {
        // A pre-#228 worker answers nothing, so the view is credited to 0 —
        // "built by a parser that would not say". Two such runs may still go
        // incrementally: 0 == 0 claims only that the parse path is the same
        // unknown one, which is true, and forcing a full walk on every index
        // would punish every not-yet-upgraded user (badly so under `--watch`).
        //
        // What must not happen is 0 being mistaken for this build's version —
        // that is the #228 failure, and the transition below is where it shows.
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");
        git_init_commit(src.path());
        let bins = TempDir::new().unwrap();
        let silent = worker_reporting(bins.path(), "silent", None);
        let filter = IndexFilter::default();

        index_with_worker(
            root.path(),
            src.path(),
            "r",
            "main",
            &filter,
            Some(silent.as_path()),
        )
        .await
        .unwrap();
        assert_eq!(fmt_of(root.path()), "0", "unknown is recorded, not assumed");

        // Installing a worker that does report repairs the view rather than
        // inheriting whatever the silent one left behind.
        let current = worker_reporting(bins.path(), "current", Some(EXTRACTION_VERSION));
        let healed = index_with_worker(
            root.path(),
            src.path(),
            "r",
            "main",
            &filter,
            Some(current.as_path()),
        )
        .await
        .unwrap();
        assert!(
            !healed.incremental,
            "a view built by an unknown parser must be rebuilt once a known one appears"
        );
        assert_eq!(fmt_of(root.path()), EXTRACTION_VERSION.to_string());
    }

    #[tokio::test]
    async fn in_process_parsing_is_credited_to_this_build() {
        // With no worker at all the CLI parses in-process, so its own constant
        // does describe what was recorded (#212's fallback path).
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");
        git_init_commit(src.path());

        index_with_worker(
            root.path(),
            src.path(),
            "r",
            "main",
            &IndexFilter::default(),
            None,
        )
        .await
        .unwrap();
        assert_eq!(fmt_of(root.path()), EXTRACTION_VERSION.to_string());
    }

    #[tokio::test]
    async fn a_format_change_forces_a_full_walk() {
        // Without this, an existing view keeps pre-upgrade extraction forever:
        // the incremental driver carries unchanged slices forward untouched, so
        // a parser improvement never reaches files that did not change (#223).
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");
        git_init_commit(src.path());
        index(root.path(), src.path(), "r", "main").await.unwrap();

        // A second run would normally take the incremental path...
        let fmt = view_db_path(&root.path().join("graphs"), "r", "main").with_extension("fmt");
        assert_eq!(
            std::fs::read_to_string(&fmt).unwrap(),
            EXTRACTION_VERSION.to_string()
        );
        assert!(
            index(root.path(), src.path(), "r", "main")
                .await
                .unwrap()
                .incremental
        );

        // ...but not when the view was built by an older extraction format.
        std::fs::write(&fmt, "1").unwrap();
        let summary = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert!(!summary.incremental, "a format change must rebuild in full");
        assert_eq!(
            std::fs::read_to_string(&fmt).unwrap(),
            EXTRACTION_VERSION.to_string(),
            "and must record the version it rebuilt with"
        );
    }

    #[tokio::test]
    async fn first_index_of_git_repo_is_full_then_reindex_is_incremental() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");
        write_file(src.path(), "b.rs", "fn b() {}");
        git_init_commit(src.path());

        // First run: no recorded base yet → full walk, records the base.
        let first = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert!(!first.incremental, "first index is a full walk");
        assert_eq!(first.files, 2);
        assert_eq!(first.added, 2);

        // Change the working tree: modify a.rs, add untracked c.rs, leave b.rs.
        write_file(src.path(), "a.rs", "fn a() { helper(); }");
        write_file(src.path(), "c.rs", "fn c() {}");

        let second = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert!(second.incremental, "second index uses the git diff driver");
        assert_eq!(second.files, 2, "only a.rs and c.rs are re-parsed");
        assert_eq!(second.added, 1, "c.rs added");
        assert_eq!(second.modified, 1, "a.rs modified");
        assert_eq!(second.deleted, 0);

        // b.rs was carried forward unchanged and is still queryable.
        let g =
            SqliteGraphStore::open(view_db_path(&root.path().join("graphs"), "r", "main")).unwrap();
        assert_eq!(g.definitions("b")[0].path, "b.rs");
        assert_eq!(g.definitions("c")[0].path, "c.rs");
    }

    #[tokio::test]
    async fn incremental_index_drops_deleted_files() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "keep.rs", "fn keep() {}");
        write_file(src.path(), "gone.rs", "fn gone() {}");
        git_init_commit(src.path());
        index(root.path(), src.path(), "r", "v").await.unwrap();

        std::fs::remove_file(src.path().join("gone.rs")).unwrap();
        let summary = index(root.path(), src.path(), "r", "v").await.unwrap();
        assert!(summary.incremental);
        assert_eq!(summary.deleted, 1);

        let g =
            SqliteGraphStore::open(view_db_path(&root.path().join("graphs"), "r", "v")).unwrap();
        assert_eq!(g.definitions("keep").len(), 1);
        assert!(g.definitions("gone").is_empty(), "deleted file is gone");
    }

    #[tokio::test]
    async fn non_git_src_stays_full_walk() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");
        let summary = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert!(!summary.incremental, "a non-git tree cannot go incremental");
    }

    // ── gc: sweep blobs no record references (gonzalo#94, #292) ─────────────

    #[tokio::test]
    async fn gc_on_empty_store_frees_nothing() {
        let root = TempDir::new().unwrap();
        let summary = gc(root.path(), Duration::ZERO).await.unwrap();
        assert_eq!(summary.scanned, 0);
        assert_eq!(summary.freed, 0);
        assert_eq!(summary.retained, 0);
    }

    #[tokio::test]
    async fn gc_frees_slices_orphaned_by_a_reindex() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");
        index(root.path(), src.path(), "r", "main").await.unwrap();

        // Reindex with different content: the original slice is now orphaned.
        write_file(src.path(), "a.rs", "fn a() { b(); }");
        index(root.path(), src.path(), "r", "main").await.unwrap();

        let summary = gc(root.path(), Duration::ZERO).await.unwrap();
        assert_eq!(summary.scanned, 1, "the view's manifest record");
        assert_eq!(summary.freed, 1, "the pre-edit slice is unreferenced");
        assert_eq!(summary.retained, 1, "the current slice stays");

        // GC did not corrupt the live view.
        let g =
            SqliteGraphStore::open(view_db_path(&root.path().join("graphs"), "r", "main")).unwrap();
        assert_eq!(g.definitions("a")[0].path, "a.rs");
    }

    // The flag's default is a string clap parses at startup. A typo here is a
    // runtime failure on every `gonzalo gc`, which no other test would catch.
    #[test]
    fn the_default_min_age_spelling_parses_to_an_hour() {
        assert_eq!(
            parse_duration("1h").unwrap(),
            std::time::Duration::from_secs(3600)
        );
        assert_eq!(parse_duration("1h").unwrap(), gonzalo_core::DEFAULT_MIN_AGE);
    }

    #[tokio::test]
    async fn gc_defers_a_blob_younger_than_min_age_and_counts_it() {
        let dir = tempfile::tempdir().unwrap();
        let store = FsStore::new(dir.path());
        store.put_blob(b"brand new").await.unwrap();

        // The default horizon: nothing written moments ago is collectable.
        let held = gc(dir.path(), gonzalo_core::DEFAULT_MIN_AGE).await.unwrap();
        assert_eq!(held.freed, 0);
        assert_eq!(held.deferred, 1);

        // An operator asking for the old behaviour gets it.
        let swept = gc(dir.path(), std::time::Duration::ZERO).await.unwrap();
        assert_eq!(swept.freed, 1);
        assert_eq!(swept.deferred, 0);
    }

    // ── index: view membership (#209) ────────────────────────────────────────

    #[tokio::test]
    async fn index_excludes_vendored_bundles_from_the_view() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");
        write_file(src.path(), "mermaid.min.js", "var a=1,e=2,t=3;");

        let summary = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert_eq!(summary.files, 1, "only the hand-written source");
        assert_eq!(summary.ignored.files, 1, "the minified bundle, reported");

        let graph =
            SqliteGraphStore::open(view_db_path(&root.path().join("graphs"), "r", "main")).unwrap();
        assert!(
            graph.all_symbols().iter().all(|s| s.path == "a.rs"),
            "no symbol may come from a vendored bundle"
        );
    }

    #[tokio::test]
    async fn incremental_reindex_prunes_paths_a_laxer_run_admitted() {
        // The upgrade path: a view indexed before the filter existed still holds
        // vendored bundles. They never change, so they never appear in the git
        // diff, and once a base commit is recorded there is no full walk — so
        // without pruning the carried-forward set they would persist forever.
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");
        write_file(src.path(), "vendor.min.js", "var a=1,e=2;");

        // Index once with a filter that admits the bundle, standing in for the
        // pre-#209 behaviour.
        let lax = IndexFilter::new(&["vendor.min.js".to_string()]);
        let first = index_with(root.path(), src.path(), "r", "main", &lax)
            .await
            .unwrap();
        assert_eq!(first.files, 2, "the bundle is in the view to begin with");

        // Re-index with the default rules. The bundle is unchanged, so only the
        // carried-forward prune can remove it.
        let second = index_with(
            root.path(),
            src.path(),
            "r",
            "main",
            &IndexFilter::default(),
        )
        .await
        .unwrap();
        assert_eq!(second.deleted, 1, "the bundle is dropped from the view");
        assert_eq!(second.ignored.files, 1);

        let graph =
            SqliteGraphStore::open(view_db_path(&root.path().join("graphs"), "r", "main")).unwrap();
        assert!(
            graph.all_symbols().iter().all(|s| s.path == "a.rs"),
            "no vendored symbol may survive the re-index"
        );
    }

    #[tokio::test]
    async fn incremental_reindex_keeps_paths_the_filter_still_admits() {
        // The prune must not eat ordinary carried-forward files.
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");
        write_file(src.path(), "b.rs", "fn b() {}");
        index(root.path(), src.path(), "r", "main").await.unwrap();

        let second = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert_eq!(second.deleted, 0, "nothing legitimate is pruned");
        assert_eq!(second.ignored.files, 0);
    }

    #[tokio::test]
    async fn index_can_be_told_to_keep_a_vendored_path() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");
        write_file(src.path(), "keep.min.js", "function f(){}");

        let filter = IndexFilter::new(&["keep.min.js".to_string()]);
        let summary = index_with(root.path(), src.path(), "r", "main", &filter)
            .await
            .unwrap();
        assert_eq!(summary.files, 2, "the override re-admits it");
        assert_eq!(summary.ignored.files, 0);
    }

    #[tokio::test]
    async fn index_with_gc_sweeps_orphans_when_enabled() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");
        index(root.path(), src.path(), "r", "main").await.unwrap();

        // Reindex changed content with the post-index sweep on: the pre-edit
        // slice is orphaned and should be freed in the same call.
        write_file(src.path(), "a.rs", "fn a() { b(); }");
        let (_summary, swept) =
            index_with_gc(root.path(), src.path(), "r", "main", true, Duration::ZERO)
                .await
                .unwrap();
        let swept = swept.expect("gc runs when enabled");
        assert_eq!(swept.freed, 1, "orphaned slice swept during the index");

        // A follow-up gc finds nothing left to free.
        assert_eq!(gc(root.path(), Duration::ZERO).await.unwrap().freed, 0);
    }

    #[tokio::test]
    async fn index_with_gc_skips_sweep_when_disabled() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn a() {}");
        index(root.path(), src.path(), "r", "main").await.unwrap();

        write_file(src.path(), "a.rs", "fn a() { b(); }");
        let (_summary, swept) =
            index_with_gc(root.path(), src.path(), "r", "main", false, Duration::ZERO)
                .await
                .unwrap();
        assert!(swept.is_none(), "no gc when disabled");

        // The orphan survived: an explicit gc still has one to free.
        assert_eq!(gc(root.path(), Duration::ZERO).await.unwrap().freed, 1);
    }

    // ── watch: debounce core (gonzalo#100) ──────────────────────────────────

    #[test]
    fn debouncer_not_due_without_events() {
        let d = Debouncer::new(Duration::from_millis(500));
        assert!(!d.is_pending());
        assert!(!d.is_due(Instant::now()));
    }

    #[test]
    fn debouncer_due_only_after_quiet_window() {
        let t0 = Instant::now();
        let mut d = Debouncer::new(Duration::from_millis(500));
        d.on_event(t0);
        assert!(d.is_pending());
        // Still inside the window → not due.
        assert!(!d.is_due(t0 + Duration::from_millis(499)));
        // Window elapsed → due.
        assert!(d.is_due(t0 + Duration::from_millis(500)));
    }

    #[test]
    fn debouncer_coalesces_a_burst() {
        let t0 = Instant::now();
        let mut d = Debouncer::new(Duration::from_millis(500));
        d.on_event(t0);
        d.on_event(t0 + Duration::from_millis(200)); // second edit resets the clock
        // 500ms after the *first* event is not enough — the burst is still hot.
        assert!(!d.is_due(t0 + Duration::from_millis(500)));
        // 500ms after the *last* event → due, and only once for the whole burst.
        assert!(d.is_due(t0 + Duration::from_millis(700)));
    }

    #[test]
    fn debouncer_clear_resets_pending() {
        let t0 = Instant::now();
        let mut d = Debouncer::new(Duration::from_millis(500));
        d.on_event(t0);
        d.clear();
        assert!(!d.is_pending());
        assert!(!d.is_due(t0 + Duration::from_secs(10)));
    }

    #[tokio::test]
    async fn gc_retains_slice_still_referenced_by_another_view() {
        let root = TempDir::new().unwrap();

        // Two views index the *same* file content → one shared, deduped slice.
        let src_a = TempDir::new().unwrap();
        write_file(src_a.path(), "shared.rs", "fn shared() {}");
        index(root.path(), src_a.path(), "r", "a").await.unwrap();

        let src_b = TempDir::new().unwrap();
        write_file(src_b.path(), "shared.rs", "fn shared() {}");
        index(root.path(), src_b.path(), "r", "b").await.unwrap();

        // Remove the file from view A only; view B still references the slice.
        std::fs::remove_file(src_a.path().join("shared.rs")).unwrap();
        index(root.path(), src_a.path(), "r", "a").await.unwrap();

        let summary = gc(root.path(), Duration::ZERO).await.unwrap();
        assert_eq!(summary.scanned, 2, "one manifest record per view");
        assert_eq!(
            summary.freed, 0,
            "the shared slice is live via view B and must not be swept"
        );

        // View B still assembles and answers with the shared symbol.
        let store = FsStore::new(root.path());
        let manifest = Manifest::from_body(
            &store
                .get(&Manifest::key("r", "b"))
                .await
                .unwrap()
                .unwrap()
                .body,
        )
        .unwrap();
        let graph = gonzalo_graph::assemble(&manifest, &store).await.unwrap();
        assert_eq!(graph.definitions("shared")[0].path, "shared.rs");
    }

    // ---- files gonzalo cannot parse (#259) ---------------------------------

    #[tokio::test]
    async fn a_tree_of_unparseable_files_says_so_instead_of_reporting_nothing() {
        // Indexing a Jupyter-notebook-only repository reported every count at
        // zero with no hint why, so an empty view read as "the repo is empty"
        // rather than "gonzalo does not parse this" (#259).
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "analysis.ipynb", "{ \"cells\": [] }");
        write_file(src.path(), "notes.org", "* heading");

        let summary = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert_eq!(summary.files, 0, "nothing was indexable");
        assert_eq!(summary.unindexed.files, 2, "but two files were seen");
        assert_eq!(
            summary.unindexed.extensions(),
            vec![("ipynb".to_string(), 1), ("org".to_string(), 1)],
            "and the message can name them"
        );
    }

    #[tokio::test]
    async fn unparseable_files_are_counted_beside_ones_that_index() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn helper() {}");
        write_file(src.path(), "README.md", "# hi");

        let summary = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert_eq!(summary.files, 1);
        assert_eq!(summary.unindexed.files, 1);
        assert_eq!(summary.unindexed.extensions(), vec![("md".to_string(), 1)]);
    }

    #[tokio::test]
    async fn extensions_are_reported_most_common_first() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        for name in ["a.md", "b.md", "c.md", "d.txt"] {
            write_file(src.path(), name, "x");
        }
        let summary = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert_eq!(
            summary.unindexed.extensions(),
            vec![("md".to_string(), 3), ("txt".to_string(), 1)]
        );
    }

    #[tokio::test]
    async fn an_extensionless_file_is_counted_without_an_extension() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "LICENSE", "text");
        let summary = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert_eq!(summary.unindexed.files, 1);
        assert!(summary.unindexed.extensions().is_empty(), "nothing to name");
    }

    #[tokio::test]
    async fn a_genuinely_empty_tree_reports_no_unindexed_files() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        let summary = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert_eq!(summary.files, 0);
        assert_eq!(summary.unindexed.files, 0);
    }

    // ---- naming a skipped file (#270) --------------------------------------

    #[test]
    fn a_skip_message_names_the_file() {
        // Without the path there is no way to find the offender but bisecting
        // by directory, which is what landing #266 actually cost.
        let msg = skip_message("src/gpu/pipeline_state_cache.h", "parse worker died");
        assert!(msg.contains("src/gpu/pipeline_state_cache.h"), "{msg}");
        assert!(msg.contains("parse worker died"), "{msg}");
    }

    #[tokio::test]
    async fn a_clean_index_reports_no_skipped_paths() {
        let root = TempDir::new().unwrap();
        let src = TempDir::new().unwrap();
        write_file(src.path(), "a.rs", "fn helper() {}");

        let summary = index(root.path(), src.path(), "r", "main").await.unwrap();
        assert_eq!(summary.skipped, 0);
        assert!(summary.skipped_paths.is_empty());
    }

    #[test]
    fn the_skipped_listing_is_bounded_and_says_how_many_it_left_out() {
        let few: Vec<String> = vec!["a.rs".into(), "b.rs".into()];
        let listed = named_skips(&few);
        assert!(
            listed.contains("a.rs") && listed.contains("b.rs"),
            "{listed}"
        );
        assert!(!listed.contains("more"), "nothing was left out: {listed}");

        let many: Vec<String> = (0..12).map(|i| format!("f{i}.rs")).collect();
        let listed = named_skips(&many);
        assert!(listed.contains("f0.rs"), "{listed}");
        assert!(
            listed.contains("7 more"),
            "must say what it left out: {listed}"
        );
    }

    #[test]
    fn no_skips_lists_nothing() {
        assert_eq!(named_skips(&[]), String::new());
    }
}

#[cfg(test)]
mod tombstone_cli_tests {
    use super::*;
    use tempfile::TempDir;

    #[test]
    fn parse_duration_accepts_each_unit() {
        assert_eq!(parse_duration("30d"), Ok(Duration::from_secs(30 * 86_400)));
        assert_eq!(parse_duration("12h"), Ok(Duration::from_secs(12 * 3_600)));
        assert_eq!(parse_duration("90m"), Ok(Duration::from_secs(90 * 60)));
        assert_eq!(parse_duration("45s"), Ok(Duration::from_secs(45)));
        assert_eq!(parse_duration(" 7d "), Ok(Duration::from_secs(7 * 86_400)));
    }

    #[test]
    fn parse_duration_rejects_malformed_input() {
        for (raw, needle) in [
            ("", "missing unit"),
            ("30", "missing unit"),
            ("d", "expected a number"),
            ("-5d", "expected a number"),
            ("30x", "unknown unit"),
            ("30dd", "unknown unit"),
            ("1d2h", "unknown unit"),
            ("0d", "greater than zero"),
            ("99999999999999999999d", "too large"),
            ("18446744073709551615d", "too large"),
        ] {
            let err = parse_duration(raw).expect_err(raw);
            assert!(
                err.contains(needle),
                "{raw:?}: {err:?} should mention {needle:?}"
            );
        }
    }

    #[test]
    fn parse_horizon_keeps_the_operator_spelling() {
        let h = parse_horizon(" 30d").unwrap();
        assert_eq!(h.raw, "30d");
        assert_eq!(h.duration, Duration::from_secs(2_592_000));
    }

    #[test]
    fn parse_revision_reads_the_json_that_get_prints() {
        let rev = parse_revision(r#"{"counter":3,"hash":"abc"}"#).unwrap();
        assert_eq!(
            rev,
            Revision {
                counter: 3,
                hash: ContentHash("abc".into())
            }
        );
        let err = parse_revision("3").unwrap_err();
        assert!(err.contains("expected a revision as JSON"), "{err}");
    }

    #[test]
    fn open_store_rejects_a_zero_ancestor_cap() {
        let root = TempDir::new().unwrap();
        let err = open_store(root.path(), 0).err().expect("cap 0 rejected");
        assert!(
            format!("{err:#}").contains("ancestor cap must be at least 1"),
            "{err:#}"
        );
        assert!(open_store(root.path(), 1).is_ok());
    }

    #[tokio::test]
    async fn delete_conflicts_on_a_stale_revision_and_attributes_the_tombstone() {
        let root = TempDir::new().unwrap();
        let store = FsStore::new(root.path());
        let body = Body::Inline(b"x".to_vec());
        let record = Record {
            key: RecordKey::new("ns", "col", "k"),
            kind: RecordKind::Topic,
            revision: Revision::initial(body.bytes()),
            parent: None,
            body,
            meta: Meta {
                author: Identity::new("t"),
                origin_system: "test".into(),
                created: 0,
                updated: 0,
                labels: BTreeMap::new(),
            },
            links: Vec::new(),
            ancestors: Vec::new(),
            deleted_at: None,
            deleted_blob: None,
            deleted_kind: None,
        };
        let live = record.revision.clone();
        assert!(matches!(
            store.put(record, None).await.unwrap(),
            PutResult::Committed(_)
        ));
        let stale = Revision {
            counter: 99,
            hash: ContentHash("deadbeef".into()),
        };

        let outcome = delete(root.path(), 32, "ns", "col", "k", Some(stale))
            .await
            .unwrap();
        assert!(matches!(outcome, DeleteOutcome::Conflict { ref current } if *current == live));

        let outcome = delete(root.path(), 32, "ns", "col", "k", Some(live))
            .await
            .unwrap();
        assert!(matches!(outcome, DeleteOutcome::Deleted));
        assert_eq!(get(root.path(), "ns", "col", "k").await.unwrap(), None);

        // The seeded record was authored by "t"; the tombstone names the CLI.
        let tomb = store
            .get_raw(&RecordKey::new("ns", "col", "k"))
            .await
            .unwrap()
            .expect("tombstone stored");
        assert!(tomb.is_tombstone());
        assert_eq!(tomb.meta.author, Identity::new(CLI_AUTHOR));
    }

    #[tokio::test]
    async fn undelete_stamps_the_cli_author_not_the_deleter() {
        let root = TempDir::new().unwrap();
        let store = FsStore::new(root.path());
        let body = gonzalo_core::Manifest::new().to_body();
        let key = RecordKey::new("ns", "col", "k");
        let record = Record {
            key: key.clone(),
            kind: RecordKind::GraphManifest,
            revision: Revision::initial(body.bytes()),
            parent: None,
            body,
            meta: Meta {
                author: Identity::new("not-the-cli"),
                origin_system: "test".into(),
                created: 0,
                updated: 0,
                labels: BTreeMap::new(),
            },
            links: Vec::new(),
            ancestors: Vec::new(),
            deleted_at: None,
            deleted_blob: None,
            deleted_kind: None,
        };
        assert!(matches!(
            store.put(record, None).await.unwrap(),
            PutResult::Committed(_)
        ));
        let deleted = store
            .delete_as(&key, None, Some(Identity::new("someone-else")))
            .await
            .unwrap();
        assert!(matches!(deleted, gonzalo_core::DeleteResult::Deleted));

        undelete(
            root.path(),
            gonzalo_core::DEFAULT_ANCESTOR_CAP,
            "ns",
            "col",
            "k",
        )
        .await
        .unwrap();

        let restored = store.get(&key).await.unwrap().expect("restored");
        assert_eq!(restored.meta.author, Identity::new(CLI_AUTHOR));
    }

    #[tokio::test]
    async fn reset_stamps_the_cli_author() {
        let root = TempDir::new().unwrap();
        let store = FsStore::new(root.path());
        let body = Body::Inline(b"x".to_vec());
        let record = Record {
            key: RecordKey::new("ns", "col", "k"),
            kind: RecordKind::Topic,
            revision: Revision::initial(body.bytes()),
            parent: None,
            body,
            meta: Meta {
                author: Identity::new("not-the-cli"),
                origin_system: "test".into(),
                created: 0,
                updated: 0,
                labels: BTreeMap::new(),
            },
            links: Vec::new(),
            ancestors: Vec::new(),
            deleted_at: None,
            deleted_blob: None,
            deleted_kind: None,
        };
        assert!(matches!(
            store.put(record, None).await.unwrap(),
            PutResult::Committed(_)
        ));

        let report = reset(
            root.path(),
            gonzalo_core::DEFAULT_ANCESTOR_CAP,
            "ns".to_string(),
            None,
        )
        .await
        .unwrap();
        assert_eq!(report.deleted, vec![RecordKey::new("ns", "col", "k")]);

        let tomb = store
            .get_raw(&RecordKey::new("ns", "col", "k"))
            .await
            .unwrap()
            .expect("tombstone stored");
        assert!(tomb.is_tombstone());
        assert_eq!(tomb.meta.author, Identity::new(CLI_AUTHOR));
    }

    #[test]
    fn reset_exit_code_is_three_only_with_conflicts() {
        let clean = gonzalo_core::ResetReport {
            deleted: vec![RecordKey::new("ns", "col", "a")],
            conflicts: Vec::new(),
        };
        assert_eq!(reset_exit_code(&clean), 0);

        let conflicted = gonzalo_core::ResetReport {
            deleted: Vec::new(),
            conflicts: vec![RecordKey::new("ns", "col", "b")],
        };
        assert_eq!(reset_exit_code(&conflicted), EXIT_CONFLICT);
    }

    #[test]
    fn sync_exit_code_is_three_for_conflicts_or_non_convergence() {
        let clean = SyncSummary {
            copied_to_a: 1,
            copied_to_b: 2,
            fast_forwarded: 3,
            merged: 4,
            conflicts: Vec::new(),
            unconverged: Vec::new(),
        };
        assert_eq!(sync_exit_code(&clean), 0);

        let conflicted = SyncSummary {
            conflicts: vec![RecordKey::new("ns", "col", "a")],
            ..clean.clone()
        };
        assert_eq!(sync_exit_code(&conflicted), EXIT_CONFLICT);

        // A run that gave up before converging is not "in sync", even though it
        // reports no conflicts (#290).
        let unconverged = SyncSummary {
            unconverged: vec![RecordKey::new("ns", "col", "b")],
            ..clean
        };
        assert_eq!(sync_exit_code(&unconverged), EXIT_CONFLICT);
    }

    #[tokio::test]
    async fn ensure_slices_present_restores_a_blob_that_was_swept() {
        let dir = tempfile::tempdir().unwrap();
        let store = FsStore::new(dir.path());

        // Parse one real file so staging holds a Slice whose bytes we can restore.
        let src = dir.path().join("src");
        std::fs::create_dir_all(&src).unwrap();
        std::fs::write(src.join("lib.rs"), "pub fn alpha() {}\n").unwrap();

        let mut staging = GraphStaging::default();
        let desired = build_desired_full(&store, &mut staging, None, &src, &IndexFilter::default())
            .await
            .unwrap();

        let wanted: Vec<(String, ContentHash)> = desired
            .entries
            .iter()
            .map(|(p, h)| (p.clone(), h.clone()))
            .collect();
        assert!(!wanted.is_empty(), "the parsed file produced a slice");

        // A sweep takes it between the upload and the manifest commit.
        store.delete_blob(&wanted[0].1).await.unwrap();
        assert!(!store.has_blob(&wanted[0].1).await.unwrap());

        let restored = ensure_slices_present(&store, &wanted, &staging)
            .await
            .unwrap();
        assert_eq!(restored, 1);
        assert!(store.has_blob(&wanted[0].1).await.unwrap());
    }
}