//! HNSW rebuild orchestration: in-flight rebuild state, backoff for
//! embedder init, model resolution, threshold helpers, and the
//! foreground/background rebuild paths.
//!
//! The watch loop calls into this module to kick off a background rebuild
//! (`spawn_hnsw_rebuild`), drain a completed one (`drain_pending_rebuild`), or
//! fall back to retrying a flaky embedder (`try_init_embedder` +
//! `EmbedderBackoff`). The model resolution helper lives here because it feeds
//! the rebuild thread and the watch loop's main `Embedder` from the same
//! source of truth.

use std::path::PathBuf;
use std::time::Duration;

use anyhow::Result;
use tracing::{info, warn};

use cqs::embedder::{Embedder, Embedding, ModelConfig};
use cqs::hnsw::HnswIndex;
use cqs::store::Store;

use super::{WatchConfig, WatchState};

/// Whether the tiered index backend is the active served backend
/// (`CQS_TIERED_INDEX=1`, compiled with the `tiered-index` feature).
///
/// When true, the served vector index is the cuVS tiered index, which absorbs
/// incremental adds into its brute-force tier and compacts its CAGRA ANN tier
/// internally — so the watch loop's *periodic* full HNSW rebuild (the
/// `incremental_count >= threshold` trigger) is redundant work and is skipped.
/// The cheap incremental HNSW insert path still runs as a CPU-side fallback
/// (and to keep the base-index router fed); only the expensive periodic
/// rebuild becomes a no-op. Always `false` without the feature, so default and
/// CAGRA builds are unaffected.
pub(super) fn tiered_index_active() -> bool {
    #[cfg(feature = "tiered-index")]
    {
        std::env::var("CQS_TIERED_INDEX").as_deref() == Ok("1")
    }
    #[cfg(not(feature = "tiered-index"))]
    {
        false
    }
}

/// Full HNSW rebuild after this many incremental inserts to clean orphaned vectors.
/// Override with CQS_WATCH_REBUILD_THRESHOLD env var.
pub(super) fn hnsw_rebuild_threshold() -> usize {
    static CACHE: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
    *CACHE.get_or_init(|| {
        std::env::var("CQS_WATCH_REBUILD_THRESHOLD")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(100)
    })
}
/// Handle to an in-flight background HNSW rebuild.
///
/// The rebuild thread streams embeddings from a read-only Store opened on
/// the same `index.db`, builds a fresh `Owned` `HnswIndex`, and saves it to
/// disk before sending the index back through the channel. While the
/// thread runs, the watch loop continues to commit new chunks to SQLite —
/// any (id, embedding) pair indexed during the rebuild window is captured
/// in `delta` and replayed into the new index just before the atomic swap,
/// closing the TOCTOU between the rebuild thread's snapshot and `recv`.
pub(super) struct PendingRebuild {
    pub(super) rx: std::sync::mpsc::Receiver<RebuildOutcome>,
    /// Each entry carries the chunk's `content_hash` alongside (id, embedding)
    /// so the swap-time drain can compare against the rebuild thread's
    /// snapshot. An id-only dedup would silently drop the fresh embedding for
    /// any chunk that was re-embedded mid-rebuild (snapshot has the OLD vector
    /// under the same id; delta has the NEW one) — the HNSW would carry the
    /// stale vector until the next threshold rebuild.
    pub(super) delta: Vec<(String, Embedding, String)>,
    pub(super) started_at: std::time::Instant,
    /// Held so daemon shutdown can `join` (or detect the thread is finished)
    /// instead of leaking a detached worker. `None` if the spawn itself failed
    /// — the channel disconnect path then handles cleanup.
    pub(super) handle: Option<std::thread::JoinHandle<()>>,
    /// Latched once `delta` exceeds the dim-aware [`pending_rebuild_delta_max`]
    /// cap. When set, the drain path discards the rebuilt index instead of
    /// swapping (the missed embeddings would silently disappear); the next
    /// threshold rebuild reads fresh state from SQLite and recovers cleanly.
    pub(super) delta_saturated: bool,
    /// Shutdown signal for the rebuild thread. Set by the daemon shutdown
    /// path before its bounded join; the thread checks it before entering
    /// each sidecar save so a process exit after the join window detaches
    /// a thread that is no longer mid-save. A skipped save leaves the dirty
    /// flag set and the previous on-disk index intact — the next daemon
    /// start rebuilds.
    pub(super) shutdown: std::sync::Arc<std::sync::atomic::AtomicBool>,
}

impl PendingRebuild {
    /// Queue (id, embedding, content_hash) triples for replay into the
    /// rebuilt index at swap time, respecting the dim-aware
    /// [`pending_rebuild_delta_max`] cap. Once the cap would be exceeded the
    /// `delta_saturated` latch is set and the triples are dropped — SQLite
    /// remains the source of truth and the drain discards a saturated
    /// rebuild instead of swapping it.
    ///
    /// Ordering contract: pairs committed in the same cycle a drain may
    /// complete MUST be queued BEFORE `drain_pending_rebuild` runs. The
    /// drain's sidecar save is stamped with the watch loop's observed store
    /// state, which already includes those pairs; a save without them would
    /// clear the dirty flag on an index that is missing chunks its stamp
    /// claims to contain.
    pub(super) fn queue_delta(
        &mut self,
        pairs: Vec<(String, Embedding, String)>,
        dim: usize,
        quiet: bool,
    ) {
        if pairs.is_empty() {
            return;
        }
        let delta_cap = pending_rebuild_delta_max(dim);
        if self.delta.len() + pairs.len() > delta_cap {
            if !self.delta_saturated {
                tracing::warn!(
                    cap = delta_cap,
                    current = self.delta.len(),
                    dim,
                    "Pending HNSW rebuild delta saturated; \
                     abandoning in-flight rebuild — next threshold \
                     rebuild will pick up changes from SQLite"
                );
                self.delta_saturated = true;
            }
            // Drop the new pairs; SQLite is the source of truth.
            return;
        }
        let added = pairs.len();
        self.delta.extend(pairs);
        tracing::debug!(
            added,
            total_delta = self.delta.len(),
            "Captured chunks in pending rebuild delta"
        );
        if !quiet {
            println!(
                "  HNSW index: +{} vectors queued for in-flight rebuild ({} total deferred)",
                added,
                self.delta.len()
            );
        }
    }
}

/// What [`drain_pending_rebuild`] did with the pending rebuild this tick.
/// The caller uses this to decide whether this cycle's pairs were consumed
/// (queued into the delta pre-drain and replayed/retained) or still need the
/// post-drain handling (new rebuild's delta or the fast incremental path).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum DrainOutcome {
    /// No rebuild was pending.
    NoPending,
    /// Rebuild still in flight — pending (and its delta) retained.
    StillRunning,
    /// New index swapped in; the pending delta was replayed into it first.
    Swapped,
    /// Pending cleared without a swap (thread error, empty store, saturated
    /// delta, stale-stamp discard, or channel disconnect). Any delta entries
    /// were dropped — SQLite remains the source of truth and the next
    /// threshold rebuild recovers them.
    Cleared,
}

/// The rebuild thread reports both the freshly-built `HnswIndex` AND a
/// fingerprint of every (id, content_hash) the build consumed. The drain path
/// needs the snapshot to detect mid-rebuild re-embeddings — without it, a
/// hash-aware dedup would have to issue a second SQL query (and lose snapshot
/// consistency under concurrent writers).
///
/// `snapshot_keys` is a `HashSet<u64>` of `snapshot_fingerprint(id, hash)` —
/// far smaller than a `HashMap<String, String>` held across the rebuild
/// window (~3 MB on a 17 k-chunk corpus).
pub(crate) struct RebuildResult {
    pub index: HnswIndex,
    pub snapshot_keys: std::collections::HashSet<u64>,
}

/// Cap on per-rebuild delta size. A stale-rebuild that runs longer than
/// expected (very large index, slow disk) accumulates one entry per chunk
/// re-embedded by the watch loop. Without a cap a multi-GB embedding vector
/// backlog is possible — every entry is `Vec<f32>` of `dim` floats. 5,000
/// entries × 1024 dim × 4 bytes ≈ 20 MB worst case, recoverable by the next
/// threshold rebuild's fresh SQLite scan.
///
/// The constant below is the *baseline at 1024-dim*. The actual cap is
/// dim-scaled by [`pending_rebuild_delta_max`] so a 4096-dim Qwen3-style
/// backbone gets ~1,250 entries (still ~20 MB) instead of the dim-blind 5,000
/// (which would be ~80 MB). Bound on both sides via
/// `dim_scaled_batch(_, dim, 500, 50_000)` so a tiny-dim preset doesn't go
/// unbounded either. Operators can override the baseline via
/// `CQS_PENDING_REBUILD_DELTA_MAX`.
pub(super) const MAX_PENDING_REBUILD_DELTA_BASELINE: usize = 5_000;

/// Dim-aware resolution of the pending-rebuild delta cap. Reads
/// `CQS_PENDING_REBUILD_DELTA_MAX` for a baseline override (rejected if 0 or
/// unparseable, falls back to the compiled-in baseline), then dim-scales via
/// [`crate::limits::dim_scaled_batch`] so wider models don't blow the
/// per-rebuild memory budget. Returns the absolute cap (entries, not bytes).
pub(super) fn pending_rebuild_delta_max(dim: usize) -> usize {
    let baseline = std::env::var("CQS_PENDING_REBUILD_DELTA_MAX")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .filter(|n| *n > 0)
        .unwrap_or(MAX_PENDING_REBUILD_DELTA_BASELINE);
    cqs::limits::dim_scaled_batch(baseline, dim, 500, 50_000)
}

pub(super) type RebuildOutcome = Result<Option<RebuildResult>, anyhow::Error>;

/// Track exponential backoff state for embedder initialization retries.
///
/// On repeated failures, backs off from 0s to max 5 minutes between attempts
/// to avoid burning CPU retrying a broken ONNX model load every ~2s cycle.
pub(super) struct EmbedderBackoff {
    /// Number of consecutive failures
    pub(super) failures: u32,
    /// Instant when the next retry is allowed
    next_retry: std::time::Instant,
}

impl EmbedderBackoff {
    pub(super) fn new() -> Self {
        Self {
            failures: 0,
            next_retry: std::time::Instant::now(),
        }
    }

    /// Record a failure and compute the next retry time with exponential backoff.
    /// Backoff: 2^failures seconds, capped at 300s (5 min).
    pub(super) fn record_failure(&mut self) {
        self.failures = self.failures.saturating_add(1);
        let delay_secs = 2u64.saturating_pow(self.failures).min(300);
        self.next_retry = std::time::Instant::now() + Duration::from_secs(delay_secs);
        warn!(
            failures = self.failures,
            next_retry_secs = delay_secs,
            "Embedder init failed, backing off"
        );
    }

    /// Reset backoff on success.
    pub(super) fn reset(&mut self) {
        self.failures = 0;
        self.next_retry = std::time::Instant::now();
    }

    /// Whether we should attempt initialization (backoff expired).
    pub(super) fn should_retry(&self) -> bool {
        std::time::Instant::now() >= self.next_retry
    }
}

/// Try to initialize the shared embedder, returning a reference from the
/// Arc-backed OnceLock. Uses `backoff` to apply exponential backoff on
/// repeated failures.
///
/// The OnceLock is shared with the daemon thread; whichever side initializes
/// first wins, and the other reuses the same Arc.
pub(super) fn try_init_embedder<'a>(
    embedder: &'a std::sync::OnceLock<std::sync::Arc<Embedder>>,
    backoff: &mut EmbedderBackoff,
    model_config: &ModelConfig,
) -> Option<&'a Embedder> {
    match embedder.get() {
        Some(e) => Some(e.as_ref()),
        None => {
            if !backoff.should_retry() {
                return None;
            }
            match Embedder::new(model_config.clone()) {
                Ok(e) => {
                    backoff.reset();
                    Some(embedder.get_or_init(|| std::sync::Arc::new(e)).as_ref())
                }
                Err(e) => {
                    warn!(error = %e, "Failed to initialize embedder");
                    backoff.record_failure();
                    None
                }
            }
        }
    }
}

/// Resolve the embedding model for the watch / daemon path, preferring the
/// model recorded in the open store's metadata over CLI flag / env / config.
///
/// Quick standalone read of `Store::stored_model_name()` — opened in
/// read-only mode and dropped before the caller's main store handle is built,
/// so this does not interfere with the watch loop's read-write store. The
/// extra open is cheap (a single SELECT against the `metadata` table) and
/// runs at most twice per `cmd_watch` invocation (once for the daemon
/// thread's `daemon_model_config`, once for the watch loop's
/// `cli.try_model_config()` override below).
///
/// See [`cqs::embedder::ModelConfig::resolve_for_query`] for the resolution
/// chain when no stored name is present (CLI > env > config > default).
pub(super) fn resolve_index_aware_model_for_watch(
    index_path: &std::path::Path,
    root: &std::path::Path,
    cli_model: Option<&str>,
) -> Result<ModelConfig> {
    // Use the strict variant so a corrupt-metadata read surfaces as an error
    // in journald instead of silently collapsing to None and triggering a
    // wrong-dim embedder pick. The open-readonly Err arm already warns; mirror
    // that on the metadata read.
    let stored = match cqs::Store::open_readonly(index_path) {
        Ok(s) => match s.try_stored_model_name() {
            Ok(opt) => opt,
            Err(e) => {
                tracing::error!(
                    error = %e,
                    index_path = %index_path.display(),
                    "Daemon thread failed to read stored_model_name from metadata — \
                     falling back to CLI/env/config resolution. Wrong-dim \
                     embedder risk; investigate metadata health before next reindex."
                );
                None
            }
        },
        Err(e) => {
            tracing::warn!(
                error = %e,
                "Quick read-only open for stored_model_name() failed; \
                 falling back to CLI/env/config resolution"
            );
            None
        }
    };
    let project_config = cqs::config::Config::load(root);
    let resolved = ModelConfig::resolve_for_query(
        stored.as_deref(),
        cli_model,
        project_config.embedding.as_ref(),
    )
    .apply_env_overrides();
    tracing::info!(
        stored = stored.as_deref().unwrap_or("<none>"),
        resolved = %resolved.name,
        dim = resolved.dim,
        "Watch resolved index-aware model config"
    );
    Ok(resolved)
}
/// Spawn a background thread to rebuild the enriched HNSW from the store and
/// save it to disk. Returns a `PendingRebuild` whose `rx` will receive the new
/// `Owned` index (or an error) when the build completes.
///
/// The thread opens its own read-only Store on the same `index.db` so the
/// main watch loop's `&Store` isn't borrowed across thread boundaries.
/// SQLite WAL gives the thread a consistent read snapshot; new commits made
/// by the watch loop while the rebuild is in flight are tracked in the
/// returned `PendingRebuild::delta` (filled by the caller) and replayed
/// into the new index just before the swap, closing the TOCTOU.
///
/// `context` is logged at info on completion to help operators distinguish
/// startup-owned-swap rebuilds from threshold-triggered rebuilds.
pub(super) fn spawn_hnsw_rebuild(
    cqs_dir: PathBuf,
    index_path: PathBuf,
    expected_dim: usize,
    context: &'static str,
) -> PendingRebuild {
    let (tx, rx) = std::sync::mpsc::channel();
    let started_at = std::time::Instant::now();
    let shutdown = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let shutdown_for_thread = std::sync::Arc::clone(&shutdown);
    let span = tracing::info_span!(
        "hnsw_rebuild_bg",
        context,
        cqs_dir = %cqs_dir.display(),
    );
    let thread_result = std::thread::Builder::new()
        .name(format!("cqs-hnsw-rebuild-{}", context))
        .spawn(move || {
            let _enter = span.entered();
            let result: RebuildOutcome = (|| -> RebuildOutcome {
                // Use the lower-footprint readonly open instead of
                // `open_readonly_pooled`. The watch loop's main (read-write)
                // store is already resident with 256 MiB mmap + 16 MiB cache;
                // re-opening the rebuild store at the same size would peak the
                // rebuild window at 2× SQLite mapping (272 MiB) on top of the
                // dual HNSW indexes (~50–200 MiB each). `open_readonly` keeps
                // mmap at 64 MiB + cache at 4 MiB — `build_hnsw_index_owned`
                // streams chunks via the batched embedding API, so the smaller
                // mapping doesn't hurt rebuild throughput on the corpora that
                // hit this path.
                let store = cqs::Store::open_readonly(&index_path).map_err(anyhow::Error::from)?;
                if store.dim() != expected_dim {
                    anyhow::bail!(
                        "store dim ({}) does not match expected ({}); refusing rebuild",
                        store.dim(),
                        expected_dim
                    );
                }
                let enriched = crate::cli::commands::build_hnsw_index_owned(
                    &store,
                    &cqs_dir,
                    Some(&shutdown_for_thread),
                )?;
                // The thread-side save may come back DiscardedStale when the
                // watch loop (or an external `cqs index`) committed chunks
                // during the build. That's fine here: the drain replays the
                // watch-observed delta and re-attempts the save under its own
                // stamp check, which is the authoritative one. The discard
                // only means this intermediate on-disk write was skipped.
                if let Some((_, _, cqs::hnsw::SaveOutcome::DiscardedStale)) = &enriched {
                    tracing::warn!(
                        context,
                        "background HNSW thread save discarded (store moved past build \
                         snapshot); drain will re-attempt after delta replay"
                    );
                }
                // Phase 5: also rebuild the base (non-enriched) HNSW so the
                // dual-index router stays in sync. The base index is loaded
                // fresh from disk by search processes — no in-memory swap
                // needed. Best-effort: a base rebuild failure shouldn't block
                // the enriched swap, so log + continue.
                match crate::cli::commands::build_hnsw_base_index(
                    &store,
                    &cqs_dir,
                    Some(&shutdown_for_thread),
                ) {
                    Ok(Some((n, cqs::hnsw::SaveOutcome::Saved))) => {
                        tracing::info!(vectors = n, "base HNSW rebuilt in background")
                    }
                    Ok(Some((_, cqs::hnsw::SaveOutcome::DiscardedStale))) => tracing::warn!(
                        "base HNSW save discarded (store moved past build snapshot); \
                         base dirty flag stays set until the next rebuild"
                    ),
                    Ok(None) => tracing::debug!("base HNSW skipped (no embedding_base rows yet)"),
                    Err(e) => tracing::warn!(
                        error = %e,
                        "base HNSW rebuild failed in background; router falls back to enriched-only"
                    ),
                }
                // Package the (index, snapshot_keys) pair so the drain can
                // detect mid-rebuild re-embeddings via fingerprint set.
                Ok(
                    enriched.map(|(index, snapshot_keys, _outcome)| RebuildResult {
                        index,
                        snapshot_keys,
                    }),
                )
            })();
            let elapsed_ms = started_at.elapsed().as_millis();
            match &result {
                Ok(Some(r)) => tracing::info!(
                    vectors = r.index.len(),
                    elapsed_ms,
                    context,
                    "background HNSW rebuild complete"
                ),
                Ok(None) => tracing::info!(
                    elapsed_ms,
                    context,
                    "background HNSW rebuild: store empty, nothing to build"
                ),
                Err(e) => tracing::warn!(
                    error = %e,
                    elapsed_ms,
                    context,
                    "background HNSW rebuild failed"
                ),
            }
            // Receiver may have been dropped if the daemon shut down — that's fine.
            let _ = tx.send(result);
        });
    let handle = match thread_result {
        Ok(h) => Some(h),
        Err(e) => {
            // Spawn failed (rare — only on resource exhaustion). Log and
            // return a PendingRebuild whose channel will hang up on first
            // poll, which the caller treats as "no rebuild in flight."
            tracing::warn!(error = %e, context, "Failed to spawn HNSW rebuild thread");
            None
        }
    };
    PendingRebuild {
        rx,
        delta: Vec::new(),
        started_at,
        handle,
        delta_saturated: false,
        shutdown,
    }
}

/// Try to consume a completed background HNSW rebuild and swap it into
/// `state.hnsw_index`. Replays any chunks captured in
/// `pending.delta` into the new index before saving + swapping so chunks
/// committed during the rebuild window aren't dropped.
///
/// Behaviour:
/// - Channel ready, `Ok(Some(idx))`: replay delta → save → swap, clear pending.
/// - Channel ready, `Ok(None)`: store was empty when the thread ran; clear
///   pending without swapping. Next reindex cycle will spawn a fresh one.
/// - Channel ready, `Err(_)`: thread reported an error; clear pending so
///   the next threshold trigger can retry.
/// - Channel empty: rebuild still in flight; leave pending alone so the
///   caller continues to capture delta entries.
/// - Channel disconnected: spawn failed earlier or thread panicked; clear.
///
/// Returns a [`DrainOutcome`] so the caller can tell whether pairs it queued
/// into the delta pre-drain were consumed (`Swapped` / retained
/// `StillRunning`) or dropped (`Cleared`) and need re-handling.
pub(super) fn drain_pending_rebuild(
    cfg: &WatchConfig,
    store: &Store,
    state: &mut WatchState,
) -> DrainOutcome {
    let Some(pending) = state.pending_rebuild.as_mut() else {
        return DrainOutcome::NoPending;
    };
    let outcome = match pending.rx.try_recv() {
        Ok(o) => o,
        Err(std::sync::mpsc::TryRecvError::Empty) => return DrainOutcome::StillRunning,
        Err(std::sync::mpsc::TryRecvError::Disconnected) => {
            tracing::warn!("Background rebuild thread channel disconnected; clearing pending");
            state.pending_rebuild = None;
            return DrainOutcome::Cleared;
        }
    };
    let pending = state
        .pending_rebuild
        .take()
        .expect("pending_rebuild was Some when we held a borrow");

    match outcome {
        Ok(Some(RebuildResult {
            index: mut new_index,
            snapshot_keys,
        })) => {
            // If the delta saturated, the rebuilt index is missing events that
            // we dropped on the floor; swapping it in would silently lose
            // those chunks. Discard instead — the next threshold rebuild reads
            // fresh state from SQLite and recovers.
            if pending.delta_saturated {
                let elapsed_ms = pending.started_at.elapsed().as_millis();
                tracing::warn!(
                    elapsed_ms,
                    discarded_delta = pending.delta.len(),
                    "Discarding rebuilt HNSW: delta saturated during rebuild — \
                     next threshold rebuild will pick up changes from SQLite"
                );
                if !cfg.quiet {
                    println!(
                        "  HNSW index: rebuild discarded (delta saturated; {}ms wasted, will retry)",
                        elapsed_ms
                    );
                }
                return DrainOutcome::Cleared;
            }
            // Replay captured delta — skip only entries whose (id,
            // content_hash) match the snapshot. An id that was re-embedded
            // mid-rebuild has a NEW hash in delta but the snapshot baked in the
            // OLD vector; the entry must replay so the fresh embedding lands in
            // the swapped HNSW. Pure id-only dedup would silently drop these
            // and leave the HNSW serving stale vectors until the next threshold
            // rebuild.
            //
            // Trade-off when an id IS re-embedded: `insert_batch` on
            // hnsw_rs adds a duplicate node (no deletion API). That's the
            // same orphan situation as the existing fast-incremental path
            // — search post-filters by SQLite, so the orphan is invisible
            // to callers. The next threshold rebuild cleans it up.
            let to_replay: Vec<(String, Embedding)> = pending
                .delta
                .into_iter()
                .filter(|(id, _, hash)| {
                    // `snapshot_keys` contains the fingerprint for every (id,
                    // hash) the rebuild consumed; replay only when the (id,
                    // hash) we observed mid-rebuild was NOT in the rebuild's
                    // snapshot.
                    let fp = crate::cli::commands::snapshot_fingerprint(id, hash);
                    !snapshot_keys.contains(&fp)
                })
                .map(|(id, emb, _)| (id, emb))
                .collect();
            if !to_replay.is_empty() {
                let items: Vec<(String, &[f32])> = to_replay
                    .iter()
                    .map(|(id, emb)| (id.clone(), emb.as_slice()))
                    .collect();
                match new_index.insert_batch(&items) {
                    Ok(n) => {
                        tracing::info!(replayed = n, "Replayed delta into rebuilt HNSW before swap")
                    }
                    Err(e) => tracing::warn!(
                        error = %e,
                        replayed_attempt = items.len(),
                        "Failed to replay delta into rebuilt HNSW; new chunks will surface on next rebuild"
                    ),
                }
            }
            // Save under a store-state stamp. The snapshot is the stamp the
            // watch loop captured after its own most recent write cycle:
            // everything the loop has committed is either in the rebuild
            // thread's SQLite snapshot or in the delta replayed above, so
            // the rebuilt index reflects exactly that store state. If the
            // live stamp (re-read under the exclusive save lock) differs, a
            // writer this process never observed — a concurrent `cqs index`
            // is the documented case — committed chunks the rebuilt index
            // does not contain. Saving would overwrite that writer's newer
            // sidecars with stale contents stamped as current; instead the
            // save is discarded and the rebuilt index is dropped (swapping
            // it into memory would serve the same stale view in-process).
            // The dirty flag stays set, and even a stale save that raced
            // through the snapshot-capture window carries an old stamp that
            // the dirty-path load check refuses.
            let snapshot = state.observed_stamp.map(Ok).unwrap_or_else(|| {
                // First cycle (no writes observed yet): fall back to the
                // live stamp. No foreign-writer detection on this save, but
                // the under-lock re-read still bounds the race window.
                cqs::hnsw::StoreStamp::read(store)
            });
            match snapshot {
                Ok(snapshot) => {
                    match new_index.save_stamped(cfg.cqs_dir, "index", store, snapshot) {
                        Ok(cqs::hnsw::SaveOutcome::Saved) => {
                            clear_hnsw_dirty_with_retry(
                                store,
                                cqs::HnswKind::Enriched,
                                "background_rebuild_swap",
                            );
                        }
                        Ok(cqs::hnsw::SaveOutcome::DiscardedStale) => {
                            let elapsed_ms = pending.started_at.elapsed().as_millis();
                            tracing::warn!(
                                elapsed_ms,
                                "Discarding rebuilt HNSW: store moved past the watch loop's \
                                 observed state (concurrent indexer) — keeping the current \
                                 in-memory index; next threshold rebuild reads fresh SQLite"
                            );
                            if !cfg.quiet {
                                println!(
                                    "  HNSW index: rebuild discarded (concurrent indexer; {}ms, will retry)",
                                    elapsed_ms
                                );
                            }
                            return DrainOutcome::Cleared;
                        }
                        Err(e) => {
                            tracing::warn!(
                                error = %e,
                                "Failed to save rebuilt HNSW after delta replay; in-memory swap proceeds anyway"
                            );
                        }
                    }
                }
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        "Failed to read store stamp for rebuilt HNSW save; skipping disk save, \
                         in-memory swap proceeds anyway"
                    );
                }
            }
            let elapsed_ms = pending.started_at.elapsed().as_millis();
            let n = new_index.len();
            state.hnsw_index = Some(new_index);
            state.incremental_count = 0;
            info!(
                vectors = n,
                elapsed_ms, "Background HNSW rebuild swapped in"
            );
            if !cfg.quiet {
                println!(
                    "  HNSW index: {} vectors (background rebuild swapped in, {}ms)",
                    n, elapsed_ms
                );
            }
            DrainOutcome::Swapped
        }
        Ok(None) => {
            tracing::debug!("Background rebuild reported empty store; cleared pending");
            DrainOutcome::Cleared
        }
        Err(e) => {
            tracing::warn!(
                error = %e,
                "Background HNSW rebuild failed; will retry on next threshold trigger"
            );
            DrainOutcome::Cleared
        }
    }
}

/// Clear the HNSW-dirty flag for `kind`, retrying once on a
/// transient error (e.g. `SQLITE_BUSY` when a concurrent writer is
/// finishing a transaction). If both attempts fail, emits a warn and
/// returns — the caller keeps the in-memory HNSW but the persistent
/// dirty flag stays `1`, forcing a full rebuild on the next daemon
/// start. The single retry is enough to absorb the common lock-window
/// race without extending the write-side lock hold time.
pub(super) fn clear_hnsw_dirty_with_retry(store: &Store, kind: cqs::HnswKind, context: &str) {
    if let Err(e1) = store.set_hnsw_dirty(kind, false) {
        tracing::debug!(
            error = %e1,
            kind = ?kind,
            context,
            "First clear-dirty attempt failed, retrying once"
        );
        if let Err(e2) = store.set_hnsw_dirty(kind, false) {
            tracing::warn!(
                first_error = %e1,
                retry_error = %e2,
                kind = ?kind,
                context,
                "Failed to clear HNSW dirty flag after retry — unnecessary rebuild on next daemon start"
            );
        }
    }
}
