lean_ctx/core/

index_orchestrator.rs

use std::collections::HashMap;
use std::path::Path;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{SystemTime, UNIX_EPOCH};

use serde::Serialize;

use crate::core::bm25_index::BM25Index;
use crate::core::graph_index::{self, ProjectIndex};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum State {
    Idle,
    Building,
    Ready,
    Failed,
}

#[derive(Debug, Clone)]
struct Component {
    state: State,
    started_ms: Option<u64>,
    finished_ms: Option<u64>,
    duration_ms: Option<u64>,
    last_error: Option<String>,
    /// Human-readable outcome detail surfaced to operators (e.g. doc count +
    /// persisted size, or the "not persisted: too large …" remedy). Independent
    /// of `last_error` so a *successful* build can still carry a warning note.
    note: Option<String>,
}

impl Component {
    fn new() -> Self {
        Self {
            state: State::Idle,
            started_ms: None,
            finished_ms: None,
            duration_ms: None,
            last_error: None,
            note: None,
        }
    }
}

#[derive(Debug)]
struct ProjectBuild {
    worker_running: bool,
    /// Cleared on worker claim; set when the graph phase of this run finishes.
    /// Used by CLI wait loops so they do not exit during the claim→start gap
    /// or the graph→BM25 handoff (search pre-warm keeps `worker_running`).
    graph_run_done: bool,
    /// Cleared on worker claim; set when the BM25 phase of this run finishes.
    bm25_run_done: bool,
    /// Set the first time a heavy-index tool lazily pre-warms this root (#152).
    /// Prevents re-triggering a full rebuild on every subsequent dispatch — the
    /// tools' own `load_or_build` paths handle staleness from then on.
    warm_triggered: bool,
    graph: Component,
    bm25: Component,
    /// Dense embedding index (semantic search). Built after BM25 as Phase 3.
    /// Tracked separately so the orchestrator does not block on a missing ONNX
    /// model — the status lets users see why semantic stays cold (#249).
    semantic: Component,
}

impl ProjectBuild {
    fn new() -> Self {
        Self {
            worker_running: false,
            graph_run_done: true,
            bm25_run_done: true,
            warm_triggered: false,
            graph: Component::new(),
            bm25: Component::new(),
            semantic: Component::new(),
        }
    }
}

// Lock ordering (see rust/LOCK_ORDERING.md):
//   L1 = REGISTRY outer Mutex  (the HashMap guard)
//   L2 = per-project Arc<Mutex<ProjectBuild>>  (inner guard)
//
// Invariant: L1 must NEVER be held while locking L2.
// `entry_for()` enforces this by cloning the Arc and dropping L1 before
// the caller acquires L2.
static REGISTRY: OnceLock<Mutex<HashMap<String, Arc<Mutex<ProjectBuild>>>>> = OnceLock::new();

fn registry() -> &'static Mutex<HashMap<String, Arc<Mutex<ProjectBuild>>>> {
    REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
}

fn entry_for(project_root: &str) -> Arc<Mutex<ProjectBuild>> {
    let mut map = registry()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    map.entry(project_root.to_string())
        .or_insert_with(|| Arc::new(Mutex::new(ProjectBuild::new())))
        .clone()
}

fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}

/// Per-repo lock name for serializing the BM25 build across processes, mirroring
/// the `graph-idx-<hash>` lock (see LOCK_ORDERING.md). The distinct `bm25-` vs
/// `graph-` prefix keeps the graph and BM25 builds from serializing against each
/// other while still preventing N processes from rebuilding either in parallel.
fn bm25_index_lock_name(root: &Path) -> String {
    format!(
        "bm25-idx-{}",
        &crate::core::index_namespace::namespace_hash(root)[..8]
    )
}

/// Per-repo lock name for the semantic (embedding) build. Two concurrent builds
/// would each checkpoint `embeddings.bin` and overwrite the other's progress.
fn semantic_index_lock_name(root: &Path) -> String {
    format!(
        "semantic-idx-{}",
        &crate::core::index_namespace::namespace_hash(root)[..8]
    )
}

/// A CPU-only embed of a large corpus can run for hours; a crashed holder is
/// reclaimed at once through its PID, so this only bounds a frozen process.
const SEMANTIC_LOCK_STALE_AFTER: std::time::Duration = std::time::Duration::from_hours(24);

fn start_component(c: &mut Component) {
    c.state = State::Building;
    c.started_ms = Some(now_ms());
    c.finished_ms = None;
    c.duration_ms = None;
    c.last_error = None;
    c.note = None;
}

fn finish_ok(c: &mut Component) {
    c.state = State::Ready;
    let end = now_ms();
    c.finished_ms = Some(end);
    c.duration_ms = c.started_ms.map(|s| end.saturating_sub(s));
}

fn finish_err(c: &mut Component, e: String) {
    c.state = State::Failed;
    let end = now_ms();
    c.finished_ms = Some(end);
    c.duration_ms = c.started_ms.map(|s| end.saturating_sub(s));
    c.last_error = Some(e);
}

/// The index warmth a tool benefits from. Drives lazy, demand-driven warming
/// (issue #152) so the server no longer scans the whole project eagerly on every
/// `initialize` — a session that only uses `ctx_read`/`ctx_shell`/`ctx_tree`
/// pays zero indexing cost.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WarmNeed {
    /// No prebuilt index needed.
    None,
    /// Only the resident line-search (trigram) index — cheap, used by `ctx_search`.
    Search,
    /// Full project indices (graph + BM25; this also warms the search index).
    Heavy,
}

/// Classify a tool by the index warmth it benefits from. Unknown tools default
/// to [`WarmNeed::None`]; a heavy tool mis-classified as `None` still works — it
/// just builds its index synchronously on first use instead of being pre-warmed.
#[must_use]
pub fn warm_need_for_tool(tool: &str) -> WarmNeed {
    match tool {
        "ctx_search" => WarmNeed::Search,
        // Tools that build/consume the graph, call-graph, BM25 or artifact index.
        "ctx_graph" | "ctx_callgraph" | "ctx_artifacts" | "ctx_provider" | "ctx_compose" => {
            WarmNeed::Heavy
        }
        _ => WarmNeed::None,
    }
}

/// Lazily warm the indices a tool needs, deduped per root. Never blocks (all
/// work is spawned in the background) and is safe to call on every dispatch.
///
/// Returns `true` only when this call is the *first* heavy pre-warm for `root`
/// in this process — the caller can use that signal to warm secondary roots once
/// without re-reading session state on every dispatch.
pub fn ensure_warm_for_tool(project_root: &str, tool: &str) -> bool {
    if project_root.is_empty() {
        return false;
    }
    match warm_need_for_tool(tool) {
        WarmNeed::None => false,
        WarmNeed::Search => {
            // The search index has its own TTL + background-rebuild dedup, so it
            // is safe (and cheap) to nudge on every `ctx_search`.
            crate::core::search_index::ensure_background(project_root, true, false);
            false
        }
        WarmNeed::Heavy => {
            let entry = entry_for(project_root);
            let first_warm = {
                let mut s = entry
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                if s.warm_triggered {
                    false
                } else {
                    s.warm_triggered = true;
                    true
                }
            };
            if first_warm {
                ensure_all_background(project_root);
            }
            first_warm
        }
    }
}

/// Stack size for background index workers. Large enough that deep ASTs and
/// graph traversals cannot overflow it (the #378 SIGABRT class). The AST walks
/// are iterative now too, so this is defense-in-depth.
const INDEXER_STACK_BYTES: usize = 16 * 1024 * 1024;

/// Fire-and-forget: ask a running daemon to own the index build for `root`
/// (#460, shared-indexer-daemon / thin clients).
///
/// The daemon is the single long-lived, machine-wide indexer. Once it holds the
/// per-repo `graph-idx`/`bm25-idx` build locks, every session load-shares its
/// on-disk result instead of each running a full scan during a cold boot wave —
/// turning N simultaneous index passes into ~one. Strictly additive and
/// best-effort: it runs on its own thread (no ambient runtime to nest into), is
/// skipped when we *are* the daemon or when no daemon is reachable, and never
/// blocks the caller. The local build started right after remains the fallback,
/// so indexing always works with no daemon present.
fn nudge_daemon_index(project_root: &str) {
    // The daemon must never delegate the build to itself.
    if crate::daemon::is_foreground_daemon() {
        return;
    }
    let root = project_root.to_string();
    let _ = std::thread::Builder::new()
        .name("leanctx-index-nudge".to_string())
        .spawn(move || {
            if !crate::daemon::is_daemon_running() {
                return;
            }
            let Ok(rt) = tokio::runtime::Runtime::new() else {
                return;
            };
            let body = serde_json::json!({ "root": root }).to_string();
            rt.block_on(async {
                let _ = crate::daemon_client::try_daemon_request("POST", "/v1/index/ensure", &body)
                    .await;
            });
        });
}

/// Try to claim the per-root build slot. Returns `false` when a worker is
/// already running for `project_root` (the claim is released by the build
/// worker itself when it finishes).
fn try_claim_worker(project_root: &str) -> bool {
    let state = entry_for(project_root);
    let mut s = state
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    if s.worker_running {
        false
    } else {
        s.worker_running = true;
        s.graph_run_done = false;
        s.bm25_run_done = false;
        true
    }
}

pub fn ensure_all_background(project_root: &str) {
    if !try_claim_worker(project_root) {
        return;
    }

    // #460: hand the build to the daemon (the single machine-wide indexer) when
    // one is running and we aren't it. Deduped naturally — we only reach here
    // when this process is actually about to start a build. Purely additive: the
    // local build below still runs and load-shares via the per-repo locks.
    //
    // #735 exception: with a per-run CLI filter overlay active, the daemon
    // (which builds with *its* config, not the overlay) could overwrite the
    // filtered result — the one-off run keeps the build local instead.
    if !crate::core::index_filter::cli_overlay_active() {
        nudge_daemon_index(project_root);
    }

    let state = entry_for(project_root);
    let root = project_root.to_string();
    let indexer = move || run_build_worker(&root);

    // Indexing parses large ASTs and traverses graphs; give the worker a
    // generous stack as defense-in-depth against deep-recursion overflow (the
    // #378 SIGABRT class) and a name so it is identifiable in crash dumps.
    let spawned = std::thread::Builder::new()
        .name("leanctx-index".to_string())
        .stack_size(INDEXER_STACK_BYTES)
        .spawn(indexer);
    if spawned.is_err() {
        // The OS refused a new thread (rare). Clear the in-flight flag so a
        // later trigger retries instead of assuming a build runs forever.
        let mut s = state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        s.worker_running = false;
    }
}

/// Actual per-root build work. Heavy phases run in a fixed, memory-bounded order:
/// graph → reclaim → BM25 → reclaim → resident search pre-warm. Blocking — callers
/// own the threading. Requires the caller to have claimed the worker slot via
/// [`try_claim_worker`]; releases it on exit.
fn run_build_worker(root: &str) {
    // The build's project is the root it indexes: its files are the
    // project's own, not unknown stored text (pz21 §1.4), whatever the
    // process directory of the detached worker is.
    crate::core::policy::runtime::REQUEST_PROJECT.sync_scope(
        std::cell::RefCell::new(Some(std::path::PathBuf::from(root))),
        || run_build_worker_in_project(root),
    );
}

fn run_build_worker_in_project(root: &str) {
    // Released on every exit, a panic outside the per-phase `catch_unwind`
    // included: `index build` waits for the slot without a deadline, so a slot
    // left claimed would hang the CLI instead of reporting the failure.
    let _slot = WorkerSlot(root.to_string());

    // #790: start memory guardian for daemon-spawned builds. Previously only
    // the CLI path (index_cmd.rs) started the guardian, leaving background
    // builds (MCP `ensure_all_background`, watcher, graph_coordinator) to run
    // without RSS pressure checks — the guardian's abort/pressure flags never
    // fired. `start_guard` is idempotent: a second call within the same
    // process is a no-op.
    start_index_memory_guard("build_worker");

    // Graph, BM25, and the resident search index each retain substantial live
    // state. Running them concurrently defeats cache eviction because active
    // builders still own their allocations. Keep phases strictly ordered and
    // reclaim between them (#918); intra-phase Rayon parallelism remains intact.

    let graph_state = entry_for(root);
    let graph_root = root.to_string();
    let build_graph = move || {
        {
            let mut s = graph_state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            start_component(&mut s.graph);
        }
        // Indeterminate while the scan walks the tree; the scan reports its
        // file counts once it knows them. Guard clears on exit.
        let guard = crate::core::index_progress::ProgressGuard::new(
            graph_root.clone(),
            crate::core::index_progress::IndexComponent::Graph,
        );
        guard.report(0, 0);
        let graph_result = std::panic::catch_unwind(|| {
            let (idx, _cache) = graph_index::scan_with_content_cache(&graph_root);
            // scan_with_content_cache persists the freshly built ProjectIndex
            // before returning it; mirroring the same unchanged index again here
            // only repeats the property-graph mirror.
            crate::core::code_health::persist::refresh_if_stale(&graph_root, &idx);
        });
        if graph_result.is_ok() {
            // A resident call graph predates the rebuilt index: the next
            // request compares the persisted one with the files again.
            crate::core::call_graph::CallGraph::invalidate(&graph_root);
            // Evidence-graded call/implements edges on top of the fresh
            // structure; a no-op unless a semantic backend can contribute.
            let _ = std::panic::catch_unwind(|| {
                crate::core::graph_enricher::refresh_semantic_edges_in_background(&graph_root);
            });
        }
        drop(guard);
        crate::core::telemetry_features::record(
            crate::core::telemetry_features::Feature::GraphIndex,
            graph_result.is_ok(),
        );
        {
            let mut s = graph_state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            if let Ok(()) = graph_result {
                finish_ok(&mut s.graph);
            } else {
                finish_err(&mut s.graph, "graph index build panicked".to_string());
            }
            s.graph_run_done = true;
        }
    };

    let bm25_state = entry_for(root);
    let bm25_root = root.to_string();
    let build_bm25 = move || {
        {
            let mut s = bm25_state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            start_component(&mut s.bm25);
        }
        let _progress = crate::core::index_progress::ProgressGuard::new(
            bm25_root.clone(),
            crate::core::index_progress::IndexComponent::Bm25,
        );
        let bm = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let root_pb = Path::new(&bm25_root);
            let lock_name = bm25_index_lock_name(root_pb);
            let _lock = crate::core::startup_guard::try_acquire_lock(
                &lock_name,
                std::time::Duration::from_millis(800),
                std::time::Duration::from_mins(3),
            );
            if _lock.is_none() {
                tracing::info!(
                    "[bm25: another process is building {bm25_root} — loading the shared index]"
                );
                let idx = BM25Index::load(root_pb).unwrap_or_default();
                return (idx.doc_count, None);
            }
            let idx = BM25Index::load_or_build(root_pb);
            let outcome = idx.save(root_pb);
            (idx.doc_count, Some(outcome))
        }));
        // A build another process owns is not counted here; that one counts it.
        if !matches!(bm, Ok((_, None))) {
            crate::core::telemetry_features::record(
                crate::core::telemetry_features::Feature::Bm25Index,
                bm.is_ok(),
            );
        }
        {
            let mut s = bm25_state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            if let Ok((doc_count, save_res)) = bm {
                finish_ok(&mut s.bm25);
                s.bm25.note = Some(match save_res {
                    Some(outcome) => bm25_build_note(doc_count, &outcome),
                    None => format!(
                        "loaded shared BM25 index ({doc_count} chunks) — build in progress in another process"
                    ),
                });
            } else {
                finish_err(&mut s.bm25, "bm25 build panicked".to_string());
            }
            s.bm25_run_done = true;
        }
    };

    run_memory_bounded_phases(
        build_graph,
        build_bm25,
        || crate::core::search_index::ensure_background(root, true, false),
        || {
            crate::core::content_cache::clear();
            crate::core::memory_guard::force_purge();
        },
    );

    // Post-build memory reclamation: the parallel build allocates large
    // transient structures (file contents, token vectors, inverted postings).
    // After merge+save these are freed, but jemalloc retains dirty pages in
    // its arena — inflating RSS far past the logical working set. Purging
    // immediately returns those pages to the OS, keeping the serve process
    // close to the configured max_ram_percent target.
    crate::core::content_cache::trim_oldest_percent(75);
    crate::core::memory_guard::force_purge();
}

/// A claimed per-root build slot (see [`try_claim_worker`]); dropping it
/// releases the slot.
struct WorkerSlot(String);

impl Drop for WorkerSlot {
    fn drop(&mut self) {
        let state = entry_for(&self.0);
        let mut s = state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        s.worker_running = false;
    }
}

/// Execute index phases without overlapping their retained working sets.
/// Kept as a small pure coordinator so ordering remains regression-testable.
fn run_memory_bounded_phases<G, B, S, R>(graph: G, bm25: B, search: S, mut reclaim: R)
where
    G: FnOnce(),
    B: FnOnce(),
    S: FnOnce(),
    R: FnMut(),
{
    graph();
    reclaim();
    bm25();
    reclaim();
    search();
}

/// Build only the semantic (dense embedding) index from the existing BM25 index.
/// The BM25 index must already exist on disk — this function loads it and runs
/// `embedding_index::build_or_update`. Updates the in-memory semantic component
/// state on completion.
pub fn build_semantic(project_root: &str) {
    let state = entry_for(project_root);
    let root = Path::new(project_root);

    let Some(_lock) = crate::core::startup_guard::try_acquire_lock(
        &semantic_index_lock_name(root),
        std::time::Duration::ZERO,
        SEMANTIC_LOCK_STALE_AFTER,
    ) else {
        let mut s = state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        s.semantic.note = Some(
            "another lean-ctx process is already building the semantic index for this project; \
             it saves progress as it goes — check `lean-ctx index status` later"
                .to_string(),
        );
        return;
    };

    {
        let mut s = state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        start_component(&mut s.semantic);
    }

    // Guard clears progress on every exit (incl. early returns / panics).
    let _progress = crate::core::index_progress::ProgressGuard::new(
        project_root.to_string(),
        crate::core::index_progress::IndexComponent::Semantic,
    );

    let bm25_idx = try_load_bm25_index(project_root);
    match bm25_idx.as_ref() {
        Some(idx) if idx.doc_count > 0 => {
            let outcome = crate::core::embedding_index::build_or_update(root, idx);
            match outcome {
                crate::core::embedding_index::EmbeddingBuildOutcome::Ready => {
                    crate::core::telemetry_features::record(
                        crate::core::telemetry_features::Feature::SemanticIndex,
                        true,
                    );
                }
                crate::core::embedding_index::EmbeddingBuildOutcome::Failed => {
                    crate::core::telemetry_features::record(
                        crate::core::telemetry_features::Feature::SemanticIndex,
                        false,
                    );
                }
                _ => {}
            }
            let mut s = state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            match outcome {
                crate::core::embedding_index::EmbeddingBuildOutcome::Ready => {
                    finish_ok(&mut s.semantic);
                }
                crate::core::embedding_index::EmbeddingBuildOutcome::Skipped => {
                    finish_ok(&mut s.semantic);
                    s.semantic.note = Some(
                        "embeddings disabled by feature flag or config (search.dense_enabled / memory_profile)"
                            .to_string(),
                    );
                }
                crate::core::embedding_index::EmbeddingBuildOutcome::ModelNotAvailable(
                    ref reason,
                ) => {
                    // Not a hard failure — semantic is optional. Surface the reason
                    // so users know why dense search is cold (#249).
                    s.semantic.state = State::Idle;
                    s.semantic.note = Some(format!("embedding model not available: {reason}"));
                }
                crate::core::embedding_index::EmbeddingBuildOutcome::Failed => {
                    finish_err(
                        &mut s.semantic,
                        "embedding build failed (see logs)".to_string(),
                    );
                }
            }
        }
        _ => {
            let mut s = state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            // No BM25 docs → nothing to embed. Leave semantic Idle with a note.
            s.semantic.state = State::Idle;
            s.semantic.note =
                Some("BM25 index is empty or unavailable — nothing to embed".to_string());
        }
    }
}

/// Build the semantic index on a background thread, after the graph + BM25 run
/// when one is in flight or the BM25 index is missing. For MCP calls: a CPU
/// embed of a large corpus outlives any tool-call timeout, and starting the
/// embed before BM25 is on disk would find nothing to embed.
pub fn build_semantic_background(project_root: &str) {
    let root = project_root.to_string();
    let spawned = std::thread::Builder::new()
        .name("leanctx-semantic".to_string())
        .stack_size(INDEXER_STACK_BYTES)
        .spawn(move || {
            start_index_memory_guard("semantic_build");
            if !disk_status_for_bm25(&root).exists {
                ensure_all_background(&root);
            }
            while progress_view(&root).graph_bm25_active() {
                std::thread::sleep(std::time::Duration::from_millis(250));
            }
            build_semantic(&root);
        });
    if let Err(e) = spawned {
        tracing::warn!("[index_orchestrator: could not spawn semantic worker: {e}]");
    }
}

/// Start the memory guardian for an index build. Idempotent per process: the
/// first `start_guard` call owns the callback, so a daemon whose first request
/// is `/v1/index/ensure` gets only this one. It must reach the same
/// process-global caches as the server callback, or the daemon cannot shed its
/// resident search indexes under pressure (#2075).
fn start_index_memory_guard(label: &'static str) {
    crate::core::memory_guard::start_guard(std::sync::Arc::new(move |level| {
        tracing::warn!("[{label}] memory pressure: {level:?} — background tasks will throttle");
        let reclaimed = crate::core::eviction_orchestrator::on_memory_pressure(level);
        crate::core::memory_guard::force_purge();
        reclaimed
    }));
}

/// Ensure background indexing for all extra roots (in addition to the primary).
/// Each extra root that is not a subdirectory of `primary_root` gets its own
/// graph + BM25 index. Capped at `MAX_EXTRA_ROOT_BUILDS` to prevent runaway.
const MAX_EXTRA_ROOT_BUILDS: usize = 8;

pub fn ensure_extra_roots_background(primary_root: &str, extra_roots: &[String]) {
    let primary = Path::new(primary_root);
    let mut queue: Vec<String> = Vec::new();
    for root in extra_roots {
        if queue.len() >= MAX_EXTRA_ROOT_BUILDS {
            break;
        }
        let rp = Path::new(root);
        if !rp.is_dir() {
            continue;
        }
        // Skip if extra_root is inside primary (already indexed by the primary scan)
        if rp.starts_with(primary) {
            continue;
        }
        // Skip if primary is inside this extra_root (avoid double-indexing the parent)
        if primary.starts_with(rp) {
            continue;
        }
        queue.push(root.clone());
    }
    if queue.is_empty() {
        return;
    }

    // #685: build extra roots *sequentially* on one supervisor thread. The old
    // per-root `ensure_all_background` fan-out ran up to MAX_EXTRA_ROOT_BUILDS
    // graph+BM25 pairs concurrently (each with rayon pools inside) — on the
    // reported multi-root setup (1M+ files across 6+ roots) the combined
    // transient build state outran the guardian to 75 GB RSS. One root at a
    // time keeps peak memory bounded to a single build while still warming
    // every root; the guardian check between roots stops the queue as soon as
    // pressure appears.
    let spawned = std::thread::Builder::new()
        .name("leanctx-extra-roots".to_string())
        .stack_size(INDEXER_STACK_BYTES)
        .spawn(move || {
            for root in queue {
                if crate::core::memory_guard::is_under_pressure()
                    || crate::core::memory_guard::abort_requested()
                {
                    tracing::warn!(
                        "[index_orchestrator: skipping remaining extra-root builds under memory pressure]"
                    );
                    break;
                }
                if !try_claim_worker(&root) {
                    continue; // already building elsewhere
                }
                nudge_daemon_index(&root);
                run_build_worker(&root);
            }
        });
    if let Err(e) = spawned {
        tracing::warn!("[index_orchestrator: could not spawn extra-roots worker: {e}]");
    }
}

/// Build a human-readable outcome note for a finished BM25 build, including the
/// indexed chunk count and whether the index was persisted to disk. A
/// "too large" refusal carries the exact remedy so the operator (or agent) is
/// never left guessing why search/ranking stays cold (issue #249).
fn bm25_build_note(
    doc_count: usize,
    save: &std::io::Result<crate::core::bm25_index::SaveOutcome>,
) -> String {
    use crate::core::bm25_index::SaveOutcome;
    match save {
        Ok(SaveOutcome::Persisted { compressed_bytes }) => format!(
            "indexed {doc_count} chunks, {:.1} MB persisted",
            *compressed_bytes as f64 / 1_048_576.0
        ),
        Ok(SaveOutcome::SkippedTooLarge {
            compressed_bytes,
            limit_bytes,
        }) => format!(
            "indexed {doc_count} chunks but NOT persisted to disk: compressed {:.1} MB exceeds the {:.0} MB cap. \
             Raise it via LEAN_CTX_BM25_MAX_CACHE_MB (or bm25_max_cache_mb in config) or add extra_ignore_patterns, \
             then run `lean-ctx reindex`. Until then the index is rebuilt from scratch on every cold start.",
            *compressed_bytes as f64 / 1_048_576.0,
            *limit_bytes as f64 / 1_048_576.0
        ),
        Err(e) => format!("indexed {doc_count} chunks but persisting failed: {e}"),
    }
}

/// Lightweight, allocation-frugal snapshot of the BM25 component for the
/// in-call composer/search messaging. Avoids the heavier [`disk_status`] walk.
#[derive(Debug, Clone)]
pub struct Bm25Summary {
    pub state: &'static str,
    /// While building: elapsed so far. Otherwise: last build duration.
    pub elapsed_ms: Option<u64>,
    pub note: Option<String>,
    pub last_error: Option<String>,
}

/// Lightweight snapshot of the semantic (dense embedding) component.
#[derive(Debug, Clone)]
pub struct SemanticSummary {
    pub state: &'static str,
    pub elapsed_ms: Option<u64>,
    pub note: Option<String>,
    pub last_error: Option<String>,
}

/// Shared helper: compute (state_str, elapsed_ms) for a component.
/// Deduplicates the elapsed-while-building logic and state-to-string mapping
/// between bm25_summary and semantic_summary.
fn component_elapsed_and_state(c: &Component) -> (&'static str, Option<u64>) {
    let elapsed_ms = if matches!(c.state, State::Building) {
        c.started_ms.map(|start| now_ms().saturating_sub(start))
    } else {
        c.duration_ms
    };
    let state = match c.state {
        State::Idle => "idle",
        State::Building => "building",
        State::Ready => "ready",
        State::Failed => "failed",
    };
    (state, elapsed_ms)
}

pub fn semantic_summary(project_root: &str) -> SemanticSummary {
    let entry = entry_for(project_root);
    let s = entry
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let c = &s.semantic;
    let (state, elapsed_ms) = component_elapsed_and_state(c);
    SemanticSummary {
        state,
        elapsed_ms,
        note: c.note.clone(),
        last_error: c.last_error.clone(),
    }
}

pub fn bm25_summary(project_root: &str) -> Bm25Summary {
    let entry = entry_for(project_root);
    let s = entry
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let c = &s.bm25;
    let (state, elapsed_ms) = component_elapsed_and_state(c);
    Bm25Summary {
        state,
        elapsed_ms,
        note: c.note.clone(),
        last_error: c.last_error.clone(),
    }
}

pub fn try_load_graph_index(project_root: &str) -> Option<ProjectIndex> {
    // Resident cache: avoids re-materializing the index from the property graph
    // (SQLite query) on every graph-touching query. Returns an in-memory clone.
    crate::core::graph_cache::get_cached(project_root).map(|arc| (*arc).clone())
}

pub fn try_load_bm25_index(project_root: &str) -> Option<BM25Index> {
    BM25Index::load(Path::new(project_root))
}

/// Typed progress snapshot for one index component.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct SlotProgress {
    pub building: bool,
    pub done: u64,
    pub total: u64,
}

impl SlotProgress {
    pub fn is_determinate(&self) -> bool {
        self.building && self.total > 0
    }
}

/// Root-scoped view of index build progress for CLI wait loops.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProgressView {
    pub worker_running: bool,
    /// Graph phase of the current worker run has finished (ok or err).
    pub graph_run_done: bool,
    /// BM25 phase of the current worker run has finished (ok or err).
    pub bm25_run_done: bool,
    pub graph: SlotProgress,
    pub bm25: SlotProgress,
    pub semantic: SlotProgress,
}

impl ProgressView {
    /// Graph + BM25 pipeline still in flight for this root.
    ///
    /// True from worker claim through BM25 completion — covers the claim→start
    /// race and the graph→BM25 handoff without waiting for search pre-warm.
    pub fn graph_bm25_active(&self) -> bool {
        self.worker_running && !(self.graph_run_done && self.bm25_run_done)
    }

    pub fn semantic_active(&self) -> bool {
        self.semantic.building
    }
}

/// Snapshot progress for a project root (typed; no JSON scrape).
pub fn progress_view(project_root: &str) -> ProgressView {
    use crate::core::index_progress::{self, IndexComponent};
    let (worker_running, graph_run_done, bm25_run_done, graph_b, bm25_b, sem_b) = {
        let state = entry_for(project_root);
        let s = state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        (
            s.worker_running,
            s.graph_run_done,
            s.bm25_run_done,
            matches!(s.graph.state, State::Building),
            matches!(s.bm25.state, State::Building),
            matches!(s.semantic.state, State::Building),
        )
    };
    let g = index_progress::get(project_root, IndexComponent::Graph);
    let b = index_progress::get(project_root, IndexComponent::Bm25);
    let sem = index_progress::get(project_root, IndexComponent::Semantic);
    ProgressView {
        worker_running,
        graph_run_done,
        bm25_run_done,
        graph: SlotProgress {
            building: graph_b,
            done: g.done,
            total: g.total,
        },
        bm25: SlotProgress {
            building: bm25_b,
            done: b.done,
            total: b.total,
        },
        semantic: SlotProgress {
            building: sem_b,
            done: sem.done,
            total: sem.total,
        },
    }
}

/// Returns true if any project is currently building its indices.
pub fn is_building() -> bool {
    let map = registry()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    map.values().any(|entry| {
        let st = entry
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        matches!(st.bm25.state, State::Building)
            || matches!(st.graph.state, State::Building)
            || matches!(st.semantic.state, State::Building)
    })
}

#[derive(Debug, Serialize)]
struct ComponentStatus<'a> {
    state: &'a str,
    started_ms: Option<u64>,
    finished_ms: Option<u64>,
    duration_ms: Option<u64>,
    last_error: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    note: Option<&'a str>,
    /// Units completed while `building` (files/chunks). Omitted when idle/ready.
    #[serde(skip_serializing_if = "Option::is_none")]
    progress_done: Option<u64>,
    /// Total units; `Some(0)` while building means indeterminate.
    #[serde(skip_serializing_if = "Option::is_none")]
    progress_total: Option<u64>,
}

fn component_status(
    c: &Component,
    progress: crate::core::index_progress::ProgressSnapshot,
) -> ComponentStatus<'_> {
    let (progress_done, progress_total) = if matches!(c.state, State::Building) {
        (Some(progress.done), Some(progress.total))
    } else {
        (None, None)
    };
    ComponentStatus {
        state: match c.state {
            State::Idle => "idle",
            State::Building => "building",
            State::Ready => "ready",
            State::Failed => "failed",
        },
        started_ms: c.started_ms,
        finished_ms: c.finished_ms,
        duration_ms: c.duration_ms,
        last_error: c.last_error.as_deref(),
        note: c.note.as_deref(),
        progress_done,
        progress_total,
    }
}

#[derive(Debug, Serialize)]
struct StatusResponse<'a> {
    project_root: &'a str,
    graph_index: ComponentStatus<'a>,
    bm25_index: ComponentStatus<'a>,
    /// Dense embedding index built after BM25.  "idle" means the ONNX model
    /// has not been downloaded yet or the embeddings feature was not compiled
    /// in; "ready" means embeddings are persisted and search will use them.
    semantic_index: ComponentStatus<'a>,
    disk: DiskStatusAll,
    /// Active corpus filter summary (#735). Omitted for the unfiltered
    /// default, keeping default output byte-identical.
    #[serde(skip_serializing_if = "Option::is_none")]
    index_filters: Option<String>,
}

#[derive(Debug, Serialize, Default)]
pub struct DiskStatus {
    pub exists: bool,
    pub size_bytes: Option<u64>,
    pub file_count: Option<u64>,
    pub modified_at: Option<String>,
    /// Semantic index only: total chunks when the persisted index is an
    /// interrupted build's checkpoint (`file_count` then counts the embedded
    /// ones). Absent for a complete index.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub partial_of: Option<u64>,
}

#[derive(Debug, Serialize, Default)]
pub struct DiskStatusAll {
    pub graph_index: DiskStatus,
    pub bm25_index: DiskStatus,
    pub code_graph: DiskStatus,
    /// On-disk embedding index (`embeddings.bin`).  Present when dense search
    /// has been built at least once; absent when the model is not downloaded
    /// yet or embeddings are disabled by config.
    pub semantic_index: DiskStatus,
}

/// On-disk state of the graph index: its catalogued files, timed by
/// `graph.meta.json`. Reads row counts only — never materializes the index
/// (#2179: that checked every catalogued file through the serving gate and
/// left `doctor` running for minutes on a large project).
pub fn disk_status_for_graph(project_root: &str) -> DiskStatus {
    // #696 C4: the property graph is the sole store. The logical graph-index
    // view (file count) is timed by `graph.meta.json`, which the mirror stamps
    // on every build; `disk_status_for_code_graph` reports the raw SQLite store
    // (nodes, graph.db size) as a distinct facet. The meta file's own size is
    // not the index size, so none is reported here (#1914).
    let Some(dir) = graph_index::ProjectIndex::index_dir(project_root) else {
        return DiskStatus::default();
    };
    let meta_file = dir.join("graph.meta.json");
    if !meta_file.exists() {
        return DiskStatus::default();
    }
    let meta = std::fs::metadata(&meta_file).ok();
    // An empty catalog is an unbuilt index, as `ProjectIndex::load` reads it.
    let file_count = crate::core::property_graph::stored_counts(project_root)
        .map(|counts| counts.files)
        .filter(|files| *files > 0);
    DiskStatus {
        exists: true,
        size_bytes: None,
        file_count,
        modified_at: meta.and_then(|m| m.modified().ok()).map(format_time),
        partial_of: None,
    }
}

/// On-disk state of the BM25 index: file size and the chunk count from its
/// header, without loading it.
pub fn disk_status_for_bm25(project_root: &str) -> DiskStatus {
    let root = Path::new(project_root);
    let path = BM25Index::index_file_path(root);
    if !path.exists() {
        return DiskStatus::default();
    }
    let meta = std::fs::metadata(&path).ok();
    DiskStatus {
        exists: true,
        size_bytes: meta.as_ref().map(std::fs::Metadata::len),
        file_count: BM25Index::persisted_chunk_count(root),
        modified_at: meta.and_then(|m| m.modified().ok()).map(format_time),
        partial_of: None,
    }
}

fn disk_status_for_code_graph(project_root: &str) -> DiskStatus {
    let dir = crate::core::property_graph::graph_dir(project_root);
    let db_path = dir.join("graph.db");
    if !db_path.exists() {
        return DiskStatus::default();
    }
    let meta = std::fs::metadata(&db_path).ok();
    // A status read opens the store without migrating it and gives up on a
    // graph another process holds busy (#2179).
    let node_count =
        crate::core::property_graph::stored_counts(project_root).map(|counts| counts.nodes);
    DiskStatus {
        exists: true,
        size_bytes: meta.as_ref().map(std::fs::Metadata::len),
        file_count: node_count,
        modified_at: meta.and_then(|m| m.modified().ok()).map(format_time),
        partial_of: None,
    }
}

fn format_time(t: SystemTime) -> String {
    let secs = t.duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
    let dt = chrono::DateTime::from_timestamp(secs as i64, 0);
    dt.map_or_else(
        || format!("{secs}"),
        |d| d.format("%Y-%m-%d %H:%M:%S UTC").to_string(),
    )
}

pub fn disk_status_for_semantic(project_root: &str) -> DiskStatus {
    let root = Path::new(project_root);
    let dir = crate::core::index_namespace::vectors_dir(root);
    let bin_path = dir.join("embeddings.bin");
    if !bin_path.exists() {
        return DiskStatus::default();
    }
    let meta = std::fs::metadata(&bin_path).ok();
    let partial = crate::core::embedding_index::partial_progress(root);
    DiskStatus {
        exists: true,
        size_bytes: meta.as_ref().map(std::fs::Metadata::len),
        file_count: partial.map(|(done, _)| done),
        modified_at: meta.and_then(|m| m.modified().ok()).map(format_time),
        partial_of: partial.map(|(_, total)| total),
    }
}

pub fn disk_status(project_root: &str) -> DiskStatusAll {
    DiskStatusAll {
        graph_index: disk_status_for_graph(project_root),
        bm25_index: disk_status_for_bm25(project_root),
        code_graph: disk_status_for_code_graph(project_root),
        semantic_index: disk_status_for_semantic(project_root),
    }
}

pub fn status_json(project_root: &str) -> String {
    // Compute disk status first — may do SQLite I/O and must NOT hold L2
    // (per-project Mutex) while doing so, or the background index worker
    // cannot call finish_ok / set worker_running = false (#deadlock).
    let disk = disk_status(project_root);
    let state = entry_for(project_root);
    let s = state
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let graph_p = crate::core::index_progress::get(
        project_root,
        crate::core::index_progress::IndexComponent::Graph,
    );
    let bm25_p = crate::core::index_progress::get(
        project_root,
        crate::core::index_progress::IndexComponent::Bm25,
    );
    let semantic_p = crate::core::index_progress::get(
        project_root,
        crate::core::index_progress::IndexComponent::Semantic,
    );
    let res = StatusResponse {
        project_root,
        graph_index: component_status(&s.graph, graph_p),
        bm25_index: component_status(&s.bm25, bm25_p),
        semantic_index: component_status(&s.semantic, semantic_p),
        disk,
        index_filters: crate::core::index_filter::IndexFileFilter::effective().summary(),
    };
    serde_json::to_string(&res).unwrap_or_else(|_| "{}".to_string())
}

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

    #[test]
    fn progress_view_graph_bm25_active_logic() {
        let idle = SlotProgress::default();
        let mut v = ProgressView {
            worker_running: true,
            graph_run_done: false,
            bm25_run_done: false,
            graph: idle,
            bm25: idle,
            semantic: idle,
        };
        assert!(v.graph_bm25_active(), "claimed, phases not done");
        v.graph_run_done = true;
        assert!(v.graph_bm25_active(), "graph done, bm25 pending");
        v.bm25_run_done = true;
        assert!(
            !v.graph_bm25_active(),
            "both phases done — search pre-warm must not keep wait open"
        );
        v.worker_running = false;
        assert!(!v.graph_bm25_active());
        // Idle project (never claimed this process)
        v.graph_run_done = true;
        v.bm25_run_done = true;
        assert!(!v.graph_bm25_active());
    }

    /// `index build` waits for the slot without a deadline; a worker that
    /// panics must still release it, or the CLI would wait forever.
    #[test]
    fn worker_slot_is_released_when_the_worker_panics() {
        let root = "/tmp/leanctx-worker-slot-panic-test";
        assert!(try_claim_worker(root));
        assert!(progress_view(root).graph_bm25_active());
        let result = std::panic::catch_unwind(|| {
            let _slot = WorkerSlot(root.to_string());
            panic!("simulated failure outside the phase catch_unwind");
        });
        assert!(result.is_err());
        let view = progress_view(root);
        assert!(!view.worker_running, "slot released by the guard");
        assert!(!view.graph_bm25_active(), "the CLI wait loop ends");
        assert!(try_claim_worker(root), "a later build can claim it again");
        drop(WorkerSlot(root.to_string()));
    }

    #[test]
    fn status_json_is_valid_json() {
        let s = status_json("/tmp");
        let _: serde_json::Value = serde_json::from_str(&s).unwrap();
    }

    #[test]
    fn background_build_phases_are_serialized_with_reclamation() {
        let events = std::cell::RefCell::new(Vec::new());
        run_memory_bounded_phases(
            || events.borrow_mut().push("graph"),
            || events.borrow_mut().push("bm25"),
            || events.borrow_mut().push("search"),
            || events.borrow_mut().push("reclaim"),
        );
        assert_eq!(
            events.into_inner(),
            ["graph", "reclaim", "bm25", "reclaim", "search"]
        );
    }

    #[test]
    fn warm_need_classifies_tools() {
        // Lightweight tools must never trigger a project scan (#152).
        for light in [
            "ctx_read",
            "ctx_shell",
            "ctx_tree",
            "ctx_knowledge",
            "unknown_tool",
        ] {
            assert_eq!(warm_need_for_tool(light), WarmNeed::None, "{light}");
        }
        // ctx_search only needs the cheap trigram index.
        assert_eq!(warm_need_for_tool("ctx_search"), WarmNeed::Search);
        for heavy in [
            "ctx_graph",
            "ctx_callgraph",
            "ctx_artifacts",
            "ctx_provider",
            "ctx_compose",
        ] {
            assert_eq!(warm_need_for_tool(heavy), WarmNeed::Heavy, "{heavy}");
        }
    }

    #[test]
    fn ensure_warm_lightweight_and_search_never_signal_first_warm() {
        assert!(!ensure_warm_for_tool("", "ctx_graph"));
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().to_string_lossy().to_string();
        assert!(!ensure_warm_for_tool(&root, "ctx_read"));
        assert!(!ensure_warm_for_tool(&root, "ctx_search"));
    }

    #[test]
    fn ensure_warm_heavy_is_once_per_root() {
        // The first heavy pre-warm signals `true` (so the caller warms extra
        // roots once); every subsequent call is a no-op `false`, preventing a
        // rebuild-on-every-dispatch storm.
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().to_string_lossy().to_string();
        assert!(
            ensure_warm_for_tool(&root, "ctx_callgraph"),
            "first heavy warm must signal true"
        );
        assert!(
            !ensure_warm_for_tool(&root, "ctx_callgraph"),
            "second heavy warm must be deduped to false"
        );
        assert!(
            !ensure_warm_for_tool(&root, "ctx_compose"),
            "any later heavy tool on the same root is also deduped"
        );
    }

    #[test]
    fn build_note_persisted_reports_size() {
        let note = bm25_build_note(
            42,
            &Ok(crate::core::bm25_index::SaveOutcome::Persisted {
                compressed_bytes: 3 * 1024 * 1024,
            }),
        );
        assert!(
            note.contains("42 chunks"),
            "note should report chunk count: {note}"
        );
        assert!(
            note.contains("persisted"),
            "note should report persistence: {note}"
        );
    }

    #[test]
    fn build_note_too_large_carries_remedy() {
        let note = bm25_build_note(
            1000,
            &Ok(crate::core::bm25_index::SaveOutcome::SkippedTooLarge {
                compressed_bytes: 600 * 1024 * 1024,
                limit_bytes: 512 * 1024 * 1024,
            }),
        );
        assert!(
            note.contains("NOT persisted"),
            "must flag non-persistence: {note}"
        );
        assert!(
            note.contains("LEAN_CTX_BM25_MAX_CACHE_MB") && note.contains("reindex"),
            "too-large note must carry an actionable remedy: {note}"
        );
    }

    #[test]
    fn build_note_persist_error_is_reported() {
        let note = bm25_build_note(7, &Err(std::io::Error::other("disk full")));
        assert!(note.contains("persisting failed"), "note: {note}");
        assert!(
            note.contains("disk full"),
            "note should include the io error: {note}"
        );
    }

    #[test]
    fn bm25_summary_unknown_project_is_idle() {
        let tmp = tempfile::tempdir().unwrap();
        let summary = bm25_summary(tmp.path().to_string_lossy().as_ref());
        assert_eq!(summary.state, "idle");
        assert!(summary.note.is_none());
        assert!(summary.last_error.is_none());
    }

    #[test]
    fn extra_roots_skips_subdirs_of_primary() {
        let tmp = tempfile::tempdir().unwrap();
        let primary = tmp.path().join("primary");
        std::fs::create_dir_all(&primary).unwrap();
        let sub = primary.join("subdir");
        std::fs::create_dir_all(&sub).unwrap();
        let external = tmp.path().join("external");
        std::fs::create_dir_all(&external).unwrap();

        let primary_str = primary.to_string_lossy().to_string();
        let extra = vec![
            sub.to_string_lossy().to_string(),
            external.to_string_lossy().to_string(),
        ];

        // Should not panic; subdirs are skipped, external is attempted
        ensure_extra_roots_background(&primary_str, &extra);
    }

    #[test]
    fn extra_roots_caps_at_max() {
        let tmp = tempfile::tempdir().unwrap();
        let primary = tmp.path().join("primary");
        std::fs::create_dir_all(&primary).unwrap();

        let mut extra = Vec::new();
        for i in 0..20 {
            let d = tmp.path().join(format!("ext-{i}"));
            std::fs::create_dir_all(&d).unwrap();
            extra.push(d.to_string_lossy().to_string());
        }

        let primary_str = primary.to_string_lossy().to_string();
        // Should not spawn more than MAX_EXTRA_ROOT_BUILDS threads
        ensure_extra_roots_background(&primary_str, &extra);
    }

    #[test]
    fn bm25_index_lock_name_is_per_repo_and_distinct_from_graph() {
        let a = bm25_index_lock_name(Path::new("/tmp/repo-a"));
        let b = bm25_index_lock_name(Path::new("/tmp/repo-b"));
        assert!(a.starts_with("bm25-idx-"), "unexpected lock name: {a}");
        assert_ne!(a, b, "lock name must be per-repo");
        // Stable for the same repo across calls.
        assert_eq!(a, bm25_index_lock_name(Path::new("/tmp/repo-a")));
        // Must NOT collide with the graph lock for the same repo, or the two
        // builds would serialize against each other unnecessarily.
        let graph = format!(
            "graph-idx-{}",
            &crate::core::index_namespace::namespace_hash(Path::new("/tmp/repo-a"))[..8]
        );
        assert_ne!(a, graph, "bm25 and graph locks must be independent");
    }
}