basemind 0.29.1

//! Code-search branch helpers for the scanner (`code-search` feature).
//!
//! Mirrors `scanner_docs.rs` but for source files: after a file's L1/L2 filemap is written,
//! [`chunk_and_embed`] derives [`crate::chunk::CodeChunk`]s from the cached extraction + source
//! bytes (no re-parse), embeds each chunk's `searchable_text`, persists a content-addressed
//! `.chunk.msgpack` sidecar (the embedding cache), and hands back a [`PendingCodeBatch`] carrying
//! only lightweight metadata (path + blob hash + embedding dim) — never the LanceDB rows. The
//! single-threaded apply pass drains those descriptors and [`flush_code_batches`] re-reads each
//! file's sidecar to rebuild + write its rows one file at a time.
//!
//! The embed compute runs in the parallel per-file worker; the LanceDB write is deferred + serial
//! — exactly the document-tier pattern. Deferring the row-building (not just the write) keeps peak
//! embed-write RAM at one file's rows, independent of corpus size.

#![cfg(feature = "code-search")]

use anyhow::Result;

use crate::chunk::{ChunkOptions, CodeChunkBlob, chunk_file};
use crate::config::{CodeSearchConfig, Config};
use crate::embeddings::SharedEmbedder;
use crate::extract::{FileMapL1, FileMapL2, SCHEMA_VER};
use crate::lance::CodeRow;
use crate::scanner::EmbedMode;
use crate::scanner_filter::Filters;
use crate::search::bm25::{ChunkPosting, build_chunk_postings};
use crate::store::Store;

/// Per-file deferred-LanceDB-write descriptor for the code-search tier. Built inside the parallel
/// worker; consumed by [`flush_code_batches`] in the single-threaded apply pass.
///
/// Metadata only: it carries the content hash needed to re-read the persisted `.chunk.msgpack`
/// sidecar at flush time, NOT the `CodeRow`s or their embedding vectors. Holding the vectors here
/// (one `Vec<f32>` per chunk, across the whole corpus) was the memory leak this shape fixes.
///
/// The `bm25` postings are the exception — they are consumed by the worker itself (staged into
/// Fjall immediately after this descriptor is built) and cleared before the descriptor is
/// accumulated, so they never ride along the corpus-wide Vec either.
#[derive(Debug, Clone)]
pub(crate) struct PendingCodeBatch {
    /// Repository-relative path, forward-slash separated.
    pub rel_path: String,
    /// Content hash (hex) of the source file — the key under which the `.chunk.msgpack` sidecar is
    /// content-addressed. The flush re-reads that sidecar to rebuild rows on demand.
    pub blob_hash: String,
    /// Embedding vector length; `0` when embeddings were disabled (no rows to emit — BM25-only).
    pub embedding_dim: u16,
    /// BM25 keyword postings for this file's chunks — staged into the Fjall index by the worker
    /// (via [`crate::index::writer::IndexWriter::upsert_bm25_file`]), independent of embeddings, then
    /// cleared so the accumulated descriptor stays metadata-only.
    pub bm25: Vec<ChunkPosting>,
}

/// Feature + config gate: chunk this scan only when `[code_search] enabled`.
pub(crate) fn should_chunk(config: &Config) -> bool {
    config.code_search.enabled
}

/// True when `chunk_count` exceeds `[code_search] max_chunks_per_file`, warning once at the point
/// of decision. Over-cap files still get their `.chunk.msgpack` sidecar (outline and grep are
/// unaffected); what they lose is the per-chunk fan-out — keyword postings, vector rows, embeds.
/// Mirrors `scanner_docs::pending_from_doc`'s `max_chunks_per_document` handling.
fn over_chunk_cap(rel: &str, chunk_count: usize, cfg: &CodeSearchConfig) -> bool {
    let over = chunk_count > cfg.max_chunks_per_file;
    if over {
        tracing::warn!(
            rel,
            chunks = chunk_count,
            cap = cfg.max_chunks_per_file,
            "file exceeds max_chunks_per_file; caching chunk sidecar but skipping keyword postings and vector rows"
        );
    }
    over
}

/// BM25 postings for `chunks`, or none at all when the file is over [`over_chunk_cap`].
fn postings_for(rel: &str, chunks: &[crate::chunk::CodeChunk], cfg: &CodeSearchConfig) -> Vec<ChunkPosting> {
    if over_chunk_cap(rel, chunks.len(), cfg) {
        return Vec::new();
    }
    build_chunk_postings(chunks)
}

/// A rows-empty [`PendingCodeBatch`] (`embedding_dim: 0`) carrying only the BM25 postings for
/// `chunks` — the graceful-degradation result when embeddings were requested but the embedder was
/// unavailable. `None` for a chunkless file (nothing to index).
fn bm25_batch_from_chunks(
    rel: &str,
    blob_hash: &str,
    chunks: &[crate::chunk::CodeChunk],
    cfg: &CodeSearchConfig,
) -> Option<PendingCodeBatch> {
    if chunks.is_empty() {
        return None;
    }
    Some(PendingCodeBatch {
        rel_path: rel.to_string(),
        blob_hash: blob_hash.to_string(),
        embedding_dim: 0,
        bm25: postings_for(rel, chunks, cfg),
    })
}

/// True when a cached code-chunk blob can be reused as-is: identical content already chunked +
/// embedded under the current preset. Both the embedding **dimension** AND the **model/preset** must
/// match — `balanced` and `multilingual` share dim 768, so a dim-only check would falsely reuse
/// stale-model vectors across that switch — and the blob must actually carry one embedding per chunk.
/// Mirrors the documents tier's `scanner_docs::cached_doc_is_reusable`.
fn code_cache_is_reusable(blob: &CodeChunkBlob, dim: u16, preset: &str) -> bool {
    blob.embedding_dim == dim
        && blob.embedding_model == preset
        && !blob.chunks.is_empty()
        && blob.embeddings.len() == blob.chunks.len()
}

/// True when a cached chunk sidecar was produced under the current chunking settings and for the
/// same grammar. A blob that predates `config_digest` (empty) is accepted here and stamped by
/// [`stamp_legacy_digest`]; a blob whose first chunk names a different language was derived from
/// another grammar's outline (the file was remapped via `[languages]`) and must be rebuilt.
fn chunk_blob_is_current(blob: &CodeChunkBlob, digest: &str, lang: &str) -> bool {
    (blob.config_digest.is_empty() || blob.config_digest == digest)
        && blob.chunks.first().is_none_or(|c| c.lang == lang)
}

/// Adopt a legacy (digest-less) sidecar under the current settings by writing it back stamped, so a
/// later config change is detected against it. Costs one sidecar write per legacy file, no re-embed.
fn stamp_legacy_digest(
    store: &Store,
    rel: &str,
    hash_hex: &str,
    mut blob: CodeChunkBlob,
    digest: &str,
) -> CodeChunkBlob {
    if blob.config_digest.is_empty() {
        blob.config_digest = digest.to_string();
        if let Err(error) = store.write_chunks_hex(hash_hex, &blob) {
            tracing::warn!(rel, ?error, "stamp code-chunk sidecar with config digest failed");
        }
    }
    blob
}

/// Chunk + embed one source file. Reuses the `.chunk` sidecar when an identical-content blob
/// already exists (the embedding cache); otherwise chunks, embeds in bulk, and writes the blob.
/// Returns `Ok(None)` when the file yields no chunks. Never opens LanceDB — the write is deferred.
#[allow(clippy::too_many_arguments)]
pub(crate) fn chunk_and_embed(
    store: &Store,
    rel: &str,
    bytes: &[u8],
    l1: &FileMapL1,
    l2: Option<&FileMapL2>,
    hash_hex: &str,
    config: &Config,
    filters: &Filters,
    mode: EmbedMode,
) -> Result<Option<PendingCodeBatch>> {
    let cfg = &config.code_search;
    let embed = matches!(mode, EmbedMode::Inline) && cfg.embed && filters.code_embed_allowed(rel);
    let opts = ChunkOptions {
        max_characters: cfg.max_characters,
        overlap: cfg.overlap,
    };
    let digest = filters.code_digest.as_str();
    let cached = store
        .read_chunks_by_hex(hash_hex)
        .ok()
        .flatten()
        .filter(|blob| chunk_blob_is_current(blob, digest, &l1.language))
        .map(|blob| stamp_legacy_digest(store, rel, hash_hex, blob, digest));

    if !embed {
        let chunks = match cached {
            Some(blob) => blob.chunks,
            None => {
                let chunks = chunk_file(rel, hash_hex, l1, l2, bytes, opts);
                let blob = CodeChunkBlob {
                    config_digest: digest.to_string(),
                    schema_ver: SCHEMA_VER,
                    embedding_dim: 0,
                    embedding_model: String::new(),
                    chunks: chunks.clone(),
                    embeddings: Vec::new(),
                };
                if let Err(error) = store.write_chunks_hex(hash_hex, &blob) {
                    tracing::warn!(rel, ?error, "write code-chunk sidecar (chunk-only) failed");
                }
                chunks
            }
        };
        if chunks.is_empty() {
            return Ok(None);
        }
        let bm25 = postings_for(rel, &chunks, cfg);
        return Ok(Some(PendingCodeBatch {
            rel_path: rel.to_string(),
            blob_hash: hash_hex.to_string(),
            embedding_dim: 0,
            bm25,
        }));
    }

    let embedder = match SharedEmbedder::load_with_provider(
        &config.documents.embedding_preset,
        config
            .resources
            .effective_embed_threads(config.documents.embed_max_threads),
        config.resources.embed_batch_size,
        config.resources.onnx_provider,
    ) {
        Ok(embedder) => embedder,
        Err(error) => {
            tracing::warn!(
                rel,
                ?error,
                "load code-search embedder failed; indexing BM25 keyword lane only"
            );
            let chunks = match cached {
                Some(blob) if !blob.chunks.is_empty() => blob.chunks,
                _ => chunk_file(rel, hash_hex, l1, l2, bytes, opts),
            };
            return Ok(bm25_batch_from_chunks(rel, hash_hex, &chunks, cfg));
        }
    };
    let dim = embedder.dim();

    if let Some(blob) = &cached
        && code_cache_is_reusable(blob, dim, &config.documents.embedding_preset)
    {
        let over_cap = over_chunk_cap(rel, blob.chunks.len(), cfg);
        return Ok(Some(PendingCodeBatch {
            rel_path: rel.to_string(),
            blob_hash: hash_hex.to_string(),
            embedding_dim: if over_cap { 0 } else { dim },
            bm25: if over_cap {
                Vec::new()
            } else {
                build_chunk_postings(&blob.chunks)
            },
        }));
    }

    let chunks = chunk_file(rel, hash_hex, l1, l2, bytes, opts);
    if chunks.is_empty() {
        let blob = CodeChunkBlob {
            config_digest: digest.to_string(),
            schema_ver: SCHEMA_VER,
            embedding_dim: 0,
            embedding_model: String::new(),
            chunks: Vec::new(),
            embeddings: Vec::new(),
        };
        if let Err(error) = store.write_chunks_hex(hash_hex, &blob) {
            tracing::warn!(rel, ?error, "write empty code-chunk sidecar failed");
        }
        return Ok(None);
    }
    if over_chunk_cap(rel, chunks.len(), cfg) {
        let blob = CodeChunkBlob {
            config_digest: digest.to_string(),
            schema_ver: SCHEMA_VER,
            embedding_dim: 0,
            embedding_model: String::new(),
            chunks,
            embeddings: Vec::new(),
        };
        if let Err(error) = store.write_chunks_hex(hash_hex, &blob) {
            tracing::warn!(rel, ?error, "write over-cap code-chunk sidecar failed");
        }
        return Ok(Some(PendingCodeBatch {
            rel_path: rel.to_string(),
            blob_hash: hash_hex.to_string(),
            embedding_dim: 0,
            bm25: Vec::new(),
        }));
    }
    let texts: Vec<&str> = chunks.iter().map(|c| c.searchable_text.as_str()).collect();
    crate::backpressure::FootprintGate::new(config.resources.max_footprint_mb).admit();
    let embeddings = match embedder.embed_batch(&texts) {
        Ok(embeddings) if embeddings.len() == chunks.len() => embeddings,
        Ok(embeddings) => {
            tracing::warn!(
                rel,
                got = embeddings.len(),
                want = chunks.len(),
                "embedder returned wrong vector count; indexing BM25 keyword lane only"
            );
            return Ok(bm25_batch_from_chunks(rel, hash_hex, &chunks, cfg));
        }
        Err(error) => {
            tracing::warn!(rel, ?error, "embed code chunks failed; indexing BM25 keyword lane only");
            return Ok(bm25_batch_from_chunks(rel, hash_hex, &chunks, cfg));
        }
    };
    let bm25 = build_chunk_postings(&chunks);
    let blob = CodeChunkBlob {
        config_digest: digest.to_string(),
        schema_ver: SCHEMA_VER,
        embedding_dim: dim,
        embedding_model: config.documents.embedding_preset.clone(),
        chunks,
        embeddings,
    };
    if let Err(error) = store.write_chunks_hex(hash_hex, &blob) {
        tracing::warn!(rel, ?error, "write code-chunk sidecar failed; embedding cache skipped");
    }
    Ok(Some(PendingCodeBatch {
        rel_path: rel.to_string(),
        blob_hash: hash_hex.to_string(),
        embedding_dim: dim,
        bm25,
    }))
}

/// Assemble the LanceDB rows for a file's chunks + embeddings (parallel arrays), **consuming** both
/// so the chunk text and embedding vectors move into the rows instead of being cloned. Called at
/// flush time with the owned sidecar blob; `scope` is stamped directly (no second pass needed).
fn build_rows_owned(
    rel: &str,
    scope: &str,
    chunks: Vec<crate::chunk::CodeChunk>,
    embeddings: Vec<Vec<f32>>,
) -> Vec<CodeRow> {
    chunks
        .into_iter()
        .zip(embeddings)
        .map(|(c, emb)| CodeRow {
            scope: scope.to_string(),
            path: rel.to_string(),
            chunk_id: c.chunk_id,
            symbol: c.symbol.unwrap_or_default(),
            kind: c.kind.unwrap_or_default(),
            lang: c.lang,
            line_start: c.line_start,
            line_end: c.line_end,
            byte_start: c.byte_start,
            byte_end: c.byte_end,
            text: c.text,
            embedding: emb,
        })
        .collect()
}

/// Delete the `code_chunks` rows of files that no longer exist (or are no longer allowed). Runs in
/// the serial apply pass after the batch flush, so it reuses an already-open LanceStore when the
/// same scan wrote chunks. Best-effort: never opens (i.e. never *creates*) the vector store when it
/// does not already exist, and logs — never propagates — a delete failure.
pub(crate) fn delete_stale_code_chunks(store: &mut Store, config: &Config, scope: &str, stale: &[String]) {
    if !should_chunk(config) {
        return;
    }
    delete_code_rows(store, config, scope, stale);
}

/// Delete the `code_chunks` rows of `paths`, whether they left the scan set or stopped being
/// embed-eligible. Shared tail of the stale purge and the embed-policy purge; same best-effort,
/// never-create contract as [`delete_stale_code_chunks`].
///
/// Returns `false` only when rows that exist could not be deleted (unknown preset, store open or
/// delete error); "nothing to delete" and "no vector store" are both `true`.
pub(crate) fn delete_code_rows(store: &mut Store, config: &Config, scope: &str, paths: &[String]) -> bool {
    let stale = paths;
    if stale.is_empty() {
        return true;
    }
    if store.lance.is_none() && !store.lance_dir_exists() {
        return true;
    }
    let model = &config.documents.embedding_preset;
    let dim = match SharedEmbedder::load(model, 0, config.resources.embed_batch_size) {
        Ok(embedder) => embedder.dim(),
        Err(error) => {
            tracing::warn!(?error, preset = %model, "code-chunk stale purge: unknown embedding preset; skipping");
            return false;
        }
    };
    let lance = match store.lance_or_open(dim, model) {
        Ok(lance) => lance.clone(),
        Err(error) => {
            tracing::warn!(?error, "code-chunk stale purge: open LanceStore failed; skipping");
            return false;
        }
    };
    if let Err(error) = lance.delete_code_chunks_for_paths(scope, stale) {
        tracing::warn!(
            paths = stale.len(),
            ?error,
            "code-chunk purge failed; search_code may return a removed or no-longer-embedded path"
        );
        return false;
    }
    true
}

/// Drop the derived rows of every indexed source file that is no longer eligible under the current
/// config: the `code_chunks` vectors of files that are not `embed`-eligible (`embed = false`, or
/// outside `embed_include` / inside `embed_exclude`), and — when `[code_search] enabled = false` —
/// the BM25 keyword postings too, so a disabled tier leaves nothing queryable. Run when the embed
/// policy changed since the last scan; see [`crate::scanner_policy`].
///
/// Returns whether the purge fully applied; the caller records the policy only then, so a failed
/// purge is retried by the next scan instead of being marked done.
pub(crate) fn purge_unembedded_code(store: &mut Store, config: &Config, filters: &Filters, scope: &str) -> bool {
    let cfg = &config.code_search;
    let paths: Vec<String> = store
        .index
        .files
        .keys()
        .map(|k| k.to_str_lossy().into_owned())
        .filter(|p| !(cfg.enabled && cfg.embed && filters.code_embed_allowed(p)))
        .collect();
    if paths.is_empty() {
        return true;
    }
    let rows = delete_code_rows(store, config, scope, &paths);
    let postings = cfg.enabled || drop_keyword_postings(store, &paths);
    rows && postings
}

fn drop_keyword_postings(store: &Store, paths: &[String]) -> bool {
    let Some(db) = store.index_db.as_ref() else {
        return true;
    };
    let mut writer = db.writer();
    for path in paths {
        if let Err(error) = writer.remove_bm25_file(&crate::path::RelPath::from(path.as_str())) {
            tracing::warn!(
                ?error,
                "drop keyword postings failed; code search may return stale chunks"
            );
            return false;
        }
    }
    if let Err(error) = writer.commit() {
        tracing::warn!(?error, "commit keyword posting removal failed");
        return false;
    }
    if let Err(error) = db.recompute_bm25_stats() {
        tracing::warn!(?error, "recompute bm25 stats failed; keyword search may be stale");
    }
    true
}

/// Stream every pending code batch into the `code_chunks` LanceDB table, one file at a time. Opens
/// the store lazily — if no batch carries vectors (embeddings off) no LanceDB connection is made.
///
/// For each embedded file this re-reads the file's already-persisted `.chunk.msgpack` sidecar,
/// rebuilds its rows, writes them, and drops them before the next file — so peak embed-write RAM is
/// one file's rows plus one Arrow batch, independent of corpus size. (The sidecar is guaranteed on
/// disk here: every `embedding_dim > 0` batch either wrote it in `chunk_and_embed` or reused an
/// existing one, both before this post-barrier lane runs.)
///
/// Errors are logged and skipped per file so one malformed embedding never aborts the scan. Returns
/// the number of files for which rows were written.
pub(crate) fn flush_code_batches(
    store: &mut Store,
    scope: &str,
    batches: Vec<PendingCodeBatch>,
    embedding_model: &str,
) -> usize {
    let Some(dim) = batches.iter().find(|b| b.embedding_dim > 0).map(|b| b.embedding_dim) else {
        return 0;
    };
    match SharedEmbedder::load(
        embedding_model,
        0,
        crate::config::ResourcesConfig::default().embed_batch_size,
    ) {
        Ok(embedder) if embedder.dim() != dim => {
            tracing::error!(
                preset = %embedding_model,
                expected = embedder.dim(),
                actual = dim,
                "preset/runtime dim mismatch — refusing to write code_chunks batch"
            );
            return 0;
        }
        Ok(_) => {}
        Err(error) => {
            tracing::error!(?error, preset = %embedding_model, "unknown embedding preset — refusing to write code_chunks batch");
            return 0;
        }
    }

    let lance = match store.lance_or_open(dim, embedding_model) {
        Ok(s) => s.clone(),
        Err(error) => {
            tracing::error!(?error, "open LanceStore for code_chunks batch failed");
            return 0;
        }
    };

    let mut inserted = 0usize;
    for batch in batches {
        if batch.embedding_dim == 0 {
            continue;
        }
        let blob = match store.read_chunks_by_hex(&batch.blob_hash) {
            Ok(Some(blob)) if blob.embedding_dim == dim && blob.embeddings.len() == blob.chunks.len() => blob,
            Ok(Some(_)) => {
                continue;
            }
            Ok(None) => {
                tracing::warn!(rel = %batch.rel_path, "chunk sidecar missing at flush; skipping vector rows");
                continue;
            }
            Err(error) => {
                tracing::warn!(rel = %batch.rel_path, ?error, "re-read chunk sidecar failed; skipping vector rows");
                continue;
            }
        };
        if blob.chunks.is_empty() {
            continue;
        }
        let rows = build_rows_owned(&batch.rel_path, scope, blob.chunks, blob.embeddings);
        match lance.replace_code_chunks(scope, &batch.rel_path, rows) {
            Ok(()) => inserted += 1,
            Err(error) => {
                tracing::warn!(rel = %batch.rel_path, ?error, "lance replace_code_chunks failed; code search may be incomplete");
            }
        }
    }
    inserted
}

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

    fn chunk(chunk_id: &str, searchable_text: &str) -> CodeChunk {
        CodeChunk {
            chunk_id: chunk_id.to_string(),
            path: "src/lib.rs".to_string(),
            lang: "rust".to_string(),
            kind: None,
            symbol: None,
            signature: None,
            doc: None,
            byte_start: 0,
            byte_end: 0,
            line_start: 1,
            line_end: 1,
            text: searchable_text.to_string(),
            searchable_text: searchable_text.to_string(),
        }
    }

    #[test]
    fn bm25_batch_from_chunks_is_rows_empty_with_postings() {
        let cfg = CodeSearchConfig::default();
        let batch = bm25_batch_from_chunks("src/lib.rs", "deadbeef", &[chunk("h:0", "alpha beta alpha")], &cfg)
            .expect("non-empty chunks must yield a batch");
        assert_eq!(batch.rel_path, "src/lib.rs");
        assert_eq!(
            batch.blob_hash, "deadbeef",
            "carries the content hash for the flush re-read"
        );
        assert_eq!(batch.embedding_dim, 0, "no embeddings on the degraded path");
        assert_eq!(batch.bm25.len(), 1, "one posting per chunk");
        assert_eq!(batch.bm25[0].doclen, 3, "three tokens incl. the repeat");
    }

    #[test]
    fn bm25_batch_from_chunks_is_none_for_chunkless_file() {
        assert!(bm25_batch_from_chunks("src/empty.rs", "deadbeef", &[], &CodeSearchConfig::default()).is_none());
    }

    /// Structural regression guard for the streaming-flush memory fix: the accumulated per-file
    /// descriptor must stay metadata-only. This exhaustive struct literal names EXACTLY the metadata
    /// fields — re-introducing a `rows: Vec<CodeRow>` (or any embedding payload) field would break
    /// this compile, catching a regression back to the corpus-wide accumulation that leaked GBs.
    #[test]
    fn pending_code_batch_is_metadata_only() {
        let batch = PendingCodeBatch {
            rel_path: "src/lib.rs".to_string(),
            blob_hash: "deadbeef".to_string(),
            embedding_dim: 768,
            bm25: Vec::new(),
        };
        assert_eq!(batch.embedding_dim, 768);
    }

    fn embedded_blob(model: &str, dim: u16) -> CodeChunkBlob {
        CodeChunkBlob {
            config_digest: String::new(),
            schema_ver: 0,
            embedding_dim: dim,
            embedding_model: model.to_string(),
            chunks: vec![chunk("h:0", "alpha")],
            embeddings: vec![vec![0.0_f32; dim as usize]],
        }
    }

    #[test]
    fn code_cache_reuse_requires_matching_dim_and_model() {
        let blob = embedded_blob("balanced", 768);
        assert!(code_cache_is_reusable(&blob, 768, "balanced"));
        assert!(
            !code_cache_is_reusable(&blob, 768, "multilingual"),
            "same-dim different-model must miss and force a re-embed"
        );
        assert!(!code_cache_is_reusable(&blob, 384, "balanced"));
        assert!(!code_cache_is_reusable(&embedded_blob("", 768), 768, "balanced"));
    }

    #[test]
    fn code_cache_reuse_rejects_chunkless_or_unembedded_blob() {
        let mut blob = embedded_blob("balanced", 768);
        blob.chunks.clear();
        blob.embeddings.clear();
        assert!(!code_cache_is_reusable(&blob, 768, "balanced"));
        let chunk_only = CodeChunkBlob {
            config_digest: String::new(),
            schema_ver: 0,
            embedding_dim: 768,
            embedding_model: "balanced".to_string(),
            chunks: vec![chunk("h:0", "alpha")],
            embeddings: Vec::new(),
        };
        assert!(!code_cache_is_reusable(&chunk_only, 768, "balanced"));
    }

    #[test]
    fn postings_are_dropped_for_a_file_over_max_chunks_per_file() {
        let cfg = CodeSearchConfig {
            max_chunks_per_file: 2,
            ..CodeSearchConfig::default()
        };
        let under: Vec<CodeChunk> = (0..2).map(|i| chunk(&format!("h:{i}"), "alpha beta")).collect();
        let over: Vec<CodeChunk> = (0..3).map(|i| chunk(&format!("h:{i}"), "alpha beta")).collect();

        assert_eq!(
            postings_for("src/lib.rs", &under, &cfg).len(),
            2,
            "a file at the cap keeps every posting"
        );
        assert!(
            postings_for("src/generated.rs", &over, &cfg).is_empty(),
            "one chunk over the cap must stage no BM25 postings at all — this is the bound on the \
             per-file staging residual the byte budget cannot see, because a whole file is staged \
             in one call"
        );
    }

    #[test]
    fn over_cap_degraded_batch_carries_neither_postings_nor_vector_rows() {
        let cfg = CodeSearchConfig {
            max_chunks_per_file: 1,
            ..CodeSearchConfig::default()
        };
        let chunks: Vec<CodeChunk> = (0..4).map(|i| chunk(&format!("h:{i}"), "alpha beta")).collect();
        let batch = bm25_batch_from_chunks("src/generated.rs", "deadbeef", &chunks, &cfg)
            .expect("an over-cap file is still tracked; only its per-chunk fan-out is dropped");
        assert!(batch.bm25.is_empty());
        assert_eq!(batch.embedding_dim, 0, "no vector rows either");
    }

    #[test]
    fn default_max_chunks_per_file_mirrors_the_document_tier() {
        assert_eq!(
            CodeSearchConfig::default().max_chunks_per_file,
            crate::config::DocumentsConfig::default().max_chunks_per_document
        );
    }

    fn seeded_code_lance(store: &mut Store, paths: &[&str]) -> crate::lance::LanceStore {
        let lance = store.lance_or_open(768, "balanced").expect("open lance").clone();
        for path in paths {
            let row = CodeRow {
                scope: "repo:s".to_string(),
                path: (*path).to_string(),
                chunk_id: format!("{path}:0"),
                symbol: "sym".to_string(),
                kind: "function".to_string(),
                lang: "rust".to_string(),
                line_start: 1,
                line_end: 1,
                byte_start: 0,
                byte_end: 4,
                text: "body".to_string(),
                embedding: vec![0.25; 768],
            };
            lance.replace_code_chunks("repo:s", path, vec![row]).expect("seed row");
            store.upsert(
                *path,
                crate::store::FileEntry {
                    hash_hex: "h".to_string(),
                    language: "rust".to_string(),
                    size_bytes: 1,
                    mtime: 0,
                    extract_epoch: crate::extract::EXTRACT_EPOCH,
                },
            );
        }
        lance
    }

    fn code_rows(lance: &crate::lance::LanceStore) -> Vec<String> {
        let mut paths: Vec<String> = lance
            .search_code_chunks("repo:s", vec![0.25; 768], 50)
            .expect("search")
            .into_iter()
            .map(|hit| hit.path)
            .collect();
        paths.sort();
        paths
    }

    #[test]
    fn embed_policy_purge_deletes_vectors_of_newly_ineligible_files_only() {
        crate::store::init_isolated_cache();
        let tmp = tempfile::tempdir().unwrap();
        let mut store = Store::open(tmp.path(), crate::store::VIEW_WORKING).unwrap();
        let lance = seeded_code_lance(&mut store, &["src/a.rs", "src/generated/b.rs", "lib/c.rs"]);
        assert_eq!(code_rows(&lance).len(), 3);

        let mut config = crate::config::default_for_root(std::path::Path::new("."));
        config.code_search.embed = true;
        config.code_search.embed_exclude = vec!["**/generated/**".to_string()];
        let filters = Filters::build(&config, Vec::new()).unwrap();
        purge_unembedded_code(&mut store, &config, &filters, "repo:s");
        assert_eq!(code_rows(&lance), vec!["lib/c.rs".to_string(), "src/a.rs".to_string()]);

        config.code_search.embed_include = vec!["src/**".to_string()];
        let filters = Filters::build(&config, Vec::new()).unwrap();
        purge_unembedded_code(&mut store, &config, &filters, "repo:s");
        assert_eq!(
            code_rows(&lance),
            vec!["src/a.rs".to_string()],
            "embed_include narrows retroactively"
        );

        config.code_search.embed = false;
        let filters = Filters::build(&config, Vec::new()).unwrap();
        purge_unembedded_code(&mut store, &config, &filters, "repo:s");
        assert!(code_rows(&lance).is_empty(), "embed = false removes every vector row");
        assert!(store.lookup("src/a.rs").is_some(), "files stay indexed");
    }

    #[test]
    fn a_purge_that_cannot_delete_reports_failure_and_leaves_the_rows() {
        crate::store::init_isolated_cache();
        let tmp = tempfile::tempdir().unwrap();
        let mut store = Store::open(tmp.path(), crate::store::VIEW_WORKING).unwrap();
        let lance = seeded_code_lance(&mut store, &["src/a.rs"]);

        let mut config = crate::config::default_for_root(std::path::Path::new("."));
        config.code_search.embed = false;
        let filters = Filters::build(&config, Vec::new()).unwrap();

        config.documents.embedding_preset = "no-such-preset".to_string();
        assert!(
            !purge_unembedded_code(&mut store, &config, &filters, "repo:s"),
            "an unknown preset means the delete never ran: not applied"
        );
        assert_eq!(code_rows(&lance), vec!["src/a.rs".to_string()], "nothing was deleted");

        config.documents.embedding_preset = "balanced".to_string();
        assert!(purge_unembedded_code(&mut store, &config, &filters, "repo:s"));
        assert!(code_rows(&lance).is_empty(), "the retry removes the row");
    }

    #[test]
    fn disabling_code_search_also_drops_keyword_postings() {
        crate::store::init_isolated_cache();
        let tmp = tempfile::tempdir().unwrap();
        let mut store = Store::open(tmp.path(), crate::store::VIEW_WORKING).unwrap();
        let rel = crate::path::RelPath::from("src/a.rs");
        let db = store.index_db.as_ref().expect("index db open");
        let mut writer = db.writer();
        writer
            .upsert_bm25_file(&rel, &build_chunk_postings(&[chunk("h:0", "alpha beta")]))
            .unwrap();
        writer.commit().unwrap();
        db.recompute_bm25_stats().unwrap();
        assert_eq!(db.bm25_stats().map(|(n, _)| n), Some(1));
        store.upsert(
            "src/a.rs",
            crate::store::FileEntry {
                hash_hex: "h".to_string(),
                language: "rust".to_string(),
                size_bytes: 1,
                mtime: 0,
                extract_epoch: crate::extract::EXTRACT_EPOCH,
            },
        );

        let mut config = crate::config::default_for_root(std::path::Path::new("."));
        config.code_search.enabled = false;
        let filters = Filters::build(&config, Vec::new()).unwrap();
        purge_unembedded_code(&mut store, &config, &filters, "repo:s");
        let db = store.index_db.as_ref().unwrap();
        assert_eq!(
            db.bm25_stats().map(|(n, _)| n).unwrap_or(0),
            0,
            "keyword postings are gone"
        );
    }

    #[test]
    fn chunk_blob_currency_tracks_digest_and_grammar() {
        let mut blob = embedded_blob("balanced", 768);
        blob.chunks[0].lang = "rust".to_string();
        assert!(
            chunk_blob_is_current(&blob, "d1", "rust"),
            "legacy blob (no digest) is adoptable"
        );
        blob.config_digest = "d1".to_string();
        assert!(chunk_blob_is_current(&blob, "d1", "rust"));
        assert!(!chunk_blob_is_current(&blob, "d2", "rust"), "different chunk settings");
        assert!(
            !chunk_blob_is_current(&blob, "d1", "python"),
            "built for another grammar"
        );
    }
}